Adding upstream version 0.12.0.

Signed-off-by: Daniel Baumann <daniel@debian.org>

tests/unit/logics/test_arista_server.py

@@ -0,0 +1,115 @@
+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

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}"

tests/unit/logics/test_download.py

@@ -0,0 +1,143 @@
+import os
+import pytest
+from unittest.mock import Mock, patch, mock_open
+from eos_downloader.logics.download import SoftManager
+from eos_downloader.logics.arista_xml_server import EosXmlObject
+
+
+@pytest.fixture
+def soft_manager():
+    return SoftManager()
+
+
+@pytest.fixture
+def mock_eos_object():
+    mock = Mock(spec=EosXmlObject)
+    mock.version = "4.28.0F"
+    mock.filename = "EOS-4.28.0F.swi"
+    mock.urls = {
+        "image": "http://example.com/EOS-4.28.0F.swi",
+        "md5sum": "http://example.com/EOS-4.28.0F.swi.md5",
+        "sha512sum": "http://example.com/EOS-4.28.0F.swi.sha512",
+    }
+    mock.hash_filename = Mock(return_value="EOS-4.28.0F.swi.md5")
+    return mock
+
+
+@pytest.mark.parametrize("dry_run", [True, False])
+def test_soft_manager_init(dry_run):
+    manager = SoftManager(dry_run=dry_run)
+    assert manager.dry_run == dry_run
+    assert manager.file == {"name": None, "md5sum": None, "sha512sum": None}
+
+
+@patch("requests.get")
+@patch("tqdm.tqdm")
+def test_download_file_raw(mock_tqdm, mock_requests):
+    # Setup mock response
+    mock_response = Mock()
+    mock_response.headers = {"Content-Length": "1024"}
+    mock_response.iter_content.return_value = [b"data"]
+    mock_requests.return_value = mock_response
+
+    with patch("builtins.open", mock_open()) as mock_file:
+        result = SoftManager._download_file_raw("http://test.com/file", "/tmp/file")
+        assert result == "/tmp/file"
+        mock_file().write.assert_called_with(b"data")
+
+
+@patch("os.makedirs")
+def test_create_destination_folder(mock_makedirs):
+    SoftManager._create_destination_folder("/test/path")
+    mock_makedirs.assert_called_once_with("/test/path", exist_ok=True)
+
+
+def test_compute_hash_md5sum(soft_manager):
+    test_data = b"test data"
+    expected_hash = "eb733a00c0c9d336e65691a37ab54293"
+
+    with patch("builtins.open", mock_open(read_data=test_data)):
+        result = soft_manager._compute_hash_md5sum("test_file", expected_hash)
+        assert result is True
+
+        # Test with incorrect hash
+        result = soft_manager._compute_hash_md5sum("test_file", "wrong_hash")
+        assert result is False
+
+
+# @pytest.mark.parametrize(
+#     "check_type,valid_hash", [("md5sum", True), ("sha512sum", True)]
+# )
+# def test_checksum(soft_manager, check_type, valid_hash):
+#     soft_manager.file = {
+#         "name": "test.swi",
+#         "md5sum": "test.swi.md5",
+#         "sha512sum": "test.swi.sha512",
+#     }
+
+#     mock_hash = "a" * (128 if check_type == "sha512sum" else 32)
+
+#     with patch(
+#         "builtins.open",
+#         mock_open(read_data=f"{mock_hash if valid_hash else 'wrong'} test.swi"),
+#     ) as mock_file:
+#         if valid_hash:
+#             with patch(
+#                 "hashlib.sha512" if check_type == "sha512sum" else "hashlib.md5"
+#             ) as mock_hash_func:
+#                 mock_hash_instance = Mock()
+#                 mock_hash_instance.hexdigest.return_value = mock_hash
+#                 mock_hash_func.return_value = mock_hash_instance
+#                 assert soft_manager.checksum(check_type) is True
+#         else:
+#             with pytest.raises(ValueError):
+#                 soft_manager.checksum(check_type)
+
+
+@patch("eos_downloader.logics.download.SoftManager._download_file_raw")
+@patch("eos_downloader.helpers.DownloadProgressBar")
+def test_download_file(mock_progress_bar, mock_download_raw, soft_manager):
+    url = "http://test.com/file"
+    file_path = "/tmp"
+    filename = "test.swi"
+
+    # Test with rich interface
+    result = soft_manager.download_file(url, file_path, filename, rich_interface=True)
+    assert result == os.path.join(file_path, filename)
+    mock_progress_bar.assert_called_once()
+
+
+@patch("eos_downloader.logics.download.SoftManager.download_file")
+def test_downloads(mock_download, soft_manager, mock_eos_object):
+    result = soft_manager.downloads(
+        mock_eos_object, "/tmp/downloads", rich_interface=True
+    )
+    assert result == "/tmp/downloads"
+    assert mock_download.call_count == len(mock_eos_object.urls)
+
+
+@patch("shutil.which")
+@patch("os.system")
+def test_import_docker(mock_system, mock_which, soft_manager):
+    mock_which.return_value = "/usr/bin/docker"
+
+    # Test with existing file
+    with patch("os.path.exists", return_value=True):
+        soft_manager.import_docker("/tmp/test.swi", "arista/ceos", "latest")
+        mock_system.assert_called_once()
+
+    # Test with non-existing file
+    with patch("os.path.exists", return_value=False):
+        with pytest.raises(FileNotFoundError):
+            soft_manager.import_docker("/tmp/nonexistent.swi")
+
+
+@patch("os.system")
+@patch("os.path.exists")
+def test_provision_eve(mock_exists, mock_system, soft_manager, mock_eos_object):
+    mock_exists.return_value = False
+
+    with patch("eos_downloader.logics.download.SoftManager.download_file"):
+        soft_manager.provision_eve(mock_eos_object, noztp=False)
+        # Check if qemu-img convert and unl_wrapper commands were called
+        assert mock_system.call_count == 2