From 5b6b634a8142e2f65c841a0960a6e8cce29644da Mon Sep 17 00:00:00 2001 From: James Williams Date: Wed, 22 Apr 2026 09:28:46 -0500 Subject: [PATCH 1/4] Add JunOS pre-transfer free-space check (NAPPS-1085) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the fail-open pre-transfer free-space check on Juniper Junos using the seam added in NAPPS-1091 (PR #370). Image transfers now fail fast when the target filesystem lacks room instead of half-writing flash. Lab-validated against an SRX340 (Junos 24.x). Driver (pyntc/devices/jnpr_device.py): - JunosDevice._get_free_space uses PyEZ FS.storage_usage() and parses the human-readable avail format (e.g., "126M", "1.0G"). PyEZ does not expose a native block size and Junos block semantics vary by release, so parsing the normalised human-readable field avoids the ambiguity. - Mount resolution uses longest-prefix match (same logic df uses) via _mount_encloses_path. SRX hardware does not expose /var/tmp as its own mount — it lives inside /var — so strict equality would raise; / becomes the universal fallback. Directory-boundary semantics prevent /vari from matching /var/tmp. - file_copy calls _check_free_space(os.path.getsize(src)) before any SCP put. - remote_file_copy gains optional file_system + **kwargs for BaseDevice parity, and calls _pre_transfer_space_check before fs.cp so the check fires only when a transfer would actually happen; still fail-open when src.file_size_bytes is None. - remote_file_copy appends src.file_name to the URL when the URL carries no path (mirroring ASA), so callers can point at a bare host like ftp://server. Tests: - Unit tests cover avail parse across K/M/G/T/P units, longest-prefix match (exact hit, most-specific wins, root fallback, /vari boundary rejected, empty storage raises), file_copy raising before SCP.put, remote_file_copy raising before fs.cp, fail-open path, and URL append / keep-intact. - Existing test_file_copy and test_remote_file_copy prime fs.storage_usage so the new probe succeeds. - New integration suite (tests/integration/test_jnpr_device.py) mirrors the EOS / ASA patterns for manual lab runs. TFTP is excluded because PyEZ fs.cp does not accept TFTP URLs. Shared integration helper: - integration_hash_algo() reads FILE_HASH_ALGO (default "sha512") so lab runs can pick the algorithm their devices support. Junos does not implement sha512 — SRX/MX runs set FILE_HASH_ALGO=sha256. Backward-compatible for EOS / ASA runs. Co-Authored-By: Claude Opus 4.7 (1M context) --- changes/373.added | 1 + pyntc/devices/jnpr_device.py | 143 +++++++++++- tests/integration/_helpers.py | 16 +- tests/integration/conftest.py | 4 +- tests/integration/test_jnpr_device.py | 245 ++++++++++++++++++++ tests/unit/test_devices/test_jnpr_device.py | 192 ++++++++++++++- 6 files changed, 593 insertions(+), 8 deletions(-) create mode 100644 changes/373.added create mode 100644 tests/integration/test_jnpr_device.py diff --git a/changes/373.added b/changes/373.added new file mode 100644 index 00000000..fd8c28c3 --- /dev/null +++ b/changes/373.added @@ -0,0 +1 @@ +Added a pre-transfer free-space check to Juniper JunOS ``file_copy`` and ``remote_file_copy`` that raises ``NotEnoughFreeSpaceError`` when the target filesystem lacks room for the image. diff --git a/pyntc/devices/jnpr_device.py b/pyntc/devices/jnpr_device.py index 635dbcec..d654e2fe 100644 --- a/pyntc/devices/jnpr_device.py +++ b/pyntc/devices/jnpr_device.py @@ -6,6 +6,7 @@ import time import warnings from tempfile import NamedTemporaryFile +from urllib.parse import urlparse from jnpr.junos import Device as JunosNativeDevice from jnpr.junos.exception import ConfigLoadError @@ -18,9 +19,48 @@ from pyntc import log from pyntc.devices.base_device import BaseDevice, fix_docs from pyntc.devices.tables.jnpr.loopback import LoopbackTable # pylint: disable=no-name-in-module -from pyntc.errors import CommandError, CommandListError, FileTransferError, OSInstallError, RebootTimeoutError +from pyntc.errors import ( + CommandError, + CommandListError, + FileSystemNotFoundError, + FileTransferError, + OSInstallError, + RebootTimeoutError, +) from pyntc.utils.models import FileCopyModel +# Multipliers for Junos ``df``-style size suffixes. Junos formats available +# space with binary (1024-based) units in its ```` +# XML attribute (e.g., "126M", "1.0G"). +_JUNOS_SIZE_UNIT_MULTIPLIERS = { + "": 1, + "B": 1, + "K": 1024, + "M": 1024**2, + "G": 1024**3, + "T": 1024**4, + "P": 1024**5, +} +_JUNOS_AVAIL_FORMAT_RE = re.compile(r"^\s*(\d+(?:\.\d+)?)\s*([BKMGTP]?)\s*$", re.IGNORECASE) +# Default mount point to probe when callers do not specify one. ``/var/tmp`` is +# the standard destination for ``fs.cp`` transfers on Junos (remote device +# mount point, not a local temp directory). +_JUNOS_DEFAULT_FILE_SYSTEM = "/var/tmp" # noqa: S108 + + +def _mount_encloses_path(mount, path): + """Return True if ``mount`` is the filesystem that contains ``path``. + + Matches with directory-boundary semantics (the same rule ``df`` uses) so + ``/vari`` is not mistaken for a prefix of ``/var/tmp``. ``/`` encloses + every path. + """ + if mount == "/": + return True + if path == mount: + return True + return path.startswith(mount.rstrip("/") + "/") + @fix_docs class JunosDevice(BaseDevice): @@ -62,6 +102,83 @@ def _file_copy_local_md5(self, filepath, blocksize=2**20): buf = file_name.read(blocksize) return md5_hash.hexdigest() + def _get_free_space(self, file_system=None): + """Return free bytes on the filesystem containing ``file_system``. + + Probes the device via ``get-system-storage-information`` (invoked by + PyEZ ``FS.storage_usage``) and parses the human-readable + ``available-blocks`` ``format`` attribute (e.g., ``"126M"``, ``"1.0G"``) + into bytes. The human-readable string is used rather than the raw + block count because PyEZ does not expose a native block size and + Junos block semantics can vary by release. + + ``file_system`` is resolved by **longest-prefix mount match** — the + same logic ``df`` uses — so a caller asking about ``/var/tmp`` on a + platform that only mounts ``/var`` (e.g., SRX hardware) still gets + back the correct filesystem's free space. ``/`` is always a fallback + when nothing more specific matches. + + Args: + file_system (str, optional): Target path. Defaults to ``/var/tmp`` + — the standard destination for ``fs.cp`` copies on Junos. + + Returns: + int: Free bytes available on the resolved filesystem. + + Raises: + FileSystemNotFoundError: When no mount point encloses ``file_system`` + (i.e., not even ``/`` is present in ``storage_usage``). + CommandError: When the ``avail`` format string cannot be parsed. + """ + if file_system is None: + file_system = _JUNOS_DEFAULT_FILE_SYSTEM + + usage = self.fs.storage_usage() + best_info = None + best_mount = None + best_len = -1 + for _dev, info in usage.items(): + mount = info.get("mount") + if not mount or not _mount_encloses_path(mount, file_system): + continue + if len(mount) > best_len: + best_info = info + best_mount = mount + best_len = len(mount) + + if best_info is None: + log.error( + "Host %s: no mount encloses %s in storage_usage output.", + self.host, + file_system, + ) + raise FileSystemNotFoundError(hostname=self.host, command="show system storage") + + avail = best_info.get("avail", "") + match = _JUNOS_AVAIL_FORMAT_RE.match(str(avail)) + if match is None: + log.error( + "Host %s: could not parse avail %r for mount %s.", + self.host, + avail, + best_mount, + ) + raise CommandError( + command="show system storage", + message=f"Unable to parse available space {avail!r} for {best_mount}.", + ) + size = float(match.group(1)) + multiplier = _JUNOS_SIZE_UNIT_MULTIPLIERS[match.group(2).upper()] + free_bytes = int(size * multiplier) + log.debug( + "Host %s: %s bytes free on %s (resolved from %s).", + self.host, + free_bytes, + best_mount, + file_system, + ) + return free_bytes + def _get_interfaces(self): eth_ifaces = EthPortTable(self.native) eth_ifaces.get() @@ -304,11 +421,15 @@ def file_copy(self, src, dest=None, **kwargs): Raises: FileTransferError: Raised when unable to verify file was transferred succesfully. + NotEnoughFreeSpaceError: When the target filesystem has fewer free bytes + than ``src`` requires. """ if not self.file_copy_remote_exists(src, dest, **kwargs): if dest is None: dest = os.path.basename(src) + self._check_free_space(os.path.getsize(src)) + with SCP(self.native) as scp: scp.put(src, remote_path=dest) @@ -525,16 +646,24 @@ def compare_file_checksum(self, checksum, filename, hashing_algorithm="md5"): """ return checksum == self.get_remote_checksum(filename, hashing_algorithm) - def remote_file_copy(self, src: FileCopyModel = None, dest=None): + def remote_file_copy(self, src: FileCopyModel = None, dest=None, file_system: str | None = None, **kwargs): """Copy a file to a remote device. Args: src (FileCopyModel): The source file model. dest (str): The destination file path on the remote device. + file_system (str, optional): Mount point used for the pre-transfer + free-space check. Defaults to ``/var/tmp``. + **kwargs (Any): Accepted for parity with ``BaseDevice.remote_file_copy``; + other drivers may forward extra options. Raises: TypeError: If src is not an instance of FileCopyModel. FileTransferError: If there is an error during file transfer or if the file cannot be verified after transfer. + NotEnoughFreeSpaceError: If ``src.file_size_bytes`` is set and the + target mount point has fewer free bytes than ``src.file_size_bytes``. + When ``file_size`` is omitted from ``src`` the pre-transfer space + check is skipped entirely. """ if not isinstance(src, FileCopyModel): raise TypeError("src must be an instance of FileCopyModel") @@ -542,7 +671,15 @@ def remote_file_copy(self, src: FileCopyModel = None, dest=None): if self.verify_file(src.checksum, dest, hashing_algorithm=src.hashing_algorithm): return - if not self.fs.cp(from_path=src.download_url, to_path=dest, dev_timeout=src.timeout): + self._pre_transfer_space_check(src, file_system=file_system) + + # Junos ``fs.cp`` requires the filename in the URL; append ``src.file_name`` + # when the URL carries no path so callers can point at a bare host. + source_url = src.download_url + if not urlparse(source_url).path.strip("/"): + source_url = f"{source_url.rstrip('/')}/{src.file_name}" + + if not self.fs.cp(from_path=source_url, to_path=dest, dev_timeout=src.timeout): raise FileTransferError(message=f"Unable to copy file from remote url {src.clean_url}") # Some devices take a while to sync the filesystem after a copy but netconf returns before the sync completes diff --git a/tests/integration/_helpers.py b/tests/integration/_helpers.py index 24e33553..c0f663d4 100644 --- a/tests/integration/_helpers.py +++ b/tests/integration/_helpers.py @@ -19,11 +19,23 @@ } +def integration_hash_algo(): + """Return the hashing algorithm configured for the current integration run. + + Reads ``FILE_HASH_ALGO`` so a lab can pick whatever its device supports + (Junos SRX, for example, does not implement sha512; sha256 works). + Defaults to ``"sha512"`` for backward compatibility with existing EOS + and ASA integration runs. + """ + return os.environ.get("FILE_HASH_ALGO", "sha512") + + def build_file_copy_model(url_env_var): """Build a ``FileCopyModel`` from a per-protocol URL env var. Calls ``pytest.skip`` if the URL, ``FILE_CHECKSUM``, or ``FILE_SIZE`` env - vars are not set. + vars are not set. The hashing algorithm defaults to ``FILE_HASH_ALGO`` + (``sha512`` when unset). """ url = os.environ.get(url_env_var) checksum = os.environ.get("FILE_CHECKSUM") @@ -40,7 +52,7 @@ def build_file_copy_model(url_env_var): file_name=file_name, file_size=file_size, file_size_unit=file_size_unit, - hashing_algorithm="sha512", + hashing_algorithm=integration_hash_algo(), timeout=900, ) diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 0ded2a40..5de93ea6 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -7,7 +7,7 @@ from pyntc.utils.models import FileCopyModel -from ._helpers import PROTOCOL_URL_VARS +from ._helpers import PROTOCOL_URL_VARS, integration_hash_algo @pytest.fixture(scope="module") @@ -31,7 +31,7 @@ def any_file_copy_model(): file_name=file_name, file_size=file_size, file_size_unit=file_size_unit, - hashing_algorithm="sha512", + hashing_algorithm=integration_hash_algo(), timeout=900, ) pytest.skip("No protocol URL / FILE_CHECKSUM / FILE_SIZE environment variables not set") diff --git a/tests/integration/test_jnpr_device.py b/tests/integration/test_jnpr_device.py new file mode 100644 index 00000000..07a40681 --- /dev/null +++ b/tests/integration/test_jnpr_device.py @@ -0,0 +1,245 @@ +"""Integration tests for JunosDevice.remote_file_copy. + +These tests connect to an actual Juniper Junos device in the lab and are run manually. +They are NOT part of the CI unit test suite. + +Usage (from project root): + export JUNOS_HOST= + export JUNOS_USER= + export JUNOS_PASS= + export FTP_URL=ftp://:@/ + export SCP_URL=scp://:@:2222/ + export HTTP_URL=http://:@:8081/ + export HTTPS_URL=https://:@:8443/ + export FILE_CHECKSUM= + export FILE_HASH_ALGO=sha256 # Junos doesn't implement sha512; sha256 is typical + export FILE_SIZE= + export FILE_SIZE_UNIT=megabytes # optional; defaults to "bytes" + poetry run pytest tests/integration/test_jnpr_device.py -v + +Set only the protocol URL vars for the servers you have available; each +protocol test will skip automatically if its URL is not set. + +Environment variables: + JUNOS_HOST - IP address or hostname of the lab Junos device + JUNOS_USER - NETCONF / SSH username + JUNOS_PASS - NETCONF / SSH password + FTP_URL - FTP URL of the file to transfer + SCP_URL - SCP URL of the file to transfer + HTTP_URL - HTTP URL of the file to transfer + HTTPS_URL - HTTPS URL of the file to transfer + FILE_NAME - Destination filename on the device (default: basename of URL path) + FILE_CHECKSUM - Expected checksum of the file (algorithm set by FILE_HASH_ALGO) + FILE_HASH_ALGO - Hashing algorithm (md5 / sha1 / sha256). Defaults to sha512 for + cross-platform parity but Junos does not implement sha512; set to + sha256 for SRX/MX runs. + FILE_SIZE - Expected size of the file expressed in FILE_SIZE_UNIT units; used for + the pre-transfer free-space check + FILE_SIZE_UNIT - One of "bytes", "megabytes", or "gigabytes" (default: "bytes") +""" + +import os +from unittest import mock + +import pytest + +from pyntc.devices import JunosDevice +from pyntc.errors import NotEnoughFreeSpaceError +from pyntc.utils.models import FILE_SIZE_UNITS, FileCopyModel + +from ._helpers import PROTOCOL_URL_VARS, build_file_copy_model, first_available_url, integration_hash_algo + +# Junos ``fs.cp`` does not accept TFTP URLs, so narrow the protocol set before +# any protocol-aware fixture/test reads from it. +JUNOS_PROTOCOL_URL_VARS = {k: v for k, v in PROTOCOL_URL_VARS.items() if k != "tftp"} + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def device(): + """Connect to the lab Junos device. Skips all tests if credentials are not set.""" + host = os.environ.get("JUNOS_HOST") + user = os.environ.get("JUNOS_USER") + password = os.environ.get("JUNOS_PASS") + + if not all([host, user, password]): + pytest.skip("JUNOS_HOST / JUNOS_USER / JUNOS_PASS environment variables not set") + + dev = JunosDevice(host, user, password) + yield dev + dev.close() + + +def _junos_dest(file_name): + """Return the absolute destination path on Junos for ``file_name``.""" + return f"/var/tmp/{file_name}" + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +def test_device_connects(device): + """Verify the device is reachable and responds to facts queries.""" + assert device.hostname + assert device.os_version + + +def test_check_file_exists_false(device, any_file_copy_model): + """Before the copy, the file should not exist (or this test is a no-op if it does).""" + result = device.check_file_exists(_junos_dest(any_file_copy_model.file_name)) + assert isinstance(result, bool) + + +def test_remote_file_copy_ftp(device): + """Transfer the file using FTP and verify it exists on the device.""" + model = build_file_copy_model("FTP_URL") + dest = _junos_dest(model.file_name) + device.remote_file_copy(model, dest=dest) + assert device.check_file_exists(dest) + + +def test_remote_file_copy_scp(device): + """Transfer the file using SCP and verify it exists on the device.""" + model = build_file_copy_model("SCP_URL") + dest = _junos_dest(model.file_name) + device.remote_file_copy(model, dest=dest) + assert device.check_file_exists(dest) + + +def test_remote_file_copy_http(device): + """Transfer the file using HTTP and verify it exists on the device.""" + model = build_file_copy_model("HTTP_URL") + dest = _junos_dest(model.file_name) + device.remote_file_copy(model, dest=dest) + assert device.check_file_exists(dest) + + +def test_remote_file_copy_https(device): + """Transfer the file using HTTPS and verify it exists on the device.""" + model = build_file_copy_model("HTTPS_URL") + dest = _junos_dest(model.file_name) + device.remote_file_copy(model, dest=dest) + assert device.check_file_exists(dest) + + +def test_verify_file_after_copy(device, any_file_copy_model): + """After a successful copy the file should verify cleanly.""" + dest = _junos_dest(any_file_copy_model.file_name) + if not device.check_file_exists(dest): + pytest.skip("File does not exist on device; run a copy test first") + assert device.verify_file(any_file_copy_model.checksum, dest, hashing_algorithm=integration_hash_algo()) + + +# --------------------------------------------------------------------------- +# Free-space / pre-transfer tests (NAPPS-1085) +# --------------------------------------------------------------------------- + + +def test_get_free_space_returns_positive_int(device): + """``_get_free_space`` returns a positive int parsed from storage_usage.""" + free = device._get_free_space() # pylint: disable=protected-access + assert isinstance(free, int) + assert free > 0 + + +def test_check_free_space_succeeds_for_small_request(device): + """A 1-byte request must always fit; ``_check_free_space`` returns ``None``.""" + # pylint: disable=protected-access + assert device._check_free_space(required_bytes=1) is None + + +def test_check_free_space_raises_when_required_exceeds_free(device): + """When required bytes exceed what the device reports, raise NotEnoughFreeSpaceError.""" + # pylint: disable=protected-access + free = device._get_free_space() + with pytest.raises(NotEnoughFreeSpaceError): + device._check_free_space(required_bytes=free + 1) + + +def test_file_size_unit_conversion_matches_device_free_space(device): + """A megabyte-denominated request converts through ``FILE_SIZE_UNITS`` correctly.""" + # pylint: disable=protected-access + free_bytes = device._get_free_space() + one_mb = FILE_SIZE_UNITS["megabytes"] + if free_bytes < one_mb: + pytest.skip("Device has less than 1 MB free; conversion sanity test not meaningful") + assert device._check_free_space(required_bytes=one_mb) is None + + +def test_remote_file_copy_rejects_oversized_transfer(device): + """remote_file_copy raises NotEnoughFreeSpaceError and never copies the file.""" + checksum = os.environ.get("FILE_CHECKSUM") + scheme, url = first_available_url(JUNOS_PROTOCOL_URL_VARS) + if not (url and checksum): + pytest.skip("No protocol URL / FILE_CHECKSUM environment variables set") + + # pylint: disable=protected-access + free_bytes = device._get_free_space() + free_gb = free_bytes // FILE_SIZE_UNITS["gigabytes"] + oversized_gb = max(free_gb * 10, 10) + + unique_name = f"pyntc_integration_space_check_{os.getpid()}_{scheme}.bin" + dest = _junos_dest(unique_name) + model = FileCopyModel( + download_url=url, + checksum=checksum, + file_name=unique_name, + file_size=oversized_gb, + file_size_unit="gigabytes", + hashing_algorithm=integration_hash_algo(), + timeout=60, + ) + + assert not device.check_file_exists(dest), "Unique filename unexpectedly exists before test" + + with pytest.raises(NotEnoughFreeSpaceError): + device.remote_file_copy(model, dest=dest) + + assert not device.check_file_exists(dest) + + +def test_remote_file_copy_accepts_declared_size_within_free_space(device): + """A correctly-sized FileCopyModel copies without the space check interfering.""" + scheme, _url = first_available_url(JUNOS_PROTOCOL_URL_VARS) + if scheme is None: + pytest.skip("No protocol URL environment variables set") + model = build_file_copy_model(JUNOS_PROTOCOL_URL_VARS[scheme]) + # pylint: disable=protected-access + free_bytes = device._get_free_space() + assert model.file_size_bytes <= free_bytes, ( + "Configured FILE_SIZE/FILE_SIZE_UNIT exceeds device free space; update env vars" + ) + dest = _junos_dest(model.file_name) + device.remote_file_copy(model, dest=dest) + assert device.check_file_exists(dest) + + +def test_remote_file_copy_skips_space_check_when_file_size_omitted(device): + """When FileCopyModel has no file_size, _check_free_space is never called.""" + checksum = os.environ.get("FILE_CHECKSUM") + file_name = os.environ.get("FILE_NAME") + _, url = first_available_url(JUNOS_PROTOCOL_URL_VARS) + if not (url and checksum and file_name): + pytest.skip("URL / FILE_CHECKSUM / FILE_NAME environment variables not set") + + model = FileCopyModel( + download_url=url, + checksum=checksum, + file_name=file_name, + hashing_algorithm=integration_hash_algo(), + timeout=60, + ) # file_size intentionally omitted + assert model.file_size is None + assert model.file_size_bytes is None + + dest = _junos_dest(file_name) + with mock.patch.object(JunosDevice, "_check_free_space") as spy: + device.remote_file_copy(model, dest=dest) + + spy.assert_not_called() + assert device.check_file_exists(dest) diff --git a/tests/unit/test_devices/test_jnpr_device.py b/tests/unit/test_devices/test_jnpr_device.py index 540d05f3..9d71a4f4 100644 --- a/tests/unit/test_devices/test_jnpr_device.py +++ b/tests/unit/test_devices/test_jnpr_device.py @@ -7,9 +7,34 @@ from jnpr.junos.exception import ConfigLoadError from pyntc.devices import JunosDevice -from pyntc.errors import CommandError, CommandListError, FileTransferError, OSInstallError, RebootTimeoutError +from pyntc.errors import ( + CommandError, + CommandListError, + FileSystemNotFoundError, + FileTransferError, + NotEnoughFreeSpaceError, + OSInstallError, + RebootTimeoutError, +) from pyntc.utils.models import FileCopyModel +# Shared fake storage_usage() return value so tests exercising file_copy and +# remote_file_copy do not trip the new pre-transfer free-space probe. "10G" +# on /var/tmp is plenty of room for the tiny temp files and fake file sizes +# these tests use. +STORAGE_USAGE_PLENTY = { + "/dev/ad0s1f": { + "mount": "/var/tmp", + "total": "10G", + "total_blocks": 20971520, + "used": "0B", + "used_blocks": 0, + "used_pct": "0%", + "avail": "10G", + "avail_block": 20971520, + }, +} + DEVICE_FACTS = { "domain": "ntc.com", "hostname": "vmx3", @@ -209,6 +234,7 @@ def test_file_copy(self, mock_scp): local_checksum = "4a8ec4fa5f01b4ab1a0ab8cbccb709f0" self.device.fs.checksum.side_effect = ["", local_checksum] + self.device.fs.storage_usage.return_value = STORAGE_USAGE_PLENTY self.device.file_copy(temp_file.name, "dest") mock_scp.assert_called_with(self.device.native) @@ -388,6 +414,7 @@ def test_remote_file_copy(self, mock_sleep): ) self.device.fs.cp.return_value = True + self.device.fs.storage_usage.return_value = STORAGE_USAGE_PLENTY with mock.patch.object(self.device, "verify_file") as mock_verify_file: with self.subTest("invalid src argument"): @@ -470,5 +497,168 @@ def test_verify_file(self): self.assertTrue(result) +class TestJnprFreeSpace(unittest.TestCase): + """Tests for JunOS pre-transfer free-space verification (NAPPS-1085).""" + + def setUp(self): + self.mock_sw = mock.patch("pyntc.devices.jnpr_device.JunosNativeSW", autospec=True) + self.mock_fs = mock.patch("pyntc.devices.jnpr_device.JunosNativeFS", autospec=True) + self.mock_config = mock.patch("pyntc.devices.jnpr_device.JunosNativeConfig", autospec=True) + self.mock_device = mock.patch("pyntc.devices.jnpr_device.JunosNativeDevice", autospec=True) + + self.mock_sw.start() + self.mock_fs.start() + self.mock_config.start() + self.mock_device.start() + + self.device = JunosDevice("host", "user", "pass") + self.device.native.facts = DEVICE_FACTS + + def tearDown(self): + self.mock_sw.stop() + self.mock_fs.stop() + self.mock_config.stop() + self.mock_device.stop() + + def test_get_free_space_parses_avail_format(self): + """_get_free_space returns the bytes value parsed from the avail format string.""" + self.device.fs.storage_usage.return_value = { + "/dev/ad0s1f": {"mount": "/var/tmp", "avail": "1.0G"}, + } + self.assertEqual(self.device._get_free_space(), 1024**3) + + def test_get_free_space_honours_file_system_argument(self): + """_get_free_space uses the mount argument instead of the /var/tmp default.""" + self.device.fs.storage_usage.return_value = { + "/dev/ad0s1a": {"mount": "/", "avail": "500M"}, + "/dev/ad0s1f": {"mount": "/var/tmp", "avail": "1.0G"}, + } + self.assertEqual(self.device._get_free_space(file_system="/"), 500 * 1024**2) + + def test_get_free_space_falls_back_to_root_mount(self): + """When the requested mount is absent, fall back to the enclosing ``/`` mount.""" + self.device.fs.storage_usage.return_value = { + "/dev/ad0s1a": {"mount": "/", "avail": "500M"}, + } + # /var/tmp is not a mount here; / is the longest prefix and wins. + self.assertEqual(self.device._get_free_space(), 500 * 1024**2) + + def test_get_free_space_longest_prefix_match_wins(self): + """When multiple mounts enclose the path, the most specific one wins.""" + self.device.fs.storage_usage.return_value = { + "/dev/ad0s1a": {"mount": "/", "avail": "500M"}, + "/dev/ad0s1f": {"mount": "/var", "avail": "2.0G"}, + } + # /var/tmp is under /var (not its own mount); /var is more specific than /. + self.assertEqual(self.device._get_free_space(), 2 * 1024**3) + + def test_get_free_space_does_not_match_prefix_that_is_not_a_directory_boundary(self): + """A mount named ``/vari`` must not match ``/var/tmp`` as a prefix.""" + self.device.fs.storage_usage.return_value = { + "/dev/ad0s1a": {"mount": "/vari", "avail": "2.0G"}, + "/dev/ad0s1b": {"mount": "/", "avail": "500M"}, + } + # /var/tmp must fall through to / — it does NOT live under /vari. + self.assertEqual(self.device._get_free_space(), 500 * 1024**2) + + def test_get_free_space_raises_when_no_mount_matches(self): + """_get_free_space raises FileSystemNotFoundError when storage_usage is empty.""" + self.device.fs.storage_usage.return_value = {} + with self.assertRaises(FileSystemNotFoundError): + self.device._get_free_space() + + def test_get_free_space_raises_on_unparseable_avail(self): + """_get_free_space raises CommandError when the avail format cannot be parsed.""" + self.device.fs.storage_usage.return_value = { + "/dev/ad0s1f": {"mount": "/var/tmp", "avail": "garbage"}, + } + with self.assertRaises(CommandError): + self.device._get_free_space() + + @mock.patch("pyntc.devices.jnpr_device.os.path.getsize", return_value=10**12) + @mock.patch("pyntc.devices.jnpr_device.SCP") + def test_file_copy_raises_not_enough_free_space(self, mock_scp, _getsize): + """file_copy raises NotEnoughFreeSpaceError and never runs SCP.put().""" + self.device.fs.checksum.return_value = "" # file_copy_remote_exists -> False + self.device.fs.storage_usage.return_value = { + "/dev/ad0s1f": {"mount": "/var/tmp", "avail": "10M"}, + } + with self.assertRaises(NotEnoughFreeSpaceError): + self.device.file_copy("path/to/image.bin", "dest") + mock_scp.assert_not_called() + + def test_remote_file_copy_raises_not_enough_free_space(self): + """remote_file_copy raises NotEnoughFreeSpaceError and never invokes fs.cp.""" + self.device.fs.storage_usage.return_value = { + "/dev/ad0s1f": {"mount": "/var/tmp", "avail": "10M"}, + } + oversized = FileCopyModel( + download_url="ftp://example.com/file.bin", + checksum="c0ffee", + file_name="file.bin", + file_size=1, + file_size_unit="gigabytes", + ) + with mock.patch.object(self.device, "verify_file", return_value=False): + with self.assertRaises(NotEnoughFreeSpaceError): + self.device.remote_file_copy(oversized, dest="/var/tmp/file.bin") + self.device.fs.cp.assert_not_called() + + def test_remote_file_copy_skips_space_check_when_file_size_omitted(self): + """When FileCopyModel has no file_size, _check_free_space is NOT called.""" + self.device.fs.cp.return_value = True + model = FileCopyModel( + download_url="ftp://example.com/file.bin", + checksum="c0ffee", + file_name="file.bin", + ) + self.assertIsNone(model.file_size_bytes) + with ( + mock.patch.object(self.device, "verify_file", side_effect=[False, True]), + mock.patch.object(JunosDevice, "_check_free_space") as mock_check, + ): + self.device.remote_file_copy(model, dest="/var/tmp/file.bin") + mock_check.assert_not_called() + self.device.fs.cp.assert_called_once() + + def test_remote_file_copy_appends_filename_when_url_has_no_path(self): + """A bare ``ftp://host`` URL gets ``/`` appended before ``fs.cp``.""" + self.device.fs.cp.return_value = True + self.device.fs.storage_usage.return_value = STORAGE_USAGE_PLENTY + model = FileCopyModel( + download_url="ftp://ntc:pw@10.1.100.220", # no path + checksum="c0ffee", + file_name="image.bin", + file_size=1, + file_size_unit="megabytes", + ) + with mock.patch.object(self.device, "verify_file", side_effect=[False, True]): + self.device.remote_file_copy(model, dest="/var/tmp/image.bin") + self.device.fs.cp.assert_called_once_with( + from_path="ftp://ntc:pw@10.1.100.220/image.bin", + to_path="/var/tmp/image.bin", + dev_timeout=mock.ANY, + ) + + def test_remote_file_copy_keeps_url_intact_when_path_is_present(self): + """When the URL already contains a path, ``fs.cp`` receives it unchanged.""" + self.device.fs.cp.return_value = True + self.device.fs.storage_usage.return_value = STORAGE_USAGE_PLENTY + model = FileCopyModel( + download_url="ftp://ntc:pw@10.1.100.220/subdir/image.bin", + checksum="c0ffee", + file_name="image.bin", + file_size=1, + file_size_unit="megabytes", + ) + with mock.patch.object(self.device, "verify_file", side_effect=[False, True]): + self.device.remote_file_copy(model, dest="/var/tmp/image.bin") + self.device.fs.cp.assert_called_once_with( + from_path="ftp://ntc:pw@10.1.100.220/subdir/image.bin", + to_path="/var/tmp/image.bin", + dev_timeout=mock.ANY, + ) + + if __name__ == "__main__": unittest.main() From f8a1ca8ad00b423c5b554e1ed65633f395ee18c2 Mon Sep 17 00:00:00 2001 From: James Williams Date: Wed, 22 Apr 2026 12:28:55 -0500 Subject: [PATCH 2/4] Address PR #373 review feedback from jeffkala and mattmiller87 - Add JUNOS_SUPPORTED_HASHING_ALGORITHMS = {"md5", "sha1", "sha256"} module constant mirroring EOS_SUPPORTED_HASHING_ALGORITHMS and NXOS_SUPPORTED_HASHING_ALGORITHMS, plus a matching guard in get_remote_checksum that raises ValueError before reaching PyEZ. Callers asking for sha512 now fail fast with a clear driver-level message instead of tripping over PyEZ's raw ValueError. (mattmiller87) - Clarify _get_free_space and remote_file_copy docstrings so the ``/var/tmp`` default is attributed to the underlying ``_JUNOS_DEFAULT_FILE_SYSTEM`` constant rather than implying the public signature carries the literal default. (jeffkala) - Rename single-letter ``k, v`` loop variables in JUNOS_PROTOCOL_URL_VARS to ``scheme, env_var``. NTC style prefers descriptive names. (jeffkala) - New unit test covers the ValueError path on an unsupported algo. Co-Authored-By: Claude Opus 4.7 (1M context) --- pyntc/devices/jnpr_device.py | 27 ++++++++++++++++++--- tests/integration/test_jnpr_device.py | 2 +- tests/unit/test_devices/test_jnpr_device.py | 7 ++++++ 3 files changed, 31 insertions(+), 5 deletions(-) diff --git a/pyntc/devices/jnpr_device.py b/pyntc/devices/jnpr_device.py index d654e2fe..13a4189d 100644 --- a/pyntc/devices/jnpr_device.py +++ b/pyntc/devices/jnpr_device.py @@ -47,6 +47,12 @@ # mount point, not a local temp directory). _JUNOS_DEFAULT_FILE_SYSTEM = "/var/tmp" # noqa: S108 +# Hashing algorithms that Junos implements for the ``file checksum`` RPC. +# Junos does NOT implement sha512; callers passing it will be rejected at the +# driver boundary rather than surfacing PyEZ's raw ValueError deeper in the +# stack. Mirrors the pattern used by EOS and NXOS drivers. +JUNOS_SUPPORTED_HASHING_ALGORITHMS = {"md5", "sha1", "sha256"} + def _mount_encloses_path(mount, path): """Return True if ``mount`` is the filesystem that contains ``path``. @@ -119,8 +125,10 @@ def _get_free_space(self, file_system=None): when nothing more specific matches. Args: - file_system (str, optional): Target path. Defaults to ``/var/tmp`` - — the standard destination for ``fs.cp`` copies on Junos. + file_system (str, optional): Target path. When ``None`` (the + default), the probe uses ``_JUNOS_DEFAULT_FILE_SYSTEM`` + (``/var/tmp`` — the standard destination for ``fs.cp`` copies + on Junos). Returns: int: Free bytes available on the resolved filesystem. @@ -626,11 +634,21 @@ def get_remote_checksum(self, filename, hashing_algorithm="md5"): Args: filename (str): The name of the file to check for on the remote device. - hashing_algorithm (str): The hashing algorithm to use. Valid values are 'md5', 'sha1', and 'sha256'. Defaults to 'md5'. + hashing_algorithm (str): The hashing algorithm to use. Valid values are + those in ``JUNOS_SUPPORTED_HASHING_ALGORITHMS`` (``md5``, ``sha1``, + ``sha256``). Defaults to ``md5``. Returns: (str): The checksum of the remote file or None if the file is not found. + + Raises: + ValueError: When ``hashing_algorithm`` is not one Junos implements. """ + if hashing_algorithm.lower() not in JUNOS_SUPPORTED_HASHING_ALGORITHMS: + raise ValueError( + f"Unsupported hashing algorithm '{hashing_algorithm}' for Junos. " + f"Supported algorithms: {sorted(JUNOS_SUPPORTED_HASHING_ALGORITHMS)}" + ) return self.fs.checksum(path=filename, calc=hashing_algorithm) def compare_file_checksum(self, checksum, filename, hashing_algorithm="md5"): @@ -653,7 +671,8 @@ def remote_file_copy(self, src: FileCopyModel = None, dest=None, file_system: st src (FileCopyModel): The source file model. dest (str): The destination file path on the remote device. file_system (str, optional): Mount point used for the pre-transfer - free-space check. Defaults to ``/var/tmp``. + free-space check. When ``None`` (the default), the probe uses + ``_JUNOS_DEFAULT_FILE_SYSTEM`` (``/var/tmp``). **kwargs (Any): Accepted for parity with ``BaseDevice.remote_file_copy``; other drivers may forward extra options. diff --git a/tests/integration/test_jnpr_device.py b/tests/integration/test_jnpr_device.py index 07a40681..8157a404 100644 --- a/tests/integration/test_jnpr_device.py +++ b/tests/integration/test_jnpr_device.py @@ -51,7 +51,7 @@ # Junos ``fs.cp`` does not accept TFTP URLs, so narrow the protocol set before # any protocol-aware fixture/test reads from it. -JUNOS_PROTOCOL_URL_VARS = {k: v for k, v in PROTOCOL_URL_VARS.items() if k != "tftp"} +JUNOS_PROTOCOL_URL_VARS = {scheme: env_var for scheme, env_var in PROTOCOL_URL_VARS.items() if scheme != "tftp"} # --------------------------------------------------------------------------- # Fixtures diff --git a/tests/unit/test_devices/test_jnpr_device.py b/tests/unit/test_devices/test_jnpr_device.py index 9d71a4f4..5b9d325d 100644 --- a/tests/unit/test_devices/test_jnpr_device.py +++ b/tests/unit/test_devices/test_jnpr_device.py @@ -496,6 +496,13 @@ def test_verify_file(self): result = self.device.verify_file(checksum, filename, hashing_algorithm=hashing_algorithm) self.assertTrue(result) + def test_get_remote_checksum_rejects_unsupported_algorithm(self): + """Junos does not implement sha512; the driver rejects it at the boundary.""" + with self.assertRaises(ValueError) as ctx: + self.device.get_remote_checksum("file.bin", hashing_algorithm="sha512") + assert "sha512" in str(ctx.exception) + self.device.fs.checksum.assert_not_called() + class TestJnprFreeSpace(unittest.TestCase): """Tests for JunOS pre-transfer free-space verification (NAPPS-1085).""" From 537fcb8d06991c44755d9186957374145a226ac1 Mon Sep 17 00:00:00 2001 From: James Williams Date: Wed, 22 Apr 2026 12:34:00 -0500 Subject: [PATCH 3/4] Drop integration_hash_algo helper; pin Junos tests to sha256 constant Per @mattmiller87's review: each device should carry its own default algorithm rather than routing through a shared env-var helper. - Delete ``integration_hash_algo()`` from ``tests/integration/_helpers.py``. - ``build_file_copy_model`` now takes ``hashing_algorithm`` as a kwarg (default ``"sha512"`` preserves EOS / ASA behaviour without any caller changes). - ``any_file_copy_model`` fixture in ``conftest.py`` reverts to the literal ``"sha512"`` default. - ``test_jnpr_device.py`` exposes ``JUNOS_INTEGRATION_HASH_ALGO = "sha256"`` at module level and passes it explicitly to every site that builds or verifies a checksum; driver-level validation (added in the prior commit) ensures Junos never sees sha512. - Docstring updated: no more ``FILE_HASH_ALGO`` env var; labs that ship md5 / sha1 binaries edit the module constant. Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/integration/_helpers.py | 20 ++++------------ tests/integration/conftest.py | 4 ++-- tests/integration/test_jnpr_device.py | 33 +++++++++++++++------------ 3 files changed, 25 insertions(+), 32 deletions(-) diff --git a/tests/integration/_helpers.py b/tests/integration/_helpers.py index c0f663d4..5be9a268 100644 --- a/tests/integration/_helpers.py +++ b/tests/integration/_helpers.py @@ -19,23 +19,13 @@ } -def integration_hash_algo(): - """Return the hashing algorithm configured for the current integration run. - - Reads ``FILE_HASH_ALGO`` so a lab can pick whatever its device supports - (Junos SRX, for example, does not implement sha512; sha256 works). - Defaults to ``"sha512"`` for backward compatibility with existing EOS - and ASA integration runs. - """ - return os.environ.get("FILE_HASH_ALGO", "sha512") - - -def build_file_copy_model(url_env_var): +def build_file_copy_model(url_env_var, hashing_algorithm="sha512"): """Build a ``FileCopyModel`` from a per-protocol URL env var. Calls ``pytest.skip`` if the URL, ``FILE_CHECKSUM``, or ``FILE_SIZE`` env - vars are not set. The hashing algorithm defaults to ``FILE_HASH_ALGO`` - (``sha512`` when unset). + vars are not set. ``hashing_algorithm`` defaults to ``"sha512"`` for EOS + / ASA parity; drivers whose devices do not support sha512 (Junos, older + IOS) pass their own supported algorithm explicitly. """ url = os.environ.get(url_env_var) checksum = os.environ.get("FILE_CHECKSUM") @@ -52,7 +42,7 @@ def build_file_copy_model(url_env_var): file_name=file_name, file_size=file_size, file_size_unit=file_size_unit, - hashing_algorithm=integration_hash_algo(), + hashing_algorithm=hashing_algorithm, timeout=900, ) diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 5de93ea6..0ded2a40 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -7,7 +7,7 @@ from pyntc.utils.models import FileCopyModel -from ._helpers import PROTOCOL_URL_VARS, integration_hash_algo +from ._helpers import PROTOCOL_URL_VARS @pytest.fixture(scope="module") @@ -31,7 +31,7 @@ def any_file_copy_model(): file_name=file_name, file_size=file_size, file_size_unit=file_size_unit, - hashing_algorithm=integration_hash_algo(), + hashing_algorithm="sha512", timeout=900, ) pytest.skip("No protocol URL / FILE_CHECKSUM / FILE_SIZE environment variables not set") diff --git a/tests/integration/test_jnpr_device.py b/tests/integration/test_jnpr_device.py index 8157a404..c885b036 100644 --- a/tests/integration/test_jnpr_device.py +++ b/tests/integration/test_jnpr_device.py @@ -11,8 +11,7 @@ export SCP_URL=scp://:@:2222/ export HTTP_URL=http://:@:8081/ export HTTPS_URL=https://:@:8443/ - export FILE_CHECKSUM= - export FILE_HASH_ALGO=sha256 # Junos doesn't implement sha512; sha256 is typical + export FILE_CHECKSUM= export FILE_SIZE= export FILE_SIZE_UNIT=megabytes # optional; defaults to "bytes" poetry run pytest tests/integration/test_jnpr_device.py -v @@ -29,10 +28,9 @@ HTTP_URL - HTTP URL of the file to transfer HTTPS_URL - HTTPS URL of the file to transfer FILE_NAME - Destination filename on the device (default: basename of URL path) - FILE_CHECKSUM - Expected checksum of the file (algorithm set by FILE_HASH_ALGO) - FILE_HASH_ALGO - Hashing algorithm (md5 / sha1 / sha256). Defaults to sha512 for - cross-platform parity but Junos does not implement sha512; set to - sha256 for SRX/MX runs. + FILE_CHECKSUM - Expected sha256 checksum of the file (Junos does not implement sha512; + the hashing algorithm is pinned to sha256 via ``JUNOS_INTEGRATION_HASH_ALGO`` + at module level — edit that constant for md5 / sha1 labs) FILE_SIZE - Expected size of the file expressed in FILE_SIZE_UNIT units; used for the pre-transfer free-space check FILE_SIZE_UNIT - One of "bytes", "megabytes", or "gigabytes" (default: "bytes") @@ -47,12 +45,17 @@ from pyntc.errors import NotEnoughFreeSpaceError from pyntc.utils.models import FILE_SIZE_UNITS, FileCopyModel -from ._helpers import PROTOCOL_URL_VARS, build_file_copy_model, first_available_url, integration_hash_algo +from ._helpers import PROTOCOL_URL_VARS, build_file_copy_model, first_available_url # Junos ``fs.cp`` does not accept TFTP URLs, so narrow the protocol set before # any protocol-aware fixture/test reads from it. JUNOS_PROTOCOL_URL_VARS = {scheme: env_var for scheme, env_var in PROTOCOL_URL_VARS.items() if scheme != "tftp"} +# Junos ``file checksum`` RPC does not implement sha512. The integration run is +# pinned to sha256; labs that ship ``md5`` / ``sha1`` binaries edit this constant +# and regenerate ``FILE_CHECKSUM`` accordingly. +JUNOS_INTEGRATION_HASH_ALGO = "sha256" + # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @@ -97,7 +100,7 @@ def test_check_file_exists_false(device, any_file_copy_model): def test_remote_file_copy_ftp(device): """Transfer the file using FTP and verify it exists on the device.""" - model = build_file_copy_model("FTP_URL") + model = build_file_copy_model("FTP_URL", hashing_algorithm=JUNOS_INTEGRATION_HASH_ALGO) dest = _junos_dest(model.file_name) device.remote_file_copy(model, dest=dest) assert device.check_file_exists(dest) @@ -105,7 +108,7 @@ def test_remote_file_copy_ftp(device): def test_remote_file_copy_scp(device): """Transfer the file using SCP and verify it exists on the device.""" - model = build_file_copy_model("SCP_URL") + model = build_file_copy_model("SCP_URL", hashing_algorithm=JUNOS_INTEGRATION_HASH_ALGO) dest = _junos_dest(model.file_name) device.remote_file_copy(model, dest=dest) assert device.check_file_exists(dest) @@ -113,7 +116,7 @@ def test_remote_file_copy_scp(device): def test_remote_file_copy_http(device): """Transfer the file using HTTP and verify it exists on the device.""" - model = build_file_copy_model("HTTP_URL") + model = build_file_copy_model("HTTP_URL", hashing_algorithm=JUNOS_INTEGRATION_HASH_ALGO) dest = _junos_dest(model.file_name) device.remote_file_copy(model, dest=dest) assert device.check_file_exists(dest) @@ -121,7 +124,7 @@ def test_remote_file_copy_http(device): def test_remote_file_copy_https(device): """Transfer the file using HTTPS and verify it exists on the device.""" - model = build_file_copy_model("HTTPS_URL") + model = build_file_copy_model("HTTPS_URL", hashing_algorithm=JUNOS_INTEGRATION_HASH_ALGO) dest = _junos_dest(model.file_name) device.remote_file_copy(model, dest=dest) assert device.check_file_exists(dest) @@ -132,7 +135,7 @@ def test_verify_file_after_copy(device, any_file_copy_model): dest = _junos_dest(any_file_copy_model.file_name) if not device.check_file_exists(dest): pytest.skip("File does not exist on device; run a copy test first") - assert device.verify_file(any_file_copy_model.checksum, dest, hashing_algorithm=integration_hash_algo()) + assert device.verify_file(any_file_copy_model.checksum, dest, hashing_algorithm=JUNOS_INTEGRATION_HASH_ALGO) # --------------------------------------------------------------------------- @@ -191,7 +194,7 @@ def test_remote_file_copy_rejects_oversized_transfer(device): file_name=unique_name, file_size=oversized_gb, file_size_unit="gigabytes", - hashing_algorithm=integration_hash_algo(), + hashing_algorithm=JUNOS_INTEGRATION_HASH_ALGO, timeout=60, ) @@ -208,7 +211,7 @@ def test_remote_file_copy_accepts_declared_size_within_free_space(device): scheme, _url = first_available_url(JUNOS_PROTOCOL_URL_VARS) if scheme is None: pytest.skip("No protocol URL environment variables set") - model = build_file_copy_model(JUNOS_PROTOCOL_URL_VARS[scheme]) + model = build_file_copy_model(JUNOS_PROTOCOL_URL_VARS[scheme], hashing_algorithm=JUNOS_INTEGRATION_HASH_ALGO) # pylint: disable=protected-access free_bytes = device._get_free_space() assert model.file_size_bytes <= free_bytes, ( @@ -231,7 +234,7 @@ def test_remote_file_copy_skips_space_check_when_file_size_omitted(device): download_url=url, checksum=checksum, file_name=file_name, - hashing_algorithm=integration_hash_algo(), + hashing_algorithm=JUNOS_INTEGRATION_HASH_ALGO, timeout=60, ) # file_size intentionally omitted assert model.file_size is None From 7bf9c94c6e4504e65713334361ed077221fd8478 Mon Sep 17 00:00:00 2001 From: James Williams Date: Wed, 22 Apr 2026 13:07:45 -0500 Subject: [PATCH 4/4] Auto-select integration hash algo per platform in conftest.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Supports the ``run every device's integration suite in one pytest invocation'' workflow: each platform test module maps to the hashing algorithm its device family implements, and an autouse module-scoped fixture copies the matching ``FILE_CHECKSUM_`` env var into ``FILE_CHECKSUM`` (and ``FILE_HASH_ALGO`` to the algo name) before that module's tests run. The env is restored on teardown. - ``tests/integration/conftest.py`` adds ``_PLATFORM_HASH_ALGOS`` (module name → algo) and ``_HASH_ALGO_ENV_SUFFIXES`` (algo → env suffix), plus the ``_configure_integration_env`` autouse fixture that applies the pair for each module. When the suffix env var is missing the fixture leaves both vars untouched so the user's own shell-level pair wins (or ``build_file_copy_model`` skips cleanly on no env) — guards against the silent algo/checksum mismatch that would otherwise fail later at ``verify_file``. - ``tests/integration/_helpers.py`` simplifies ``build_file_copy_model``: it now reads ``FILE_HASH_ALGO`` and ``FILE_CHECKSUM`` from the environment (the autouse fixture populates both), dropping the ``hashing_algorithm`` kwarg it grew in the previous commit. - ``tests/integration/test_jnpr_device.py`` drops the ``JUNOS_INTEGRATION_HASH_ALGO`` module constant — the autouse fixture now sets ``FILE_HASH_ALGO="sha256"`` for this module. Direct ``FileCopyModel(...)`` constructions read the algo from env; ``test_verify_file_after_copy`` uses ``any_file_copy_model.hashing_algorithm`` instead of a hardcode. Docstring updated to reference ``FILE_CHECKSUM_256``. User setup: export ``FILE_CHECKSUM_512`` / ``FILE_CHECKSUM_256`` / ``FILE_CHECKSUM_MD5`` once; ``FILE_HASH_ALGO`` and ``FILE_CHECKSUM`` no longer need to live in .env at all. Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/integration/_helpers.py | 12 +++-- tests/integration/conftest.py | 73 ++++++++++++++++++++++++++- tests/integration/test_jnpr_device.py | 57 +++++++++++---------- 3 files changed, 108 insertions(+), 34 deletions(-) diff --git a/tests/integration/_helpers.py b/tests/integration/_helpers.py index 5be9a268..5ba7ff57 100644 --- a/tests/integration/_helpers.py +++ b/tests/integration/_helpers.py @@ -19,16 +19,18 @@ } -def build_file_copy_model(url_env_var, hashing_algorithm="sha512"): +def build_file_copy_model(url_env_var): """Build a ``FileCopyModel`` from a per-protocol URL env var. - Calls ``pytest.skip`` if the URL, ``FILE_CHECKSUM``, or ``FILE_SIZE`` env - vars are not set. ``hashing_algorithm`` defaults to ``"sha512"`` for EOS - / ASA parity; drivers whose devices do not support sha512 (Junos, older - IOS) pass their own supported algorithm explicitly. + Reads ``FILE_HASH_ALGO`` and ``FILE_CHECKSUM`` from the environment. + An autouse fixture in ``conftest.py`` sets both to the running test + module's platform default before each module runs, so individual + tests never have to hardcode an algorithm. Calls ``pytest.skip`` when + any required env var is missing. """ url = os.environ.get(url_env_var) checksum = os.environ.get("FILE_CHECKSUM") + hashing_algorithm = os.environ.get("FILE_HASH_ALGO", "sha512") file_name = os.environ.get("FILE_NAME") or (posixpath.basename(url.split("?")[0]) if url else None) file_size = int(os.environ.get("FILE_SIZE", "0")) file_size_unit = os.environ.get("FILE_SIZE_UNIT", "bytes") diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 0ded2a40..1a17977e 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -9,6 +9,76 @@ from ._helpers import PROTOCOL_URL_VARS +# Each driver's integration test module is mapped to the hashing algorithm +# its device family implements. ``conftest.py`` owns this mapping (rather +# than per-module constants or .env entries) so the user can run every +# driver's integration suite in one ``pytest tests/integration`` invocation +# and each module automatically picks up the algorithm its hardware +# supports. Junos does not implement sha512 — sha256 is typical on SRX / +# MX. EOS and ASA both implement sha512. Extend the map as new drivers get +# integration tests. +_PLATFORM_HASH_ALGOS = { + "test_eos_device": "sha512", + "test_asa_device": "sha512", + "test_jnpr_device": "sha256", + "test_ios_device": "md5", + "test_nxos_device": "sha256", +} + +# Maps each hashing algorithm to the suffix convention used on the +# per-algorithm checksum env vars (``FILE_CHECKSUM_512`` / ``_256`` / +# ``_MD5``). The user exports one checksum per algorithm once and +# ``_configure_integration_env`` copies the right one into +# ``FILE_CHECKSUM`` before the module runs. +_HASH_ALGO_ENV_SUFFIXES = {"sha512": "512", "sha256": "256", "md5": "MD5"} + + +@pytest.fixture(scope="module", autouse=True) +def _configure_integration_env(request): + """Set ``FILE_HASH_ALGO`` / ``FILE_CHECKSUM`` per test module. + + Resolves the module-specific hashing algorithm from + ``_PLATFORM_HASH_ALGOS``, copies the matching ``FILE_CHECKSUM_`` + env var into ``FILE_CHECKSUM``, and restores any prior values when + the module finishes. Test files not listed in the map are left alone — + they either carry no hashing dependency or set their own env. + """ + module_name = request.module.__name__.split(".")[-1] + algo = _PLATFORM_HASH_ALGOS.get(module_name) + if algo is None: + yield + return + + suffix = _HASH_ALGO_ENV_SUFFIXES.get(algo) + checksum = os.environ.get(f"FILE_CHECKSUM_{suffix}") if suffix else None + + # Skip the env overwrite entirely when the suffix env var is missing. + # Otherwise we would pin FILE_HASH_ALGO to the platform's algo while + # leaving FILE_CHECKSUM inherited from the shell (possibly a different + # algo's hash) — tests would then fail at verify with a confusing + # mismatch. Leaving both vars alone lets the user's own pair win, or + # lets ``build_file_copy_model`` skip cleanly on missing env. + if checksum is None: + yield + return + + prior_algo = os.environ.get("FILE_HASH_ALGO") + prior_checksum = os.environ.get("FILE_CHECKSUM") + + os.environ["FILE_HASH_ALGO"] = algo + os.environ["FILE_CHECKSUM"] = checksum + + yield + + if prior_algo is None: + os.environ.pop("FILE_HASH_ALGO", None) + else: + os.environ["FILE_HASH_ALGO"] = prior_algo + if prior_checksum is None: + os.environ.pop("FILE_CHECKSUM", None) + else: + os.environ["FILE_CHECKSUM"] = prior_checksum + @pytest.fixture(scope="module") def any_file_copy_model(): @@ -19,6 +89,7 @@ def any_file_copy_model(): protocol URL / ``FILE_CHECKSUM`` / ``FILE_SIZE`` env vars are set. """ checksum = os.environ.get("FILE_CHECKSUM") + checksum_algo = os.environ.get("FILE_HASH_ALGO", "sha512") file_size = int(os.environ.get("FILE_SIZE", "0")) file_size_unit = os.environ.get("FILE_SIZE_UNIT", "bytes") for env_var in PROTOCOL_URL_VARS.values(): @@ -31,7 +102,7 @@ def any_file_copy_model(): file_name=file_name, file_size=file_size, file_size_unit=file_size_unit, - hashing_algorithm="sha512", + hashing_algorithm=checksum_algo, timeout=900, ) pytest.skip("No protocol URL / FILE_CHECKSUM / FILE_SIZE environment variables not set") diff --git a/tests/integration/test_jnpr_device.py b/tests/integration/test_jnpr_device.py index c885b036..85bb9aac 100644 --- a/tests/integration/test_jnpr_device.py +++ b/tests/integration/test_jnpr_device.py @@ -11,7 +11,7 @@ export SCP_URL=scp://:@:2222/ export HTTP_URL=http://:@:8081/ export HTTPS_URL=https://:@:8443/ - export FILE_CHECKSUM= + export FILE_CHECKSUM_256= export FILE_SIZE= export FILE_SIZE_UNIT=megabytes # optional; defaults to "bytes" poetry run pytest tests/integration/test_jnpr_device.py -v @@ -20,20 +20,20 @@ protocol test will skip automatically if its URL is not set. Environment variables: - JUNOS_HOST - IP address or hostname of the lab Junos device - JUNOS_USER - NETCONF / SSH username - JUNOS_PASS - NETCONF / SSH password - FTP_URL - FTP URL of the file to transfer - SCP_URL - SCP URL of the file to transfer - HTTP_URL - HTTP URL of the file to transfer - HTTPS_URL - HTTPS URL of the file to transfer - FILE_NAME - Destination filename on the device (default: basename of URL path) - FILE_CHECKSUM - Expected sha256 checksum of the file (Junos does not implement sha512; - the hashing algorithm is pinned to sha256 via ``JUNOS_INTEGRATION_HASH_ALGO`` - at module level — edit that constant for md5 / sha1 labs) - FILE_SIZE - Expected size of the file expressed in FILE_SIZE_UNIT units; used for - the pre-transfer free-space check - FILE_SIZE_UNIT - One of "bytes", "megabytes", or "gigabytes" (default: "bytes") + JUNOS_HOST - IP address or hostname of the lab Junos device + JUNOS_USER - NETCONF / SSH username + JUNOS_PASS - NETCONF / SSH password + FTP_URL - FTP URL of the file to transfer + SCP_URL - SCP URL of the file to transfer + HTTP_URL - HTTP URL of the file to transfer + HTTPS_URL - HTTPS URL of the file to transfer + FILE_NAME - Destination filename on the device (default: basename of URL path) + FILE_CHECKSUM_256 - Expected sha256 checksum of the file. ``conftest.py`` maps Junos to + sha256 and copies this into ``FILE_CHECKSUM`` for the module — Junos + does not implement sha512. + FILE_SIZE - Expected size of the file expressed in FILE_SIZE_UNIT units; used for + the pre-transfer free-space check + FILE_SIZE_UNIT - One of "bytes", "megabytes", or "gigabytes" (default: "bytes") """ import os @@ -51,11 +51,6 @@ # any protocol-aware fixture/test reads from it. JUNOS_PROTOCOL_URL_VARS = {scheme: env_var for scheme, env_var in PROTOCOL_URL_VARS.items() if scheme != "tftp"} -# Junos ``file checksum`` RPC does not implement sha512. The integration run is -# pinned to sha256; labs that ship ``md5`` / ``sha1`` binaries edit this constant -# and regenerate ``FILE_CHECKSUM`` accordingly. -JUNOS_INTEGRATION_HASH_ALGO = "sha256" - # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @@ -100,7 +95,7 @@ def test_check_file_exists_false(device, any_file_copy_model): def test_remote_file_copy_ftp(device): """Transfer the file using FTP and verify it exists on the device.""" - model = build_file_copy_model("FTP_URL", hashing_algorithm=JUNOS_INTEGRATION_HASH_ALGO) + model = build_file_copy_model("FTP_URL") dest = _junos_dest(model.file_name) device.remote_file_copy(model, dest=dest) assert device.check_file_exists(dest) @@ -108,7 +103,7 @@ def test_remote_file_copy_ftp(device): def test_remote_file_copy_scp(device): """Transfer the file using SCP and verify it exists on the device.""" - model = build_file_copy_model("SCP_URL", hashing_algorithm=JUNOS_INTEGRATION_HASH_ALGO) + model = build_file_copy_model("SCP_URL") dest = _junos_dest(model.file_name) device.remote_file_copy(model, dest=dest) assert device.check_file_exists(dest) @@ -116,7 +111,7 @@ def test_remote_file_copy_scp(device): def test_remote_file_copy_http(device): """Transfer the file using HTTP and verify it exists on the device.""" - model = build_file_copy_model("HTTP_URL", hashing_algorithm=JUNOS_INTEGRATION_HASH_ALGO) + model = build_file_copy_model("HTTP_URL") dest = _junos_dest(model.file_name) device.remote_file_copy(model, dest=dest) assert device.check_file_exists(dest) @@ -124,7 +119,7 @@ def test_remote_file_copy_http(device): def test_remote_file_copy_https(device): """Transfer the file using HTTPS and verify it exists on the device.""" - model = build_file_copy_model("HTTPS_URL", hashing_algorithm=JUNOS_INTEGRATION_HASH_ALGO) + model = build_file_copy_model("HTTPS_URL") dest = _junos_dest(model.file_name) device.remote_file_copy(model, dest=dest) assert device.check_file_exists(dest) @@ -135,7 +130,11 @@ def test_verify_file_after_copy(device, any_file_copy_model): dest = _junos_dest(any_file_copy_model.file_name) if not device.check_file_exists(dest): pytest.skip("File does not exist on device; run a copy test first") - assert device.verify_file(any_file_copy_model.checksum, dest, hashing_algorithm=JUNOS_INTEGRATION_HASH_ALGO) + assert device.verify_file( + any_file_copy_model.checksum, + dest, + hashing_algorithm=any_file_copy_model.hashing_algorithm, + ) # --------------------------------------------------------------------------- @@ -177,6 +176,7 @@ def test_file_size_unit_conversion_matches_device_free_space(device): def test_remote_file_copy_rejects_oversized_transfer(device): """remote_file_copy raises NotEnoughFreeSpaceError and never copies the file.""" checksum = os.environ.get("FILE_CHECKSUM") + hashing_algorithm = os.environ.get("FILE_HASH_ALGO", "sha512") scheme, url = first_available_url(JUNOS_PROTOCOL_URL_VARS) if not (url and checksum): pytest.skip("No protocol URL / FILE_CHECKSUM environment variables set") @@ -194,7 +194,7 @@ def test_remote_file_copy_rejects_oversized_transfer(device): file_name=unique_name, file_size=oversized_gb, file_size_unit="gigabytes", - hashing_algorithm=JUNOS_INTEGRATION_HASH_ALGO, + hashing_algorithm=hashing_algorithm, timeout=60, ) @@ -211,7 +211,7 @@ def test_remote_file_copy_accepts_declared_size_within_free_space(device): scheme, _url = first_available_url(JUNOS_PROTOCOL_URL_VARS) if scheme is None: pytest.skip("No protocol URL environment variables set") - model = build_file_copy_model(JUNOS_PROTOCOL_URL_VARS[scheme], hashing_algorithm=JUNOS_INTEGRATION_HASH_ALGO) + model = build_file_copy_model(JUNOS_PROTOCOL_URL_VARS[scheme]) # pylint: disable=protected-access free_bytes = device._get_free_space() assert model.file_size_bytes <= free_bytes, ( @@ -225,6 +225,7 @@ def test_remote_file_copy_accepts_declared_size_within_free_space(device): def test_remote_file_copy_skips_space_check_when_file_size_omitted(device): """When FileCopyModel has no file_size, _check_free_space is never called.""" checksum = os.environ.get("FILE_CHECKSUM") + hashing_algorithm = os.environ.get("FILE_HASH_ALGO", "sha512") file_name = os.environ.get("FILE_NAME") _, url = first_available_url(JUNOS_PROTOCOL_URL_VARS) if not (url and checksum and file_name): @@ -234,7 +235,7 @@ def test_remote_file_copy_skips_space_check_when_file_size_omitted(device): download_url=url, checksum=checksum, file_name=file_name, - hashing_algorithm=JUNOS_INTEGRATION_HASH_ALGO, + hashing_algorithm=hashing_algorithm, timeout=60, ) # file_size intentionally omitted assert model.file_size is None