|
| 1 | +"""Integration tests for NXOSDevice.remote_file_copy. |
| 2 | +
|
| 3 | +These tests connect to an actual Cisco NXOS device in the lab and are run manually. |
| 4 | +They are NOT part of the CI unit test suite. |
| 5 | +
|
| 6 | +Usage (from project root): |
| 7 | + export NXOS_HOST=<nxos_ip> |
| 8 | + export NXOS_USER=<user> |
| 9 | + export NXOS_PASS=<pass> |
| 10 | + export FTP_URL=ftp://<ftp_user>:<ftp_password>@<server_ip>/<file_name> |
| 11 | + export TFTP_URL=tftp://<server_ip>/<file_name> |
| 12 | + export SCP_URL=scp://<scp_user>:<scp_password>@<server_ip>/<file_name> |
| 13 | + export HTTP_URL=http://<http_user>:<http_password>@<server_ip>:8081/<file_name> |
| 14 | + export HTTPS_URL=https://<https_user>:<https_password>@<server_ip>:8443/<file_name> |
| 15 | + export SFTP_URL=sftp://<sftp_user>:<sftp_password>@<server_ip>/<file_name> |
| 16 | + export FILE_CHECKSUM_MD5=<md5_hash> |
| 17 | + export FILE_SIZE=<image_size> |
| 18 | + export FILE_SIZE_UNIT=megabytes # optional; defaults to "bytes" |
| 19 | + # export NXOS_VRF=management # optional; applied to every copy test |
| 20 | + poetry run pytest tests/integration/test_nxos_device.py -v |
| 21 | +
|
| 22 | +Set only the protocol URL vars for the servers you have available; each |
| 23 | +protocol test will skip automatically if its URL is not set. ``conftest.py`` |
| 24 | +maps this module to md5 (older NXOS releases only support md5/cksum) and |
| 25 | +copies ``FILE_CHECKSUM_MD5`` into ``FILE_CHECKSUM`` automatically. |
| 26 | +
|
| 27 | +Environment variables: |
| 28 | + NXOS_HOST - IP address or hostname of the lab NXOS device |
| 29 | + NXOS_USER - SSH username |
| 30 | + NXOS_PASS - SSH password |
| 31 | + NXOS_VRF - Optional VRF name; when set, every copy test routes through this VRF |
| 32 | + (needed when the file servers are only reachable via the management VRF) |
| 33 | + FTP_URL - FTP URL of the file to transfer |
| 34 | + TFTP_URL - TFTP URL of the file to transfer |
| 35 | + SCP_URL - SCP URL of the file to transfer |
| 36 | + HTTP_URL - HTTP URL of the file to transfer |
| 37 | + HTTPS_URL - HTTPS URL of the file to transfer |
| 38 | + SFTP_URL - SFTP URL of the file to transfer |
| 39 | + FILE_NAME - Destination filename on the device (default: basename of URL path) |
| 40 | + FILE_CHECKSUM_MD5 - Expected md5 checksum of the file (shared across all protocols) |
| 41 | + FILE_SIZE - Expected size of the file expressed in FILE_SIZE_UNIT units; used for |
| 42 | + the pre-transfer free-space check |
| 43 | + FILE_SIZE_UNIT - One of "bytes", "megabytes", or "gigabytes" (default: "bytes") |
| 44 | +""" |
| 45 | + |
| 46 | +import os |
| 47 | +from unittest import mock |
| 48 | + |
| 49 | +import pytest |
| 50 | + |
| 51 | +from pyntc.devices import NXOSDevice |
| 52 | +from pyntc.errors import NotEnoughFreeSpaceError |
| 53 | +from pyntc.utils.models import FILE_SIZE_UNITS, FileCopyModel |
| 54 | + |
| 55 | +from ._helpers import PROTOCOL_URL_VARS, build_file_copy_model, first_available_url |
| 56 | + |
| 57 | +# --------------------------------------------------------------------------- |
| 58 | +# Fixtures |
| 59 | +# --------------------------------------------------------------------------- |
| 60 | + |
| 61 | + |
| 62 | +@pytest.fixture(scope="module") |
| 63 | +def device(): |
| 64 | + """Connect to the lab NXOS device. Skips all tests if credentials are not set.""" |
| 65 | + host = os.environ.get("NXOS_HOST") |
| 66 | + user = os.environ.get("NXOS_USER") |
| 67 | + password = os.environ.get("NXOS_PASS") |
| 68 | + |
| 69 | + if not all([host, user, password]): |
| 70 | + pytest.skip("NXOS_HOST / NXOS_USER / NXOS_PASS environment variables not set") |
| 71 | + |
| 72 | + dev = NXOSDevice(host, user, password) |
| 73 | + yield dev |
| 74 | + dev.close() |
| 75 | + |
| 76 | + |
| 77 | +def _build_nxos_file_copy_model(env_var): |
| 78 | + """Wrap ``build_file_copy_model`` to stamp ``NXOS_VRF`` onto the model. |
| 79 | +
|
| 80 | + NXOS file servers are typically only reachable via the management VRF, so |
| 81 | + the device's ``copy`` command needs ``vrf <name>`` appended. The shared |
| 82 | + helper has no concept of VRF; this driver-local wrapper bridges that gap |
| 83 | + without leaking NXOS specifics into the shared helper. |
| 84 | + """ |
| 85 | + model = build_file_copy_model(env_var) |
| 86 | + vrf = os.environ.get("NXOS_VRF") |
| 87 | + if vrf: |
| 88 | + model.vrf = vrf |
| 89 | + return model |
| 90 | + |
| 91 | + |
| 92 | +# --------------------------------------------------------------------------- |
| 93 | +# Tests |
| 94 | +# --------------------------------------------------------------------------- |
| 95 | + |
| 96 | + |
| 97 | +def test_device_connects(device): |
| 98 | + """Verify the device is reachable and responds to show commands.""" |
| 99 | + assert device.hostname |
| 100 | + assert device.os_version |
| 101 | + |
| 102 | + |
| 103 | +def test_check_file_exists_false(device, any_file_copy_model): |
| 104 | + """Before the copy, the file should not exist (or this test is a no-op if it does).""" |
| 105 | + result = device.check_file_exists(any_file_copy_model.file_name) |
| 106 | + assert isinstance(result, bool) |
| 107 | + |
| 108 | + |
| 109 | +def test_get_remote_checksum_after_exists(device, any_file_copy_model): |
| 110 | + """If the file already exists, verify get_remote_checksum returns a non-empty string.""" |
| 111 | + if not device.check_file_exists(any_file_copy_model.file_name): |
| 112 | + pytest.skip("File does not exist on device; run test_remote_file_copy_* first") |
| 113 | + checksum = device.get_remote_checksum( |
| 114 | + any_file_copy_model.file_name, hashing_algorithm=any_file_copy_model.hashing_algorithm |
| 115 | + ) |
| 116 | + assert checksum and len(checksum) > 0 |
| 117 | + |
| 118 | + |
| 119 | +def test_remote_file_copy_ftp(device): |
| 120 | + """Transfer the file using FTP and verify it exists on the device.""" |
| 121 | + model = _build_nxos_file_copy_model("FTP_URL") |
| 122 | + device.remote_file_copy(model) |
| 123 | + assert device.check_file_exists(model.file_name) |
| 124 | + |
| 125 | + |
| 126 | +def test_remote_file_copy_tftp(device): |
| 127 | + """Transfer the file using TFTP and verify it exists on the device.""" |
| 128 | + model = _build_nxos_file_copy_model("TFTP_URL") |
| 129 | + device.remote_file_copy(model) |
| 130 | + assert device.check_file_exists(model.file_name) |
| 131 | + |
| 132 | + |
| 133 | +def test_remote_file_copy_scp(device): |
| 134 | + """Transfer the file using SCP and verify it exists on the device.""" |
| 135 | + model = _build_nxos_file_copy_model("SCP_URL") |
| 136 | + device.remote_file_copy(model) |
| 137 | + assert device.check_file_exists(model.file_name) |
| 138 | + |
| 139 | + |
| 140 | +def test_remote_file_copy_http(device): |
| 141 | + """Transfer the file using HTTP and verify it exists on the device.""" |
| 142 | + model = _build_nxos_file_copy_model("HTTP_URL") |
| 143 | + device.remote_file_copy(model) |
| 144 | + assert device.check_file_exists(model.file_name) |
| 145 | + |
| 146 | + |
| 147 | +def test_remote_file_copy_https(device): |
| 148 | + """Transfer the file using HTTPS and verify it exists on the device.""" |
| 149 | + model = _build_nxos_file_copy_model("HTTPS_URL") |
| 150 | + device.remote_file_copy(model) |
| 151 | + assert device.check_file_exists(model.file_name) |
| 152 | + |
| 153 | + |
| 154 | +def test_remote_file_copy_sftp(device): |
| 155 | + """Transfer the file using SFTP and verify it exists on the device.""" |
| 156 | + model = _build_nxos_file_copy_model("SFTP_URL") |
| 157 | + device.remote_file_copy(model) |
| 158 | + assert device.check_file_exists(model.file_name) |
| 159 | + |
| 160 | + |
| 161 | +def test_verify_file_after_copy(device, any_file_copy_model): |
| 162 | + """After a successful copy the file should verify cleanly.""" |
| 163 | + if not device.check_file_exists(any_file_copy_model.file_name): |
| 164 | + pytest.skip("File does not exist on device; run a copy test first") |
| 165 | + assert device.verify_file( |
| 166 | + any_file_copy_model.checksum, |
| 167 | + any_file_copy_model.file_name, |
| 168 | + hashing_algorithm=any_file_copy_model.hashing_algorithm, |
| 169 | + ) |
| 170 | + |
| 171 | + |
| 172 | +# --------------------------------------------------------------------------- |
| 173 | +# Free-space / pre-transfer tests |
| 174 | +# --------------------------------------------------------------------------- |
| 175 | + |
| 176 | + |
| 177 | +def test_get_free_space_returns_positive_int(device): |
| 178 | + """``_get_free_space`` parses the ``dir`` trailer into a positive int.""" |
| 179 | + free = device._get_free_space() # pylint: disable=protected-access |
| 180 | + assert isinstance(free, int) |
| 181 | + assert free > 0 |
| 182 | + |
| 183 | + |
| 184 | +def test_check_free_space_succeeds_for_small_request(device): |
| 185 | + """A 1-byte request must always fit; ``_check_free_space`` returns ``None``.""" |
| 186 | + # pylint: disable=protected-access |
| 187 | + assert device._check_free_space(required_bytes=1) is None |
| 188 | + |
| 189 | + |
| 190 | +def test_check_free_space_raises_when_required_exceeds_free(device): |
| 191 | + """When required bytes exceed what the device reports, raise NotEnoughFreeSpaceError.""" |
| 192 | + # pylint: disable=protected-access |
| 193 | + free = device._get_free_space() |
| 194 | + with pytest.raises(NotEnoughFreeSpaceError): |
| 195 | + device._check_free_space(required_bytes=free + 1) |
| 196 | + |
| 197 | + |
| 198 | +def test_file_size_unit_conversion_matches_device_free_space(device): |
| 199 | + """A megabyte-denominated request converts through ``FILE_SIZE_UNITS`` correctly.""" |
| 200 | + # pylint: disable=protected-access |
| 201 | + free_bytes = device._get_free_space() |
| 202 | + one_mb = FILE_SIZE_UNITS["megabytes"] |
| 203 | + if free_bytes < one_mb: |
| 204 | + pytest.skip("Device has less than 1 MB free; conversion sanity test not meaningful") |
| 205 | + # 1 MB should always fit when free space is at least that large. |
| 206 | + assert device._check_free_space(required_bytes=one_mb) is None |
| 207 | + |
| 208 | + |
| 209 | +def test_remote_file_copy_rejects_oversized_transfer(device): |
| 210 | + """remote_file_copy raises NotEnoughFreeSpaceError and never copies the file.""" |
| 211 | + checksum = os.environ.get("FILE_CHECKSUM") |
| 212 | + scheme, url = first_available_url() |
| 213 | + if not (url and checksum): |
| 214 | + pytest.skip("No protocol URL / FILE_CHECKSUM environment variables not set") |
| 215 | + |
| 216 | + # pylint: disable=protected-access |
| 217 | + free_bytes = device._get_free_space() |
| 218 | + free_gb = free_bytes // FILE_SIZE_UNITS["gigabytes"] |
| 219 | + # Ask for ten times the currently-free capacity (minimum 10 GB), expressed in |
| 220 | + # gigabytes so this also exercises the unit conversion end-to-end. |
| 221 | + oversized_gb = max(free_gb * 10, 10) |
| 222 | + |
| 223 | + unique_name = f"pyntc_integration_space_check_{os.getpid()}_{scheme}.bin" |
| 224 | + model = FileCopyModel( |
| 225 | + download_url=url, |
| 226 | + checksum=checksum, |
| 227 | + file_name=unique_name, |
| 228 | + file_size=oversized_gb, |
| 229 | + file_size_unit="gigabytes", |
| 230 | + hashing_algorithm="md5", |
| 231 | + vrf=os.environ.get("NXOS_VRF"), |
| 232 | + timeout=60, |
| 233 | + ) |
| 234 | + |
| 235 | + assert not device.check_file_exists(unique_name), "Unique filename unexpectedly exists before test" |
| 236 | + |
| 237 | + with pytest.raises(NotEnoughFreeSpaceError): |
| 238 | + device.remote_file_copy(model) |
| 239 | + |
| 240 | + # The transfer must never have started — file should still be absent. |
| 241 | + assert not device.check_file_exists(unique_name) |
| 242 | + |
| 243 | + |
| 244 | +def test_remote_file_copy_accepts_declared_size_within_free_space(device): |
| 245 | + """A correctly-sized FileCopyModel copies without the space check interfering.""" |
| 246 | + scheme, _url = first_available_url() |
| 247 | + if scheme is None: |
| 248 | + pytest.skip("No protocol URL environment variables set") |
| 249 | + model = _build_nxos_file_copy_model(PROTOCOL_URL_VARS[scheme]) |
| 250 | + # pylint: disable=protected-access |
| 251 | + free_bytes = device._get_free_space() |
| 252 | + assert model.file_size_bytes <= free_bytes, ( |
| 253 | + "Configured FILE_SIZE/FILE_SIZE_UNIT exceeds device free space; update env vars" |
| 254 | + ) |
| 255 | + device.remote_file_copy(model) |
| 256 | + assert device.check_file_exists(model.file_name) |
| 257 | + |
| 258 | + |
| 259 | +def test_remote_file_copy_skips_space_check_when_file_size_omitted(device): |
| 260 | + """When FileCopyModel has no file_size, _check_free_space is never called. |
| 261 | +
|
| 262 | + Spies on ``NXOSDevice._check_free_space`` for the duration of the |
| 263 | + transfer and asserts it was not invoked. The transfer itself uses the |
| 264 | + same canonical ``FILE_NAME`` that the other copy tests use. The file |
| 265 | + already existing from a prior test run is fine — the assertion that |
| 266 | + matters is ``spy.assert_not_called()`` combined with the transfer |
| 267 | + completing without raising ``FileTransferError``. |
| 268 | + """ |
| 269 | + checksum = os.environ.get("FILE_CHECKSUM") |
| 270 | + file_name = os.environ.get("FILE_NAME") |
| 271 | + _, url = first_available_url() |
| 272 | + if not (url and checksum and file_name): |
| 273 | + pytest.skip("URL / FILE_CHECKSUM / FILE_NAME environment variables not set") |
| 274 | + |
| 275 | + model = FileCopyModel( |
| 276 | + download_url=url, |
| 277 | + checksum=checksum, |
| 278 | + file_name=file_name, |
| 279 | + hashing_algorithm="md5", |
| 280 | + vrf=os.environ.get("NXOS_VRF"), |
| 281 | + timeout=60, |
| 282 | + ) # file_size intentionally omitted |
| 283 | + assert model.file_size is None |
| 284 | + assert model.file_size_bytes is None |
| 285 | + |
| 286 | + with mock.patch.object(NXOSDevice, "_check_free_space") as spy: |
| 287 | + device.remote_file_copy(model) |
| 288 | + |
| 289 | + spy.assert_not_called() |
| 290 | + assert device.check_file_exists(model.file_name) |
0 commit comments