From fb6efa65f59b4d29b2c1ef0af90223693c47acdf Mon Sep 17 00:00:00 2001
From: Stephen Kiely <stephenkiely87@gmail.com>
Date: Wed, 18 Feb 2026 12:13:35 -0600
Subject: [PATCH 1/4] Add file_pull option to Cisco IOS devices

---
 pyntc/devices/base_device.py               |   4 +-
 pyntc/devices/ios_device.py                | 141 +++++++++++-
 pyntc/utils/models.py                      |  55 +++++
 tests/unit/test_devices/test_ios_device.py | 241 ++++++++++++++++++++-
 4 files changed, 421 insertions(+), 20 deletions(-)
 create mode 100644 pyntc/utils/models.py

diff --git a/pyntc/devices/base_device.py b/pyntc/devices/base_device.py
index a6273579..db0b8daa 100644
--- a/pyntc/devices/base_device.py
+++ b/pyntc/devices/base_device.py
@@ -221,7 +221,7 @@ def file_copy(self, src, dest=None, **kwargs):
 
         Keyword Args:
             file_system (str): Supported only for IOS and NXOS. The file system for the
-                remote fle. If no file_system is provided, then the ``get_file_system``
+                remote file. If no file_system is provided, then the ``get_file_system``
                 method is used to determine the correct file system to use.
         """
         raise NotImplementedError
@@ -241,7 +241,7 @@ def file_copy_remote_exists(self, src, dest=None, **kwargs):
 
         Keyword Args:
             file_system (str): Supported only for IOS and NXOS. The file system for the
-                remote fle. If no file_system is provided, then the ``get_file_system``
+                remote file. If no file_system is provided, then the ``get_file_system``
                 method is used to determine the correct file system to use.
 
         Returns:
diff --git a/pyntc/devices/ios_device.py b/pyntc/devices/ios_device.py
index c5c78e22..c2c7a7b4 100644
--- a/pyntc/devices/ios_device.py
+++ b/pyntc/devices/ios_device.py
@@ -1,12 +1,19 @@
 """Module for using a Cisco IOS device over SSH."""
 
+from typing import Callable, Optional, Any, Type, Union
+from typing import TYPE_CHECKING
+import contextlib
 import os
 import re
 import time
 
-from netmiko import ConnectHandler, FileTransfer
+from netmiko import ConnectHandler
+from netmiko.cisco import CiscoIosFileTransfer
 from netmiko.exceptions import ReadTimeout
 
+if TYPE_CHECKING:
+    from netmiko.base_connection import BaseConnection
+
 from pyntc import log
 from pyntc.devices.base_device import BaseDevice, RollbackError, fix_docs
 from pyntc.errors import (
@@ -22,6 +29,7 @@
     SocketClosedError,
 )
 from pyntc.utils import get_structured_data
+from pyntc.utils.models import FilePullSpec
 
 BASIC_FACTS_KM = {"model": "hardware", "os_version": "version", "serial_number": "serial", "hostname": "hostname"}
 RE_SHOW_REDUNDANCY = re.compile(
@@ -36,6 +44,117 @@
 INSTALL_MODE_FILE_NAME = "packages.conf"
 
 
+class FileTransferURLPull(CiscoIosFileTransfer):
+    """Custom FileTransfer class to optionally allow downloading files from a url."""
+
+    def __init__(
+        self,
+        ssh_conn: "BaseConnection",
+        source_file: Union[str, "FilePullSpec"],
+        dest_file: str,
+        file_system: Optional[str] = None,
+        direction: str = "put",
+        socket_timeout: float = 10.0,
+        progress: Optional[Callable[..., Any]] = None,
+        progress4: Optional[Callable[..., Any]] = None,
+        hash_supported: bool = True,
+    ) -> None:
+        self.ssh_ctl_chan = ssh_conn
+        self.source_file = source_file
+        self.dest_file = dest_file
+        self.direction = direction
+        self.socket_timeout = socket_timeout
+        self.progress = progress
+        self.progress4 = progress4
+        self.pull_from_url = direction == "url_pull"
+        self.source_sha1 = None
+
+        auto_flag = (
+            "cisco_ios" in ssh_conn.device_type
+            or "cisco_xe" in ssh_conn.device_type
+            or "cisco_xr" in ssh_conn.device_type
+        )
+        if not file_system:
+            if auto_flag:
+                self.file_system = self.ssh_ctl_chan._autodetect_fs()
+            else:
+                raise ValueError("Destination file system not specified")
+        else:
+            self.file_system = file_system
+
+        if direction == "put":
+            self.source_md5 = self.file_md5(source_file) if hash_supported else None
+            self.file_size = os.stat(source_file).st_size
+        elif direction == "get":
+            self.source_md5 = (
+                self.remote_md5(remote_file=source_file) if hash_supported else None
+            )
+            self.file_size = self.remote_file_size(remote_file=source_file)
+        elif direction == "url_pull":
+            if not isinstance(source_file, FilePullSpec):
+                raise ValueError("When direction is 'url_pull', source_file must be a FilePullSpec instance.")
+            # For url_pull, source_file is the URL and dest_file is the filename to save as on the device
+            if source_file.hashing_algorithm and source_file.hashing_algorithm.lower() not in {"md5", "sha512"}:
+                raise ValueError("When direction is 'url_pull', hashing_algorithm must be either 'md5' or 'sha512'.")
+            self.source_md5 = source_file.checksum if hash_supported and source_file.hashing_algorithm.lower() == "md5" else None
+            self.source_sha512 = source_file.checksum if hash_supported and source_file.hashing_algorithm.lower() == "sha512" else None
+            self.file_size = source_file.file_size
+            self.direction = "put"  # FileTransfer only supports put and get, so treat url_pull as put for the transfer process
+        else:
+            raise ValueError("Invalid direction specified")
+
+    def pull_file(self) -> None:
+        """Download the file from the URL to the device using the device's own download capabilities."""
+        url = self.source_file.clean_url
+        current_prompt = self.ssh_ctl_chan.find_prompt()
+        expect_regex = rf"(Destination filename|{re.escape(current_prompt)}|confirm|Password|Address or name of remote host|Source username|Source filename|yes/no|Are you sure you want to continue connecting)"
+        command = f"copy {url} {self.file_system}{self.dest_file}"
+        # Use VRF specified, unless using http or https, which don't support vrf in the copy command
+        if self.source_file.vrf and self.source_file.scheme not in ["http", "https"]:
+            command = f"{command} vrf {self.source_file.vrf}"
+        output = self.ssh_ctl_chan.send_command(command, expect_string=expect_regex, read_timeout=300)
+        for _ in range(10):
+            if current_prompt in output:
+                return
+            # Assume that the filename and address are sent with the url.
+            if re.search(r"(confirm|Address or name of remote host|Source filename|Destination filename)", output, re.IGNORECASE):
+                output = self.ssh_ctl_chan.send_command("", expect_string=expect_regex, read_timeout=300)
+            if re.search(r"Password", output, re.IGNORECASE):
+                output = self.ssh_ctl_chan.send_command(self.source_file.token, expect_string=expect_regex, read_timeout=300, cmd_verify=False)
+            if re.search(r"Source username", output, re.IGNORECASE):
+                output = self.ssh_ctl_chan.send_command(self.source_file.username, expect_string=expect_regex, read_timeout=300)
+            if re.search(r"yes/no|Are you sure you want to continue connecting", output, re.IGNORECASE):
+                output = self.ssh_ctl_chan.send_command("yes", expect_string=expect_regex, read_timeout=300)
+
+    def transfer_file(self) -> None:
+        """Override the transfer_file method to pull from URL if pull_from_url is True."""
+        if self.pull_from_url:
+            self.pull_file()
+        else:
+            super().transfer_file()
+
+    def remote_sha512(
+        self, base_cmd: str = "verify /sha512", remote_file: Optional[str] = None
+    ) -> str:
+        """Calculate the sha512 hash of the file on the device."""
+        if remote_file is None:
+            if self.direction == "put":
+                remote_file = self.dest_file
+            elif self.direction == "get":
+                remote_file = self.source_file
+        remote_sha512_cmd = f"{base_cmd} {self.file_system}/{remote_file}"
+        dest_sha512 = self.ssh_ctl_chan._send_command_str(remote_sha512_cmd, read_timeout=300)
+        dest_sha512 = self.process_md5(dest_sha512) # Process MD5 still does what we want for parsing the output.
+        return dest_sha512
+
+    def compare_md5(self) -> bool:
+        """Overload compare_md5 to handle sha512."""
+        if self.source_sha512:
+            dest_sha512 = self.remote_sha512()
+            return self.source_sha512 == dest_sha512
+        else:
+            return super().compare_md5()
+
 @fix_docs
 class IOSDevice(BaseDevice):
     """Cisco IOS Device Implementation."""
