From e3329968ab1d7b56048a4a3210f7c638c208203a Mon Sep 17 00:00:00 2001 From: Thomas Grimonet Date: Sat, 14 Dec 2024 16:51:06 +0100 Subject: [PATCH 1/3] doc: Implement mkdoc to expose eos-downloader documentation --- docs/README.md | 43 ++ docs/api/helpers.md | 1 + docs/api/logics/arista_server.md | 1 + docs/api/logics/arista_xml_server.md | 1 + docs/api/logics/download.md | 1 + docs/api/models/custom_types.md | 1 + docs/api/models/data.md | 3 + docs/api/models/version.md | 1 + docs/imgs/favicon.ico | Bin 0 -> 4286 bytes docs/stylesheets/extra.material.css | 245 ++++++++ docs/stylesheets/highlight.js | 3 + docs/stylesheets/tables.js | 6 + docs/usage/eos.md | 53 ++ docs/usage/info.md | 89 +++ eos_downloader/cli/debug/commands.py | 4 +- eos_downloader/cli/get/commands.py | 2 +- eos_downloader/cli/get/utils.py | 2 +- eos_downloader/cli/info/commands.py | 6 +- eos_downloader/helpers/__init__.py | 100 ++-- eos_downloader/logics/arista_server.py | 661 ++++++++------------- eos_downloader/logics/arista_xml_server.py | 536 +++++++++++++++++ eos_downloader/logics/download.py | 208 ++++--- eos_downloader/logics/server.py | 263 -------- eos_downloader/models/data.py | 92 ++- eos_downloader/models/types.py | 35 +- eos_downloader/models/version.py | 369 +++++++----- mkdocs.yml | 177 ++++++ pyproject.toml | 18 + tests/unit/logics/test_arista_server.py | 4 +- tests/unit/logics/test_download.py | 2 +- tests/unit/logics/test_server.py | 2 +- 31 files changed, 1913 insertions(+), 1016 deletions(-) create mode 100644 docs/README.md create mode 100644 docs/api/helpers.md create mode 100644 docs/api/logics/arista_server.md create mode 100644 docs/api/logics/arista_xml_server.md create mode 100644 docs/api/logics/download.md create mode 100644 docs/api/models/custom_types.md create mode 100644 docs/api/models/data.md create mode 100644 docs/api/models/version.md create mode 100644 docs/imgs/favicon.ico create mode 100644 docs/stylesheets/extra.material.css create mode 100644 docs/stylesheets/highlight.js create mode 100644 docs/stylesheets/tables.js create mode 100644 docs/usage/eos.md create mode 100644 docs/usage/info.md create mode 100644 eos_downloader/logics/arista_xml_server.py delete mode 100644 eos_downloader/logics/server.py create mode 100644 mkdocs.yml diff --git a/docs/README.md b/docs/README.md new file mode 100644 index 0000000..b6a515e --- /dev/null +++ b/docs/README.md @@ -0,0 +1,43 @@ + + +# Arista Software Downloader + +A project to download Arista softwares to local folder, Cloudvision or EVE-NG. It comes in 2 way: a framework with object to automate Arista software download and a CLI for human activities. + +> [!CAUTION] +> This script should not be deployed on EOS device. If you do that, there is no support to expect from Arista TAC team. + +```bash +# install eos-downloader from pypi +pip install eos-downloader + +# download EOS swi for EOS 64bits +ardl --token get eos --format 64 --latest --release-type M +``` + +> [!NOTE] +> The main branch is not the stable branch and can be broken between releases. It is safe to consider using tags for stable versions. All versions on pypi servers are considered stable. + +## FAQ + +On EVE-NG, you may have to install/upgrade __pyOpenSSL__ in version `23.0.0`: + +```bash +# Error when running ardl: AttributeError: module 'lib' has no attribute 'X509_V_FLAG_CB_ISSUER_CHECK' + +$ pip install pyopenssl --upgrade +``` + +## Author + +From an original idea of [@Mark Rayson](https://github.com/Sparky-python) in [arista-netdevops-community/eos-scripts](https://github.com/arista-netdevops-community/eos-scripts) + +## License + +Code is under [Apache2](LICENSE) License diff --git a/docs/api/helpers.md b/docs/api/helpers.md new file mode 100644 index 0000000..339c22d --- /dev/null +++ b/docs/api/helpers.md @@ -0,0 +1 @@ +## ::: eos_downloader.helpers diff --git a/docs/api/logics/arista_server.md b/docs/api/logics/arista_server.md new file mode 100644 index 0000000..716080d --- /dev/null +++ b/docs/api/logics/arista_server.md @@ -0,0 +1 @@ +## ::: eos_downloader.logics.arista_server diff --git a/docs/api/logics/arista_xml_server.md b/docs/api/logics/arista_xml_server.md new file mode 100644 index 0000000..89d9d1e --- /dev/null +++ b/docs/api/logics/arista_xml_server.md @@ -0,0 +1 @@ +## ::: eos_downloader.logics.arista_xml_server diff --git a/docs/api/logics/download.md b/docs/api/logics/download.md new file mode 100644 index 0000000..d159f33 --- /dev/null +++ b/docs/api/logics/download.md @@ -0,0 +1 @@ +## ::: eos_downloader.logics.download diff --git a/docs/api/models/custom_types.md b/docs/api/models/custom_types.md new file mode 100644 index 0000000..a1d9118 --- /dev/null +++ b/docs/api/models/custom_types.md @@ -0,0 +1 @@ +## ::: eos_downloader.models.types diff --git a/docs/api/models/data.md b/docs/api/models/data.md new file mode 100644 index 0000000..3f8025c --- /dev/null +++ b/docs/api/models/data.md @@ -0,0 +1,3 @@ +## ::: eos_downloader.models.data.software_mapping + +## ::: eos_downloader.models.data diff --git a/docs/api/models/version.md b/docs/api/models/version.md new file mode 100644 index 0000000..93cdd4b --- /dev/null +++ b/docs/api/models/version.md @@ -0,0 +1 @@ +## ::: eos_downloader.models.version diff --git a/docs/imgs/favicon.ico b/docs/imgs/favicon.ico new file mode 100644 index 0000000000000000000000000000000000000000..cf55b0c85d981ebce5ef33b723be1ede8f26cbc1 GIT binary patch literal 4286 zcmeH|yK7WI6o+SnCWS<;)Fi^D5FfQ7iHK{FbShZ*2Z&K(VX64Q-8NtmA7CTcCLqKn zWe~L4LJ=QmXOYCxM?#2RaNY4c!GXhJW*>J$s=eXM+;ir4e&^oFor!Y=eujsg{C2au zo!jG_I}D`UX^?#P>aIX+1tyQVIryP?&3#`?+er@3e)z2VNYAHm2!4ZClsoAatHt}i z@|RveVMw%$cg$QyEL&((F1(II&GeWI=Hz?fgPDIx+!Hi;M?AK}JLCPzdJ~VrU!%L6 zw{NJ!@0RgQt}VJpMl0XQK=5jJr1l{EGM;x+KInGBlF`o-UueRZ^+aEQ7oflEkHo!J zt7GQRnHpW6x|hG(3FC9g_{jG+sx`5DG_oI_nc7reotN)-!SB!`Be*)JUFu~keu+MrX)hCt5qX*;+&|Ytdw>7Ad zGgjVJ*US*-x~Mw>vR}2|^^_U92S)picyCJIN}l?%@mwc9U~9>ZfIVX`T3x#_o_+8M z^cu*%E(9;`<@|ncGz(f2w0TB+PJ%s~CxRDqw?RK&bd7>O&xrqH(?j;>AjlbO|9{`e z)1y%hJ$AuY(C_)2*q*U($tCwlqaMxX@I7z(>t~VsKX9~}Pu_XFaywe}=|o4LtEPXQc)SzOc53M*XY&-SfviRHOl-%utL?79R .admonition-title, .md-typeset .note > summary { + background-color: var(--md-accent-bg-color); + color: var(--md-default-fg-color--lighter) + } */ + .md-typeset__table { + min-width: 80%; + } + .md-typeset table:not([class]) { + display: table; + } + + .mdx-content__footer { + margin-top: 20px; + text-align: center; + } + .mdx-content__footer a { + display: inline-block; + transition: transform 250ms cubic-bezier(0.1, 0.7, 0.1, 1), color 125ms; + } + .mdx-content__footer a:focus, .mdx-content__footer a:hover { + transform: scale(1.2); + } + + .md-typeset table:not([class]) th { + min-width: 5rem; + padding: .6rem .8rem; + } + + .md-footer-copyright { + color: var(--md-footer-fg-color--lighter); + font-size: .64rem; + margin: auto 0.6rem; + padding: 0.4rem; + width: 100%; + text-align: center; + } + .img_center { + display: block; + margin-left: auto; + margin-right: auto; + border-radius: 1%; + } +} + +/* mkdocstrings css from official repo to indent sub-elements nicely */ +/* Indentation. */ +div.doc-contents { + padding-left: 25px; + border-left: .05rem solid var(--md-typeset-table-color); +} +h5.doc-heading { + /* Avoid to capitalize h5 headers for mkdocstrings */ + text-transform: none; +} diff --git a/docs/stylesheets/highlight.js b/docs/stylesheets/highlight.js new file mode 100644 index 0000000..86e50b9 --- /dev/null +++ b/docs/stylesheets/highlight.js @@ -0,0 +1,3 @@ +document$.subscribe(() => { + hljs.highlightAll() +}) diff --git a/docs/stylesheets/tables.js b/docs/stylesheets/tables.js new file mode 100644 index 0000000..e848f07 --- /dev/null +++ b/docs/stylesheets/tables.js @@ -0,0 +1,6 @@ +document$.subscribe(function() { + var tables = document.querySelectorAll("article table") + tables.forEach(function(table) { + new Tablesort(table) + }) +}) diff --git a/docs/usage/eos.md b/docs/usage/eos.md new file mode 100644 index 0000000..79cb611 --- /dev/null +++ b/docs/usage/eos.md @@ -0,0 +1,53 @@ +# Download EOS package from arista website + +This command gives you option to download EOS images localy. Some options are available based on image type like importing your cEOS container in your local registry + +```bash +# Get latest version of EOS using docker format. +ardl get eos --latest --format cEOS + +# Get latest version of maintenance type in specific branch 4.29 +ardl get eos --branch 4.29 --format cEOS --release-type M + +# Get a specific version +ardl get eos --version 4.29.4M + +# Get a specific version and import to docker using default arista/ceos:{version}{release_type} +ardl get eos --version 4.29.4M --import-docker + +# Get a specific version and import to EVE-NG +ardl get eos --version 4.33.0F --eve-ng +``` + +## ardl get eos options + +Below are all the options available to get EOS package: + +```bash +$ ardl get eos --help +Usage: ardl get eos [OPTIONS] + + Download EOS image from Arista server. + +Options: + --format TEXT Image format [default: vmdk] + --output PATH Path to save image [default: .] + --latest Get latest version. If --branch is not use, get the + latest branch with specific release type + --eve-ng Run EVE-NG vEOS provisioning (only if CLI runs on an + EVE-NG server) + --import-docker Import docker image to local docker + --skip-download Skip download process - for debug only + --docker-name TEXT Docker image name [default: arista/ceos] + --docker-tag TEXT Docker image tag + --version TEXT EOS version to download + --release-type TEXT Release type (M for Maintenance, F for Feature) + [default: F] + --branch TEXT Branch to download + --dry-run Enable dry-run mode: only run code without system + changes + --help Show this message and exit. +``` + +!!! info + You can get information about available version using the [`ardl info version` cli](./info.md) \ No newline at end of file diff --git a/docs/usage/info.md b/docs/usage/info.md new file mode 100644 index 0000000..6d56f7c --- /dev/null +++ b/docs/usage/info.md @@ -0,0 +1,89 @@ +# Get information about softwares versions + +`ardl` comes with a tool to get version information from Arista website. It is valid for both __CloudVision__ and __EOS__ packages. + +## Get information about available versions + +```bash +ardl info versions --help +Usage: ardl info versions [OPTIONS] + + List available versions of Arista packages (eos or CVP) packages + +Options: + --format [json|text|fancy] Output format + --package [eos|cvp] + -b, --branch TEXT + --release-type TEXT + --help Show this message and exit. +``` + +## Usage example + +With this CLI, you can specify either a branch or a release type when applicable to filter information: + +### Fancy format (default) + +```bash +# Get F version in branch 4.29 using default fancy mode +ardl info versions --branch 4.29 --release-type F + +╭──────────────────────────── Available versions ──────────────────────────────╮ +│ │ +│ - version: 4.29.2F │ +│ - version: 4.29.1F │ +│ - version: 4.29.0.2F │ +│ - version: 4.29.2F │ +│ - version: 4.29.1F │ +│ - version: 4.29.0.2F │ +│ │ +╰──────────────────────────────────────────────────────────────────────────────╯ +``` + +### Text Format + +```bash +# Get M version in branch 4.29 using text output +❯ ardl info versions --branch 4.29 --release-type M --format text +Listing versions + - version: 4.29.10M + - version: 4.29.9.1M + - version: 4.29.9M + - version: 4.29.8M + - version: 4.29.7.1M + ... +``` + +### JSON format + +You can also specify JSON as output format: + +```bash +ardl info versions --branch 4.29 --release-type F --format json +[ + { + "version": "4.29.2F", + "branch": "4.29" + }, + { + "version": "4.29.1F", + "branch": "4.29" + }, + { + "version": "4.29.0.2F", + "branch": "4.29" + }, + { + "version": "4.29.2F", + "branch": "4.29" + }, + { + "version": "4.29.1F", + "branch": "4.29" + }, + { + "version": "4.29.0.2F", + "branch": "4.29" + } +] +``` diff --git a/eos_downloader/cli/debug/commands.py b/eos_downloader/cli/debug/commands.py index 24604c3..54bfbe1 100644 --- a/eos_downloader/cli/debug/commands.py +++ b/eos_downloader/cli/debug/commands.py @@ -19,7 +19,7 @@ # Local imports import eos_downloader.defaults -import eos_downloader.logics.server +import eos_downloader.logics.arista_server from eos_downloader.cli.utils import cli_logging @@ -63,7 +63,7 @@ def xml(ctx: click.Context, output: str, log_level: str) -> None: log = cli_logging(log_level) token = ctx.obj["token"] - server = eos_downloader.logics.server.AristaServer( + server = eos_downloader.logics.arista_server.AristaServer( token=token, session_server=eos_downloader.defaults.DEFAULT_SERVER_SESSION ) try: diff --git a/eos_downloader/cli/get/commands.py b/eos_downloader/cli/get/commands.py index eaa4543..4882327 100644 --- a/eos_downloader/cli/get/commands.py +++ b/eos_downloader/cli/get/commands.py @@ -15,7 +15,7 @@ import click from eos_downloader.models.data import RTYPE_FEATURE from eos_downloader.logics.download import SoftManager -from eos_downloader.logics.arista_server import ( +from eos_downloader.logics.arista_xml_server import ( EosXmlObject, AristaXmlQuerier, CvpXmlObject, diff --git a/eos_downloader/cli/get/utils.py b/eos_downloader/cli/get/utils.py index e953bb4..4261138 100644 --- a/eos_downloader/cli/get/utils.py +++ b/eos_downloader/cli/get/utils.py @@ -12,7 +12,7 @@ from eos_downloader.cli.utils import cli_logging, console_configuration from eos_downloader.models.data import RTYPE_FEATURE, RTYPES from eos_downloader.models.types import ReleaseType -from eos_downloader.logics.arista_server import AristaXmlQuerier, AristaXmlObjects +from eos_downloader.logics.arista_xml_server import AristaXmlQuerier, AristaXmlObjects def initialize(ctx: click.Context) -> tuple[Console, str, bool, str]: diff --git a/eos_downloader/cli/info/commands.py b/eos_downloader/cli/info/commands.py index 8e4d19b..721a93f 100644 --- a/eos_downloader/cli/info/commands.py +++ b/eos_downloader/cli/info/commands.py @@ -34,7 +34,7 @@ from eos_downloader.models.data import software_mapping from eos_downloader.models.types import AristaPackage, ReleaseType, AristaMapping -import eos_downloader.logics.arista_server +import eos_downloader.logics.arista_xml_server from eos_downloader.cli.utils import console_configuration from eos_downloader.cli.utils import cli_logging @@ -71,7 +71,7 @@ def versions( log_level = ctx.obj["log_level"] cli_logging(log_level) - querier = eos_downloader.logics.arista_server.AristaXmlQuerier(token=token) + querier = eos_downloader.logics.arista_xml_server.AristaXmlQuerier(token=token) received_versions = None try: @@ -144,7 +144,7 @@ def latest( debug = ctx.obj["debug"] log_level = ctx.obj["log_level"] cli_logging(log_level) - querier = eos_downloader.logics.arista_server.AristaXmlQuerier(token=token) + querier = eos_downloader.logics.arista_xml_server.AristaXmlQuerier(token=token) received_version = None try: received_version = querier.latest( diff --git a/eos_downloader/helpers/__init__.py b/eos_downloader/helpers/__init__.py index 34c8620..151324f 100644 --- a/eos_downloader/helpers/__init__.py +++ b/eos_downloader/helpers/__init__.py @@ -5,12 +5,13 @@ using the Rich library. It includes a signal handler for graceful interruption and a DownloadProgressBar class for concurrent file downloads with progress tracking. -Classes: +Classes +------- DownloadProgressBar: A class that provides visual progress tracking for file downloads. -Functions: +Functions +------- handle_sigint: Signal handler for SIGINT (Ctrl+C) to enable graceful termination. - console (Console): Rich Console instance for output rendering. done_event (Event): Threading Event used for signaling download interruption. """ @@ -52,12 +53,16 @@ def handle_sigint(signum: Any, frame: Any) -> None: This function sets the done_event flag when SIGINT is received, allowing for graceful termination of the program. - Args: - signum (Any): Signal number - frame (Any): Current stack frame object + Parameters + ---------- + signum : Any + Signal number. + frame : Any + Current stack frame object. - Returns: - None + Returns + ------- + None """ done_event.set() @@ -72,16 +77,16 @@ class DownloadProgressBar: It supports downloading multiple files concurrently with a progress bar showing download speed, completion percentage, and elapsed time. - Attributes: - progress (Progress): A Rich Progress instance configured with custom columns for - displaying download information. + Attributes + ---------- + progress : Progress + A Rich Progress instance configured with custom columns for displaying download information. - Example: - ```python - downloader = DownloadProgressBar() - urls = ['http://example.com/file1.zip', 'http://example.com/file2.zip'] - downloader.download(urls, '/path/to/destination') - ``` + Examples + -------- + >>> downloader = DownloadProgressBar() + >>> urls = ['http://example.com/file1.zip', 'http://example.com/file2.zip'] + >>> downloader.download(urls, '/path/to/destination') """ def __init__(self) -> None: @@ -110,19 +115,30 @@ def _copy_url( specified local path while updating a progress bar. The download can be interrupted via a done event. - Args: - task_id (TaskID): Identifier for the progress tracking task - url (str): URL to download the file from - path (str): Local path where the file should be saved - block_size (int, optional): Size of chunks to download at a time. Defaults to 1024 bytes - - Returns: - bool: True if download was interrupted by done_event, False if completed successfully - - Raises: - requests.exceptions.RequestException: If the download request fails - IOError: If there are issues writing to the local file - KeyError: If the response doesn't contain Content-Length header + Parameters + ---------- + task_id : TaskID + Identifier for the progress tracking task. + url : str + URL to download the file from. + path : str + Local path where the file should be saved. + block_size : int, optional + Size of chunks to download at a time. Defaults to 1024 bytes. + + Returns + ------- + bool + True if download was interrupted by done_event, False if completed successfully. + + Raises + ------ + requests.exceptions.RequestException + If the download request fails. + IOError + If there are issues writing to the local file. + KeyError + If the response doesn't contain Content-Length header. """ response = requests.get( url, @@ -148,19 +164,23 @@ def download(self, urls: Iterable[str], dest_dir: str) -> None: This method downloads files from the provided URLs in parallel using a thread pool, displaying progress for each download in the console. - Args: - urls (Iterable[str]): An iterable of URLs to download files from. - dest_dir (str): The destination directory where files will be saved. + Parameters + ---------- + urls : Iterable[str] + An iterable of URLs to download files from. + dest_dir : str + The destination directory where files will be saved. - Returns: - None + Returns + ------- + None - Example: - >>> downloader = Downloader() - >>> urls = ["http://example.com/file1.txt", "http://example.com/file2.txt"] - >>> downloader.download(urls, "/path/to/destination") + Examples + -------- + >>> downloader = DownloadProgressBar() + >>> urls = ["http://example.com/file1.txt", "http://example.com/file2.txt"] + >>> downloader.download(urls, "/path/to/destination") """ - with self.progress: with ThreadPoolExecutor(max_workers=4) as pool: futures = [] diff --git a/eos_downloader/logics/arista_server.py b/eos_downloader/logics/arista_server.py index 38d3102..c1b87c2 100644 --- a/eos_downloader/logics/arista_server.py +++ b/eos_downloader/logics/arista_server.py @@ -1,461 +1,272 @@ +#!/usr/bin/python # coding: utf-8 -*- - -"""This module provides classes for managing and querying Arista XML data. - -Classes: - AristaXmlBase: Base class for Arista XML data management. - AristaXmlQuerier: Class to query Arista XML data for Software versions. - AristaXmlObject: Base class for Arista XML data management with specific software and version. - EosXmlObject: Class to query Arista XML data for EOS versions. - -Classes and Methods: - AristaXmlBase: - - __init__(self, token: str) -> None: Initializes the AristaXmlBase class with a token. - - _get_xml_root(self): Retrieves the XML root from the Arista server. - - AristaXmlQuerier(AristaXmlBase): - - available_public_eos_version(self, branch: Union[str, None] = None, rtype: Union[str, None] = None) - -> List[eos_downloader.models.version.EosVersion]: Extracts a list of available EOS versions from Arista.com. - - latest(self, branch: Union[str, None] = None, rtype: str = eos_downloader.models.version.RTYPE_FEATURE) - -> eos_downloader.models.version.EosVersion: Gets the latest branch from a semver standpoint. - - branches(self, latest: bool = False) -> List[str]: Returns a list of valid EOS version branches. - - _get_branches(self, versions: Union[List[eos_downloader.models.version.EosVersion], List[eos_downloader.models.version.CvpVersion]]) - -> Union[List[eos_downloader.models.version.EosVersion], List[eos_downloader.models.version.CvpVersion]]: Extracts unique branch names from a list of version objects. - - AristaXmlObject(AristaXmlBase): - - __init__(self, searched_version: str, image_type: str, token: str) -> None: Initializes the AristaXmlObject class with a searched version, image type, and token. - - filename(self) -> Union[str, None]: Builds the filename to search on arista.com. - - hashfile(self, hashtype: str = 'md5sum') -> Union[str, None]: Builds the hash filename to search on arista.com. - - path_from_xml(self, search_file: str) -> Union[str, None]: Parses XML to find the path for a given file. - - _url(self, xml_path: str) -> str: Gets the URL to download a file from the Arista server. - - urls(self) -> Dict[str, str]: Gets URLs to download files from the Arista server for given software and version. - - available_public_eos_version(self): Raises NotImplementedError. - - EosXmlObject(AristaXmlObject): - - Class to query Arista XML data for EOS versions. -""" # noqa: E501 +# pylint: disable=too-many-positional-arguments +# pylint: disable=dangerous-default-value + +"""Server module for handling interactions with Arista software download portal. + +This module provides the AristaServer class which manages authentication and +file retrieval operations with the Arista software download portal. It handles +session management, XML data retrieval, and download URL generation. + +Classes +------- +AristaServer + Main class for interacting with the Arista software portal. + +Dependencies +----------- +- base64: For encoding authentication tokens +- json: For handling JSON data in requests +- xml.etree.ElementTree: For parsing XML responses +- loguru: For logging +- requests: For making HTTP requests + +Example +------- + >>> from eos_downloader.logics.server import AristaServer + >>> server = AristaServer(token='my_auth_token') + >>> server.authenticate() + >>> xml_data = server.get_xml_data() + >>> download_url = server.get_url('/path/to/file') + +Notes +----- +The module requires valid authentication credentials to interact with the Arista portal. +All server interactions are performed over HTTPS and follow Arista's API specifications. +""" from __future__ import annotations +import base64 import logging -import xml.etree.ElementTree as ET -from typing import ClassVar, Union, List, Dict - -import eos_downloader.logics.server -import eos_downloader.models.version -import eos_downloader.models.data -from eos_downloader.models.types import AristaPackage, AristaVersions, AristaMapping - - -class AristaXmlBase: - # pylint: disable=too-few-public-methods - """Base class for Arista XML data management.""" - - # File extensions supported to be downloaded from arista server. - # Should cover: image file (image) and has files (md5sum and/or sha512sum) - supported_role_types: ClassVar[List[str]] = ["image", "md5sum", "sha512sum"] - - def __init__( - self, token: Union[str, None] = None, xml_path: Union[str, None] = None - ) -> None: - logging.info("Initializing AristXmlBase.") - self.server = eos_downloader.logics.server.AristaServer(token=token) - if xml_path is not None: - try: - self.xml_data = ET.parse(xml_path) - except ET.ParseError as error: - logging.error(f"Error while parsing XML data: {error}") - else: - if self.server.authenticate(): - data = self._get_xml_root() - if data is None: - logging.error("Unable to get XML data from Arista server") - raise ValueError("Unable to get XML data from Arista server") - self.xml_data = data - else: - logging.error("Unable to authenticate to Arista server") - raise ValueError("Unable to authenticate to Arista server") - - def _get_xml_root(self) -> Union[ET.ElementTree, None]: - logging.info("Getting XML root from Arista server.") - try: - return self.server.get_xml_data() - except Exception as error: # pylint: disable=broad-except - logging.error(f"Error while getting XML data from Arista server: {error}") - return None - - -class AristaXmlQuerier(AristaXmlBase): - """Class to query Arista XML data for Software versions.""" - - def available_public_versions( - self, - branch: Union[str, None] = None, - rtype: Union[str, None] = None, - package: AristaPackage = "eos", - ) -> List[AristaVersions]: - """Get list of available public EOS versions from Arista website. - - This method parses XML data to extract available EOS or CVP versions based on specified criteria. - - Args: - branch (Union[str, None], optional): Branch number to filter versions (e.g. "4.29"). - Defaults to None. - rtype (Union[str, None], optional): Release type to filter versions. - Must be one of the valid release types defined in RTYPES. Defaults to None. - package (AristaPackage, optional): Type of package to look for - either 'eos' or 'cvp'. - Defaults to 'eos'. - Returns: - List[eos_downloader.models.types.AristaVersions]: List of version objects (EosVersion or CvpVersion) matching the criteria. - List[AristaVersions]: List of version objects (EosVersion or CvpVersion) matching the criteria. - - Raises: - ValueError: If provided rtype is not in the list of valid release types. - - Example: - >>> server.available_public_eos_version(branch="4.29", rtype="INT", package="eos") - [EosVersion('4.29.0F-INT'), EosVersion('4.29.1F-INT'), ...] - """ - - logging.info(f"Getting available versions for {package} package") - - xpath_query = './/dir[@label="Active Releases"]//dir[@label]' - regexp = eos_downloader.models.version.EosVersion.regex_version - - if package == "cvp": - xpath_query = './/dir[@label="Active Releases"]//dir[@label]' - regexp = eos_downloader.models.version.CvpVersion.regex_version - - package_versions = [] - - if rtype is not None and rtype not in eos_downloader.models.data.RTYPES: - raise ValueError( - f"Invalid release type: {rtype}. Expected one of {eos_downloader.models.data.RTYPES}" - ) - nodes = self.xml_data.findall(xpath_query) - for node in nodes: - if "label" in node.attrib and node.get("label") is not None: - label = node.get("label") - if label is not None and regexp.match(label): - package_version = None - if package == "eos": - package_version = ( - eos_downloader.models.version.EosVersion.from_str(label) - ) - elif package == "cvp": - package_version = ( - eos_downloader.models.version.CvpVersion.from_str(label) - ) - package_versions.append(package_version) - if rtype is not None or branch is not None: - package_versions = [ - version - for version in package_versions - if version is not None - and (rtype is None or version.rtype == rtype) - and (branch is None or str(version.branch) == branch) - ] - - return package_versions - - def latest( - self, - package: eos_downloader.models.types.AristaPackage = "eos", - branch: Union[str, None] = None, - rtype: Union[eos_downloader.models.types.ReleaseType, None] = None, - ) -> AristaVersions: - """ - Get latest branch from semver standpoint +import json +from typing import Dict, Union, Any - Args: - branch (str): Branch to search for - rtype (str): Release type to search for - - Returns: - eos_downloader.models.version.EosVersion: Latest version found - """ - if package == "eos": - if rtype is not None and rtype not in eos_downloader.models.data.RTYPES: - raise ValueError( - f"Invalid release type: {rtype}. Expected {eos_downloader.models.data.RTYPES}" - ) - - versions = self.available_public_versions( - package=package, branch=branch, rtype=rtype - ) - if len(versions) == 0: - raise ValueError("No versions found to run the max() function") - return max(versions) - - def branches( - self, - package: eos_downloader.models.types.AristaPackage = "eos", - latest: bool = False, - ) -> List[str]: - """Returns a list of valid EOS version branches. - - The branches are determined based on the available public EOS versions. - When latest=True, only the most recent branch is returned. - - Args: - latest: If True, returns only the latest branch version. - If False, returns all available branches sorted in descending order. - - Returns: - List[str]: A list of branch version strings. - Contains single latest version if latest=True, - otherwise all available versions sorted descendingly. - """ - if latest: - latest_branch = max( - self._get_branches(self.available_public_versions(package=package)) - ) - return [str(latest_branch)] - return sorted( - self._get_branches(self.available_public_versions(package=package)), - reverse=True, - ) - - def _get_branches( - self, - versions: Union[ - List[eos_downloader.models.version.EosVersion], - List[eos_downloader.models.version.CvpVersion], - ], - ) -> List[str]: - """ - Extracts unique branch names from a list of version objects. - Args: - versions (Union[List[EosVersion], List[CvpVersion]]): A list of version objects, - either EosVersion or CvpVersion types. - Returns: - Union[List[EosVersion], List[CvpVersion]]: A list of unique branch names. - """ - branch = [version.branch for version in versions] - return list(set(branch)) - - -class AristaXmlObject(AristaXmlBase): - """Base class for Arista XML data management.""" - - software: ClassVar[AristaMapping] - base_xpath_active_version: ClassVar[str] - base_xpath_filepath: ClassVar[str] - checksum_file_extension: ClassVar[str] = "sha512sum" +import xml.etree.ElementTree as ET +from loguru import logger +import requests + +import eos_downloader.exceptions +import eos_downloader.defaults + + +class AristaServer: + """AristaServer class to handle authentication and interactions with Arista software download portal. + + This class provides methods to authenticate with the Arista software portal, + retrieve XML data containing available software packages, and generate download URLs + for specific files. + + Attributes + ---------- + token : str, optional + Authentication token for Arista portal access + timeout : int, default=5 + Timeout in seconds for HTTP requests + session_server : str + URL of the authentication server + headers : Dict[str, any] + HTTP headers to use in requests + xml_url : str + URL to retrieve software package XML data + download_server : str + Base URL for file downloads + _session_id : str + Session ID obtained after authentication + + Methods + ------- + authenticate(token: Union[bool, None] = None) -> bool + Authenticates with the Arista portal using provided or stored token + get_xml_data() -> ET.ElementTree + Retrieves XML data containing available software packages + get_url(remote_file_path: str) -> Union[str, None] + Generates download URL for a specific file path + + Raises + ------ + eos_downloader.exceptions.AuthenticationError + When authentication fails due to invalid or expired token + """ def __init__( self, - searched_version: str, - image_type: str, token: Union[str, None] = None, - xml_path: Union[str, None] = None, + timeout: int = 5, + session_server: str = eos_downloader.defaults.DEFAULT_SERVER_SESSION, + headers: Dict[str, Any] = eos_downloader.defaults.DEFAULT_REQUEST_HEADERS, + xml_url: str = eos_downloader.defaults.DEFAULT_SOFTWARE_FOLDER_TREE, + download_server: str = eos_downloader.defaults.DEFAULT_DOWNLOAD_URL, ) -> None: - self.search_version = searched_version - self.image_type = image_type - super().__init__(token=token, xml_path=xml_path) - - @property - def filename(self) -> Union[str, None]: - """ - _build_filename Helper to build filename to search on arista.com + """Initialize the Server class with optional parameters. + + Parameters + ---------- + token : Union[str, None], optional + Authentication token. Defaults to None. + timeout : int, optional + Request timeout in seconds. Defaults to 5. + session_server : str, optional + URL of the session server. Defaults to DEFAULT_SERVER_SESSION. + headers : Dict[str, any], optional + HTTP headers for requests. Defaults to DEFAULT_REQUEST_HEADERS. + xml_url : str, optional + URL of the software folder tree XML. Defaults to DEFAULT_SOFTWARE_FOLDER_TREE. + download_server : str, optional + Base URL for downloads. Defaults to DEFAULT_DOWNLOAD_URL. Returns ------- - str: - Filename to search for on Arista.com + None """ - logging.info( - f"Building filename for {self.image_type} package: {self.search_version}." - ) - try: - filename = eos_downloader.models.data.software_mapping.filename( - self.software, self.image_type, self.search_version - ) - return filename - except ValueError as e: - logging.error(f"Error: {e}") - return None + self.token: Union[str, None] = token + self._session_server = session_server + self._headers = headers + self._timeout = timeout + self._xml_url = xml_url + self._download_server = download_server + self._session_id = None - def hash_filename(self) -> Union[str, None]: - """ - hash_filename Helper to build filename for checksum to search on arista.com + logging.info(f"Initialized AristaServer with headers: {self._headers}") - Returns - ------- - str: - Filename to search for on Arista.com - """ - - logging.info(f"Building hash filename for {self.software} package.") - - if self.filename is not None: - return f"{self.filename}.{self.checksum_file_extension}" - return None - - def path_from_xml(self, search_file: str) -> Union[str, None]: - """Parse XML to find path for a given file. - - Args: - search_file (str): File to search for - - Returns: - Union[str, None]: Path from XML if found, None otherwise - """ - - logging.info(f"Building path from XML for {search_file}.") - - # Build xpath with provided file - xpath_query = self.base_xpath_filepath.format(search_file) - # Find the element using XPath - path_element = self.xml_data.find(xpath_query) - - if path_element is not None: - logging.debug(f'found path: {path_element.get("path")} for {search_file}') + def authenticate(self, token: Union[str, None] = None) -> bool: + """Authenticate to the API server using access token. - # Return the path if found, otherwise return None - return path_element.get("path") if path_element is not None else None + The token is encoded in base64 and sent to the server for authentication. + A session ID is retrieved from the server response if authentication is successful. - def _url(self, xml_path: str) -> Union[str, None]: - """Get URL to download a file from Arista server. + Parameters + ---------- + token : Union[str, None], optional + Access token for authentication. If None, uses existing token stored in instance. Defaults to None. - Args: - xml_path (str): Path to the file in the XML + Returns + ------- + bool + True if authentication successful, False otherwise - Returns: - str: URL to download the file + Raises + ------ + eos_downloader.exceptions.AuthenticationError + If access token is invalid or expired """ - logging.info(f"Getting URL for {xml_path}.") - - return self.server.get_url(xml_path) - - @property - def urls(self) -> Dict[str, Union[str, None]]: - """Get URLs to download files from Arista server for given software and version. + if token is not None: + self.token = token + if self.token is None: + logger.error("No token provided for authentication") + return False + credentials = (base64.b64encode(self.token.encode())).decode("utf-8") + jsonpost = {"accessToken": credentials} + result = requests.post( + self._session_server, + data=json.dumps(jsonpost), + timeout=self._timeout, + headers=self._headers, + ) + if result.json()["status"]["message"] in [ + "Access token expired", + "Invalid access token", + ]: + logging.critical( + f"Authentication failed: {result.json()['status']['message']}" + ) + raise eos_downloader.exceptions.AuthenticationError + # return False + try: + if "data" in result.json(): + self._session_id = result.json()["data"]["session_code"] + logging.info(f"Authenticated with session ID: {self._session_id}") + return True + except KeyError as error: + logger.error( + f"Key Error in parsing server response ({result.json()}): {error}" + ) + return False + return False - This method will return a dictionary with file type as key and URL as value. - It returns URL for the following items: 'image', 'md5sum', and 'sha512sum'. + def get_xml_data(self) -> Union[ET.ElementTree, None]: + """Retrieves XML data from the server. - Returns: - Dict[str, str]: Dictionary with file type as key and URL as value - """ - logging.info(f"Getting URLs for {self.software} package.") - - urls = {} - - if self.filename is None: - raise ValueError("Filename not found") - - for role in self.supported_role_types: - file_path = None - logging.debug(f"working on {role}") - hash_filename = self.hash_filename() - if hash_filename is None: - raise ValueError("Hash file not found") - if role == "image": - file_path = self.path_from_xml(self.filename) - elif role == self.checksum_file_extension: - file_path = self.path_from_xml(hash_filename) - if file_path is not None: - logging.info(f"Adding {role} with {file_path} to urls dict") - urls[role] = self._url(file_path) - logging.debug(f"URLs dict contains: {urls}") - return urls - - -class EosXmlObject(AristaXmlObject): - """Class to query Arista XML data for EOS versions.""" - - software: ClassVar[AristaMapping] = "EOS" - base_xpath_active_version: ClassVar[ - str - ] = './/dir[@label="Active Releases"]/dir/dir/[@label]' - base_xpath_filepath: ClassVar[str] = './/file[.="{}"]' - - # File extensions supported to be downloaded from arista server. - # Should cover: image file (image) and has files (md5sum and/or sha512sum) - supported_role_types: ClassVar[List[str]] = ["image", "md5sum", "sha512sum"] - checksum_file_extension: ClassVar[str] = "sha512sum" + This method fetches XML data by making a POST request to the server's XML endpoint. + If not already authenticated, it will initiate the authentication process first. - def __init__( - self, - searched_version: str, - image_type: str, - token: Union[str, None] = None, - xml_path: Union[str, None] = None, - ) -> None: - """Initialize an instance of the EosXmlObject class. - - Args: - searched_version (str): The version of the software to search for. - image_type (str): The type of image to download. - token (Union[str, None], optional): The authentication token. Defaults to None. - xml_path (Union[str, None], optional): The path to the XML file. Defaults to None. - - Returns: - None + Returns + ------- + ET.ElementTree + An ElementTree object containing the parsed XML data from the server response. + + Raises + ------ + KeyError + If the server response doesn't contain the expected data structure. + + Notes + ----- + The method requires a valid session ID which is obtained through authentication. + The XML data is expected to be in the response JSON under data.xml path. """ - self.search_version = searched_version - self.image_type = image_type - self.version = eos_downloader.models.version.EosVersion().from_str( - searched_version - ) - - super().__init__( - searched_version=searched_version, - image_type=image_type, - token=token, - xml_path=xml_path, + logging.info(f"Getting XML data from server {self._session_server}") + if self._session_id is None: + logger.debug("Not authenticated to server, start authentication process") + self.authenticate() + jsonpost = {"sessionCode": self._session_id} + result = requests.post( + self._xml_url, + data=json.dumps(jsonpost), + timeout=self._timeout, + headers=self._headers, ) + try: + folder_tree = result.json()["data"]["xml"] + logging.debug("XML data received from Arista server") + return ET.ElementTree(ET.fromstring(folder_tree)) + except KeyError as error: + logger.error(f"Unkown key in server response: {error}") + return None + def get_url(self, remote_file_path: str) -> Union[str, None]: + """Get download URL for a remote file from server. -class CvpXmlObject(AristaXmlObject): - """Class to query Arista XML data for CVP versions.""" - - software: ClassVar[AristaMapping] = "CloudVision" - base_xpath_active_version: ClassVar[ - str - ] = './/dir[@label="Active Releases"]/dir/dir/[@label]' - base_xpath_filepath: ClassVar[str] = './/file[.="{}"]' + This method retrieves the download URL for a specified remote file by making a POST request + to the server. If not authenticated, it will first authenticate before making the request. - # File extensions supported to be downloaded from arista server. - # Should cover: image file (image) and has files (md5sum and/or sha512sum) - supported_role_types: ClassVar[List[str]] = ["image", "md5"] - checksum_file_extension: ClassVar[str] = "md5" + Parameters + ---------- + remote_file_path : str + Path to the remote file on server to get download URL for - def __init__( - self, - searched_version: str, - image_type: str, - token: Union[str, None] = None, - xml_path: Union[str, None] = None, - ) -> None: - """Initialize an instance of the CvpXmlObject class. - - Args: - searched_version (str): The version of the software to search for. - image_type (str): The type of image to download. - token (Union[str, None], optional): The authentication token. Defaults to None. - xml_path (Union[str, None], optional): The path to the XML file. Defaults to None. - - Returns: - None + Returns + ------- + Union[str, None] + The download URL if successful, None if request fails or URL not found in response + + Raises + ------ + requests.exceptions.RequestException + If the request to server fails + json.JSONDecodeError + If server response is not valid JSON + requests.exceptions.Timeout + If server request times out """ - self.search_version = searched_version - self.image_type = image_type - self.version = eos_downloader.models.version.CvpVersion().from_str( - searched_version + logging.info(f"Getting download URL for {remote_file_path}") + if self._session_id is None: + logger.debug("Not authenticated to server, start authentication process") + self.authenticate() + jsonpost = {"sessionCode": self._session_id, "filePath": remote_file_path} + result = requests.post( + self._download_server, + data=json.dumps(jsonpost), + timeout=self._timeout, + headers=self._headers, ) - - super().__init__( - searched_version=searched_version, - image_type=image_type, - token=token, - xml_path=xml_path, - ) - - -# Create the custom type -AristaXmlObjects = Union[CvpXmlObject, EosXmlObject] + if "data" in result.json() and "url" in result.json()["data"]: + # logger.debug('URL to download file is: {}', result.json()) + logging.info("Download URL received from server") + logging.debug(f'URL to download file is: {result.json()["data"]["url"]}') + return result.json()["data"]["url"] + return None diff --git a/eos_downloader/logics/arista_xml_server.py b/eos_downloader/logics/arista_xml_server.py new file mode 100644 index 0000000..1f3d0c3 --- /dev/null +++ b/eos_downloader/logics/arista_xml_server.py @@ -0,0 +1,536 @@ +# coding: utf-8 -*- + +"""This module provides classes for managing and querying Arista XML data. + +Classes: + AristaXmlBase: Base class for Arista XML data management. + AristaXmlQuerier: Class to query Arista XML data for Software versions. + AristaXmlObject: Base class for Arista XML data management with specific software and version. + EosXmlObject: Class to query Arista XML data for EOS versions. +""" # noqa: E501 + +from __future__ import annotations + +import logging +import xml.etree.ElementTree as ET +from typing import ClassVar, Union, List, Dict + +import eos_downloader.logics.arista_server +import eos_downloader.models.version +import eos_downloader.models.data +from eos_downloader.models.types import AristaPackage, AristaVersions, AristaMapping + + +class AristaXmlBase: + # pylint: disable=too-few-public-methods + """Base class for Arista XML data management.""" + + # File extensions supported to be downloaded from arista server. + # Should cover: image file (image) and has files (md5sum and/or sha512sum) + supported_role_types: ClassVar[List[str]] = ["image", "md5sum", "sha512sum"] + + def __init__( + self, token: Union[str, None] = None, xml_path: Union[str, None] = None + ) -> None: + """ + Initialize the AristaXmlBase class. + + Parameters + ---------- + token : Union[str, None], optional + Authentication token. Defaults to None. + xml_path : Union[str, None], optional + Path to the XML file. Defaults to None. + + Returns + ------- + None + """ + logging.info("Initializing AristXmlBase.") + self.server = eos_downloader.logics.arista_server.AristaServer(token=token) + if xml_path is not None: + try: + self.xml_data = ET.parse(xml_path) + except ET.ParseError as error: + logging.error(f"Error while parsing XML data: {error}") + else: + if self.server.authenticate(): + data = self._get_xml_root() + if data is None: + logging.error("Unable to get XML data from Arista server") + raise ValueError("Unable to get XML data from Arista server") + self.xml_data = data + else: + logging.error("Unable to authenticate to Arista server") + raise ValueError("Unable to authenticate to Arista server") + + def _get_xml_root(self) -> Union[ET.ElementTree, None]: + """ + Retrieves the XML root from the Arista server. + + Returns + ------- + Union[ET.ElementTree, None] + The XML root element tree if successful, None otherwise. + """ + logging.info("Getting XML root from Arista server.") + try: + return self.server.get_xml_data() + except Exception as error: # pylint: disable=broad-except + logging.error(f"Error while getting XML data from Arista server: {error}") + return None + + +class AristaXmlQuerier(AristaXmlBase): + """Class to query Arista XML data for Software versions.""" + + def available_public_versions( + self, + branch: Union[str, None] = None, + rtype: Union[str, None] = None, + package: AristaPackage = "eos", + ) -> List[AristaVersions]: + """ + Get list of available public EOS versions from Arista website. + + This method parses XML data to extract available EOS or CVP versions based on specified criteria. + + Parameters + ---------- + branch : Union[str, None], optional + Branch number to filter versions (e.g. "4.29"). Defaults to None. + rtype : Union[str, None], optional + Release type to filter versions. Must be one of the valid release types defined in RTYPES. Defaults to None. + package : AristaPackage, optional + Type of package to look for - either 'eos' or 'cvp'. Defaults to 'eos'. + + Returns + ------- + List[AristaVersions] + List of version objects (EosVersion or CvpVersion) matching the criteria. + + Raises + ------ + ValueError + If provided rtype is not in the list of valid release types. + + Examples + -------- + >>> server.available_public_eos_version(branch="4.29", rtype="INT", package="eos") + [EosVersion('4.29.0F-INT'), EosVersion('4.29.1F-INT'), ...] + """ + + logging.info(f"Getting available versions for {package} package") + + xpath_query = './/dir[@label="Active Releases"]//dir[@label]' + regexp = eos_downloader.models.version.EosVersion.regex_version + + if package == "cvp": + xpath_query = './/dir[@label="Active Releases"]//dir[@label]' + regexp = eos_downloader.models.version.CvpVersion.regex_version + + package_versions = [] + + if rtype is not None and rtype not in eos_downloader.models.data.RTYPES: + raise ValueError( + f"Invalid release type: {rtype}. Expected one of {eos_downloader.models.data.RTYPES}" + ) + nodes = self.xml_data.findall(xpath_query) + for node in nodes: + if "label" in node.attrib and node.get("label") is not None: + label = node.get("label") + if label is not None and regexp.match(label): + package_version = None + if package == "eos": + package_version = ( + eos_downloader.models.version.EosVersion.from_str(label) + ) + elif package == "cvp": + package_version = ( + eos_downloader.models.version.CvpVersion.from_str(label) + ) + package_versions.append(package_version) + if rtype is not None or branch is not None: + package_versions = [ + version + for version in package_versions + if version is not None + and (rtype is None or version.rtype == rtype) + and (branch is None or str(version.branch) == branch) + ] + + return package_versions + + def latest( + self, + package: eos_downloader.models.types.AristaPackage = "eos", + branch: Union[str, None] = None, + rtype: Union[eos_downloader.models.types.ReleaseType, None] = None, + ) -> AristaVersions: + """ + Get latest branch from semver standpoint. + + Parameters + ---------- + package : eos_downloader.models.types.AristaPackage, optional + Type of package to look for - either 'eos' or 'cvp'. Defaults to 'eos'. + branch : Union[str, None], optional + Branch to search for. Defaults to None. + rtype : Union[eos_downloader.models.types.ReleaseType, None], optional + Release type to search for. Defaults to None. + + Returns + ------- + AristaVersions + Latest version found. + + Raises + ------ + ValueError + If no versions are found to run the max() function. + """ + if package == "eos": + if rtype is not None and rtype not in eos_downloader.models.data.RTYPES: + raise ValueError( + f"Invalid release type: {rtype}. Expected {eos_downloader.models.data.RTYPES}" + ) + + versions = self.available_public_versions( + package=package, branch=branch, rtype=rtype + ) + if len(versions) == 0: + raise ValueError("No versions found to run the max() function") + return max(versions) + + def branches( + self, + package: eos_downloader.models.types.AristaPackage = "eos", + latest: bool = False, + ) -> List[str]: + """ + Returns a list of valid EOS version branches. + + The branches are determined based on the available public EOS versions. + When latest=True, only the most recent branch is returned. + + Parameters + ---------- + package : eos_downloader.models.types.AristaPackage, optional + Type of package to look for - either 'eos' or 'cvp'. Defaults to 'eos'. + latest : bool, optional + If True, returns only the latest branch version. Defaults to False. + + Returns + ------- + List[str] + A list of branch version strings. Contains single latest version if latest=True, + otherwise all available versions sorted descendingly. + """ + if latest: + latest_branch = max( + self._get_branches(self.available_public_versions(package=package)) + ) + return [str(latest_branch)] + return sorted( + self._get_branches(self.available_public_versions(package=package)), + reverse=True, + ) + + def _get_branches( + self, + versions: Union[ + List[eos_downloader.models.version.EosVersion], + List[eos_downloader.models.version.CvpVersion], + ], + ) -> List[str]: + """ + Extracts unique branch names from a list of version objects. + + Parameters + ---------- + versions : Union[List[eos_downloader.models.version.EosVersion], List[eos_downloader.models.version.CvpVersion]] + A list of version objects, either EosVersion or CvpVersion types. + + Returns + ------- + List[str] + A list of unique branch names. + """ + branch = [version.branch for version in versions] + return list(set(branch)) + + +class AristaXmlObject(AristaXmlBase): + """Base class for Arista XML data management.""" + + software: ClassVar[AristaMapping] + base_xpath_active_version: ClassVar[str] + base_xpath_filepath: ClassVar[str] + checksum_file_extension: ClassVar[str] = "sha512sum" + + def __init__( + self, + searched_version: str, + image_type: str, + token: Union[str, None] = None, + xml_path: Union[str, None] = None, + ) -> None: + """ + Initialize the AristaXmlObject class. + + Parameters + ---------- + searched_version : str + The version of the software to search for. + image_type : str + The type of image to download. + token : Union[str, None], optional + Authentication token. Defaults to None. + xml_path : Union[str, None], optional + Path to the XML file. Defaults to None. + + Returns + ------- + None + """ + self.search_version = searched_version + self.image_type = image_type + super().__init__(token=token, xml_path=xml_path) + + @property + def filename(self) -> Union[str, None]: + """ + Helper to build filename to search on arista.com. + + Returns + ------- + Union[str, None] + Filename to search for on Arista.com. + """ + logging.info( + f"Building filename for {self.image_type} package: {self.search_version}." + ) + try: + filename = eos_downloader.models.data.software_mapping.filename( + self.software, self.image_type, self.search_version + ) + return filename + except ValueError as e: + logging.error(f"Error: {e}") + return None + + def hash_filename(self) -> Union[str, None]: + """ + Helper to build filename for checksum to search on arista.com. + + Returns + ------- + Union[str, None] + Filename to search for on Arista.com. + """ + + logging.info(f"Building hash filename for {self.software} package.") + + if self.filename is not None: + return f"{self.filename}.{self.checksum_file_extension}" + return None + + def path_from_xml(self, search_file: str) -> Union[str, None]: + """ + Parse XML to find path for a given file. + + Parameters + ---------- + search_file : str + File to search for. + + Returns + ------- + Union[str, None] + Path from XML if found, None otherwise. + """ + + logging.info(f"Building path from XML for {search_file}.") + + # Build xpath with provided file + xpath_query = self.base_xpath_filepath.format(search_file) + # Find the element using XPath + path_element = self.xml_data.find(xpath_query) + + if path_element is not None: + logging.debug(f'found path: {path_element.get("path")} for {search_file}') + + # Return the path if found, otherwise return None + return path_element.get("path") if path_element is not None else None + + def _url(self, xml_path: str) -> Union[str, None]: + """ + Get URL to download a file from Arista server. + + Parameters + ---------- + xml_path : str + Path to the file in the XML. + + Returns + ------- + Union[str, None] + URL to download the file. + """ + + logging.info(f"Getting URL for {xml_path}.") + + return self.server.get_url(xml_path) + + @property + def urls(self) -> Dict[str, Union[str, None]]: + """ + Get URLs to download files from Arista server for given software and version. + + This method will return a dictionary with file type as key and URL as value. + It returns URL for the following items: 'image', 'md5sum', and 'sha512sum'. + + Returns + ------- + Dict[str, Union[str, None]] + Dictionary with file type as key and URL as value. + + Raises + ------ + ValueError + If filename or hash file is not found. + """ + logging.info(f"Getting URLs for {self.software} package.") + + urls = {} + + if self.filename is None: + raise ValueError("Filename not found") + + for role in self.supported_role_types: + file_path = None + logging.debug(f"working on {role}") + hash_filename = self.hash_filename() + if hash_filename is None: + raise ValueError("Hash file not found") + if role == "image": + file_path = self.path_from_xml(self.filename) + elif role == self.checksum_file_extension: + file_path = self.path_from_xml(hash_filename) + if file_path is not None: + logging.info(f"Adding {role} with {file_path} to urls dict") + urls[role] = self._url(file_path) + logging.debug(f"URLs dict contains: {urls}") + return urls + + +class EosXmlObject(AristaXmlObject): + """Class to query Arista XML data for EOS versions.""" + + software: ClassVar[AristaMapping] = "EOS" + base_xpath_active_version: ClassVar[ + str + ] = './/dir[@label="Active Releases"]/dir/dir/[@label]' + base_xpath_filepath: ClassVar[str] = './/file[.="{}"]' + + # File extensions supported to be downloaded from arista server. + # Should cover: image file (image) and has files (md5sum and/or sha512sum) + supported_role_types: ClassVar[List[str]] = ["image", "md5sum", "sha512sum"] + checksum_file_extension: ClassVar[str] = "sha512sum" + + def __init__( + self, + searched_version: str, + image_type: str, + token: Union[str, None] = None, + xml_path: Union[str, None] = None, + ) -> None: + """ + Initialize an instance of the EosXmlObject class. + + Parameters + ---------- + searched_version : str + The version of the software to search for. + image_type : str + The type of image to download. + token : Union[str, None], optional + The authentication token. Defaults to None. + xml_path : Union[str, None], optional + The path to the XML file. Defaults to None. + + Returns + ------- + None + """ + + self.search_version = searched_version + self.image_type = image_type + self.version = eos_downloader.models.version.EosVersion().from_str( + searched_version + ) + + super().__init__( + searched_version=searched_version, + image_type=image_type, + token=token, + xml_path=xml_path, + ) + + +class CvpXmlObject(AristaXmlObject): + """Class to query Arista XML data for CVP versions.""" + + software: ClassVar[AristaMapping] = "CloudVision" + base_xpath_active_version: ClassVar[ + str + ] = './/dir[@label="Active Releases"]/dir/dir/[@label]' + base_xpath_filepath: ClassVar[str] = './/file[.="{}"]' + + # File extensions supported to be downloaded from arista server. + # Should cover: image file (image) and has files (md5sum and/or sha512sum) + supported_role_types: ClassVar[List[str]] = ["image", "md5"] + checksum_file_extension: ClassVar[str] = "md5" + + def __init__( + self, + searched_version: str, + image_type: str, + token: Union[str, None] = None, + xml_path: Union[str, None] = None, + ) -> None: + """ + Initialize an instance of the CvpXmlObject class. + + Parameters + ---------- + searched_version : str + The version of the software to search for. + image_type : str + The type of image to download. + token : Union[str, None], optional + The authentication token. Defaults to None. + xml_path : Union[str, None], optional + The path to the XML file. Defaults to None. + + Returns + ------- + None + """ + + self.search_version = searched_version + self.image_type = image_type + self.version = eos_downloader.models.version.CvpVersion().from_str( + searched_version + ) + + super().__init__( + searched_version=searched_version, + image_type=image_type, + token=token, + xml_path=xml_path, + ) + + +# Create the custom type +AristaXmlObjects = Union[CvpXmlObject, EosXmlObject] diff --git a/eos_downloader/logics/download.py b/eos_downloader/logics/download.py index ec303b7..68fb3f5 100644 --- a/eos_downloader/logics/download.py +++ b/eos_downloader/logics/download.py @@ -6,6 +6,7 @@ the download process. Methods +-------- download_file(url: str, file_path: str, filename: str, rich_interface: bool = True) -> Union[None, str] Downloads a file from the given URL to the specified path with optional rich interface. @@ -13,16 +14,18 @@ Static method that performs the actual file download with tqdm progress bar. Attributes +-------- None Example ->>> downloader = ObjectDownloader() ->>> result = downloader.download_file( -... url='http://example.com/file.zip', -... file_path='/downloads', -... filename='file.zip', -... rich_interface=True -... ) +-------- + >>> downloader = ObjectDownloader() + >>> result = downloader.download_file( + ... url='http://example.com/file.zip', + ... file_path='/downloads', + ... filename='file.zip', + ... rich_interface=True + ... ) """ import os @@ -38,7 +41,7 @@ import eos_downloader.defaults import eos_downloader.helpers import eos_downloader.logics -import eos_downloader.logics.arista_server +import eos_downloader.logics.arista_xml_server import eos_downloader.models.version @@ -48,12 +51,6 @@ class SoftManager: This class provides methods to download files using either a simple progress bar or a rich interface with enhanced visual feedback. - Methods - download_file(url: str, file_path: str, filename: str, rich_interface: bool = True) -> Union[None, str] - Downloads a file from the given URL to the specified path - _download_file_raw(url: str, file_path: str) -> str - Internal method to download file with basic progress bar - Examples -------- >>> downloader = SoftManager() @@ -77,17 +74,23 @@ def __init__(self, dry_run: bool = False) -> None: def _download_file_raw(url: str, file_path: str) -> str: """Downloads a file from a URL and saves it to a local file. - Args: - url (str): The URL of the file to download. - file_path (str): The local path where the file will be saved. - - Returns: - str: The path to the downloaded file. + Parameters + ---------- + url : str + The URL of the file to download. + file_path : str + The local path where the file will be saved. - Notes: - - Uses requests library to stream download in chunks of 1024 bytes - - Shows download progress using tqdm progress bar - - Sets timeout of 5 seconds for initial connection + Returns + ------- + str + The path to the downloaded file. + + Notes + ----- + - Uses requests library to stream download in chunks of 1024 bytes + - Shows download progress using tqdm progress bar + - Sets timeout of 5 seconds for initial connection """ chunkSize = 1024 @@ -107,7 +110,17 @@ def _download_file_raw(url: str, file_path: str) -> str: @staticmethod def _create_destination_folder(path: str) -> None: - """Creates a directory path if it doesn't already exist.""" + """Creates a directory path if it doesn't already exist. + + Parameters + ---------- + path : str + The directory path to create. + + Returns + ------- + None + """ try: os.makedirs(path, exist_ok=True) except OSError as e: @@ -115,21 +128,21 @@ def _create_destination_folder(path: str) -> None: def _compute_hash_md5sum(self, file: str, hash_expected: str) -> bool: """ - _compute_hash_md5sum Compare MD5 sum + Compare MD5 sum. - Do comparison between local md5 of the file and value provided by arista.com + Do comparison between local md5 of the file and value provided by arista.com. Parameters ---------- file : str - Local file to use for MD5 sum + Local file to use for MD5 sum. hash_expected : str - MD5 from arista.com + MD5 from arista.com. Returns ------- bool - True if both are equal, False if not + True if both are equal, False if not. """ hash_md5 = hashlib.md5() with open(file, "rb") as f: @@ -149,20 +162,26 @@ def checksum(self, check_type: Literal["md5sum", "sha512sum", "md5"]) -> bool: """ Verifies the integrity of a downloaded file using a specified checksum algorithm. - Args: - check_type (Literal['md5sum', 'sha512sum']): The type of checksum to perform. Currently supports 'md5sum' or 'sha512sum'. - - Returns: - bool: True if the checksum verification passes. - - Raises: - ValueError: If the calculated checksum does not match the expected checksum. - FileNotFoundError: If either the checksum file or the target file cannot be found. + Parameters + ---------- + check_type : Literal['md5sum', 'sha512sum', 'md5'] + The type of checksum to perform. Currently supports 'md5sum' or 'sha512sum'. - Example: - ```python - client.checksum('sha512sum') # Returns True if checksum matches - ``` + Returns + ------- + bool + True if the checksum verification passes. + + Raises + ------ + ValueError + If the calculated checksum does not match the expected checksum. + FileNotFoundError + If either the checksum file or the target file cannot be found. + + Examples + -------- + >>> client.checksum('sha512sum') # Returns True if checksum matches """ logging.info(f"Checking checksum for {self.file['name']} using {check_type}") @@ -230,14 +249,21 @@ def download_file( """ Downloads a file from a given URL to a specified location. - Args: - url (str): The URL from which to download the file. - file_path (str): The directory path where the file should be saved. - filename (str): The name to be given to the downloaded file. - rich_interface (bool, optional): Whether to use rich progress bar interface. Defaults to True. + Parameters + ---------- + url : str + The URL from which to download the file. + file_path : str + The directory path where the file should be saved. + filename : str + The name to be given to the downloaded file. + rich_interface : bool, optional + Whether to use rich progress bar interface. Defaults to True. - Returns: - Union[None, str]: The full path to the downloaded file if successful, None if download fails. + Returns + ------- + Union[None, str] + The full path to the downloaded file if successful, None if download fails. """ logging.info( f"{'[DRY-RUN] Would download' if self.dry_run else 'Downloading'} {filename} from {url}" @@ -258,26 +284,34 @@ def download_file( def downloads( self, - object_arista: eos_downloader.logics.arista_server.AristaXmlObjects, + object_arista: eos_downloader.logics.arista_xml_server.AristaXmlObjects, file_path: str, rich_interface: bool = True, ) -> Union[None, str]: - """Downloads files from Arista EOS server. + """ + Downloads files from Arista EOS server. Downloads the EOS image and optional md5/sha512 files based on the provided EOS XML object. Each file is downloaded to the specified path with appropriate filenames. - Args: - object_arista (EosXmlObject): Object containing EOS image and hash file URLs - file_path (str): Directory path where files should be downloaded - rich_interface (bool, optional): Whether to use rich console output. Defaults to True. + Parameters + ---------- + object_arista : eos_downloader.logics.arista_xml_server.AristaXmlObjects + Object containing EOS image and hash file URLs. + file_path : str + Directory path where files should be downloaded. + rich_interface : bool, optional + Whether to use rich console output. Defaults to True. - Returns: - Union[None, str]: The file path where files were downloaded, or None if download failed + Returns + ------- + Union[None, str] + The file path where files were downloaded, or None if download failed. - Example: - >>> client.downloads(eos_obj, "/tmp/downloads") - '/tmp/downloads' + Examples + -------- + >>> client.downloads(eos_obj, "/tmp/downloads") + '/tmp/downloads' """ logging.info(f"Downloading files from {object_arista.version}") @@ -317,22 +351,31 @@ def import_docker( docker_name: str = "arista/ceos", docker_tag: str = "latest", ) -> None: - """Import a local file into a Docker image. + """ + Import a local file into a Docker image. This method imports a local file into Docker with a specified image name and tag. It checks for the existence of both the local file and docker binary before proceeding. - Args: - local_file_path (str): Path to the local file to import - docker_name (str, optional): Name for the Docker image. Defaults to 'arista/ceos' - docker_tag (str, optional): Tag for the Docker image. Defaults to 'latest' - - Raises: - FileNotFoundError: If the local file doesn't exist or docker binary is not found - Exception: If the docker import operation fails + Parameters + ---------- + local_file_path : str + Path to the local file to import. + docker_name : str, optional + Name for the Docker image. Defaults to 'arista/ceos'. + docker_tag : str, optional + Tag for the Docker image. Defaults to 'latest'. + + Raises + ------ + FileNotFoundError + If the local file doesn't exist or docker binary is not found. + Exception + If the docker import operation fails. - Returns: - None + Returns + ------- + None """ logging.info( @@ -358,20 +401,27 @@ def import_docker( # pylint: disable=too-many-branches def provision_eve( self, - object_arista: eos_downloader.logics.arista_server.EosXmlObject, + object_arista: eos_downloader.logics.arista_xml_server.EosXmlObject, noztp: bool = False, ) -> None: """ Provisions EVE-NG with the specified Arista EOS object. - Args: - object_arista (eos_downloader.logics.arista_server.EosXmlObject): The Arista EOS object containing version, filename, and URLs. - noztp (bool, optional): If True, disables ZTP (Zero Touch Provisioning). Defaults to False. - checksum (bool, optional): If True, verifies the checksum of the downloaded files. Defaults to True. + Parameters + ---------- + object_arista : eos_downloader.logics.arista_xml_server.EosXmlObject + The Arista EOS object containing version, filename, and URLs. + noztp : bool, optional + If True, disables ZTP (Zero Touch Provisioning). Defaults to False. - Raises: - ValueError: If no URLs are found for download or if a URL or filename is None. + Raises + ------ + ValueError + If no URLs are found for download or if a URL or filename is None. + Returns + ------- + None """ # EVE-NG provisioning page for vEOS diff --git a/eos_downloader/logics/server.py b/eos_downloader/logics/server.py deleted file mode 100644 index a1d2527..0000000 --- a/eos_downloader/logics/server.py +++ /dev/null @@ -1,263 +0,0 @@ -#!/usr/bin/python -# coding: utf-8 -*- - -"""Server module for handling interactions with Arista software download portal. - -This module provides the AristaServer class which manages authentication and -file retrieval operations with the Arista software download portal. It handles -session management, XML data retrieval, and download URL generation. - -Classes -------- -AristaServer - Main class for interacting with the Arista software portal. - -Dependencies ------------ -- base64: For encoding authentication tokens -- json: For handling JSON data in requests -- xml.etree.ElementTree: For parsing XML responses -- loguru: For logging -- requests: For making HTTP requests - -Example -------- - >>> from eos_downloader.logics.server import AristaServer - >>> server = AristaServer(token='my_auth_token') - >>> server.authenticate() - >>> xml_data = server.get_xml_data() - >>> download_url = server.get_url('/path/to/file') - -Notes ------ -The module requires valid authentication credentials to interact with the Arista portal. -All server interactions are performed over HTTPS and follow Arista's API specifications. -""" - -from __future__ import annotations - -import base64 -import logging -import json -from typing import Dict, Union, Any - -import xml.etree.ElementTree as ET -from loguru import logger -import requests - -import eos_downloader.exceptions -import eos_downloader.defaults - - -class AristaServer: - """AristaServer class to handle authentication and interactions with Arista software download portal. - - This class provides methods to authenticate with the Arista software portal, - retrieve XML data containing available software packages, and generate download URLs - for specific files. - - Attributes - ------ - token : str, optional - Authentication token for Arista portal access - timeout : int, default=5 - Timeout in seconds for HTTP requests - session_server : str - URL of the authentication server - headers : Dict[str, any] - HTTP headers to use in requests - xml_url : str - URL to retrieve software package XML data - download_server : str - Base URL for file downloads - _session_id : str - Session ID obtained after authentication - - Methods - ------ - authenticate(token: Union[bool, None] = None) -> bool - Authenticates with the Arista portal using provided or stored token - get_xml_data() -> ET.ElementTree - Retrieves XML data containing available software packages - get_url(remote_file_path: str) -> Union[str, None] - Generates download URL for a specific file path - - Raises - ------ - eos_downloader.exceptions.AuthenticationError - When authentication fails due to invalid or expired token - """ - - def __init__( - self, - token: Union[str, None] = None, - timeout: int = 5, - session_server: str = eos_downloader.defaults.DEFAULT_SERVER_SESSION, - headers: Dict[str, Any] = eos_downloader.defaults.DEFAULT_REQUEST_HEADERS, - xml_url: str = eos_downloader.defaults.DEFAULT_SOFTWARE_FOLDER_TREE, - download_server: str = eos_downloader.defaults.DEFAULT_DOWNLOAD_URL, - ) -> None: - # pylint: disable=dangerous-default-value, - # pylint: disable=too-many-positional-arguments - """Initialize the Server class with optional parameters. - - Args: - ------ - token (Union[str, None], optional): Authentication token. Defaults to None. - timeout (int, optional): Request timeout in seconds. Defaults to 5. - session_server (str, optional): URL of the session server. Defaults to DEFAULT_SERVER_SESSION. - headers (Dict[str, any], optional): HTTP headers for requests. Defaults to DEFAULT_REQUEST_HEADERS. - xml_url (str, optional): URL of the software folder tree XML. Defaults to DEFAULT_SOFTWARE_FOLDER_TREE. - download_server (str, optional): Base URL for downloads. Defaults to DEFAULT_DOWNLOAD_URL. - - Returns: - ------ - None - """ - self.token: Union[str, None] = token - self._session_server = session_server - self._headers = headers - self._timeout = timeout - self._xml_url = xml_url - self._download_server = download_server - self._session_id = None - - logging.info(f"Initialized AristaServer with headers: {self._headers}") - - def authenticate(self, token: Union[str, None] = None) -> bool: - """Authenticate to the API server using access token. - - The token is encoded in base64 and sent to the server for authentication. - A session ID is retrieved from the server response if authentication is successful. - - Example: - ------ - >>> server = Server() - >>> server.authenticate(token="myaccesstoken") - True - - Args: - ------ - token (Union[str, None], optional): Access token for authentication. - If None, uses existing token stored in instance. Defaults to None. - - Returns: - ------ - bool: True if authentication successful, False otherwise - - Raises: - ------ - eos_downloader.exceptions.AuthenticationError: If access token is invalid or expired - """ - - if token is not None: - self.token = token - if self.token is None: - logger.error("No token provided for authentication") - return False - credentials = (base64.b64encode(self.token.encode())).decode("utf-8") - jsonpost = {"accessToken": credentials} - result = requests.post( - self._session_server, - data=json.dumps(jsonpost), - timeout=self._timeout, - headers=self._headers, - ) - if result.json()["status"]["message"] in [ - "Access token expired", - "Invalid access token", - ]: - logging.critical( - f"Authentication failed: {result.json()['status']['message']}" - ) - raise eos_downloader.exceptions.AuthenticationError - # return False - try: - if "data" in result.json(): - self._session_id = result.json()["data"]["session_code"] - logging.info(f"Authenticated with session ID: {self._session_id}") - return True - except KeyError as error: - logger.error( - f"Key Error in parsing server response ({result.json()}): {error}" - ) - return False - return False - - def get_xml_data(self) -> Union[ET.ElementTree, None]: - """Retrieves XML data from the server. - - This method fetches XML data by making a POST request to the server's XML endpoint. - If not already authenticated, it will initiate the authentication process first. - - Returns: - ------ - ET.ElementTree: An ElementTree object containing the parsed XML data from the server response. - - Raises: - ------ - KeyError: If the server response doesn't contain the expected data structure. - - Note: - ------ - The method requires a valid session ID which is obtained through authentication. - The XML data is expected to be in the response JSON under data.xml path. - """ - - logging.info(f"Getting XML data from server {self._session_server}") - if self._session_id is None: - logger.debug("Not authenticated to server, start authentication process") - self.authenticate() - jsonpost = {"sessionCode": self._session_id} - result = requests.post( - self._xml_url, - data=json.dumps(jsonpost), - timeout=self._timeout, - headers=self._headers, - ) - try: - folder_tree = result.json()["data"]["xml"] - logging.debug("XML data received from Arista server") - return ET.ElementTree(ET.fromstring(folder_tree)) - except KeyError as error: - logger.error(f"Unkown key in server response: {error}") - return None - - def get_url(self, remote_file_path: str) -> Union[str, None]: - """Get download URL for a remote file from server. - - This method retrieves the download URL for a specified remote file by making a POST request - to the server. If not authenticated, it will first authenticate before making the request. - - Args: - ------ - remote_file_path (str): Path to the remote file on server to get download URL for - - Returns: - ------ - Union[str, None]: The download URL if successful, None if request fails or URL not found in response - - Raises: - ------ - requests.exceptions.RequestException: If the request to server fails - json.JSONDecodeError: If server response is not valid JSON - requests.exceptions.Timeout: If server request times out - """ - - logging.info(f"Getting download URL for {remote_file_path}") - if self._session_id is None: - logger.debug("Not authenticated to server, start authentication process") - self.authenticate() - jsonpost = {"sessionCode": self._session_id, "filePath": remote_file_path} - result = requests.post( - self._download_server, - data=json.dumps(jsonpost), - timeout=self._timeout, - headers=self._headers, - ) - if "data" in result.json() and "url" in result.json()["data"]: - # logger.debug('URL to download file is: {}', result.json()) - logging.info("Download URL received from server") - logging.debug(f'URL to download file is: {result.json()["data"]["url"]}') - return result.json()["data"]["url"] - return None diff --git a/eos_downloader/models/data.py b/eos_downloader/models/data.py index a948d76..3897847 100644 --- a/eos_downloader/models/data.py +++ b/eos_downloader/models/data.py @@ -1,21 +1,27 @@ # coding: utf-8 -*- """This module defines data models and mappings for image types of CloudVision and EOS on Arista.com. -Classes: - ImageInfo: A Pydantic model representing image information for a specific image type. - DataMapping: A Pydantic model representing data mapping for image types of CloudVision and EOS on Arista.com. - -Constants: - RTYPE_FEATURE (ReleaseType): Represents a feature release type. - RTYPE_MAINTENANCE (ReleaseType): Represents a maintenance release type. - RTYPES (List[ReleaseType]): A list containing the feature and maintenance release types. - -Variables: - software_mapping (DataMapping): An instance of DataMapping containing the mappings for CloudVision and EOS image types. - -Methods: - DataMapping.filename(software: AristaMapping, image_type: str, version: str) -> str: - Generates a filename based on the provided software, image type, and version. +Classes +------- +ImageInfo: + A Pydantic model representing image information for a specific image type. +DataMapping: + A Pydantic model representing data mapping for image types of CloudVision and EOS on Arista.com. + +Methods +------- +DataMapping.filename(software: AristaMapping, image_type: str, version: str) -> str: + Generates a filename based on the provided software, image type, and version. + +Constants +------- +- RTYPE_FEATURE (ReleaseType): Represents a feature release type. +- RTYPE_MAINTENANCE (ReleaseType): Represents a maintenance release type. +- RTYPES (List[ReleaseType]): A list containing the feature and maintenance release types. + +Variables +------- +- software_mapping (DataMapping): An instance of DataMapping containing the mappings for CloudVision and EOS image types. """ from typing import Dict, List @@ -31,14 +37,35 @@ class ImageInfo(BaseModel): - """Image information for a specific image type.""" + """Image information for a specific image type. + + Attributes + ---------- + extension : str + The file extension for the image type. + prepend : str + The prefix to prepend to the filename. + """ extension: str prepend: str class DataMapping(BaseModel): - """Data mapping for image types of CloudVision and EOS on Arista.com.""" + """Data mapping for image types of CloudVision and EOS on Arista.com. + + Attributes + ---------- + CloudVision : Dict[str, ImageInfo] + Mapping of image types to their information for CloudVision. + EOS : Dict[str, ImageInfo] + Mapping of image types to their information for EOS. + + Methods + ------- + filename(software: AristaMapping, image_type: str, version: str) -> str + Generates a filename based on the provided software, image type, and version. + """ CloudVision: Dict[str, ImageInfo] EOS: Dict[str, ImageInfo] @@ -46,17 +73,26 @@ class DataMapping(BaseModel): def filename(self, software: AristaMapping, image_type: str, version: str) -> str: """Generates a filename based on the provided software, image type, and version. - Args: - software (str): The name of the software for which the filename is being generated. - image_type (str): The type of image for which the filename is being generated. - version (str): The version of the software or image. - - Returns: - str: The generated filename. - - Raises: - ValueError: If the software does not have a corresponding mapping. - ValueError: If no configuration is found for the given image type and no default configuration is available. + Parameters + ---------- + software : AristaMapping + The name of the software for which the filename is being generated. + image_type : str + The type of image for which the filename is being generated. + version : str + The version of the software or image. + + Returns + ------- + str + The generated filename. + + Raises + ------ + ValueError + If the software does not have a corresponding mapping. + ValueError + If no configuration is found for the given image type and no default configuration is available. """ if hasattr(self, software): diff --git a/eos_downloader/models/types.py b/eos_downloader/models/types.py index 7b902d8..f7290f7 100644 --- a/eos_downloader/models/types.py +++ b/eos_downloader/models/types.py @@ -1,12 +1,35 @@ +#!/usr/bin/python # coding: utf-8 -*- """ -This module defines various type aliases using the `Literal` and `Union` types from the `typing` module. +This module defines type aliases and literals used in the eos_downloader project. -Type Aliases: - - AristaPackage: A literal type that can be either "eos" or "cvp". - - AristaMapping: A literal type that can be either "CloudVision" or "EOS". - - AristaVersions: A union type that can be either `EosVersion` or `CvpVersion` from the `eos_downloader.models.version` module. - - ReleaseType: A literal type that can be either "M" or "F". +Attributes +---------- +AristaPackage : Literal + Literal type for Arista package types. Can be either "eos" or "cvp". +AristaMapping : Literal + Literal type for Arista mapping types. Can be either "CloudVision" or "EOS". +AristaVersions : Union + Union type for supported SemVer object types. Can be either EosVersion or CvpVersion. +ReleaseType : Literal + Literal type for release types. Can be either "M" (maintenance) or "F" (feature). + +Examples +-------- + # Example usage of AristaPackage + def get_package_type(package: AristaPackage): + if package == "eos": + return "Arista EOS package" + elif package == "cvp": + return "CloudVision Portal package" + + # Example usage of AristaVersions + def print_version(version: AristaVersions): + print(f"Version: {version}") + + # Example usage of ReleaseType + def is_feature_release(release: ReleaseType) -> bool: + return release == "F" """ from typing import Literal, Union diff --git a/eos_downloader/models/version.py b/eos_downloader/models/version.py index 6bdfd7f..4c9a133 100644 --- a/eos_downloader/models/version.py +++ b/eos_downloader/models/version.py @@ -3,27 +3,47 @@ """The module implements version management following semantic versioning principles with custom adaptations for Arista EOS and CloudVision Portal (CVP) software versioning schemes. - SemVer: Base class implementing semantic versioning with comparison and matching capabilities. - EosVersion: Specialized version handling for Arista EOS software releases. - CvpVersion: Specialized version handling for CloudVision Portal releases. - -Each class provides methods to: -- Parse version strings into structured objects -- Compare versions -- Extract branch information -- Match version patterns -- Convert versions to string representations - - Basic SemVer usage: +Classes +------- +SemVer: + Base class implementing semantic versioning with comparison and matching capabilities. +EosVersion: + Specialized version handling for Arista EOS software releases. +CvpVersion: + Specialized version handling for CloudVision Portal releases. + +Attributes +---------- +major : int + Major version number. +minor : int + Minor version number. +patch : int + Patch version number. +rtype : Optional[str] + Release type (e.g., 'M' for maintenance, 'F' for feature). +other : Any + Additional version information. +regex_version : ClassVar[Pattern[str]] + Regular expression to extract version information. +regex_branch : ClassVar[Pattern[str]] + Regular expression to extract branch information. +description : str + A basic description of this class. + + +Examples +-------- + # Basic SemVer usage: >>> version = SemVer(major=4, minor=23, patch=3) '4.23.3' - EOS version handling: + # EOS version handling: >>> eos = EosVersion.from_str('4.23.3M') >>> eos.branch '4.23' - CVP version handling: + # CVP version handling: >>> cvp = CvpVersion.from_str('2024.1.0') >>> str(cvp) @@ -31,8 +51,9 @@ comprehensive comparison operations (==, !=, <, <=, >, >=) between versions. Note: - - EOS versions follow the format: ..[M|F] - - CVP versions follow the format: .. +-------- +- EOS versions follow the format: ..[M|F] +- CVP versions follow the format: .. """ from __future__ import annotations @@ -56,69 +77,43 @@ class SemVer(BaseModel): This class provides methods to parse, compare, and manipulate semantic versions. It supports standard semantic versioning with optional release type and additional version information. - Examples: - >>> version = SemVer(major=4, minor=23, patch=3, rtype="M") - >>> str(version) - '4.23.3M' - - >>> version2 = SemVer.from_str('4.24.1F') - >>> version2.branch - '4.24' - - >>> version < version2 - True - - >>> version.match("<=4.24.0") - True - - >>> version.is_in_branch("4.23") - True - - Attributes: - major (int): Major version number. - minor (int): Minor version number. - patch (int): Patch version number. - rtype (Optional[str]): Release type (e.g., 'M' for maintenance, 'F' for feature). - other (Any): Additional version information. - regex_version (ClassVar[Pattern[str]]): Regular expression to extract version information. - regex_branch (ClassVar[Pattern[str]]): Regular expression to extract branch information. - description (str): A basic description of this class. - - Methods: - from_str(cls, semver: str) -> SemVer: - Create a SemVer instance from a version string. - - branch(self) -> str: - Extract the branch of the version. - - __str__(self) -> str: - Return a standard string representation of the version. - - _compare(self, other: SemVer) -> float: - Compare this SemVer instance with another. - - __eq__(self, other): - Implement equality comparison (==). - - __ne__(self, other): - Implement inequality comparison (!=). - - __lt__(self, other): - Implement less than comparison (<). - - __le__(self, other): - Implement less than or equal comparison (<=). - - __gt__(self, other): - Implement greater than comparison (>). - - __ge__(self, other): - Implement greater than or equal comparison (>=). - - match(self, match_expr: str) -> bool: - - is_in_branch(self, branch_str: str) -> bool: - Check if the current version is part of a branch version. + Examples + -------- + >>> version = SemVer(major=4, minor=23, patch=3, rtype="M") + >>> str(version) + '4.23.3M' + + >>> version2 = SemVer.from_str('4.24.1F') + >>> version2.branch + '4.24' + + >>> version < version2 + True + + >>> version.match("<=4.24.0") + True + + >>> version.is_in_branch("4.23") + True + + Attributes + ---------- + major : int + Major version number. + minor : int + Minor version number. + patch : int + Patch version number. + rtype : Optional[str] + Release type (e.g., 'M' for maintenance, 'F' for feature). + other : Any + Additional version information. + regex_version : ClassVar[Pattern[str]] + Regular expression to extract version information. + regex_branch : ClassVar[Pattern[str]] + Regular expression to extract branch information. + description : str + A basic description of this class. """ major: int = 0 @@ -143,19 +138,24 @@ def from_str(cls, semver: str) -> SemVer: This method parses a semantic version string or branch name into a SemVer object. It supports both standard semver format (x.y.z) and branch format. - Args: - semver (str): The version string to parse. Can be either a semantic version - string (e.g., "1.2.3") or a branch format. - - Returns: - SemVer: A SemVer object representing the parsed version. - Returns an empty SemVer object if parsing fails. - - Examples: - >>> SemVer.from_str("1.2.3") - SemVer(major=1, minor=2, patch=3) - >>> SemVer.from_str("branch-1.2.3") - SemVer(major=1, minor=2, patch=3) + Parameters + ---------- + semver : str + The version string to parse. Can be either a semantic version + string (e.g., "1.2.3") or a branch format. + + Returns + ------- + SemVer + A SemVer object representing the parsed version. + Returns an empty SemVer object if parsing fails. + + Examples + -------- + >>> SemVer.from_str("1.2.3") + SemVer(major=1, minor=2, patch=3) + >>> SemVer.from_str("branch-1.2.3") + SemVer(major=1, minor=2, patch=3) """ logging.debug(f"Creating SemVer object from string: {semver}") @@ -178,42 +178,52 @@ def from_str(cls, semver: str) -> SemVer: @property def branch(self) -> str: """ - Extract branch of version + Extract branch of version. - Returns: - str: branch from version + Returns + ------- + str + Branch from version. """ return f"{self.major}.{self.minor}" def __str__(self) -> str: """ - Standard str representation + Standard str representation. - Return string for EOS version like 4.23.3M + Return string for EOS version like 4.23.3M. - Returns: - str: A standard EOS version string representing .. + Returns + ------- + str + A standard EOS version string representing ... """ return f"{self.major}.{self.minor}.{self.patch}{self.other if self.other is not None else ''}{self.rtype if self.rtype is not None else ''}" def _compare(self, other: SemVer) -> float: """ - An internal comparison function to compare 2 EosVersion objects + An internal comparison function to compare 2 EosVersion objects. - Do a deep comparison from Major to Release Type - The return value is + Do a deep comparison from Major to Release Type. + The return value is: - negative if ver1 < ver2, - - zero if ver1 == ver2 - - strictly positive if ver1 > ver2 - - Args: - other (EosVersion): An EosVersion to compare with this object - - Raises: - ValueError: Raise ValueError if input is incorrect type - - Returns: - float: -1 if ver1 < ver2, 0 if ver1 == ver2, 1 if ver1 > ver2 + - zero if ver1 == ver2, + - strictly positive if ver1 > ver2. + + Parameters + ---------- + other : SemVer + An EosVersion to compare with this object. + + Raises + ------ + ValueError + Raise ValueError if input is incorrect type. + + Returns + ------- + float + -1 if ver1 < ver2, 0 if ver1 == ver2, 1 if ver1 > ver2. """ if not isinstance(other, SemVer): raise ValueError( @@ -283,26 +293,33 @@ def match(self, match_expr: str) -> bool: """ Compare self to match a match expression. - Example: - >>> eos_version.match("<=4.23.3M") - True - >>> eos_version.match("==4.23.3M") - False - - Args: - match_expr (str): optional operator and version; valid operators are + Parameters + ---------- + match_expr : str + Optional operator and version; valid operators are: ``<`` smaller than ``>`` greater than - ``>=`` greator or equal than + ``>=`` greater or equal than ``<=`` smaller or equal than ``==`` equal - ``!=`` not equal + ``!=`` not equal. + + Raises + ------ + ValueError + If input has no match_expr nor match_ver. - Raises: - ValueError: If input has no match_expr nor match_ver + Returns + ------- + bool + True if the expression matches the version, otherwise False. - Returns: - bool: True if the expression matches the version, otherwise False + Examples + -------- + >>> eos_version.match("<=4.23.3M") + True + >>> eos_version.match("==4.23.3M") + False """ prefix = match_expr[:2] if prefix in (">=", "<=", "==", "!="): @@ -335,15 +352,19 @@ def match(self, match_expr: str) -> bool: def is_in_branch(self, branch_str: str) -> bool: """ - Check if current version is part of a branch version + Check if current version is part of a branch version. - Comparison is done across MAJOR and MINOR + Comparison is done across MAJOR and MINOR. - Args: - branch_str (str): a string for EOS branch. It supports following formats 4.23 or 4.23.0 + Parameters + ---------- + branch_str : str + A string for EOS branch. It supports following formats 4.23 or 4.23.0. - Returns: - bool: True if current version is in provided branch, otherwise False + Returns + ------- + bool + True if current version is in provided branch, otherwise False. """ logging.info(f"Checking if {self} is in branch {branch_str}") try: @@ -361,23 +382,33 @@ class EosVersion(SemVer): Since EOS is not using strictly semver approach, this class mimics some functions from the semver library for Arista EOS versions. It is based on Pydantic and provides helpers for comparison. - Example: - >>> version = EosVersion(major=4, minor=21, patch=1, rtype="M") - >>> print(version) - EosVersion(major=4, minor=21, patch=1, rtype='M', other=None) - >>> version = EosVersion.from_str('4.32.1F') - >>> print(version) - EosVersion(major=4, minor=32, patch=1, rtype='F', other=None) - - Attributes: - major (int): Major version number, default is 4. - minor (int): Minor version number, default is 0. - patch (int): Patch version number, default is 0. - rtype (Optional[str]): Release type, default is "F". - other (Any): Any other version information. - regex_version (ClassVar[Pattern[str]]): Regular expression to extract version information. - regex_branch (ClassVar[Pattern[str]]): Regular expression to extract branch information. - description (str): A basic description of this class, default is "A Generic SemVer implementation". + Examples + -------- + >>> version = EosVersion(major=4, minor=21, patch=1, rtype="M") + >>> print(version) + EosVersion(major=4, minor=21, patch=1, rtype='M', other=None) + >>> version = EosVersion.from_str('4.32.1F') + >>> print(version) + EosVersion(major=4, minor=32, patch=1, rtype='F', other=None) + + Attributes + ---------- + major : int + Major version number, default is 4. + minor : int + Minor version number, default is 0. + patch : int + Patch version number, default is 0. + rtype : Optional[str] + Release type, default is "F". + other : Any + Any other version information. + regex_version : ClassVar[Pattern[str]] + Regular expression to extract version information. + regex_branch : ClassVar[Pattern[str]] + Regular expression to extract branch information. + description : str + A basic description of this class, default is "A Generic SemVer implementation". """ major: int = 4 @@ -405,20 +436,30 @@ class CvpVersion(SemVer): - minor version represents feature releases - patch version represents bug fixes - Attributes: - major (int): The year component of the version (e.g. 2024) - minor (int): The minor version number - patch (int): The patch version number - rtype (Optional[str]): Release type if any - other (Any): Additional version information if any - regex_version (Pattern[str]): Regular expression to parse version strings - regex_branch (Pattern[str]): Regular expression to parse branch version strings - description (str): Brief description of the class purpose - - Example: - >>> version = CvpVersion(2024, 1, 0) - >>> str(version) - '2024.1.0' + Examples + -------- + >>> version = CvpVersion(2024, 1, 0) + >>> str(version) + '2024.1.0' + + Attributes + ---------- + major : int + The year component of the version (e.g. 2024). + minor : int + The minor version number. + patch : int + The patch version number. + rtype : Optional[str] + Release type if any. + other : Any + Additional version information if any. + regex_version : ClassVar[Pattern[str]] + Regular expression to parse version strings. + regex_branch : ClassVar[Pattern[str]] + Regular expression to parse branch version strings. + description : str + Brief description of the class purpose. """ major: int = 2024 diff --git a/mkdocs.yml b/mkdocs.yml new file mode 100644 index 0000000..cc44f8f --- /dev/null +++ b/mkdocs.yml @@ -0,0 +1,177 @@ +# Project information +site_name: Arista EOS Downloader +site_author: Thomas Grimonet +site_description: A downloader CLI to download arista software images. +# copyright: Copyright © 2019 - 2024 Arista Networks + +# Repository +repo_name: eos_downloader on Github +repo_url: https://github.com/titom73/eos-downloader + +# Configuration +use_directory_urls: true +theme: + name: material + features: + - navigation.instant + - navigation.top + - content.tabs.link + - content.code.copy + # - toc.integrate + - toc.follow + - navigation.indexes + - content.tabs.link + highlightjs: true + hljs_languages: + - yaml + - python + - shell + icon: + repo: fontawesome/brands/github + logo: fontawesome/solid/network-wired + favicon: imgs/favicon.ico + font: + code: Fira Mono + language: en + include_search_page: false + search_index_only: true + palette: + # Light mode + - media: "(prefers-color-scheme: light)" + scheme: default + primary: white + accent: blue + toggle: + icon: material/weather-night + name: Switch to dark mode + # Dark mode + - media: "(prefers-color-scheme: dark)" + scheme: slate + primary: black + accent: blue + toggle: + icon: material/weather-sunny + name: Switch to light mode + # custom_dir: docs/overrides + +extra: + social: + - icon: fontawesome/brands/github + link: https://github.com/titom73/eos-downloader + - icon: fontawesome/brands/docker + link: https://github.com/titom73/eos-downloader/pkgs/container/eos-downloader + - icon: fontawesome/brands/python + link: https://pypi.org/project/eos-downloader/ + version: + provider: mike + default: + - stable + +extra_css: + - stylesheets/extra.material.css + +extra_javascript: + - https://cdnjs.cloudflare.com/ajax/libs/tablesort/5.2.1/tablesort.min.js + - https://cdnjs.cloudflare.com/ajax/libs/highlight.js/10.7.2/highlight.min.js + +watch: + - docs + - eos_downloader + +plugins: + - gh-admonitions + - mkdocstrings: + handlers: + python: + paths: [eos_downloader] + import: + - https://docs.python.org/3/objects.inv + - https://mkdocstrings.github.io/objects.inv + - https://mkdocstrings.github.io/griffe/objects.inv + options: + docstring_style: numpy + docstring_options: + ignore_init_summary: true + docstring_section_style: table + show_docstring_other_parameters: true + heading_level: 2 + inherited_members: false + merge_init_into_class: true + separate_signature: true + show_root_heading: true + show_root_full_path: false + show_signature_annotations: true + filters: ["!^_[^_]"] + + - search: + lang: en + - git-revision-date-localized: + type: date + - mike: + - glightbox: + background: none + shadow: true + touchNavigation: true + loop: false + effect: fade + slide_effect: slide + width: 90vw + +markdown_extensions: + - admonition + - attr_list + - codehilite: + guess_lang: true + - pymdownx.arithmatex + - pymdownx.betterem: + smart_enable: all + - pymdownx.caret + - pymdownx.critic + - pymdownx.details + - pymdownx.emoji: + emoji_index: !!python/name:material.extensions.emoji.twemoji + emoji_generator: !!python/name:material.extensions.emoji.to_svg + - pymdownx.highlight + - pymdownx.inlinehilite + - pymdownx.magiclink + - pymdownx.mark + - pymdownx.smartsymbols + # - pymdownx.snippets: + # base_path: + # - docs/snippets + # - examples + - pymdownx.superfences: + custom_fences: + - name: mermaid + class: mermaid + format: !!python/name:pymdownx.superfences.fence_code_format + - pymdownx.tabbed: + alternate_style: true + - pymdownx.tasklist: + custom_checkbox: true + - pymdownx.tilde + - smarty + - toc: + separator: "-" + # permalink: "#" + permalink: true + baselevel: 2 + +# TOC +docs_dir: docs/ +nav: + - Home: README.md + - Usage: + - Get EOS package: usage/eos.md + - Version information: usage/info.md + - Code documentation: + - Models: + - Version: api/models/version.md + - Data: api/models/data.md + - Custom Types: api/models/custom_types.md + - Logics: + - Arista Server: api/logics/arista_server.md + - Arista XML API: api/logics/arista_xml_server.md + - Download Management: api/logics/download.md + - Helpers: api/helpers.md + diff --git a/pyproject.toml b/pyproject.toml index 589a364..2c689f9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -84,6 +84,24 @@ dev = [ "bumpver>=2023.1126", ] +doc = [ + "fontawesome_markdown>=0.2.6", + "griffe >=1.2.0", + "mike==2.1.3", + "mkdocs>=1.6.1", + "mkdocs-autorefs>=1.2.0", + "mkdocs-bootswatch>=1.1", + "mkdocs-git-revision-date-localized-plugin>=1.2.8", + "mkdocs-git-revision-date-plugin>=0.3.2", + "mkdocs-glightbox>=0.4.0", + "mkdocs-material-extensions>=1.3.1", + "mkdocs-material>=9.5.34", + "mkdocstrings[python]>=0.26.0", + "mkdocstrings-python>=1.11.0", + "black>=24.10.0", + "mkdocs-github-admonitions-plugin" +] + [project.urls] Homepage = "https://www.github.com/titom73/eos-downloader" "Bug Tracker" = "https://www.github.com/titom73/eos-downloader/issues" diff --git a/tests/unit/logics/test_arista_server.py b/tests/unit/logics/test_arista_server.py index 917b2c9..8ec928c 100644 --- a/tests/unit/logics/test_arista_server.py +++ b/tests/unit/logics/test_arista_server.py @@ -7,10 +7,10 @@ import os import pytest import logging -from eos_downloader.logics.arista_server import AristaXmlQuerier +from eos_downloader.logics.arista_xml_server import AristaXmlQuerier from eos_downloader.models.version import EosVersion, CvpVersion from unittest.mock import patch -from eos_downloader.logics.arista_server import ( +from eos_downloader.logics.arista_xml_server import ( AristaXmlBase, AristaXmlObject, EosXmlObject, diff --git a/tests/unit/logics/test_download.py b/tests/unit/logics/test_download.py index f3e3a19..2b93c71 100644 --- a/tests/unit/logics/test_download.py +++ b/tests/unit/logics/test_download.py @@ -2,7 +2,7 @@ import pytest from unittest.mock import Mock, patch, mock_open from eos_downloader.logics.download import SoftManager -from eos_downloader.logics.arista_server import EosXmlObject +from eos_downloader.logics.arista_xml_server import EosXmlObject @pytest.fixture diff --git a/tests/unit/logics/test_server.py b/tests/unit/logics/test_server.py index 5f027a9..8404ecc 100644 --- a/tests/unit/logics/test_server.py +++ b/tests/unit/logics/test_server.py @@ -1,7 +1,7 @@ import pytest import requests from unittest.mock import patch, Mock -from eos_downloader.logics.server import AristaServer +from eos_downloader.logics.arista_server import AristaServer import eos_downloader.exceptions From 6060fcbfd6aaa1dea9eb22670d9c90920fe2d125 Mon Sep 17 00:00:00 2001 From: Thomas Grimonet Date: Sat, 14 Dec 2024 16:55:03 +0100 Subject: [PATCH 2/3] ci: Add doc autobuild --- .github/workflows/main_doc.yml | 37 ++++++++++++++++++++++++++++++++++ .github/workflows/release.yml | 22 ++++++++++++++++++++ 2 files changed, 59 insertions(+) create mode 100644 .github/workflows/main_doc.yml diff --git a/.github/workflows/main_doc.yml b/.github/workflows/main_doc.yml new file mode 100644 index 0000000..04016d1 --- /dev/null +++ b/.github/workflows/main_doc.yml @@ -0,0 +1,37 @@ +--- +# This is deploying the latest commits on main to main documentation +name: Mkdocs +on: + push: + branches: + - main + paths: + # Run only if any of the following paths are changed when pushing to main + - "docs/**" + - "mkdocs.yml" + - "eos_downloader/**" + workflow_dispatch: + +jobs: + 'build_latest_doc': + name: 'Update Public main documentation' + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 0 + + - name: 'Setup Python 3 on runner' + uses: actions/setup-python@v5 + with: + python-version: '3.x' + + - name: Setup Git config + run: | + git config --global user.name 'github-actions[bot]' + git config --global user.email 'github-actions[bot]@users.noreply.github.com' + + - name: 'Build mkdocs content and deploy to gh-pages to main' + run: | + pip install .[doc] + mike deploy --push main diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index ab5cee9..d85df38 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -114,3 +114,25 @@ jobs: tags: ${{ steps.meta.outputs.tags }} labels: ${{ steps.meta.outputs.labels }} + + release-doc: + name: "Publish documentation for release ${{github.ref_name}}" + runs-on: ubuntu-latest + needs: [docker-in-docker] + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 0 + - name: 'Setup Python 3 on runner' + uses: actions/setup-python@v5 + with: + python-version: '3.x' + - name: Setup Git config + run: | + git config --global user.name 'github-actions[bot]' + git config --global user.email 'github-actions[bot]@users.noreply.github.com' + + - name: 'Build mkdocs content to site folder' + run: | + pip install .[doc] + mike deploy --update-alias --push ${{github.ref_name}} stable \ No newline at end of file From ab72d105c7aedb7a009702c47eab1fc972e98da6 Mon Sep 17 00:00:00 2001 From: Thomas Grimonet Date: Sat, 14 Dec 2024 17:15:06 +0100 Subject: [PATCH 3/3] test(pytest): Fix module rename --- tests/unit/logics/test_arista_server.py | 398 +++++--------------- tests/unit/logics/test_arista_xml_server.py | 351 +++++++++++++++++ tests/unit/logics/test_server.py | 115 ------ 3 files changed, 435 insertions(+), 429 deletions(-) create mode 100644 tests/unit/logics/test_arista_xml_server.py delete mode 100644 tests/unit/logics/test_server.py diff --git a/tests/unit/logics/test_arista_server.py b/tests/unit/logics/test_arista_server.py index 8ec928c..8404ecc 100644 --- a/tests/unit/logics/test_arista_server.py +++ b/tests/unit/logics/test_arista_server.py @@ -1,345 +1,115 @@ -#!/usr/bin/python -# coding: utf-8 -*- - -from __future__ import absolute_import, division, print_function - -import sys -import os import pytest -import logging -from eos_downloader.logics.arista_xml_server import AristaXmlQuerier -from eos_downloader.models.version import EosVersion, CvpVersion -from unittest.mock import patch -from eos_downloader.logics.arista_xml_server import ( - AristaXmlBase, - AristaXmlObject, - EosXmlObject, -) -import xml.etree.ElementTree as ET - +import requests +from unittest.mock import patch, Mock +from eos_downloader.logics.arista_server import AristaServer -# Fixtures -@pytest.fixture -def xml_path() -> str: - """Fixture to provide path to test XML file""" - return os.path.join(os.path.dirname(os.path.dirname(__file__)), "../data.xml") +import eos_downloader.exceptions +from tests.lib.fixtures import xml_path, xml_data @pytest.fixture -def xml_data(): - xml_file = os.path.join(os.path.dirname(__file__), "../data.xml") - tree = ET.parse(xml_file) - root = tree.getroot() - return root - - -# ------------------- # -# Tests AristaXmlBase -# ------------------- # -def test_arista_xml_base_initialization(xml_path): - arista_xml_base = AristaXmlBase(xml_path=str(xml_path)) - assert arista_xml_base.xml_data.getroot().tag == "cvpFolderList", ( - f"Root tag should be 'cvpFolderList' but got" - ) - -# ---------------------- # -# Tests AristaXmlQuerier -# ---------------------- # - -# ---------------------- # -# Tests AristaXmlQuerier available_public_versions for eos - -def test_AristaXmlQuerier_available_public_versions_eos(xml_path): - xml_querier = AristaXmlQuerier(xml_path=xml_path) - versions = xml_querier.available_public_versions(package="eos") - assert len(versions) == 309, "Incorrect number of versions" - assert versions[0] == EosVersion().from_str("4.33.0F"), "First version should be 4.33.0F - got {versions[0]}" - - -def test_AristaXmlQuerier_available_public_versions_eos_f_release(xml_path): - xml_querier = AristaXmlQuerier(xml_path=xml_path) - versions = xml_querier.available_public_versions(package="eos", rtype="F") - assert len(versions) == 95, "Incorrect number of versions: got {len(versions)} expected 207" - assert versions[0] == EosVersion().from_str( - "4.33.0F" - ), "First version should be 4.33.0F - got {len(versions)}" - - -def test_AristaXmlQuerier_available_public_versions_eos_m_release(xml_path): - xml_querier = AristaXmlQuerier(xml_path=xml_path) - versions = xml_querier.available_public_versions(package="eos", rtype="M") - assert ( - len(versions) == 207 - ), "Incorrect number of versions: got {len(versions)} expected 207" - assert versions[0] == EosVersion().from_str( - "4.32.3M" - ), "First version should be 4.32.3M - got {versions[0]}" - -def test_AristaXmlQuerier_available_public_versions_eos_branch_4_29(xml_path): - xml_querier = AristaXmlQuerier(xml_path=xml_path) - versions = xml_querier.available_public_versions(package="eos", branch="4.29") - assert len(versions) == 34, "Incorrect number of versions" - for version in versions: - # logging.debug(f"Checking version {version}") - assert version.is_in_branch("4.29"), f"Version {version} is not in branch 4.29" - assert versions[0] == EosVersion().from_str("4.29.10M"), "First version should be 4.29.10M - got {versions[0]}" - - -def test_AristaXmlQuerier_available_public_versions_eos_f_release_branch_4_29(xml_path): - xml_querier = AristaXmlQuerier(xml_path=xml_path) - versions = xml_querier.available_public_versions(package="eos", rtype="F", branch="4.29") - assert len(versions) == 6, "Incorrect number of versions - expected 6" - for version in versions: - # logging.debug(f"Checking version {version}") - assert version.is_in_branch("4.29"), f"Version {version} is not in branch 4.29" - assert versions[0] == EosVersion().from_str( - "4.29.2F" - ), "First version should be 4.29.2F - got {versions[0]}" - - -def test_AristaXmlQuerier_available_public_versions_eos_m_release_branch_4_29(xml_path): - xml_querier = AristaXmlQuerier(xml_path=xml_path) - versions = xml_querier.available_public_versions(package="eos", rtype="M", branch="4.29") - assert len(versions) == 28, "Incorrect number of versions - expected 28" - for version in versions: - # logging.debug(f"Checking version {version}") - assert version.is_in_branch("4.29"), f"Version {version} is not in branch 4.29" - assert versions[0] == EosVersion().from_str( - "4.29.10M" - ), "First version should be 4.29.10M - got {versions[0]}" - - -# ---------------------- # -# Tests AristaXmlQuerier available_public_versions for cvp - - -def test_AristaXmlQuerier_available_public_versions_cvp(xml_path): - xml_querier = AristaXmlQuerier(xml_path=xml_path) - versions = xml_querier.available_public_versions(package="cvp") - assert ( - len(versions) == 12 - ), "Incorrect number of versions: got {len(versions)} expected 12" - assert versions[0] == CvpVersion().from_str( - "2024.3.0" - ), "First version should be 2024.3.0 - got {versions[0]}" - +def server(): + return AristaServer(token="testtoken") -# ---------------------- # -# Tests AristaXmlQuerier branches for eos - - -def test_AristaXmlQuerier_branch_eos(xml_path): - xml_querier = AristaXmlQuerier(xml_path=xml_path) - versions = xml_querier.branches(package="eos") - assert len(versions) == 14, "Incorrect number of branches, got {len(versions)} expected 14" - assert EosVersion().from_str("4.33.0F").branch in versions, "4.33 should be in branches {versions}" - - -def test_AristaXmlQuerier_branch_eos_latest(xml_path): - xml_querier = AristaXmlQuerier(xml_path=xml_path) - versions = xml_querier.branches(package="eos", latest=True) - assert ( - len(versions) == 1 - ), "Incorrect number of branches, got {len(versions)} expected 1" - assert ( - EosVersion().from_str("4.33.0F").branch in versions - ), "4.33 should be in branches {versions}" - - -# ---------------------- # -# Tests AristaXmlQuerier branches for cvp - - -def test_AristaXmlQuerier_branch_cvp(xml_path): - xml_querier = AristaXmlQuerier(xml_path=xml_path) - versions = xml_querier.branches(package="cvp") - assert ( - len(versions) == 5 - ), "Incorrect number of branches, got {len(versions)} expected 5" - assert ( - CvpVersion().from_str("2024.3.0").branch in versions - ), "2024.3 should be in branches {versions}" - - -def test_AristaXmlQuerier_branch_cvp_latest(xml_path): - xml_querier = AristaXmlQuerier(xml_path=xml_path) - versions = xml_querier.branches(package="cvp", latest=True) - assert ( - len(versions) == 1 - ), "Incorrect number of branches, got {len(versions)} expected 1" - assert ( - CvpVersion().from_str("2024.3.0").branch in versions - ), "2024.3 should be in branches {versions}" - - -# ---------------------- # -# Tests AristaXmlQuerier latest for eos - - -def test_AristaXmlQuerier_latest_eos(xml_path): - xml_querier = AristaXmlQuerier(xml_path=xml_path) - versions = xml_querier.latest(package="eos") - assert ( - EosVersion().from_str("4.33.0F") == versions - ), "4.33.0F should be the latest, got {versions}" +def test_authenticate_success(server): + with patch('requests.post') as mock_post: + mock_response = Mock() + mock_response.json.return_value = { + "status": {"message": "Success"}, + "data": {"session_code": "testsessioncode"} + } + mock_post.return_value = mock_response -def test_AristaXmlQuerier_latest_eos_f_release(xml_path): - xml_querier = AristaXmlQuerier(xml_path=xml_path) - versions = xml_querier.latest(package="eos", rtype="F") - assert ( - EosVersion().from_str("4.33.0F") == versions - ), "4.33.0F should be the latest, got {versions}" + assert server.authenticate() is True + assert server._session_id is not None -def test_AristaXmlQuerier_latest_eos_m_release(xml_path): - xml_querier = AristaXmlQuerier(xml_path=xml_path) - versions = xml_querier.latest(package="eos", rtype="M") - assert ( - EosVersion().from_str("4.32.3M") == versions - ), "4.32.3M should be the latest, got {versions}" +def test_authenticate_invalid_token(server): + with patch('requests.post') as mock_post: + mock_response = Mock() + mock_response.json.return_value = { + "status": {"message": "Invalid access token"} + } + mock_post.return_value = mock_response + with pytest.raises(eos_downloader.exceptions.AuthenticationError): + server.authenticate() -def test_AristaXmlQuerier_latest_eos_f_release_branch_4_29(xml_path): - xml_querier = AristaXmlQuerier(xml_path=xml_path) - versions = xml_querier.latest(package="eos", rtype="F", branch="4.29") - assert ( - EosVersion().from_str("4.29.2F") == versions - ), "4.29.2F should be the latest, got {versions}" +def test_authenticate_expired_token(server): + with patch('requests.post') as mock_post: + mock_response = Mock() + mock_response.json.return_value = { + "status": {"message": "Access token expired"} + } + mock_post.return_value = mock_response -def test_AristaXmlQuerier_latest_eos_m_release_branch_4_29(xml_path): - xml_querier = AristaXmlQuerier(xml_path=xml_path) - versions = xml_querier.latest(package="eos", rtype="M", branch="4.29") - assert ( - EosVersion().from_str("4.29.10M") == versions - ), "4.29.10M should be the latest, got {versions}" + with pytest.raises(eos_downloader.exceptions.AuthenticationError): + server.authenticate() -# ---------------------- # -# Tests AristaXmlQuerier latest for cvp +def test_authenticate_key_error(server): + with patch('requests.post') as mock_post: + mock_response = Mock() + mock_response.json.return_value = { + "status": {"message": "Success"} + } + mock_post.return_value = mock_response + assert server.authenticate() is False + assert server._session_id is None -def test_AristaXmlQuerier_latest_cvp(xml_path): - xml_querier = AristaXmlQuerier(xml_path=xml_path) - versions = xml_querier.latest(package="cvp") - assert ( - CvpVersion().from_str("2024.3.0") == versions - ), "2024.3.0 should be the latest, got {versions}" +def test_get_xml_data_success(server, xml_path): + with patch('requests.post') as mock_post: + with open(xml_path, 'r') as file: + xml_content = file.read() -def test_AristaXmlQuerier_latest_cvp(xml_path): - xml_querier = AristaXmlQuerier(xml_path=xml_path) - versions = xml_querier.latest(package="cvp", branch="2024.2") - assert ( - CvpVersion().from_str("2024.2.1") == versions - ), "2024.2.1 should be the latest, got {versions}" + mock_response = Mock() + mock_response.json.return_value = { + "status": {"message": "Success"}, + "data": {"xml": xml_content}, + } + mock_post.return_value = mock_response + xml_data = server.get_xml_data() + assert xml_data is not None + assert ( + xml_data.getroot().tag == "cvpFolderList" + ) # Assuming the root tag in data.xml is 'cvpFolderList' -# ---------------------- # -# Tests AristaXmlObject -# ---------------------- # -def test_arista_xml_object_initialization(xml_path): - arista_xml_object = AristaXmlObject(searched_version="4.29.2F", image_type="image", xml_path=xml_path) - assert arista_xml_object.search_version == "4.29.2F", "Incorrect search version" - assert arista_xml_object.image_type == "image", "Incorrect image type" +def test_get_xml_data_key_error(server): + with patch('requests.post') as mock_post: + mock_response = Mock() + mock_response.json.return_value = {} + mock_post.return_value = mock_response -def test_arista_xml_object_filename_for_ceos(xml_path): - arista_xml_object = EosXmlObject( - searched_version="4.29.2F", - image_type="cEOS", - xml_path=xml_path, - ) - filename = arista_xml_object.filename - assert filename == "cEOS-lab-4.29.2F.tar.xz", f"Incorrect filename, got {filename}" + with pytest.raises(KeyError): + server.get_xml_data() -def test_arista_xml_object_hashfile(xml_path): - arista_xml_object = EosXmlObject( - searched_version="4.29.2F", - image_type="cEOS", - xml_path=xml_path, - ) - hashfile = arista_xml_object.hash_filename() - assert ( - hashfile == "cEOS-lab-4.29.2F.tar.xz.sha512sum" - ), f"Incorrect hashfile, got {hashfile}" - hashfile = arista_xml_object.hash_filename() - assert ( - hashfile == "cEOS-lab-4.29.2F.tar.xz.sha512sum" - ), f"Incorrect hashfile, got {hashfile}" -def test_arista_xml_object_path_from_xml(xml_path): - arista_xml_object = EosXmlObject( - searched_version="4.29.2F", - image_type="cEOS", - xml_path=xml_path, - ) - path = arista_xml_object.path_from_xml(search_file="EOS-4.29.2F.swi") - assert ( - path - == "/support/download/EOS-USA/Active Releases/4.29/EOS-4.29.2F/EOS-4.29.2F.swi" - ), f"Incorrect path, got {path}" +def test_get_url_success(server): + with patch('requests.post') as mock_post: + mock_response = Mock() + mock_response.json.return_value = { + "status": {"message": "Success"}, + "data": {"url": "http://example.com/download"}, + } + mock_post.return_value = mock_response -def test_arista_xml_object_url(xml_path): - with patch('eos_downloader.logics.arista_server.AristaXmlObject._url') as mock_url: - mock_url.return_value = "https://testserver.com/path/to/EOS-4.29.2F.swi" - arista_xml_object = EosXmlObject( - searched_version="4.29.2F", - image_type="cEOS", - xml_path=xml_path, - ) - url = arista_xml_object._url(xml_path="/path/to/EOS-4.29.2F.swi") - assert url == "https://testserver.com/path/to/EOS-4.29.2F.swi", f"Incorrect URL, got {url}" + url = server.get_url("remote/file/path") + assert url == "http://example.com/download" -def test_arista_xml_object_urls(xml_path): - with patch('eos_downloader.logics.arista_server.AristaXmlObject._url') as mock_url: - mock_url.side_effect = [ - "https://arista.com/path/to/EOS-4.29.2F.swi", - "https://arista.com/path/to/EOS-4.29.2F.swi.sha512sum" - ] - arista_xml_object = EosXmlObject( - searched_version="4.29.2F", - image_type="default", - xml_path=xml_path, - ) - urls = arista_xml_object.urls - logging.warning(f"URLs are: {urls}") - expected_urls = { - "image": "https://arista.com/path/to/EOS-4.29.2F.swi", - "sha512sum": "https://arista.com/path/to/EOS-4.29.2F.swi.sha512sum" - } - assert urls == expected_urls, f"Incorrect URLs, got {urls}" -def test_arista_xml_object_urls_with_invalid_hash(xml_path): - with patch('eos_downloader.logics.arista_server.AristaXmlObject._url') as mock_url: - mock_url.side_effect = [ - "https://arista.com/path/to/EOS-4.29.2F.swi", - "https://arista.com/path/to/EOS-4.29.2F.swi.sha512sum" - ] - arista_xml_object = EosXmlObject( - searched_version="4.29.2F", - image_type="default", - xml_path=xml_path, - ) - urls = arista_xml_object.urls - expected_urls = { - "image": "https://arista.com/path/to/EOS-4.29.2F.swi", - "sha512sum": "https://arista.com/path/to/EOS-4.29.2F.swi.sha512sum" +def test_get_url_no_data(server): + with patch('requests.post') as mock_post: + mock_response = Mock() + mock_response.json.return_value = { + "status": {"message": "Success"}, } - assert urls == expected_urls, f"Incorrect URLs, got {urls}" + mock_post.return_value = mock_response -def test_arista_xml_object_urls_with_missing_files(xml_path): - with patch('eos_downloader.logics.arista_server.AristaXmlObject._url') as mock_url: - mock_url.side_effect = [None, None, None] - arista_xml_object = EosXmlObject( - searched_version="4.29.2F", - image_type="default", - xml_path=xml_path, - ) - urls = arista_xml_object.urls - expected_urls = { - "image": None, - "sha512sum": None - } - assert urls == expected_urls, f"Incorrect URLs, got {urls}" + url = server.get_url("remote/file/path") + assert url is None diff --git a/tests/unit/logics/test_arista_xml_server.py b/tests/unit/logics/test_arista_xml_server.py new file mode 100644 index 0000000..2ffae64 --- /dev/null +++ b/tests/unit/logics/test_arista_xml_server.py @@ -0,0 +1,351 @@ +#!/usr/bin/python +# coding: utf-8 -*- + +from __future__ import absolute_import, division, print_function + +import sys +import os +import pytest +import logging +from eos_downloader.models.version import EosVersion, CvpVersion +from unittest.mock import patch +from eos_downloader.logics.arista_xml_server import ( + AristaXmlBase, + AristaXmlObject, + EosXmlObject, + AristaXmlQuerier +) +import xml.etree.ElementTree as ET + + +# Fixtures +@pytest.fixture +def xml_path() -> str: + """Fixture to provide path to test XML file""" + return os.path.join(os.path.dirname(os.path.dirname(__file__)), "../data.xml") + + +@pytest.fixture +def xml_data(): + xml_file = os.path.join(os.path.dirname(__file__), "../data.xml") + tree = ET.parse(xml_file) + root = tree.getroot() + return root + + +# ------------------- # +# Tests AristaXmlBase +# ------------------- # +def test_arista_xml_base_initialization(xml_path): + arista_xml_base = AristaXmlBase(xml_path=str(xml_path)) + assert arista_xml_base.xml_data.getroot().tag == "cvpFolderList", ( + f"Root tag should be 'cvpFolderList' but got" + ) + +# ---------------------- # +# Tests AristaXmlQuerier +# ---------------------- # + +# ---------------------- # +# Tests AristaXmlQuerier available_public_versions for eos + +def test_AristaXmlQuerier_available_public_versions_eos(xml_path): + xml_querier = AristaXmlQuerier(xml_path=xml_path) + versions = xml_querier.available_public_versions(package="eos") + assert len(versions) == 309, "Incorrect number of versions" + assert versions[0] == EosVersion().from_str("4.33.0F"), "First version should be 4.33.0F - got {versions[0]}" + + +def test_AristaXmlQuerier_available_public_versions_eos_f_release(xml_path): + xml_querier = AristaXmlQuerier(xml_path=xml_path) + versions = xml_querier.available_public_versions(package="eos", rtype="F") + assert len(versions) == 95, "Incorrect number of versions: got {len(versions)} expected 207" + assert versions[0] == EosVersion().from_str( + "4.33.0F" + ), "First version should be 4.33.0F - got {len(versions)}" + + +def test_AristaXmlQuerier_available_public_versions_eos_m_release(xml_path): + xml_querier = AristaXmlQuerier(xml_path=xml_path) + versions = xml_querier.available_public_versions(package="eos", rtype="M") + assert ( + len(versions) == 207 + ), "Incorrect number of versions: got {len(versions)} expected 207" + assert versions[0] == EosVersion().from_str( + "4.32.3M" + ), "First version should be 4.32.3M - got {versions[0]}" + +def test_AristaXmlQuerier_available_public_versions_eos_branch_4_29(xml_path): + xml_querier = AristaXmlQuerier(xml_path=xml_path) + versions = xml_querier.available_public_versions(package="eos", branch="4.29") + assert len(versions) == 34, "Incorrect number of versions" + for version in versions: + # logging.debug(f"Checking version {version}") + assert version.is_in_branch("4.29"), f"Version {version} is not in branch 4.29" + assert versions[0] == EosVersion().from_str("4.29.10M"), "First version should be 4.29.10M - got {versions[0]}" + + +def test_AristaXmlQuerier_available_public_versions_eos_f_release_branch_4_29(xml_path): + xml_querier = AristaXmlQuerier(xml_path=xml_path) + versions = xml_querier.available_public_versions(package="eos", rtype="F", branch="4.29") + assert len(versions) == 6, "Incorrect number of versions - expected 6" + for version in versions: + # logging.debug(f"Checking version {version}") + assert version.is_in_branch("4.29"), f"Version {version} is not in branch 4.29" + assert versions[0] == EosVersion().from_str( + "4.29.2F" + ), "First version should be 4.29.2F - got {versions[0]}" + + +def test_AristaXmlQuerier_available_public_versions_eos_m_release_branch_4_29(xml_path): + xml_querier = AristaXmlQuerier(xml_path=xml_path) + versions = xml_querier.available_public_versions(package="eos", rtype="M", branch="4.29") + assert len(versions) == 28, "Incorrect number of versions - expected 28" + for version in versions: + # logging.debug(f"Checking version {version}") + assert version.is_in_branch("4.29"), f"Version {version} is not in branch 4.29" + assert versions[0] == EosVersion().from_str( + "4.29.10M" + ), "First version should be 4.29.10M - got {versions[0]}" + + +# ---------------------- # +# Tests AristaXmlQuerier available_public_versions for cvp + + +def test_AristaXmlQuerier_available_public_versions_cvp(xml_path): + xml_querier = AristaXmlQuerier(xml_path=xml_path) + versions = xml_querier.available_public_versions(package="cvp") + assert ( + len(versions) == 12 + ), "Incorrect number of versions: got {len(versions)} expected 12" + assert versions[0] == CvpVersion().from_str( + "2024.3.0" + ), "First version should be 2024.3.0 - got {versions[0]}" + + +# ---------------------- # +# Tests AristaXmlQuerier branches for eos + + +def test_AristaXmlQuerier_branch_eos(xml_path): + xml_querier = AristaXmlQuerier(xml_path=xml_path) + versions = xml_querier.branches(package="eos") + assert len(versions) == 14, "Incorrect number of branches, got {len(versions)} expected 14" + assert EosVersion().from_str("4.33.0F").branch in versions, "4.33 should be in branches {versions}" + + +def test_AristaXmlQuerier_branch_eos_latest(xml_path): + xml_querier = AristaXmlQuerier(xml_path=xml_path) + versions = xml_querier.branches(package="eos", latest=True) + assert ( + len(versions) == 1 + ), "Incorrect number of branches, got {len(versions)} expected 1" + assert ( + EosVersion().from_str("4.33.0F").branch in versions + ), "4.33 should be in branches {versions}" + + +# ---------------------- # +# Tests AristaXmlQuerier branches for cvp + + +def test_AristaXmlQuerier_branch_cvp(xml_path): + xml_querier = AristaXmlQuerier(xml_path=xml_path) + versions = xml_querier.branches(package="cvp") + assert ( + len(versions) == 5 + ), "Incorrect number of branches, got {len(versions)} expected 5" + assert ( + CvpVersion().from_str("2024.3.0").branch in versions + ), "2024.3 should be in branches {versions}" + + +def test_AristaXmlQuerier_branch_cvp_latest(xml_path): + xml_querier = AristaXmlQuerier(xml_path=xml_path) + versions = xml_querier.branches(package="cvp", latest=True) + assert ( + len(versions) == 1 + ), "Incorrect number of branches, got {len(versions)} expected 1" + assert ( + CvpVersion().from_str("2024.3.0").branch in versions + ), "2024.3 should be in branches {versions}" + + +# ---------------------- # +# Tests AristaXmlQuerier latest for eos + + +def test_AristaXmlQuerier_latest_eos(xml_path): + xml_querier = AristaXmlQuerier(xml_path=xml_path) + versions = xml_querier.latest(package="eos") + assert ( + EosVersion().from_str("4.33.0F") == versions + ), "4.33.0F should be the latest, got {versions}" + + +def test_AristaXmlQuerier_latest_eos_f_release(xml_path): + xml_querier = AristaXmlQuerier(xml_path=xml_path) + versions = xml_querier.latest(package="eos", rtype="F") + assert ( + EosVersion().from_str("4.33.0F") == versions + ), "4.33.0F should be the latest, got {versions}" + + +def test_AristaXmlQuerier_latest_eos_m_release(xml_path): + xml_querier = AristaXmlQuerier(xml_path=xml_path) + versions = xml_querier.latest(package="eos", rtype="M") + assert ( + EosVersion().from_str("4.32.3M") == versions + ), "4.32.3M should be the latest, got {versions}" + + +def test_AristaXmlQuerier_latest_eos_f_release_branch_4_29(xml_path): + xml_querier = AristaXmlQuerier(xml_path=xml_path) + versions = xml_querier.latest(package="eos", rtype="F", branch="4.29") + assert ( + EosVersion().from_str("4.29.2F") == versions + ), "4.29.2F should be the latest, got {versions}" + + +def test_AristaXmlQuerier_latest_eos_m_release_branch_4_29(xml_path): + xml_querier = AristaXmlQuerier(xml_path=xml_path) + versions = xml_querier.latest(package="eos", rtype="M", branch="4.29") + assert ( + EosVersion().from_str("4.29.10M") == versions + ), "4.29.10M should be the latest, got {versions}" + + +# ---------------------- # +# Tests AristaXmlQuerier latest for cvp + + +def test_AristaXmlQuerier_latest_cvp(xml_path): + xml_querier = AristaXmlQuerier(xml_path=xml_path) + versions = xml_querier.latest(package="cvp") + assert ( + CvpVersion().from_str("2024.3.0") == versions + ), "2024.3.0 should be the latest, got {versions}" + + +def test_AristaXmlQuerier_latest_cvp(xml_path): + xml_querier = AristaXmlQuerier(xml_path=xml_path) + versions = xml_querier.latest(package="cvp", branch="2024.2") + assert ( + CvpVersion().from_str("2024.2.1") == versions + ), "2024.2.1 should be the latest, got {versions}" + + +# ---------------------- # +# Tests AristaXmlObject +# ---------------------- # + +def test_arista_xml_object_initialization(xml_path): + arista_xml_object = AristaXmlObject(searched_version="4.29.2F", image_type="image", xml_path=xml_path) + assert arista_xml_object.search_version == "4.29.2F", "Incorrect search version" + assert arista_xml_object.image_type == "image", "Incorrect image type" + +def test_arista_xml_object_filename_for_ceos(xml_path): + arista_xml_object = EosXmlObject( + searched_version="4.29.2F", + image_type="cEOS", + xml_path=xml_path, + ) + filename = arista_xml_object.filename + assert filename == "cEOS-lab-4.29.2F.tar.xz", f"Incorrect filename, got {filename}" + +def test_arista_xml_object_hashfile(xml_path): + arista_xml_object = EosXmlObject( + searched_version="4.29.2F", + image_type="cEOS", + xml_path=xml_path, + ) + hashfile = arista_xml_object.hash_filename() + assert ( + hashfile == "cEOS-lab-4.29.2F.tar.xz.sha512sum" + ), f"Incorrect hashfile, got {hashfile}" + hashfile = arista_xml_object.hash_filename() + assert ( + hashfile == "cEOS-lab-4.29.2F.tar.xz.sha512sum" + ), f"Incorrect hashfile, got {hashfile}" + +def test_arista_xml_object_path_from_xml(xml_path): + arista_xml_object = EosXmlObject( + searched_version="4.29.2F", + image_type="cEOS", + xml_path=xml_path, + ) + path = arista_xml_object.path_from_xml(search_file="EOS-4.29.2F.swi") + assert ( + path + == "/support/download/EOS-USA/Active Releases/4.29/EOS-4.29.2F/EOS-4.29.2F.swi" + ), f"Incorrect path, got {path}" + +def test_arista_xml_object_url(xml_path): + with patch('eos_downloader.logics.arista_xml_server.AristaXmlObject._url') as mock_url: + mock_url.return_value = "https://testserver.com/path/to/EOS-4.29.2F.swi" + arista_xml_object = EosXmlObject( + searched_version="4.29.2F", + image_type="cEOS", + xml_path=xml_path, + ) + url = arista_xml_object._url(xml_path="/path/to/EOS-4.29.2F.swi") + assert url == "https://testserver.com/path/to/EOS-4.29.2F.swi", f"Incorrect URL, got {url}" + +def test_arista_xml_object_urls(xml_path): + with patch( + "eos_downloader.logics.arista_xml_server.AristaXmlObject._url" + ) as mock_url: + mock_url.side_effect = [ + "https://arista.com/path/to/EOS-4.29.2F.swi", + "https://arista.com/path/to/EOS-4.29.2F.swi.sha512sum" + ] + arista_xml_object = EosXmlObject( + searched_version="4.29.2F", + image_type="default", + xml_path=xml_path, + ) + urls = arista_xml_object.urls + logging.warning(f"URLs are: {urls}") + expected_urls = { + "image": "https://arista.com/path/to/EOS-4.29.2F.swi", + "sha512sum": "https://arista.com/path/to/EOS-4.29.2F.swi.sha512sum" + } + assert urls == expected_urls, f"Incorrect URLs, got {urls}" + +def test_arista_xml_object_urls_with_invalid_hash(xml_path): + with patch( + "eos_downloader.logics.arista_xml_server.AristaXmlObject._url" + ) as mock_url: + mock_url.side_effect = [ + "https://arista.com/path/to/EOS-4.29.2F.swi", + "https://arista.com/path/to/EOS-4.29.2F.swi.sha512sum" + ] + arista_xml_object = EosXmlObject( + searched_version="4.29.2F", + image_type="default", + xml_path=xml_path, + ) + urls = arista_xml_object.urls + expected_urls = { + "image": "https://arista.com/path/to/EOS-4.29.2F.swi", + "sha512sum": "https://arista.com/path/to/EOS-4.29.2F.swi.sha512sum" + } + assert urls == expected_urls, f"Incorrect URLs, got {urls}" + +def test_arista_xml_object_urls_with_missing_files(xml_path): + with patch( + "eos_downloader.logics.arista_xml_server.AristaXmlObject._url" + ) as mock_url: + mock_url.side_effect = [None, None, None] + arista_xml_object = EosXmlObject( + searched_version="4.29.2F", + image_type="default", + xml_path=xml_path, + ) + urls = arista_xml_object.urls + expected_urls = { + "image": None, + "sha512sum": None + } + assert urls == expected_urls, f"Incorrect URLs, got {urls}" diff --git a/tests/unit/logics/test_server.py b/tests/unit/logics/test_server.py deleted file mode 100644 index 8404ecc..0000000 --- a/tests/unit/logics/test_server.py +++ /dev/null @@ -1,115 +0,0 @@ -import pytest -import requests -from unittest.mock import patch, Mock -from eos_downloader.logics.arista_server import AristaServer - -import eos_downloader.exceptions - -from tests.lib.fixtures import xml_path, xml_data - -@pytest.fixture -def server(): - return AristaServer(token="testtoken") - - -def test_authenticate_success(server): - with patch('requests.post') as mock_post: - mock_response = Mock() - mock_response.json.return_value = { - "status": {"message": "Success"}, - "data": {"session_code": "testsessioncode"} - } - mock_post.return_value = mock_response - - assert server.authenticate() is True - assert server._session_id is not None - - -def test_authenticate_invalid_token(server): - with patch('requests.post') as mock_post: - mock_response = Mock() - mock_response.json.return_value = { - "status": {"message": "Invalid access token"} - } - mock_post.return_value = mock_response - - with pytest.raises(eos_downloader.exceptions.AuthenticationError): - server.authenticate() - - -def test_authenticate_expired_token(server): - with patch('requests.post') as mock_post: - mock_response = Mock() - mock_response.json.return_value = { - "status": {"message": "Access token expired"} - } - mock_post.return_value = mock_response - - with pytest.raises(eos_downloader.exceptions.AuthenticationError): - server.authenticate() - - -def test_authenticate_key_error(server): - with patch('requests.post') as mock_post: - mock_response = Mock() - mock_response.json.return_value = { - "status": {"message": "Success"} - } - mock_post.return_value = mock_response - - assert server.authenticate() is False - assert server._session_id is None - - -def test_get_xml_data_success(server, xml_path): - with patch('requests.post') as mock_post: - with open(xml_path, 'r') as file: - xml_content = file.read() - - mock_response = Mock() - mock_response.json.return_value = { - "status": {"message": "Success"}, - "data": {"xml": xml_content}, - } - mock_post.return_value = mock_response - - xml_data = server.get_xml_data() - assert xml_data is not None - assert ( - xml_data.getroot().tag == "cvpFolderList" - ) # Assuming the root tag in data.xml is 'cvpFolderList' - - -def test_get_xml_data_key_error(server): - with patch('requests.post') as mock_post: - mock_response = Mock() - mock_response.json.return_value = {} - mock_post.return_value = mock_response - - with pytest.raises(KeyError): - server.get_xml_data() - - -def test_get_url_success(server): - with patch('requests.post') as mock_post: - mock_response = Mock() - mock_response.json.return_value = { - "status": {"message": "Success"}, - "data": {"url": "http://example.com/download"}, - } - mock_post.return_value = mock_response - - url = server.get_url("remote/file/path") - assert url == "http://example.com/download" - - -def test_get_url_no_data(server): - with patch('requests.post') as mock_post: - mock_response = Mock() - mock_response.json.return_value = { - "status": {"message": "Success"}, - } - mock_post.return_value = mock_response - - url = server.get_url("remote/file/path") - assert url is None