Skip to content

sftp_mock

SFTPClientMock

The SFTPClientMock class mocks the paramiko.SFTPClient class. This is mainly an internal class and should not be used directly.

Source code in src/paramiko_mock/sftp_mock.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
class SFTPClientMock():
    """
    The SFTPClientMock class mocks the paramiko.SFTPClient class.
    __This is mainly an internal class and should not be used directly.__
    """

    def __init__(
        self,
        file_system: SFTPFileSystem | None = None,
        local_filesystem: LocalFilesystemMock | None = None
    ) -> None:
        if file_system is None:
            raise BadSetupError("file_system is required")
        if local_filesystem is None:
            raise BadSetupError("local_filesystem is required")
        self.__remote_file_system__: SFTPFileSystem = file_system
        self.__local_filesystem__: LocalFilesystemMock = local_filesystem

    def open(self, filename: str, mode: str = "r", bufsize: int = -1) -> "SFTPFileMock":
        file = self.__remote_file_system__.get_file(filename)
        if file is None:
            file = SFTPFileMock()
            self.__remote_file_system__.add_file(filename, file)
        return file

    def close(self) -> None:
        pass

    def put(
        self,
        localpath: str,
        remotepath: str,
        callback: Callable[[int, int], None] | None = None,
        prefetch: bool = True,
        max_concurrent_prefetch_requests: int | None = None,
        confirm: bool = True
    ) -> SFTPAttributes:
        mock_local_file = self.__local_filesystem__.get_file(localpath)
        if mock_local_file is None:
            raise FileNotFoundError(f"File not found: {localpath}")

        file_content = mock_local_file.file_content
        size = len(file_content)

        sftp_file_mock = SFTPFileMock()
        self.__remote_file_system__.add_file(remotepath, sftp_file_mock)

        chunk_size = 32768  # Typical SFTP chunk size
        transferred = 0

        for i in range(0, size, chunk_size):
            chunk = file_content[i:i + chunk_size]
            sftp_file_mock.write(chunk)
            transferred += len(chunk)
            if callback:
                callback(transferred, size)

        fake_stat = os.stat_result((
            33206,   # st_mode (file mode)
            1234567,  # st_ino (inode number)
            1000,    # st_dev (device)
            1,       # st_nlink (number of hard links)
            1001,    # st_uid (user ID of owner)
            1002,    # st_gid (group ID of owner)
            size,    # st_size (size in bytes)
            int(time.time()),  # st_atime (last access time)
            int(time.time()),  # st_mtime (last modification time)
            int(time.time())   # st_ctime (creation time on Windows,
                               # metadata change on Unix)
        ))

        if confirm:
            s = SFTPAttributes.from_stat(fake_stat)
            if s.st_size != size:
                raise IOError(f"size mismatch in put! {s.st_size} != {size}")
        else:
            s = SFTPAttributes()

        return s

    def get(
        self,
        remotepath: str,
        localpath: str,
        callback: Callable[[int, int], None] | None = None,
        prefetch: bool = True,
        max_concurrent_prefetch_requests: int | None = None
    ) -> None:
        file = self.__remote_file_system__.get_file(remotepath)
        if file is None:
            raise FileNotFoundError(f"File not found: {remotepath}")

        file_content = file.file_content
        size = len(file_content)

        local_file = LocalFileMock()
        self.__local_filesystem__.add_file(localpath, local_file)

        chunk_size = 32768  # Typical SFTP chunk size
        transferred = 0

        for i in range(0, size, chunk_size):
            chunk = file_content[i:i + chunk_size]
            local_file.write(chunk)
            transferred += len(chunk)
            if callback:
                callback(transferred, size)

    def listdir(self, path: str = ".") -> list[str]:
        file_path = Path(path)
        file_list = [Path(x) for x in self.__remote_file_system__.list_files()]
        return [x.name for x in file_list if x.parent == file_path]

SFTPFileMock

This class mocks a file in the remote filesystem.

Source code in src/paramiko_mock/sftp_mock.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class SFTPFileMock():
    """
    This class mocks a file in the remote filesystem.
    """
    write_history: list[Any] = []
    file_content: Any = None

    def close(self) -> None:
        pass

    def write(self, data: Any) -> None:
        self.write_history.append(data)
        self.file_content = data

    def read(self, size: int | None = None) -> Any:
        return self.file_content

SFTPFileSystem

The SFPTFileSystem class stores the file system for the SFTPClientMock. This is mainly an internal class and should not be used directly.

Source code in src/paramiko_mock/sftp_mock.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class SFTPFileSystem():
    """
    The SFPTFileSystem class stores the file system for the SFTPClientMock.
    __This is mainly an internal class and should not be used directly.__
    """
    file_system: dict[str, "SFTPFileMock"] = {}

    def add_file(self, path: str, content: "SFTPFileMock") -> None:
        self.file_system[path] = content

    def get_file(self, path: str) -> Optional["SFTPFileMock"]:
        return self.file_system.get(path)

    def remove_file(self, path: str) -> None:
        self.file_system.pop(path, None)

    def list_files(self) -> list[str]:
        return list(self.file_system.keys())