Source code for core_ftp.tests.base
# -*- coding: utf-8 -*-
"""
Base test case and Paramiko mock utilities for core-ftp
unit tests.
"""
import os
from unittest.mock import Mock
from unittest.mock import patch
from core_tests.tests.base import BaseTestCase
from paramiko.sftp_attr import SFTPAttributes
[docs]
class BaseFtpTestCase(BaseTestCase):
"""
Base class for Test Cases related to FTP connections.
This class provides comprehensive mocking infrastructure for SFTP testing
by patching Paramiko's transport and client operations. It simulates
realistic SFTP behavior using local filesystem operations.
Mock Behavior:
- Transport operations: Fully mocked with no real network connections.
- SFTP client operations: Mapped to local filesystem equivalents.
- Directory operations: Virtual directory tracking with get_cwd/chdir.
- File operations: Actual file I/O using local test resources.
- Authentication: Bypassed with successful mock responses.
Usage:
>>> class MyFtpTest(BaseFtpTestCase):
... def test_upload(self):
... client = SftpClient("host", user="user", password="pass")
... client.connect() # Uses mocked transport
... client.upload_file("local.txt", "remote.txt") # Uses mock
Test Isolation:
- No real network connections made.
- Working directory changes are virtual only.
- Clean setup/teardown of all mocks.
- Safe for parallel test execution.
"""
init_transport_mock = None
connect_transport_mock = None
from_private_key_mock = None
close_transport_mock = None
from_transport_mock = None
init_transport_patcher = patch("paramiko.transport.Transport.__init__")
connect_transport_patcher = patch("paramiko.transport.Transport.connect")
from_transport_patcher = patch("paramiko.sftp_client.SFTPClient.from_transport")
from_private_key_patcher = patch("paramiko.pkey.PKey.from_private_key_file")
close_transport_patcher = patch("paramiko.transport.Transport.close")
_cwd = ""
_root_path = ""
[docs]
@classmethod
def setUpClass(cls) -> None:
"""
Sets up class-level mocks for all test methods.
Initializes and starts all Paramiko patches to intercept SSH/SFTP operations.
Establishes the virtual filesystem root and configures mock return values
for successful connection simulation.
Mock Setup:
- Transport initialization: Returns None (bypasses real network setup)
- Transport connection: Returns None (bypasses authentication)
- Private key loading: Returns None (bypasses key file reading)
- Transport closing: Returns None (bypasses connection cleanup)
- SFTP client creation: Returns configured mock client
Virtual Environment:
- Sets root path to current working directory
- Initializes virtual current working directory as empty
"""
super(BaseFtpTestCase, cls).setUpClass()
cls._root_path = os.getcwd()
cls._cwd = ""
cls.init_transport_mock = cls.init_transport_patcher.start()
cls.connect_transport_mock = cls.connect_transport_patcher.start()
cls.from_private_key_mock = cls.from_private_key_patcher.start()
cls.close_transport_mock = cls.close_transport_patcher.start()
cls.from_transport_mock = cls.from_transport_patcher.start()
cls.init_transport_mock.return_value = None
cls.from_transport_mock.return_value = cls.get_client_mock()
[docs]
@classmethod
def tearDownClass(cls) -> None:
"""
Cleans up all mock patches after test completion.
Stops all Paramiko patches to restore original functionality
and prevent mock leakage to other test classes.
Cleanup Operations:
- Stops transport initialization patch
- Stops transport connection patch
- Stops private key loading patch
- Stops SFTP client creation patch
- Stops transport closing patch
"""
super(BaseFtpTestCase, cls).tearDownClass()
cls.init_transport_patcher.stop()
cls.connect_transport_patcher.stop()
cls.from_private_key_patcher.stop()
cls.from_transport_patcher.stop()
cls.close_transport_patcher.stop()
[docs]
@classmethod
def get_client_mock(cls):
"""
Creates and configures the mock SFTP client.
Returns a Mock object that simulates Paramiko's SFTPClient behavior
by mapping SFTP operations to local filesystem operations.
:return: Configured mock SFTP client
:rtype: Mock
"""
client_mock = Mock()
client_mock.listdir_attr.side_effect = cls.list_dir_attr
client_mock.getcwd.side_effect = cls.get_cwd
client_mock.chdir.side_effect = cls.chdir
client_mock.get.side_effect = cls.get
client_mock.put.side_effect = cls.put
client_mock.putfo.side_effect = cls.put_fo
client_mock.remove.side_effect = cls.remove
client_mock.rmdir.side_effect = cls.rmdir
return client_mock
[docs]
@classmethod
def get_cwd(cls):
"""
Simulates SFTP `getcwd()` operation.
Returns the virtual current working directory or defaults to
the test resources directory.
:return: Current working directory path
:rtype: str
"""
return cls._cwd if cls._cwd else os.path.join(os.getcwd(), "tests/resources")
[docs]
@classmethod
def chdir(cls, remote_path: str):
"""
Simulates SFTP chdir() operation with virtual directory tracking.
Changes only the virtual working directory without affecting
the real filesystem working directory.
:param remote_path: Path to change to
:type remote_path: str
"""
cls._cwd = remote_path
[docs]
@classmethod
def list_dir_attr(cls, remote_path: str):
"""
Simulates SFTP listdir_attr() operation.
Lists files in the specified local directory and wraps each
filename in an SFTPAttributes object to match Paramiko's API.
:param remote_path: Directory path to list
:type remote_path: str
:return: Generator of SFTPAttributes objects
:rtype: Iterator[SFTPAttributes]
"""
for file_name in os.listdir(remote_path):
attr = SFTPAttributes()
attr.filename = file_name
yield attr
[docs]
@staticmethod
def get(_remote_path: str, local_path: str, **_kwargs):
"""
Simulates SFTP get() operation (file download).
Creates a local file with test content to simulate downloading
a file from the remote server.
:param _remote_path: Path to remote file (unused in mock)
:type _remote_path: str
:param local_path: Local path where file will be created
:type local_path: str
:param _kwargs: Additional arguments (unused in mock)
"""
with open(local_path, "x", encoding="utf-8") as f:
f.write("This is a test!")
[docs]
@staticmethod
def put(file_path, *args, **kwargs):
"""
Simulates SFTP put() operation (file upload).
Stub method for upload operations. Override in test subclasses
if specific upload behavior testing is needed.
:param file_path: Local file path to upload
:param args: Additional arguments
:param kwargs: Additional keyword arguments
"""
[docs]
@staticmethod
def put_fo(file_like_object, remote_path, *args, **kwargs):
"""
Simulates SFTP putfo() operation (file-like object upload).
Stub method for file-like object upload operations. Override in
test subclasses if specific upload behavior testing is needed.
:param file_like_object: File-like object to upload
:param remote_path: Remote path for upload
:param args: Additional arguments
:param kwargs: Additional keyword arguments
"""
[docs]
@staticmethod
def rmdir(path: str):
"""
Simulates SFTP rmdir() operation (directory removal).
Stub method for directory removal operations. Override in
test subclasses if specific deletion behavior testing is needed.
:param path: Directory path to remove
:type path: str
"""
[docs]
@staticmethod
def remove(path: str):
"""
Simulates SFTP remove() operation (file removal).
Stub method for file removal operations. Override in
test subclasses if specific deletion behavior testing is needed.
:param path: File path to remove
:type path: str
"""