@@ -105,10 +224,16 @@ def _enter_config(self):
         log.debug("Host %s: Device entered config mode.", self.host)
 
     def _file_copy_instance(self, src, dest=None, file_system="flash:"):
+        """Create a FileTransfer instance for copying a file to the device."""
+        direction = "put"
         if dest is None:
-            dest = os.path.basename(src)
+            if isinstance(src, FilePullSpec):
+                dest = src.file_name
+                direction = "url_pull"
+            else:
+                dest = os.path.basename(src)
 
-        file_copy = FileTransfer(self.native, src, dest, file_system=file_system)
+        file_copy = FileTransferURLPull(self.native, src, dest, direction=direction, file_system=file_system)
         log.debug("Host %s: File copy instance %s.", self.host, file_copy)
         return file_copy
 
@@ -634,8 +759,9 @@ def file_copy(self, src, dest=None, file_system=None):
             #            raise FileTransferError('Not enough space available.')
 
             try:
-                file_copy.enable_scp()
-                file_copy.establish_scp_conn()
+                if not isinstance(src, FilePullSpec):
+                    file_copy.enable_scp()
+                    file_copy.establish_scp_conn()
                 file_copy.transfer_file()
                 log.info("Host %s: File %s transferred successfully.", self.host, src)
             except OSError as error:
@@ -648,7 +774,8 @@ def file_copy(self, src, dest=None, file_system=None):
                 log.error("Host %s: File transfer error %s", self.host, FileTransferError.default_message)
                 raise FileTransferError
             finally:
-                file_copy.close_scp_chan()
+                if not isinstance(src, FilePullSpec):
+                    file_copy.close_scp_chan()
 
             # Ensure connection to device is still open after long transfers
             self.open()
@@ -663,7 +790,7 @@ def file_copy(self, src, dest=None, file_system=None):
 
     # TODO: Make this an internal method since exposing file_copy should be sufficient
     def file_copy_remote_exists(self, src, dest=None, file_system=None):
-        """Copy file to device.
+        """Check if file exists on remote device.
 
         Args:
             src (str): Source of file.
diff --git a/pyntc/utils/models.py b/pyntc/utils/models.py
new file mode 100644
index 00000000..21722209
--- /dev/null
+++ b/pyntc/utils/models.py
@@ -0,0 +1,55 @@
+from dataclasses import dataclass, field, asdict
+from urllib.parse import urlparse
+from typing import Optional
+
+# Use Hashing algorithms from Nautobot's supported list.
+HASHING_ALGORITHMS = {"md5", "sha1", "sha224", "sha384", "sha256", "sha512", "sha3", "blake2", "blake3"}
+
+@dataclass
+class FilePullSpec:
+    download_url: str
+    checksum: str
+    file_name: str
+    hashing_algorithm: str = "md5"
+    file_size: Optional[int] = None # Size in bytes
+    username: Optional[str] = None
+    token: Optional[str] = None  # Password/Token
+    vrf: Optional[str] = None
+    ftp_passive: bool = True
+
+    # This field is calculated, so we don't pass it in the constructor
+    clean_url: str = field(init=False)
+    scheme: str = field(init=False)
+
+    def __post_init__(self):
+        # 1. Validate the hashing algorithm choice
+        if self.hashing_algorithm.lower() not in HASHING_ALGORITHMS:
+            raise ValueError(f"Unsupported algorithm. Choose from: {HASHING_ALGORITHMS}")
+
+        # Parse the url to extract components
+        parsed = urlparse(self.download_url)
+
+        # Extract username/password from URL if not already provided as arguments
+        if parsed.username and not self.username:
+            self.username = parsed.username
+        if parsed.password and not self.token:
+            self.token = parsed.password
+
+        # 3. Create the 'clean_url' (URL without the credentials)
+        # This is what you actually send to the device if using ip http client
+        port = f":{parsed.port}" if parsed.port else ""
+        self.clean_url = f"{parsed.scheme}://{parsed.hostname}{port}{parsed.path}"
+        self.scheme = parsed.scheme
+
+        # Handle query params if they exist (though we're avoiding '?' for Cisco)
+        if parsed.query:
+            self.clean_url += f"?{parsed.query}"
+
+    @classmethod
+    def from_dict(cls, data: dict):
+        """Allows users to just pass a dictionary if they prefer."""
+        return cls(**data)
+
+    def to_dict(self):
+        """Useful for logging or passing to other Nornir tasks."""
+        return asdict(self)
diff --git a/tests/unit/test_devices/test_ios_device.py b/tests/unit/test_devices/test_ios_device.py
index f466c2be..252e07ed 100644
--- a/tests/unit/test_devices/test_ios_device.py
+++ b/tests/unit/test_devices/test_ios_device.py
@@ -6,8 +6,10 @@
 import pytest
 
 from pyntc.devices import IOSDevice
+from pyntc.devices.ios_device import FileTransferURLPull, CiscoIosFileTransfer
 from pyntc.devices import ios_device as ios_module
 from pyntc.devices.base_device import RollbackError
+from pyntc.utils.models import FilePullSpec
 
 from .device_mocks.ios import send_command, send_command_expect
 
@@ -101,7 +103,7 @@ def test_save(self):
         self.assertTrue(result)
         self.device.native.send_command_timing.assert_any_call("copy running-config startup-config")
 
