napalm-qnap-qts/tests/test_qnap_qts.py

"""Unit tests for the QNAP QTS driver.

Command-output fixtures are captured from real hardware via tools/harvest.sh —
one QTS 4 box and one QTS 5 box — and pasted in as module constants. Parsers are
written against those, never against guessed output.
"""

from __future__ import annotations

from unittest.mock import MagicMock, patch

import pytest

from napalm_qnap_qts import QnapQtsDriver

#: Parsers need real command output to be written against. These tests are the
#: specification for the work that tools/harvest.sh unblocks. strict=True means
#: the suite goes red the moment one starts passing, which is the reminder to
#: delete the marker rather than leave it lying around.
_PENDING_HARVEST = "blocked on fixture harvest from real QTS 4 / QTS 5 hardware"


@pytest.fixture()
def driver():
    """A driver with its transport mocked out, bypassing __init__.

    Same shape as napalm-linux's fixture: nothing in __init__ needs patching,
    so constructing the object by hand is cheaper and clearer than mocking
    ConnectHandler.
    """
    d = QnapQtsDriver.__new__(QnapQtsDriver)
    d.hostname = "testnas"
    d.username = "admin"
    d.password = "pass"  # noqa: S105
    d.timeout = 60
    d.port = 22
    d._secret = "pass"  # noqa: S105
    d._forced_pkg_manager = None
    d._pkg_manager = None
    d._sudo_password = None
    d.netmiko_optional_args = {}
    d._device = MagicMock()
    d._qts_major = 5
    d._docker_path = "docker"
    d._virsh_path = None
    return d


class TestDriverIdentity:
    """Read without a connection by netOrk's discovery and /drivers endpoint."""

    def test_driver_name(self):
        assert QnapQtsDriver.DRIVER_NAME == "qnap_qts"

    def test_lowercase_driver_name_alias_matches(self):
        """netOrk's register_driver path reads the lowercase attribute."""
        assert QnapQtsDriver.driver_name == QnapQtsDriver.DRIVER_NAME

    def test_vendor(self):
        assert QnapQtsDriver.VENDOR == "QNAP"

    def test_type_label_is_storage(self):
        assert QnapQtsDriver.TYPE_LABEL == "Storage"

    def test_declares_device_class_explicitly(self):
        """Inheriting LinuxDriver would otherwise get it classified as "linux",
        and netOrk would hide the Storage tab."""
        assert QnapQtsDriver.DEVICE_CLASS == "storage"

    def test_declares_at_least_one_fingerprint_source(self):
        """Discovery silently skips a driver that declares no fingerprint data."""
        assert (
            QnapQtsDriver.HTTP_FINGERPRINT
            or QnapQtsDriver.SNMP_FINGERPRINT
            or QnapQtsDriver.SSH_FINGERPRINT
        )

    def test_snmp_object_id_is_the_qnap_enterprise_oid(self):
        assert QnapQtsDriver.SNMP_OBJECT_ID_PREFIX == "1.3.6.1.4.1.24681"

    def test_http_fingerprint_is_mandatory_to_avoid_matching_any_nas(self):
        patterns = {r.pattern: r for r in QnapQtsDriver.HTTP_FINGERPRINT}
        assert patterns["qnap"].mandatory is True


class TestMroForwarding:
    """StorageDriver precedes LinuxDriver in the MRO, so its NotImplementedError
    stubs shadow LinuxDriver's working implementations. Every collision has to be
    resolved deliberately — this is the class of bug that makes a driver look
    fine until it runs against hardware.
    """

    def test_get_services_returns_the_linux_os_service_list(self, driver):
        """netOrk's poller expects a list here (OS services). StorageDriver's
        stub would return a dict of NAS services, if it returned anything."""
        from napalm_linux.linux import LinuxDriver

        with patch.object(LinuxDriver, "get_services", return_value=[{"name": "sshd"}]) as m:
            result = driver.get_services()

        assert m.called
        assert result == [{"name": "sshd"}]

    @pytest.mark.xfail(strict=True, reason=_PENDING_HARVEST)
    def test_nas_services_live_under_a_separate_name(self):
        """get_storage_services is what netOrk's _collect.py actually reads for
        the storage snapshot — get_services is the OS list."""
        assert hasattr(QnapQtsDriver, "get_storage_services")

    @pytest.mark.xfail(strict=True, reason=_PENDING_HARVEST)
    def test_get_packages_is_not_the_storage_stub(self):
        from napalm_device_types import StorageDriver

        assert QnapQtsDriver.get_packages is not StorageDriver.get_packages

    @pytest.mark.parametrize(
        ("method", "args"),
        [
            ("install_package", ("qpkg-name",)),
            ("remove_package", ("qpkg-name",)),
            ("snapshot_create", ("DataVol1", "snap1")),
        ],
    )
    def test_out_of_scope_writers_still_raise(self, driver, method, args):
        """v1 is read-only plus safe actions. These must fail loudly rather than
        appear supported — netOrk's poller catches NotImplementedError."""
        with pytest.raises(NotImplementedError):
            getattr(driver, method)(*args)


class TestQtsVersionDetection:
    def test_parses_major_version(self, driver):
        with patch.object(driver, "_send", return_value="5.1.5"):
            assert driver._detect_qts_major() == 5

    def test_parses_qts_four(self, driver):
        with patch.object(driver, "_send", return_value="4.5.4"):
            assert driver._detect_qts_major() == 4

    def test_tolerates_a_build_suffix(self, driver):
        with patch.object(driver, "_send", return_value="5.2.0.2782 (2026/03/14)"):
            assert driver._detect_qts_major() == 5

    def test_unreadable_version_does_not_raise(self, driver):
        """An unknown version must degrade to the newer code path, not abort the
        poll — a NAS answering nothing useful here is still worth polling."""
        with patch.object(driver, "_send", return_value=""):
            assert driver._detect_qts_major() is None


class TestDockerBinDiscovery:
    """Container Station does not put docker on PATH."""

    def test_uses_the_container_station_path_when_present(self, driver):
        found = "/share/CACHEDEV1_DATA/.qpkg/container-station/bin/docker"
        with patch.object(driver, "_send", return_value=found):
            assert driver._discover_docker_path() == found

    def test_falls_back_to_plain_docker_when_nothing_is_found(self, driver):
        with patch.object(driver, "_send", return_value=""):
            assert driver._discover_docker_path() == "docker"

    def test_docker_bin_hook_returns_the_discovered_path(self, driver):
        driver._docker_path = "/opt/docker"
        assert driver._docker_bin() == "/opt/docker"