Add ASA pre-transfer free-space check (NAPPS-1087)

changes/372.added

@@ -0,0 +1 @@
+Added a pre-transfer free-space check to Cisco ASA ``file_copy`` and ``remote_file_copy`` that raises ``NotEnoughFreeSpaceError`` when the target filesystem lacks room for the image.

pyntc/devices/asa_device.py

@@ -138,6 +138,39 @@ def _get_file_system(self):
         log.debug("Host %s: File system %s.", self.host, file_system)
         return file_system
 
+    def _get_free_space(self, file_system=None):  # pylint: disable=unused-argument
+        """Return free bytes on ``file_system`` as reported by ASA's ``dir`` output.
+
+        ASA exposes a single flash filesystem in practice and ``dir`` always prints
+        a ``<total> bytes total (<free> bytes free...)`` trailer as the last line.
+        Real platforms append an ``/<pct>% free`` suffix inside the parentheses
+        (e.g., ``(3580170240 bytes free/86% free)``); older releases and some
+        emulators omit it. The regex matches both shapes. The ``file_system``
+        argument is accepted for API parity with other drivers but is otherwise
+        unused.
+
+        Args:
+            file_system (str, optional): Ignored; retained for BaseDevice API parity.
+
+        Returns:
+            int: Free bytes available on the target filesystem.
+
+        Raises:
+            CommandError: When the ``dir`` output does not contain a parseable
+                ``N bytes free`` trailer.
+        """
+        raw_data = self.show("dir")
+        # Examples seen in the wild:
+        #   16777216 bytes total (1592488 bytes free)
+        #   4118732800 bytes total (3580170240 bytes free/86% free)
+        match = re.search(r"\((\d+)\s+bytes\s+free", raw_data)
+        if match is None:
+            log.error("Host %s: could not parse free space from 'dir' output.", self.host)
+            raise CommandError(command="dir", message="Unable to parse free space from dir output.")
+        free_bytes = int(match.group(1))
+        log.debug("Host %s: %s bytes free on flash.", self.host, free_bytes)
+        return free_bytes
+
     def _get_ipv4_addresses(self, host: str) -> Dict[str, List[IPv4Address]]:
         """
         Get IPv4 Addresses for ``host``.
@@ -555,6 +588,8 @@ def file_copy(
 
         Raises:
             FileTransferError: When the ``src`` file is unable to transfer the file to any device.
+            NotEnoughFreeSpaceError: When ``file_system`` has fewer free bytes than
+                ``src`` requires.
 
         Example:
             >>> dev = ASADevice(**connection_args)
@@ -566,6 +601,8 @@ def file_copy(
         if file_system is None:
             file_system = self._get_file_system()
 
+        self._check_free_space(os.path.getsize(src), file_system=file_system)
+
         # netmiko's enable_scp
         self.enable_scp()
         self._file_copy(src, dest, file_system)
@@ -1018,6 +1055,10 @@ def remote_file_copy(self, src: FileCopyModel = None, dest=None, **kwargs: Any):
         Raises:
             TypeError: If ``src`` is not a ``FileCopyModel`` instance.
             FileTransferError: If the transfer fails or the file cannot be verified afterwards.
+            NotEnoughFreeSpaceError: If ``src.file_size_bytes`` is set and
+                ``file_system`` 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")
@@ -1028,6 +1069,7 @@ def remote_file_copy(self, src: FileCopyModel = None, dest=None, **kwargs: Any):
             dest = src.file_name
 
         if not self.verify_file(src.checksum, dest, hashing_algorithm=src.hashing_algorithm, file_system=file_system):
+            self._pre_transfer_space_check(src, file_system=file_system)
             current_prompt = self.native.find_prompt()
             prompt_answers = {
                 r"Password": src.token or "",

tests/integration/test_asa_device.py

@@ -14,6 +14,8 @@
     export HTTP_URL=http://<http_user>:<http_password>@<server_ip>:8081/<file_name>
     export HTTPS_URL=https://<https_user>:<https_password>@<server_ip>:8443/<file_name>
     export FILE_CHECKSUM=<sha512_hash>
+    export FILE_SIZE=<image_size>
+    export FILE_SIZE_UNIT=megabytes  # optional; defaults to "bytes"
     poetry run pytest tests/integration/test_asa_device.py -v
 
 Set only the protocol URL vars for the servers you have available; each
@@ -31,15 +33,21 @@
     HTTPS_URL       - HTTPS URL of the file to transfer
     FILE_NAME       - Destination filename on the device (default: basename of URL path)
     FILE_CHECKSUM   - Expected sha512 checksum of the file (shared across all protocols)
+    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 ASADevice
+from pyntc.errors import NotEnoughFreeSpaceError
+from pyntc.utils.models import FILE_SIZE_UNITS, FileCopyModel
 
-from ._helpers import build_file_copy_model
+from ._helpers import PROTOCOL_URL_VARS, build_file_copy_model, first_available_url
 
 # ---------------------------------------------------------------------------
 # Fixtures
@@ -127,3 +135,110 @@ def test_verify_file_after_copy(device, any_file_copy_model):
     if not device.check_file_exists(any_file_copy_model.file_name):
         pytest.skip("File does not exist on device; run a copy test first")
     assert device.verify_file(any_file_copy_model.checksum, any_file_copy_model.file_name, hashing_algorithm="sha512")
+
+
+# ---------------------------------------------------------------------------
+# Free-space / pre-transfer tests (NAPPS-1087)
+# ---------------------------------------------------------------------------
+
+
+def test_get_free_space_returns_positive_int(device):
+    """``_get_free_space`` parses the ``dir`` trailer into a positive int."""
+    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()
+    if not (url and checksum):
+        pytest.skip("No protocol URL / FILE_CHECKSUM environment variables not 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"
+    model = FileCopyModel(
+        download_url=url,
+        checksum=checksum,
+        file_name=unique_name,
+        file_size=oversized_gb,
+        file_size_unit="gigabytes",
+        hashing_algorithm="sha512",
+        timeout=60,
+    )
+
+    assert not device.check_file_exists(unique_name), "Unique filename unexpectedly exists before test"
+
+    with pytest.raises(NotEnoughFreeSpaceError):
+        device.remote_file_copy(model)
+
+    assert not device.check_file_exists(unique_name)
+
+
+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()
+    if scheme is None:
+        pytest.skip("No protocol URL environment variables set")
+    model = build_file_copy_model(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"
+    )
+    device.remote_file_copy(model)
+    assert device.check_file_exists(model.file_name)
+
+
+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()
+    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="sha512",
+        timeout=60,
+    )  # file_size intentionally omitted
+    assert model.file_size is None
+    assert model.file_size_bytes is None
+
+    with mock.patch.object(ASADevice, "_check_free_space") as spy:
+        device.remote_file_copy(model)
+
+    spy.assert_not_called()
+    assert device.check_file_exists(model.file_name)

tests/unit/test_devices/device_mocks/asa/send_command/dir

@@ -2,4 +2,4 @@ Directory of disk0:/
 
 1  -rw-    15183868                asa9-12-3-11-smp-k8.bin
 
-16777216 bytes total (1592488 bytes free)
+4118732800 bytes total (3580170240 bytes free/86% free)

tests/unit/test_devices/test_asa_device.py

@@ -6,7 +6,7 @@
 
 from pyntc.devices import ASADevice
 from pyntc.devices import asa_device as asa_module
-from pyntc.errors import FileTransferError
+from pyntc.errors import CommandError, FileTransferError, NotEnoughFreeSpaceError
 from pyntc.utils.models import FileCopyModel
 
 from .device_mocks.asa import send_command
@@ -536,11 +536,15 @@ def test_enable_scp_enable_fail(mock_log, mock_save, mock_config, mock_peer_devi
     mock_save.assert_not_called()
 
 
+@mock.patch("pyntc.devices.asa_device.os.path.getsize", return_value=1024)
+@mock.patch.object(ASADevice, "_check_free_space")
 @mock.patch.object(os.path, "basename", return_value="a.txt")
 @mock.patch.object(ASADevice, "_get_file_system", return_value="flash:")
 @mock.patch.object(ASADevice, "enable_scp")
 @mock.patch.object(ASADevice, "_file_copy")
-def test_file_copy_no_peer_no_args(mock_file_copy, mock_enable_scp, mock_get_file_system, mock_basename, asa_device):
+def test_file_copy_no_peer_no_args(
+    mock_file_copy, mock_enable_scp, mock_get_file_system, mock_basename, _check_space, _getsize, asa_device
+):
     asa_device.file_copy("path/to/a.txt")
     mock_basename.assert_called()
     mock_get_file_system.assert_called()
@@ -549,11 +553,15 @@ def test_file_copy_no_peer_no_args(mock_file_copy, mock_enable_scp, mock_get_fil
     mock_file_copy.assert_called_with("path/to/a.txt", "a.txt", "flash:")
 
 
+@mock.patch("pyntc.devices.asa_device.os.path.getsize", return_value=1024)
+@mock.patch.object(ASADevice, "_check_free_space")
 @mock.patch.object(os.path, "basename")
 @mock.patch.object(ASADevice, "_get_file_system")
 @mock.patch.object(ASADevice, "enable_scp")
 @mock.patch.object(ASADevice, "_file_copy")
-def test_file_copy_no_peer_pass_args(mock_file_copy, mock_enable_scp, mock_get_file_system, mock_basename, asa_device):
+def test_file_copy_no_peer_pass_args(
+    mock_file_copy, mock_enable_scp, mock_get_file_system, mock_basename, _check_space, _getsize, asa_device
+):
     args = ("path/to/a.txt", "b.txt", "bootflash:")
     asa_device.file_copy(*args)
     mock_basename.assert_not_called()
@@ -563,13 +571,22 @@ def test_file_copy_no_peer_pass_args(mock_file_copy, mock_enable_scp, mock_get_f
     mock_file_copy.assert_called_with(*args)
 
 
+@mock.patch("pyntc.devices.asa_device.os.path.getsize", return_value=1024)
+@mock.patch.object(ASADevice, "_check_free_space")
 @mock.patch.object(os.path, "basename")
 @mock.patch.object(ASADevice, "_get_file_system")
 @mock.patch.object(ASADevice, "enable_scp")
 @mock.patch.object(ASADevice, "_file_copy")
 @mock.patch.object(ASADevice, "peer_device")
 def test_file_copy_include_peer(
-    mock_peer_device, mock_file_copy, mock_enable_scp, mock_get_file_system, mock_basename, asa_device
+    mock_peer_device,
+    mock_file_copy,
+    mock_enable_scp,
+    mock_get_file_system,
+    mock_basename,
+    _check_space,
+    _getsize,
+    asa_device,
 ):
     mock_peer_device.return_value = asa_device
     args = ("path/to/a.txt", "a.txt", "flash:")
@@ -1415,3 +1432,96 @@ def test_remote_file_copy_https_clean_url_used_in_command(mock_verify, mock_fs,
     call_args = asa_device.native.send_command.call_args_list[0][0][0]
     assert "example-user:example-password" not in call_args
     assert "https://192.0.2.1/asa.bin" in call_args
+
+
+# ---------------------------------------------------------------------------
+# Pre-transfer free-space tests (NAPPS-1087)
+# ---------------------------------------------------------------------------
+
+DIR_OUTPUT_WITH_TRAILER = (
+    "Directory of disk0:/\n\n"
+    "1  -rw-    15183868                asa9-12-3-11-smp-k8.bin\n\n"
+    "4118732800 bytes total (3580170240 bytes free/86% free)"
+)
+DIR_OUTPUT_LEGACY_TRAILER = (
+    "Directory of disk0:/\n\n"
+    "1  -rw-    15183868                asa9-12-3-11-smp-k8.bin\n\n"
+    "16777216 bytes total (1592488 bytes free)"
+)
+
+
+@mock.patch.object(ASADevice, "show", return_value=DIR_OUTPUT_WITH_TRAILER)
+def test_get_free_space_parses_dir_trailer(_mock_show, asa_device):
+    """_get_free_space returns the bytes-free value from the real-device dir trailer."""
+    assert asa_device._get_free_space() == 3580170240
+
+
+@mock.patch.object(ASADevice, "show", return_value=DIR_OUTPUT_LEGACY_TRAILER)
+def test_get_free_space_parses_legacy_dir_trailer(_mock_show, asa_device):
+    """_get_free_space parses the older ``(N bytes free)`` trailer shape."""
+    assert asa_device._get_free_space() == 1592488
+
+
+@mock.patch.object(ASADevice, "show", return_value="Directory of disk0:/\nno trailer here")
+def test_get_free_space_raises_when_trailer_missing(_mock_show, asa_device):
+    """_get_free_space raises CommandError when the trailer can't be parsed."""
+    with pytest.raises(CommandError):
+        asa_device._get_free_space()
+
+
+@mock.patch("pyntc.devices.asa_device.os.path.getsize", return_value=10**12)
+@mock.patch.object(ASADevice, "_get_file_system", return_value="disk0:")
+@mock.patch.object(ASADevice, "show", return_value=DIR_OUTPUT_WITH_TRAILER)
+@mock.patch.object(ASADevice, "enable_scp")
+@mock.patch.object(ASADevice, "_file_copy")
+def test_file_copy_raises_not_enough_free_space(mock_file_copy, mock_enable_scp, _show, _fs, _getsize, asa_device):
+    """file_copy raises NotEnoughFreeSpaceError and never runs SCP transfer."""
+    with pytest.raises(NotEnoughFreeSpaceError):
+        asa_device.file_copy("path/to/image.bin")
+
+    mock_enable_scp.assert_not_called()
+    mock_file_copy.assert_not_called()
+
+
+@mock.patch.object(ASADevice, "_get_file_system", return_value="disk0:")
+@mock.patch.object(ASADevice, "verify_file", return_value=False)
+@mock.patch.object(ASADevice, "show", return_value=DIR_OUTPUT_WITH_TRAILER)
+def test_remote_file_copy_raises_not_enough_free_space(_show, _verify, _fs, asa_device):
+    """remote_file_copy raises NotEnoughFreeSpaceError and never issues a copy command."""
+    oversized = FileCopyModel(
+        download_url="ftp://192.0.2.1/asa.bin",
+        checksum=SHA512_CHECKSUM,
+        file_name="asa.bin",
+        file_size=10,
+        file_size_unit="gigabytes",
+        hashing_algorithm="sha512",
+    )
+
+    with pytest.raises(NotEnoughFreeSpaceError):
+        asa_device.remote_file_copy(oversized)
+
+    asa_device.native.send_command.assert_not_called()
+
+
+@mock.patch.object(ASADevice, "_get_file_system", return_value="disk0:")
+@mock.patch.object(ASADevice, "_check_free_space")
+@mock.patch.object(ASADevice, "verify_file", side_effect=[False, True])
+def test_remote_file_copy_skips_space_check_when_file_size_omitted(mock_verify, mock_check, _fs, asa_device):
+    """When FileCopyModel has no file_size, _check_free_space is NOT called."""
+    model = FileCopyModel(
+        download_url="ftp://192.0.2.1/asa.bin",
+        checksum=SHA512_CHECKSUM,
+        file_name="asa.bin",
+        hashing_algorithm="sha512",
+    )  # file_size intentionally omitted
+    assert model.file_size_bytes is None
+
+    asa_device.native.find_prompt.return_value = "asa5512#"
+    asa_device.native.send_command.side_effect = [
+        "94038 bytes copied in 0.90 secs",
+    ]
+
+    asa_device.remote_file_copy(model)
+
+    mock_check.assert_not_called()
+    asa_device.native.send_command.assert_called()  # transfer still happens