-    @mock.patch("pyntc.devices.ios_device.FileTransfer", autospec=True)
+    @mock.patch("pyntc.devices.ios_device.FileTransferURLPull", autospec=True)
     def test_file_copy_remote_exists(self, mock_ft):
         self.device.native.send_command.side_effect = None
         self.device.native.send_command.return_value = "flash: /dev/null"
@@ -113,7 +115,7 @@ def test_file_copy_remote_exists(self, mock_ft):
 
         self.assertTrue(result)
 
-    @mock.patch("pyntc.devices.ios_device.FileTransfer", autospec=True)
+    @mock.patch("pyntc.devices.ios_device.FileTransferURLPull", autospec=True)
     def test_file_copy_remote_exists_bad_md5(self, mock_ft):
         self.device.native.send_command_timing.side_effect = None
         self.device.native.send_command.return_value = "flash: /dev/null"
@@ -125,7 +127,7 @@ def test_file_copy_remote_exists_bad_md5(self, mock_ft):
 
         self.assertFalse(result)
 
-    @mock.patch("pyntc.devices.ios_device.FileTransfer", autospec=True)
+    @mock.patch("pyntc.devices.ios_device.FileTransferURLPull", autospec=True)
     def test_file_copy_remote_exists_not(self, mock_ft):
         self.device.native.send_command_timing.side_effect = None
         self.device.native.send_command.return_value = "flash: /dev/null"
@@ -137,7 +139,7 @@ def test_file_copy_remote_exists_not(self, mock_ft):
 
         self.assertFalse(result)
 
-    @mock.patch("pyntc.devices.ios_device.FileTransfer", autospec=True)
+    @mock.patch("pyntc.devices.ios_device.FileTransferURLPull", autospec=True)
     @mock.patch.object(IOSDevice, "open")
     def test_file_copy(self, mock_open, mock_ft):
         self.device.native.send_command.side_effect = None
@@ -147,13 +149,13 @@ def test_file_copy(self, mock_open, mock_ft):
         mock_ft_instance.check_file_exists.side_effect = [False, True]
         self.device.file_copy("path/to/source_file")
 
-        mock_ft.assert_called_with(self.device.native, "path/to/source_file", "source_file", file_system="flash:")
+        mock_ft.assert_called_with(self.device.native, "path/to/source_file", "source_file", direction="put", file_system="flash:")
         mock_ft_instance.enable_scp.assert_any_call()
         mock_ft_instance.establish_scp_conn.assert_any_call()
         mock_ft_instance.transfer_file.assert_any_call()
         mock_open.assert_called_once()
 
-    @mock.patch("pyntc.devices.ios_device.FileTransfer", autospec=True)
+    @mock.patch("pyntc.devices.ios_device.FileTransferURLPull", autospec=True)
     @mock.patch.object(IOSDevice, "open")
     def test_file_copy_different_dest(self, mock_open, mock_ft):
         self.device.native.send_command_timing.side_effect = None
@@ -163,13 +165,13 @@ def test_file_copy_different_dest(self, mock_open, mock_ft):
         mock_ft_instance.check_file_exists.side_effect = [False, True]
         self.device.file_copy("source_file", "dest_file")
 
-        mock_ft.assert_called_with(self.device.native, "source_file", "dest_file", file_system="flash:")
+        mock_ft.assert_called_with(self.device.native, "source_file", "dest_file", direction="put", file_system="flash:")
         mock_ft_instance.enable_scp.assert_any_call()
         mock_ft_instance.establish_scp_conn.assert_any_call()
         mock_ft_instance.transfer_file.assert_any_call()
         mock_open.assert_called_once()
 
-    @mock.patch("pyntc.devices.ios_device.FileTransfer", autospec=True)
+    @mock.patch("pyntc.devices.ios_device.FileTransferURLPull", autospec=True)
     @mock.patch.object(IOSDevice, "open")
     def test_file_copy_fail(self, mock_open, mock_ft):
         self.device.native.send_command_timing.side_effect = None
@@ -183,7 +185,7 @@ def test_file_copy_fail(self, mock_open, mock_ft):
 
         mock_open.assert_not_called()
 
-    @mock.patch("pyntc.devices.ios_device.FileTransfer", autospec=True)
+    @mock.patch("pyntc.devices.ios_device.FileTransferURLPull", autospec=True)
     @mock.patch.object(IOSDevice, "open")
     def test_file_copy_socket_closed_good_md5(self, mock_open, mock_ft):
         self.device.native.send_command_timing.side_effect = None
@@ -195,14 +197,14 @@ def test_file_copy_socket_closed_good_md5(self, mock_open, mock_ft):
 
         self.device.file_copy("path/to/source_file")
 
-        mock_ft.assert_called_with(self.device.native, "path/to/source_file", "source_file", file_system="flash:")
+        mock_ft.assert_called_with(self.device.native, "path/to/source_file", "source_file", direction="put", file_system="flash:")
         mock_ft_instance.enable_scp.assert_any_call()
         mock_ft_instance.establish_scp_conn.assert_any_call()
         mock_ft_instance.transfer_file.assert_any_call()
         mock_ft_instance.compare_md5.assert_has_calls([mock.call(), mock.call()])
         mock_open.assert_called_once()
 
-    @mock.patch("pyntc.devices.ios_device.FileTransfer", autospec=True)
+    @mock.patch("pyntc.devices.ios_device.FileTransferURLPull", autospec=True)
     @mock.patch.object(IOSDevice, "open")
     def test_file_copy_fail_socket_closed_bad_md5(self, mock_open, mock_ft):
         self.device.native.send_command_timing.side_effect = None
@@ -414,6 +416,223 @@ def test_install_os_not_enough_space(
         mock_reboot.assert_not_called()
 
 
+class TestFileTransferURLPull(unittest.TestCase):
+
+    @mock.patch.object(IOSDevice, "open")
+    @mock.patch.object(IOSDevice, "close")
+    @mock.patch("netmiko.cisco.cisco_ios.CiscoIosSSH", autospec=True)
+    def setUp(self, mock_miko, mock_close, mock_open):
+        self.device = IOSDevice("host", "user", "pass")
+        mock_miko.send_command_timing.side_effect = send_command
+        mock_miko.send_command_expect.side_effect = send_command_expect
+        mock_miko.device_type = "cisco_ios"
+        self.device.native = mock_miko  # Mock the native connection for testing
+        self.source_file = FilePullSpec(
+            download_url="http://example.com/file.bin",
+            checksum="abc123",
+            file_name="file.bin",
+            hashing_algorithm="md5",
+            file_size=1024,
+            username="user",
+            token="pass",
+            vrf="VRF1",
+            ftp_passive=True,
+        )
+        self.dest_file = "file.bin"
+
+    def tearDown(self):
+        # Reset the mock so we don't have transient test effects
+        self.device.native.reset_mock()
+
+    def test_init(self):
+        ft = FileTransferURLPull(self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:")
+        self.assertEqual(ft.direction, "put") # The class should set direction to "put" if using url_pull as Netmiko only supports "put" or "get"
+        self.assertEqual(ft.source_file, self.source_file)
+        self.assertEqual(ft.dest_file, self.dest_file)
+        self.assertEqual(ft.source_md5, self.source_file.checksum)
+        self.assertIsNone(ft.source_sha512)
+        self.assertTrue(ft.pull_from_url)
+
+    def test_init_sha512(self):
+        self.source_file.hashing_algorithm = "sha512"
+        ft = FileTransferURLPull(self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:")
+        self.assertEqual(ft.direction, "put") # The class should set direction to "put" if using url_pull as Netmiko only supports "put" or "get"
+        self.assertEqual(ft.source_file, self.source_file)
+        self.assertIsNone(ft.source_md5)
+        self.assertEqual(ft.source_sha512, self.source_file.checksum)
+        self.assertTrue(ft.pull_from_url)
+
+    @mock.patch.object(CiscoIosFileTransfer, "transfer_file")
+    @mock.patch("pyntc.devices.ios_device.os", return_value=1024)
+    @mock.patch.object(FileTransferURLPull, "file_md5", return_value="abc123")
+    @mock.patch.object(FileTransferURLPull, "pull_file")
+    def test_transfer_file_pull_file(self, mock_pull_file, mock_file_md5, mock_os, mock_transfer_file):
+        ft = FileTransferURLPull(self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:")
+        ft.transfer_file()
+        mock_pull_file.assert_called_once()
+        mock_transfer_file.assert_not_called()
+
+    @mock.patch.object(CiscoIosFileTransfer, "transfer_file")
+    @mock.patch("pyntc.devices.ios_device.os", return_value=1024)
+    @mock.patch.object(FileTransferURLPull, "file_md5", return_value="abc123")
+    @mock.patch.object(FileTransferURLPull, "pull_file")
+    def test_transfer_file_transfer_file(self, mock_pull_file, mock_file_md5, mock_os, mock_transfer_file):
+        ft2 = FileTransferURLPull(self.device.native, "test1.txt", self.dest_file, direction="put", file_system="flash:")
+        ft2.transfer_file()
+        mock_pull_file.assert_not_called()
+        mock_transfer_file.assert_called_once()
+
+    @mock.patch.object(CiscoIosFileTransfer, "compare_md5")
+    @mock.patch.object(FileTransferURLPull, "remote_sha512", return_value="abc123")
+    @mock.patch("pyntc.devices.ios_device.os", return_value=1024)
+    @mock.patch.object(FileTransferURLPull, "file_md5", return_value="abc123")
+    @mock.patch.object(FileTransferURLPull, "pull_file")
+    def test_compare_md5_sha512(self, mock_pull_file, mock_file_md5, mock_os, mock_remote_sha512, mock_compare_md5):
+        self.source_file.hashing_algorithm = "sha512"
+        ft = FileTransferURLPull(self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:")
+        ft.compare_md5()
+        mock_compare_md5.assert_not_called()
+        mock_remote_sha512.assert_called_once()
+
+    @mock.patch.object(CiscoIosFileTransfer, "compare_md5")
+    @mock.patch.object(FileTransferURLPull, "remote_sha512", return_value="abc123")
+    @mock.patch("pyntc.devices.ios_device.os", return_value=1024)
+    @mock.patch.object(FileTransferURLPull, "file_md5", return_value="abc123")
+    @mock.patch.object(FileTransferURLPull, "pull_file")
+    def test_compare_md5(self, mock_pull_file, mock_file_md5, mock_os, mock_remote_sha512, mock_compare_md5):
+        ft = FileTransferURLPull(self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:")
+        ft.compare_md5()
+        mock_compare_md5.assert_called_once()
+        mock_remote_sha512.assert_not_called()
+
+    def test_remote_sha512(self):
+        ft = FileTransferURLPull(self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:")
+        with mock.patch.object(self.device.native, "_send_command_str", return_value="= abc123") as mock_send_command:
+            result = ft.remote_sha512()
+            mock_send_command.assert_called_once_with("verify /sha512 flash:/file.bin", read_timeout=300)
+            self.assertEqual(result, "abc123")
+
+
+    def test_pull_file(self):
+        self.device.native.find_prompt.return_value = "router#"
+        ft = FileTransferURLPull(self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:")
+        with mock.patch.object(self.device.native, "send_command", return_value="copy http://example.com/file.bin flash:/file.bin\nCopy complete\nrouter#") as mock_send_command:
+            ft.pull_file()
+            mock_send_command.assert_called_once_with(
+                "copy http://example.com/file.bin flash:file.bin", # HTTP does not use VRF in the command
+                expect_string="(Destination filename|router\\#|confirm|Password|Address or name of remote host|Source username|Source filename|yes/no|Are you sure you want to continue connecting)",
+                read_timeout=300,
+            )
+        source_file = FilePullSpec(
+            download_url="sftp://example.com/file.bin",
+            checksum="abc123",
+            file_name="file.bin",
+            hashing_algorithm="md5",
+            file_size=1024,
+            username="user",
+            token="pass",
+            vrf="VRF1",
+            ftp_passive=True,
+        )
+        ft2 = FileTransferURLPull(self.device.native, source_file=source_file, dest_file=self.dest_file, direction="url_pull", file_system="flash:")
+        with mock.patch.object(self.device.native, "send_command", return_value="copy sftp://example.com/file.bin flash:file.bin vrf VRF1\nCopy complete\nrouter#") as mock_send_command:
+            ft2.pull_file()
+            mock_send_command.assert_called_once_with(
+                "copy sftp://example.com/file.bin flash:file.bin vrf VRF1",
+                expect_string="(Destination filename|router\\#|confirm|Password|Address or name of remote host|Source username|Source filename|yes/no|Are you sure you want to continue connecting)",
+                read_timeout=300,
+            )
+
+    def test_pull_file_all_prompts(self):
+        self.device.native.find_prompt.return_value = "router#"
+        source_file = FilePullSpec(
+            download_url="sftp://example.com/file.bin",
+            checksum="abc123",
+            file_name="file.bin",
+            hashing_algorithm="md5",
+            file_size=1024,
+            username="user",
+            token="pass",
+            vrf="VRF1",
+            ftp_passive=True,
+        )
+        ft = FileTransferURLPull(
+            self.device.native,
+            source_file=source_file,
+            dest_file=self.dest_file,
+            direction="url_pull",
+            file_system="flash:"
+        )
+        responses = [
+            "Address or name of remote host [example.com]?",
+            "Source username [user]?",
+            "Source filename [file.bin]?",
+            "Destination filename [file.bin]?",
+            "%Warning: There is a file already existing with this name\nDo you want to over write? [confirm]",
+            "Password: ",
+            "1024 bytes copied in 2 secs\nrouter#"
+        ]
+
+        # Shared expect_string to keep the assertion clean
+        expect_regex = r"(Destination filename|router\#|confirm|Password|Address or name of remote host|Source username|Source filename|yes/no|Are you sure you want to continue connecting)"
+
+        with mock.patch.object(self.device.native, "send_command", side_effect=responses) as mock_send_command:
+            ft.pull_file()
+
+            # Define the expected sequence of calls
+            expected_calls = [
+                # The initial copy command
+                mock.call(
+                    "copy sftp://example.com/file.bin flash:file.bin vrf VRF1",
+                    expect_string=expect_regex,
+                    read_timeout=300
+                ),
+                # The response to the "Address or name of remote host" prompt (sending a newline/empty string to accept the default)
+                mock.call(
+                    "",
+                    expect_string=expect_regex,
+                    read_timeout=300
+                ),
+                # The response to the "Source username" prompt (sending the username)
+                mock.call(
+                    "user",
+                    expect_string=expect_regex,
+                    read_timeout=300
+                ),
+                # The response to the "Source filename" prompt (sending a newline/empty string to accept the default)
+                mock.call(
+                    "",
+                    expect_string=expect_regex,
+                    read_timeout=300
+                ),
+                # The response to the "Destination filename" prompt (sending a newline/empty string to accept the default)
+                mock.call(
+                    "",
+                    expect_string=expect_regex,
+                    read_timeout=300
+                ),
+                # The response to the confirm overwrite prompt (sending a newline/empty string to accept)
+                mock.call(
+                    "",
+                    expect_string=expect_regex,
+                    read_timeout=300
+                ),
+                # The response to the Password prompt
+                mock.call(
+                    "pass",
+                    expect_string=expect_regex,
+                    read_timeout=300,
+                    cmd_verify=False  # Don't verify the command for password input
+                ),
+            ]
+
+            # Use any_order=False to ensure the conversation happened in the right sequence
+            mock_send_command.assert_has_calls(expected_calls, any_order=False)
+
+            # Verify it didn't loop more times than necessary
+            self.assertEqual(mock_send_command.call_count, 7)
+
+
 if __name__ == "__main__":
     unittest.main()
 

From 75ad8429560680255085c2655e7ce8600557cbfb Mon Sep 17 00:00:00 2001
From: Stephen Kiely <stephenkiely87@gmail.com>
Date: Wed, 18 Feb 2026 13:44:42 -0600
Subject: [PATCH 2/4] Add docs, ruff and change fragment.

---
 changes/345.added                          |   1 +
 docs/user/lib_getting_started.md           |  27 +++++
 pyntc/devices/ios_device.py                |  40 ++++---
 pyntc/utils/models.py                      |  24 ++++-
 tests/unit/test_devices/test_ios_device.py | 116 ++++++++++++---------
 5 files changed, 140 insertions(+), 68 deletions(-)
 create mode 100644 changes/345.added

diff --git a/changes/345.added b/changes/345.added
new file mode 100644
index 00000000..a313c2a8
--- /dev/null
+++ b/changes/345.added
@@ -0,0 +1 @@
+Added the ability to pull files from within a Cisco IOS device.
diff --git a/docs/user/lib_getting_started.md b/docs/user/lib_getting_started.md
index f164d661..812c4a0d 100644
--- a/docs/user/lib_getting_started.md
+++ b/docs/user/lib_getting_started.md
@@ -250,6 +250,33 @@ interface GigabitEthernet1
 >>>
 ```
 
+#### File Copy with URL
+
+All devices that support file copy also support copying files directly from a URL to the device.  This is useful for larger files like OS images.  To do this, you need to use the `FilePullSpec` data model to specify the source file information and then pass that to the `file_copy` method. Currently only supported on Cisco IOS devices. Tested with ftp, http, https, sftp, and tftp urls.
+
+```python
+from pyntc.utils.models import FilePullSpec
+
+>>> source_file = FilePullSpec(
+...     download_url='http://example.com/newconfig.cfg',
+...     checksum='abc123def456',
+...     hashing_algorithm='md5',
+...     file_name='newconfig.cfg'
+... )
+>>> for device in devices:
+...    device.file_copy(source_file)
+...
+>>>
+```
+
+We recommend setting up the ip client on the device to be able to use this feature. For instance, on a Cisco IOS device you would need to set the source interface for the ip client when using http or https urls.  You can do this with the `config` method:
+
+```python
+>>> csr1.config('ip http client source-interface GigabitEthernet1')
+>>>
+```
+
+
 ### Save Configs
 
 - `save` method
diff --git a/pyntc/devices/ios_device.py b/pyntc/devices/ios_device.py
index c2c7a7b4..fc0c5ba5 100644
--- a/pyntc/devices/ios_device.py
+++ b/pyntc/devices/ios_device.py
@@ -1,11 +1,9 @@
 """Module for using a Cisco IOS device over SSH."""
 
-from typing import Callable, Optional, Any, Type, Union
-from typing import TYPE_CHECKING
-import contextlib
 import os
 import re
 import time
+from typing import TYPE_CHECKING, Any, Callable, Optional, Union
 
 from netmiko import ConnectHandler
 from netmiko.cisco import CiscoIosFileTransfer
@@ -59,6 +57,7 @@ def __init__(
         progress4: Optional[Callable[..., Any]] = None,
         hash_supported: bool = True,
     ) -> None:
+        """Uses Netmiko's FileTransfer class to transfer files to and from the device, but also supports pulling files directly from a URL to the device using the device's own download capabilities."""
         self.ssh_ctl_chan = ssh_conn
         self.source_file = source_file
         self.dest_file = dest_file
@@ -86,9 +85,7 @@ def __init__(
             self.source_md5 = self.file_md5(source_file) if hash_supported else None
             self.file_size = os.stat(source_file).st_size
         elif direction == "get":
-            self.source_md5 = (
-                self.remote_md5(remote_file=source_file) if hash_supported else None
-            )
+            self.source_md5 = self.remote_md5(remote_file=source_file) if hash_supported else None
             self.file_size = self.remote_file_size(remote_file=source_file)
         elif direction == "url_pull":
             if not isinstance(source_file, FilePullSpec):
@@ -96,10 +93,16 @@ def __init__(
             # For url_pull, source_file is the URL and dest_file is the filename to save as on the device
             if source_file.hashing_algorithm and source_file.hashing_algorithm.lower() not in {"md5", "sha512"}:
                 raise ValueError("When direction is 'url_pull', hashing_algorithm must be either 'md5' or 'sha512'.")
-            self.source_md5 = source_file.checksum if hash_supported and source_file.hashing_algorithm.lower() == "md5" else None
-            self.source_sha512 = source_file.checksum if hash_supported and source_file.hashing_algorithm.lower() == "sha512" else None
+            self.source_md5 = (
+                source_file.checksum if hash_supported and source_file.hashing_algorithm.lower() == "md5" else None
+            )
+            self.source_sha512 = (
+                source_file.checksum if hash_supported and source_file.hashing_algorithm.lower() == "sha512" else None
+            )
             self.file_size = source_file.file_size
-            self.direction = "put"  # FileTransfer only supports put and get, so treat url_pull as put for the transfer process
+            self.direction = (
+                "put"  # FileTransfer only supports put and get, so treat url_pull as put for the transfer process
+            )
         else:
             raise ValueError("Invalid direction specified")
 
@@ -117,12 +120,18 @@ def pull_file(self) -> None:
             if current_prompt in output:
                 return
             # Assume that the filename and address are sent with the url.
-            if re.search(r"(confirm|Address or name of remote host|Source filename|Destination filename)", output, re.IGNORECASE):
+            if re.search(
+                r"(confirm|Address or name of remote host|Source filename|Destination filename)", output, re.IGNORECASE
+            ):
                 output = self.ssh_ctl_chan.send_command("", expect_string=expect_regex, read_timeout=300)
             if re.search(r"Password", output, re.IGNORECASE):
-                output = self.ssh_ctl_chan.send_command(self.source_file.token, expect_string=expect_regex, read_timeout=300, cmd_verify=False)
+                output = self.ssh_ctl_chan.send_command(
+                    self.source_file.token, expect_string=expect_regex, read_timeout=300, cmd_verify=False
+                )
             if re.search(r"Source username", output, re.IGNORECASE):
-                output = self.ssh_ctl_chan.send_command(self.source_file.username, expect_string=expect_regex, read_timeout=300)
+                output = self.ssh_ctl_chan.send_command(
+                    self.source_file.username, expect_string=expect_regex, read_timeout=300
+                )
             if re.search(r"yes/no|Are you sure you want to continue connecting", output, re.IGNORECASE):
                 output = self.ssh_ctl_chan.send_command("yes", expect_string=expect_regex, read_timeout=300)
 
@@ -133,9 +142,7 @@ def transfer_file(self) -> None:
         else:
             super().transfer_file()
 
-    def remote_sha512(
-        self, base_cmd: str = "verify /sha512", remote_file: Optional[str] = None
-    ) -> str:
+    def remote_sha512(self, base_cmd: str = "verify /sha512", remote_file: Optional[str] = None) -> str:
         """Calculate the sha512 hash of the file on the device."""
         if remote_file is None:
             if self.direction == "put":
@@ -144,7 +151,7 @@ def remote_sha512(
                 remote_file = self.source_file
         remote_sha512_cmd = f"{base_cmd} {self.file_system}/{remote_file}"
         dest_sha512 = self.ssh_ctl_chan._send_command_str(remote_sha512_cmd, read_timeout=300)
-        dest_sha512 = self.process_md5(dest_sha512) # Process MD5 still does what we want for parsing the output.
+        dest_sha512 = self.process_md5(dest_sha512)  # Process MD5 still does what we want for parsing the output.
         return dest_sha512
 
     def compare_md5(self) -> bool:
@@ -155,6 +162,7 @@ def compare_md5(self) -> bool:
         else:
             return super().compare_md5()
 
+
 @fix_docs
 class IOSDevice(BaseDevice):
     """Cisco IOS Device Implementation."""
diff --git a/pyntc/utils/models.py b/pyntc/utils/models.py
index 21722209..f9d80d03 100644
--- a/pyntc/utils/models.py
+++ b/pyntc/utils/models.py
@@ -1,17 +1,34 @@
-from dataclasses import dataclass, field, asdict
-from urllib.parse import urlparse
+"""Data Models for Pyntc."""
+
+from dataclasses import asdict, dataclass, field
 from typing import Optional
+from urllib.parse import urlparse
 
 # Use Hashing algorithms from Nautobot's supported list.
 HASHING_ALGORITHMS = {"md5", "sha1", "sha224", "sha384", "sha256", "sha512", "sha3", "blake2", "blake3"}
 
+
 @dataclass
 class FilePullSpec:
+    """Data class to represent the specification for pulling a file from a URL to a network device.
+
+    Args:
+        download_url (str): The URL to download the file from. Can include credentials, but it's recommended to use the username and token fields instead for security reasons.
+        checksum (str): The expected checksum of the file.
+        file_name (str): The name of the file to be saved on the device.
+        hashing_algorithm (str): The hashing algorithm to use for checksum verification. Defaults to "md5".
+        file_size (int, optional): The expected size of the file in bytes. Optional but can be used for an additional layer of verification.
+        username (str, optional): The username for authentication if required by the URL. Optional if credentials are included in the URL.
+        token (str, optional): The password or token for authentication if required by the URL. Optional if credentials are included in the URL.
+        vrf (str, optional): The VRF to use for the download if the device supports VRFs. Optional.
+        ftp_passive (bool): Whether to use passive mode for FTP downloads. Defaults to True.
+    """
+
     download_url: str
     checksum: str
     file_name: str
     hashing_algorithm: str = "md5"
-    file_size: Optional[int] = None # Size in bytes
+    file_size: Optional[int] = None  # Size in bytes
     username: Optional[str] = None
     token: Optional[str] = None  # Password/Token
     vrf: Optional[str] = None
@@ -22,6 +39,7 @@ class FilePullSpec:
     scheme: str = field(init=False)
 
     def __post_init__(self):
+        """Validate the input and prepare the clean URL after initialization."""
         # 1. Validate the hashing algorithm choice
         if self.hashing_algorithm.lower() not in HASHING_ALGORITHMS:
             raise ValueError(f"Unsupported algorithm. Choose from: {HASHING_ALGORITHMS}")
diff --git a/tests/unit/test_devices/test_ios_device.py b/tests/unit/test_devices/test_ios_device.py
index 252e07ed..77892f97 100644
--- a/tests/unit/test_devices/test_ios_device.py
+++ b/tests/unit/test_devices/test_ios_device.py
@@ -6,9 +6,9 @@
 import pytest
 
 from pyntc.devices import IOSDevice
-from pyntc.devices.ios_device import FileTransferURLPull, CiscoIosFileTransfer
 from pyntc.devices import ios_device as ios_module
 from pyntc.devices.base_device import RollbackError
+from pyntc.devices.ios_device import CiscoIosFileTransfer, FileTransferURLPull
 from pyntc.utils.models import FilePullSpec
 
 from .device_mocks.ios import send_command, send_command_expect
@@ -149,7 +149,9 @@ def test_file_copy(self, mock_open, mock_ft):
         mock_ft_instance.check_file_exists.side_effect = [False, True]
         self.device.file_copy("path/to/source_file")
 
-        mock_ft.assert_called_with(self.device.native, "path/to/source_file", "source_file", direction="put", file_system="flash:")
+        mock_ft.assert_called_with(
+            self.device.native, "path/to/source_file", "source_file", direction="put", file_system="flash:"
+        )
         mock_ft_instance.enable_scp.assert_any_call()
         mock_ft_instance.establish_scp_conn.assert_any_call()
         mock_ft_instance.transfer_file.assert_any_call()
@@ -165,7 +167,9 @@ def test_file_copy_different_dest(self, mock_open, mock_ft):
         mock_ft_instance.check_file_exists.side_effect = [False, True]
         self.device.file_copy("source_file", "dest_file")
 
-        mock_ft.assert_called_with(self.device.native, "source_file", "dest_file", direction="put", file_system="flash:")
+        mock_ft.assert_called_with(
+            self.device.native, "source_file", "dest_file", direction="put", file_system="flash:"
+        )
         mock_ft_instance.enable_scp.assert_any_call()
         mock_ft_instance.establish_scp_conn.assert_any_call()
         mock_ft_instance.transfer_file.assert_any_call()
@@ -197,7 +201,9 @@ def test_file_copy_socket_closed_good_md5(self, mock_open, mock_ft):
 
         self.device.file_copy("path/to/source_file")
 
-        mock_ft.assert_called_with(self.device.native, "path/to/source_file", "source_file", direction="put", file_system="flash:")
+        mock_ft.assert_called_with(
+            self.device.native, "path/to/source_file", "source_file", direction="put", file_system="flash:"
+        )
         mock_ft_instance.enable_scp.assert_any_call()
         mock_ft_instance.establish_scp_conn.assert_any_call()
         mock_ft_instance.transfer_file.assert_any_call()
@@ -417,7 +423,6 @@ def test_install_os_not_enough_space(
 
 
 class TestFileTransferURLPull(unittest.TestCase):
-
     @mock.patch.object(IOSDevice, "open")
     @mock.patch.object(IOSDevice, "close")
     @mock.patch("netmiko.cisco.cisco_ios.CiscoIosSSH", autospec=True)
@@ -445,8 +450,12 @@ def tearDown(self):
         self.device.native.reset_mock()
 
     def test_init(self):
-        ft = FileTransferURLPull(self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:")
-        self.assertEqual(ft.direction, "put") # The class should set direction to "put" if using url_pull as Netmiko only supports "put" or "get"
+        ft = FileTransferURLPull(
+            self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:"
+        )
+        self.assertEqual(
+            ft.direction, "put"
+        )  # The class should set direction to "put" if using url_pull as Netmiko only supports "put" or "get"
         self.assertEqual(ft.source_file, self.source_file)
         self.assertEqual(ft.dest_file, self.dest_file)
         self.assertEqual(ft.source_md5, self.source_file.checksum)
@@ -455,8 +464,12 @@ def test_init(self):
 
     def test_init_sha512(self):
         self.source_file.hashing_algorithm = "sha512"
-        ft = FileTransferURLPull(self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:")
-        self.assertEqual(ft.direction, "put") # The class should set direction to "put" if using url_pull as Netmiko only supports "put" or "get"
+        ft = FileTransferURLPull(
+            self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:"
+        )
+        self.assertEqual(
+            ft.direction, "put"
+        )  # The class should set direction to "put" if using url_pull as Netmiko only supports "put" or "get"
         self.assertEqual(ft.source_file, self.source_file)
         self.assertIsNone(ft.source_md5)
         self.assertEqual(ft.source_sha512, self.source_file.checksum)
@@ -467,7 +480,9 @@ def test_init_sha512(self):
     @mock.patch.object(FileTransferURLPull, "file_md5", return_value="abc123")
     @mock.patch.object(FileTransferURLPull, "pull_file")
     def test_transfer_file_pull_file(self, mock_pull_file, mock_file_md5, mock_os, mock_transfer_file):
-        ft = FileTransferURLPull(self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:")
+        ft = FileTransferURLPull(
+            self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:"
+        )
         ft.transfer_file()
         mock_pull_file.assert_called_once()
         mock_transfer_file.assert_not_called()
@@ -477,7 +492,9 @@ def test_transfer_file_pull_file(self, mock_pull_file, mock_file_md5, mock_os, m
     @mock.patch.object(FileTransferURLPull, "file_md5", return_value="abc123")
     @mock.patch.object(FileTransferURLPull, "pull_file")
     def test_transfer_file_transfer_file(self, mock_pull_file, mock_file_md5, mock_os, mock_transfer_file):
-        ft2 = FileTransferURLPull(self.device.native, "test1.txt", self.dest_file, direction="put", file_system="flash:")
+        ft2 = FileTransferURLPull(
+            self.device.native, "test1.txt", self.dest_file, direction="put", file_system="flash:"
+        )
         ft2.transfer_file()
         mock_pull_file.assert_not_called()
         mock_transfer_file.assert_called_once()
@@ -489,7 +506,9 @@ def test_transfer_file_transfer_file(self, mock_pull_file, mock_file_md5, mock_o
     @mock.patch.object(FileTransferURLPull, "pull_file")
     def test_compare_md5_sha512(self, mock_pull_file, mock_file_md5, mock_os, mock_remote_sha512, mock_compare_md5):
         self.source_file.hashing_algorithm = "sha512"
-        ft = FileTransferURLPull(self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:")
+        ft = FileTransferURLPull(
+            self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:"
+        )
         ft.compare_md5()
         mock_compare_md5.assert_not_called()
         mock_remote_sha512.assert_called_once()
@@ -500,26 +519,35 @@ def test_compare_md5_sha512(self, mock_pull_file, mock_file_md5, mock_os, mock_r
     @mock.patch.object(FileTransferURLPull, "file_md5", return_value="abc123")
     @mock.patch.object(FileTransferURLPull, "pull_file")
     def test_compare_md5(self, mock_pull_file, mock_file_md5, mock_os, mock_remote_sha512, mock_compare_md5):
-        ft = FileTransferURLPull(self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:")
+        ft = FileTransferURLPull(
+            self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:"
+        )
         ft.compare_md5()
         mock_compare_md5.assert_called_once()
         mock_remote_sha512.assert_not_called()
 
     def test_remote_sha512(self):
-        ft = FileTransferURLPull(self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:")
+        ft = FileTransferURLPull(
+            self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:"
+        )
         with mock.patch.object(self.device.native, "_send_command_str", return_value="= abc123") as mock_send_command:
             result = ft.remote_sha512()
             mock_send_command.assert_called_once_with("verify /sha512 flash:/file.bin", read_timeout=300)
             self.assertEqual(result, "abc123")
 
-
     def test_pull_file(self):
         self.device.native.find_prompt.return_value = "router#"
-        ft = FileTransferURLPull(self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:")
-        with mock.patch.object(self.device.native, "send_command", return_value="copy http://example.com/file.bin flash:/file.bin\nCopy complete\nrouter#") as mock_send_command:
+        ft = FileTransferURLPull(
+            self.device.native, self.source_file, self.dest_file, direction="url_pull", file_system="flash:"
+        )
+        with mock.patch.object(
+            self.device.native,
+            "send_command",
+            return_value="copy http://example.com/file.bin flash:/file.bin\nCopy complete\nrouter#",
+        ) as mock_send_command:
             ft.pull_file()
             mock_send_command.assert_called_once_with(
-                "copy http://example.com/file.bin flash:file.bin", # HTTP does not use VRF in the command
+                "copy http://example.com/file.bin flash:file.bin",  # HTTP does not use VRF in the command
                 expect_string="(Destination filename|router\\#|confirm|Password|Address or name of remote host|Source username|Source filename|yes/no|Are you sure you want to continue connecting)",
                 read_timeout=300,
             )
@@ -534,8 +562,18 @@ def test_pull_file(self):
             vrf="VRF1",
             ftp_passive=True,
         )
-        ft2 = FileTransferURLPull(self.device.native, source_file=source_file, dest_file=self.dest_file, direction="url_pull", file_system="flash:")
-        with mock.patch.object(self.device.native, "send_command", return_value="copy sftp://example.com/file.bin flash:file.bin vrf VRF1\nCopy complete\nrouter#") as mock_send_command:
+        ft2 = FileTransferURLPull(
+            self.device.native,
+            source_file=source_file,
+            dest_file=self.dest_file,
+            direction="url_pull",
+            file_system="flash:",
+        )
+        with mock.patch.object(
+            self.device.native,
+            "send_command",
+            return_value="copy sftp://example.com/file.bin flash:file.bin vrf VRF1\nCopy complete\nrouter#",
+        ) as mock_send_command:
             ft2.pull_file()
             mock_send_command.assert_called_once_with(
                 "copy sftp://example.com/file.bin flash:file.bin vrf VRF1",
@@ -561,7 +599,7 @@ def test_pull_file_all_prompts(self):
             source_file=source_file,
             dest_file=self.dest_file,
             direction="url_pull",
-            file_system="flash:"
+            file_system="flash:",
         )
         responses = [
             "Address or name of remote host [example.com]?",
@@ -570,7 +608,7 @@ def test_pull_file_all_prompts(self):
             "Destination filename [file.bin]?",
             "%Warning: There is a file already existing with this name\nDo you want to over write? [confirm]",
             "Password: ",
-            "1024 bytes copied in 2 secs\nrouter#"
+            "1024 bytes copied in 2 secs\nrouter#",
         ]
 
         # Shared expect_string to keep the assertion clean
@@ -585,44 +623,24 @@ def test_pull_file_all_prompts(self):
                 mock.call(
                     "copy sftp://example.com/file.bin flash:file.bin vrf VRF1",
                     expect_string=expect_regex,
-                    read_timeout=300
+                    read_timeout=300,
                 ),
                 # The response to the "Address or name of remote host" prompt (sending a newline/empty string to accept the default)
-                mock.call(
-                    "",
-                    expect_string=expect_regex,
-                    read_timeout=300
-                ),
+                mock.call("", expect_string=expect_regex, read_timeout=300),
                 # The response to the "Source username" prompt (sending the username)
-                mock.call(
-                    "user",
-                    expect_string=expect_regex,
-                    read_timeout=300
-                ),
+                mock.call("user", expect_string=expect_regex, read_timeout=300),
                 # The response to the "Source filename" prompt (sending a newline/empty string to accept the default)
-                mock.call(
-                    "",
-                    expect_string=expect_regex,
-                    read_timeout=300
-                ),
+                mock.call("", expect_string=expect_regex, read_timeout=300),
                 # The response to the "Destination filename" prompt (sending a newline/empty string to accept the default)
-                mock.call(
-                    "",
-                    expect_string=expect_regex,
-                    read_timeout=300
-                ),
+                mock.call("", expect_string=expect_regex, read_timeout=300),
                 # The response to the confirm overwrite prompt (sending a newline/empty string to accept)
-                mock.call(
-                    "",
-                    expect_string=expect_regex,
-                    read_timeout=300
-                ),
+                mock.call("", expect_string=expect_regex, read_timeout=300),
                 # The response to the Password prompt
                 mock.call(
                     "pass",
                     expect_string=expect_regex,
                     read_timeout=300,
-                    cmd_verify=False  # Don't verify the command for password input
+                    cmd_verify=False,  # Don't verify the command for password input
                 ),
             ]
 

From 73f74e639cecf2399ec335c02b97c0752cca96b8 Mon Sep 17 00:00:00 2001
From: Stephen Kiely <stephenkiely87@gmail.com>
Date: Wed, 18 Feb 2026 14:15:25 -0600
Subject: [PATCH 3/4] Pylint fixes

Ignore the import position as Ruff is handling that.
---
 pyntc/devices/ios_device.py | 20 +++++++++++---------
 1 file changed, 11 insertions(+), 9 deletions(-)

diff --git a/pyntc/devices/ios_device.py b/pyntc/devices/ios_device.py
index fc0c5ba5..495c6bd5 100644
--- a/pyntc/devices/ios_device.py
+++ b/pyntc/devices/ios_device.py
@@ -12,9 +12,9 @@
 if TYPE_CHECKING:
     from netmiko.base_connection import BaseConnection
 
-from pyntc import log
-from pyntc.devices.base_device import BaseDevice, RollbackError, fix_docs
-from pyntc.errors import (
+from pyntc import log  # pylint: disable=wrong-import-position
+from pyntc.devices.base_device import BaseDevice, RollbackError, fix_docs  # pylint: disable=wrong-import-position
+from pyntc.errors import (  # pylint: disable=wrong-import-position
     CommandError,
     CommandListError,
     DeviceNotActiveError,
@@ -26,8 +26,8 @@
     RebootTimeoutError,
     SocketClosedError,
 )
-from pyntc.utils import get_structured_data
-from pyntc.utils.models import FilePullSpec
+from pyntc.utils import get_structured_data  # pylint: disable=wrong-import-position
+from pyntc.utils.models import FilePullSpec  # pylint: disable=wrong-import-position
 
 BASIC_FACTS_KM = {"model": "hardware", "os_version": "version", "serial_number": "serial", "hostname": "hostname"}
 RE_SHOW_REDUNDANCY = re.compile(
@@ -45,7 +45,10 @@
 class FileTransferURLPull(CiscoIosFileTransfer):
     """Custom FileTransfer class to optionally allow downloading files from a url."""
 
-    def __init__(
+    # Netmiko's FileTransfer class requires a source_file to exist.
+    # When doing a url pull, the source_file is the FilePullSpec, which doesn't exist on the filesystem,
+    # so we have to override the __init__ method to get around this, and not call super().__init__(), since it assumes the file exists.
+    def __init__(  # pylint: disable=super-init-not-called, too-many-positional-arguments
         self,
         ssh_conn: "BaseConnection",
         source_file: Union[str, "FilePullSpec"],
@@ -150,7 +153,7 @@ def remote_sha512(self, base_cmd: str = "verify /sha512", remote_file: Optional[
             elif self.direction == "get":
                 remote_file = self.source_file
         remote_sha512_cmd = f"{base_cmd} {self.file_system}/{remote_file}"
-        dest_sha512 = self.ssh_ctl_chan._send_command_str(remote_sha512_cmd, read_timeout=300)
+        dest_sha512 = self.ssh_ctl_chan._send_command_str(remote_sha512_cmd, read_timeout=300)  # pylint: disable=protected-access
         dest_sha512 = self.process_md5(dest_sha512)  # Process MD5 still does what we want for parsing the output.
         return dest_sha512
 
@@ -159,8 +162,7 @@ def compare_md5(self) -> bool:
         if self.source_sha512:
             dest_sha512 = self.remote_sha512()
             return self.source_sha512 == dest_sha512
-        else:
-            return super().compare_md5()
+        return super().compare_md5()
 
 
 @fix_docs

From bedf623548dd84df2337e7e77d3574668fceafc7 Mon Sep 17 00:00:00 2001
From: Stephen Kiely <stephenkiely87@gmail.com>
Date: Thu, 19 Feb 2026 21:23:22 -0600
Subject: [PATCH 4/4] Address feedback

---
 changes/345.added | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/changes/345.added b/changes/345.added
index a313c2a8..f7a7e976 100644
--- a/changes/345.added
+++ b/changes/345.added
@@ -1 +1 @@
-Added the ability to pull files from within a Cisco IOS device.
+Added the ability to download files from within a Cisco IOS device.