Skip to content

paramiko_mock

This module provides a set of classes to mock the paramiko library.

Classes that do not contain a docstring will not be included in the documentation. Some classes are intended to be used internally for the and are not recommended to be used directly without the public interfaces.

We cannot guarantee that the internal classes will not change in minor release so proceed with caution when using them.

BadSetupError

Bases: Exception

This exception is raised when the setup of the ParamikoMock is incorrect

Source code in src/paramiko_mock/exceptions.py
 6
 7
 8
 9
10
class BadSetupError(Exception):
    """
    This exception is raised when the setup of the ParamikoMock is incorrect
    """
    pass

ChannelMock

Mock class that simulates paramiko channel behavior. This provides the recv_exit_status() method that real paramiko channels have.

Source code in src/paramiko_mock/channel_mock.py
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class ChannelMock:
    """
    Mock class that simulates paramiko channel behavior.
    This provides the recv_exit_status() method that real paramiko channels have.
    """

    def __init__(self, exit_status: int = 0):
        """
        Initialize the channel mock with an exit status.

        Args:
            exit_status: The exit status code to return (default: 0)
        """
        self._exit_status = exit_status
        self._closed = False

    def recv_exit_status(self) -> int:
        """
        Simulate the recv_exit_status method of paramiko channels.

        Returns:
            The exit status code
        """
        return self._exit_status

    def set_exit_status(self, exit_status: int) -> None:
        """
        Set the exit status for this channel.

        Args:
            exit_status: The exit status code to set
        """
        self._exit_status = exit_status

    def close(self) -> None:
        """Close the channel."""
        self._closed = True

    @property
    def closed(self) -> bool:
        """Check if the channel is closed."""
        return self._closed

closed property

Check if the channel is closed.

__init__(exit_status=0)

Initialize the channel mock with an exit status.

Parameters:

Name Type Description Default
exit_status int

The exit status code to return (default: 0)

0
Source code in src/paramiko_mock/channel_mock.py
12
13
14
15
16
17
18
19
20
def __init__(self, exit_status: int = 0):
    """
    Initialize the channel mock with an exit status.

    Args:
        exit_status: The exit status code to return (default: 0)
    """
    self._exit_status = exit_status
    self._closed = False

close()

Close the channel.

Source code in src/paramiko_mock/channel_mock.py
40
41
42
def close(self) -> None:
    """Close the channel."""
    self._closed = True

recv_exit_status()

Simulate the recv_exit_status method of paramiko channels.

Returns:

Type Description
int

The exit status code

Source code in src/paramiko_mock/channel_mock.py
22
23
24
25
26
27
28
29
def recv_exit_status(self) -> int:
    """
    Simulate the recv_exit_status method of paramiko channels.

    Returns:
        The exit status code
    """
    return self._exit_status

set_exit_status(exit_status)

Set the exit status for this channel.

Parameters:

Name Type Description Default
exit_status int

The exit status code to set

required
Source code in src/paramiko_mock/channel_mock.py
31
32
33
34
35
36
37
38
def set_exit_status(self, exit_status: int) -> None:
    """
    Set the exit status for this channel.

    Args:
        exit_status: The exit status code to set
    """
    self._exit_status = exit_status

LocalFileMock

This class mocks a file in the local filesystem.

Source code in src/paramiko_mock/local_filesystem_mock.py
 8
 9
10
11
12
13
14
15
16
17
18
19
20
class LocalFileMock():
    """
    This class mocks a file in the local filesystem.
    """
    write_history: list[Any] = []
    file_content: Any = None

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

LocalFilesystemMock

LocalFilesystemMock is a class that stores the mocked local filesystem. This is mainly an internal class and should not be used directly.

Source code in src/paramiko_mock/local_filesystem_mock.py
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class LocalFilesystemMock():
    """
    LocalFilesystemMock is a class that stores the mocked local filesystem.
    __This is mainly an internal class and should not be used directly.__
    """
    file_system: dict[str, LocalFileMock] = {}

    def add_file(self, path: str, file_mock: LocalFileMock) -> None:
        self.file_system[path] = file_mock

    def get_file(self, path: str) -> LocalFileMock | None:
        return self.file_system.get(path)

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

ParamikoMockEnviron

This class is the Coordinator for the ParamikoMock environment. It stores information about the remote devices and the local filesystem.

Source code in src/paramiko_mock/mocked_env.py
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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
class ParamikoMockEnviron(metaclass=SingletonMeta):
    """
    This class is the Coordinator for the ParamikoMock environment.
    It stores information about the remote devices and the local filesystem.
    """

    def __init__(self) -> None:
        self.__remote_devices__: dict[str, 'MockRemoteDevice'] = {}
        # Local filesystem
        self.__local_filesystem__: "LocalFilesystemMock" = LocalFilesystemMock()

    # Private/protected methods
    def get_remote_device(self, host: str) -> 'MockRemoteDevice':
        """
        `get_remote_device` is a method that retrieves a remote device from the
        environment.

        - host: The hostname of the remote device.
        Returns: The remote device.

        _Note: This method is protected and should not be used outside of the
        package._ (We cannot guarantee that this method will not change in the
        future)
        """
        return self.__remote_devices__.get(host) or (_ for _ in ()).throw(
            BadSetupError(
                'Remote device not registered, did you forget to call '
                'add_responses_for_host?'
            )
        )

    # Public methods

    def add_responses_for_host(
        self,
        host: str,
        port: int,
        responses: dict[str, 'SSHResponseMock'],
        username: str | None = None,
        password: str | None = None,
        connection_failure: Exception | None = None
    ) -> None:
        """
        `add_responses_for_host` is a method that adds responses for a remote
        device. Effectively, it creates a new MockRemoteDevice object and stores
        it in the environment.

        - host: The hostname of the remote device.
        - port: The port of the remote device.
        - responses: A dictionary that maps commands to responses.
        - username: The username for the remote device (optional)
        - password: The password for the remote device (optional)
        - connection_failure: An exception to raise during connection (optional)
        """
        self.__remote_devices__[f'{host}:{port}'] = MockRemoteDevice(
            host, port, responses, self.__local_filesystem__,
            username, password, connection_failure
        )

    def cleanup_environment(self) -> None:
        """
        `cleanup_environment` is a method that clears the environment.
        """
        # Clear all the responses, credentials and filesystems
        self.__remote_devices__.clear()
        self.__local_filesystem__.file_system.clear()

    def add_mock_file_for_host(
        self,
        host: str,
        port: int,
        path: str,
        file_mock: 'SFTPFileMock'
    ) -> None:
        """
        `add_mock_file_for_host` is a method that adds a mock file to the remote
        filesystem for a specific host.

        - host: The hostname of the remote device.
        - port: The port of the remote device.
        - path: The path of the file.
        - file_mock: The mock file to add.
        """
        device = self.get_remote_device(f'{host}:{port}')
        device.filesystem.add_file(path, file_mock)

    def remove_mock_file_for_host(self, host: str, port: int, path: str) -> None:
        """
        `remove_mock_file_for_host` is a method that removes a mock file from the
        remote filesystem for a specific host.

        - host: The hostname of the remote device.
        - port: The port of the remote device.
        - path: The path of the file.
        """
        device = self.get_remote_device(f'{host}:{port}')
        device.filesystem.remove_file(path)

    def get_mock_file_for_host(self, host: str, port: int, path: str) -> 'SFTPFileMock':
        """
        `get_mock_file_for_host` is a method that retrieves a mock file from the
        remote filesystem for a specific host.

        - host: The hostname of the remote device.
        - port: The port of the remote device.
        - path: The path of the file.

        Returns: The mock file.
        """
        device = self.get_remote_device(f'{host}:{port}')
        return device.filesystem.get_file(path)

    def add_local_file(self, path: str, file_mock: 'LocalFileMock') -> None:
        """
        `add_local_file` is a method that adds a mock file to the local
        filesystem.

        - path: The path of the file.
        - file_mock: The mock file to add.
        """
        self.__local_filesystem__.add_file(path, file_mock)

    def remove_local_file(self, path: str) -> None:
        """
        `remove_local_file` is a method that removes a mock file from the local
        filesystem.

        - path: The path of the file.
        """
        self.__local_filesystem__.remove_file(path)

    def get_local_file(self, path: str) -> 'LocalFileMock':
        """
        `get_local_file` is a method that retrieves a mock file from the local
        filesystem.

        - path: The path of the file.

        Returns: The mock file.
        """
        return self.__local_filesystem__.get_file(path)

    # Asserts
    def assert_command_was_executed(self, host: str, port: int, command: str) -> None:
        """
        `assert_command_was_executed` is a method that asserts that a command
        was executed

        - host: The hostname of the remote device.
        - port: The port of the remote device.
        - command: The command to assert.

        Raises: AssertionError if the command was not executed.
        """
        device = self.get_remote_device(f'{host}:{port}')
        assert command in device.command_history

    def assert_command_was_not_executed(
        self,
        host: str,
        port: int,
        command: str
    ) -> None:
        """
        `assert_command_was_not_executed` is a method that asserts that a
        command was not executed

        - host: The hostname of the remote device.
        - port: The port of the remote device.
        - command: The command to assert.

        Raises: AssertionError if the command was executed
        """
        device = self.get_remote_device(f'{host}:{port}')
        assert command not in device.command_history

    def assert_command_executed_on_index(
        self,
        host: str,
        port: int,
        command: str,
        index: int
    ) -> None:
        """
        `assert_command_executed_on_index` is a method that asserts that a
        command was executed on a specific index

        - host: The hostname of the remote device.
        - port: The port of the remote device.
        - command: The command to assert.
        - index: The index to assert.

        Raises: AssertionError if the command was not executed on the index.
        """
        device = self.get_remote_device(f'{host}:{port}')
        assert device.command_history[index] == command

    # Connection failure setup methods
    def setup_dns_failure(self, host: str, port: int = 22, hostname: str = None) -> None:
        """
        Set up a DNS resolution failure for a host.

        - host: The hostname to fail DNS resolution for
        - port: The port (default: 22)
        - hostname: Optional custom hostname for the error message
        """
        self.add_responses_for_host(
            host, port, {},
            connection_failure=ConnectionFailureConfig.dns_failure(hostname)
        )

    def setup_timeout_failure(self, host: str, port: int = 22) -> None:
        """
        Set up a connection timeout failure for a host.

        - host: The hostname to timeout for
        - port: The port (default: 22)
        """
        self.add_responses_for_host(
            host, port, {},
            connection_failure=ConnectionFailureConfig.timeout_failure()
        )

    def setup_authentication_failure(self, host: str, port: int = 22) -> None:
        """
        Set up an authentication failure for a host.

        - host: The hostname to fail authentication for
        - port: The port (default: 22)
        """
        self.add_responses_for_host(
            host, port, {},
            connection_failure=ConnectionFailureConfig.authentication_failure()
        )

    def setup_connection_refused(self, host: str, port: int = 22) -> None:
        """
        Set up a connection refused failure for a host.

        - host: The hostname to refuse connection for
        - port: The port (default: 22)
        """
        self.add_responses_for_host(
            host, port, {},
            connection_failure=ConnectionFailureConfig.connection_refused()
        )

    def setup_custom_failure(self, host: str, port: int, exception: Exception) -> None:
        """
        Set up a custom exception failure for a host.

        - host: The hostname to fail for
        - port: The port
        - exception: The custom exception to raise
        """
        self.add_responses_for_host(
            host, port, {},
            connection_failure=ConnectionFailureConfig.custom_exception(exception)
        )

    def setup_badhost_failure(
        self,
        host: str,
        port: int,
        custom_got_key: PKey | None = None,
        custom_pkey: PKey | None = None
    ) -> None:
        """
        Set up a custom exception failure for a host.

        - host: The hostname to fail for
        - port: The port
        - exception: The custom exception to raise
        """
        self.add_responses_for_host(
            host, port, {},
            connection_failure=ConnectionFailureConfig.bad_host_exception(
                host, custom_got_key, custom_pkey
            )
        )

add_local_file(path, file_mock)

add_local_file is a method that adds a mock file to the local filesystem.

  • path: The path of the file.
  • file_mock: The mock file to add.
Source code in src/paramiko_mock/mocked_env.py
221
222
223
224
225
226
227
228
229
def add_local_file(self, path: str, file_mock: 'LocalFileMock') -> None:
    """
    `add_local_file` is a method that adds a mock file to the local
    filesystem.

    - path: The path of the file.
    - file_mock: The mock file to add.
    """
    self.__local_filesystem__.add_file(path, file_mock)

add_mock_file_for_host(host, port, path, file_mock)

add_mock_file_for_host is a method that adds a mock file to the remote filesystem for a specific host.

  • host: The hostname of the remote device.
  • port: The port of the remote device.
  • path: The path of the file.
  • file_mock: The mock file to add.
Source code in src/paramiko_mock/mocked_env.py
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def add_mock_file_for_host(
    self,
    host: str,
    port: int,
    path: str,
    file_mock: 'SFTPFileMock'
) -> None:
    """
    `add_mock_file_for_host` is a method that adds a mock file to the remote
    filesystem for a specific host.

    - host: The hostname of the remote device.
    - port: The port of the remote device.
    - path: The path of the file.
    - file_mock: The mock file to add.
    """
    device = self.get_remote_device(f'{host}:{port}')
    device.filesystem.add_file(path, file_mock)

add_responses_for_host(host, port, responses, username=None, password=None, connection_failure=None)

add_responses_for_host is a method that adds responses for a remote device. Effectively, it creates a new MockRemoteDevice object and stores it in the environment.

  • host: The hostname of the remote device.
  • port: The port of the remote device.
  • responses: A dictionary that maps commands to responses.
  • username: The username for the remote device (optional)
  • password: The password for the remote device (optional)
  • connection_failure: An exception to raise during connection (optional)
Source code in src/paramiko_mock/mocked_env.py
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def add_responses_for_host(
    self,
    host: str,
    port: int,
    responses: dict[str, 'SSHResponseMock'],
    username: str | None = None,
    password: str | None = None,
    connection_failure: Exception | None = None
) -> None:
    """
    `add_responses_for_host` is a method that adds responses for a remote
    device. Effectively, it creates a new MockRemoteDevice object and stores
    it in the environment.

    - host: The hostname of the remote device.
    - port: The port of the remote device.
    - responses: A dictionary that maps commands to responses.
    - username: The username for the remote device (optional)
    - password: The password for the remote device (optional)
    - connection_failure: An exception to raise during connection (optional)
    """
    self.__remote_devices__[f'{host}:{port}'] = MockRemoteDevice(
        host, port, responses, self.__local_filesystem__,
        username, password, connection_failure
    )

assert_command_executed_on_index(host, port, command, index)

assert_command_executed_on_index is a method that asserts that a command was executed on a specific index

  • host: The hostname of the remote device.
  • port: The port of the remote device.
  • command: The command to assert.
  • index: The index to assert.

Raises: AssertionError if the command was not executed on the index.

Source code in src/paramiko_mock/mocked_env.py
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
def assert_command_executed_on_index(
    self,
    host: str,
    port: int,
    command: str,
    index: int
) -> None:
    """
    `assert_command_executed_on_index` is a method that asserts that a
    command was executed on a specific index

    - host: The hostname of the remote device.
    - port: The port of the remote device.
    - command: The command to assert.
    - index: The index to assert.

    Raises: AssertionError if the command was not executed on the index.
    """
    device = self.get_remote_device(f'{host}:{port}')
    assert device.command_history[index] == command

assert_command_was_executed(host, port, command)

assert_command_was_executed is a method that asserts that a command was executed

  • host: The hostname of the remote device.
  • port: The port of the remote device.
  • command: The command to assert.

Raises: AssertionError if the command was not executed.

Source code in src/paramiko_mock/mocked_env.py
252
253
254
255
256
257
258
259
260
261
262
263
264
def assert_command_was_executed(self, host: str, port: int, command: str) -> None:
    """
    `assert_command_was_executed` is a method that asserts that a command
    was executed

    - host: The hostname of the remote device.
    - port: The port of the remote device.
    - command: The command to assert.

    Raises: AssertionError if the command was not executed.
    """
    device = self.get_remote_device(f'{host}:{port}')
    assert command in device.command_history

assert_command_was_not_executed(host, port, command)

assert_command_was_not_executed is a method that asserts that a command was not executed

  • host: The hostname of the remote device.
  • port: The port of the remote device.
  • command: The command to assert.

Raises: AssertionError if the command was executed

Source code in src/paramiko_mock/mocked_env.py
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
def assert_command_was_not_executed(
    self,
    host: str,
    port: int,
    command: str
) -> None:
    """
    `assert_command_was_not_executed` is a method that asserts that a
    command was not executed

    - host: The hostname of the remote device.
    - port: The port of the remote device.
    - command: The command to assert.

    Raises: AssertionError if the command was executed
    """
    device = self.get_remote_device(f'{host}:{port}')
    assert command not in device.command_history

cleanup_environment()

cleanup_environment is a method that clears the environment.

Source code in src/paramiko_mock/mocked_env.py
168
169
170
171
172
173
174
def cleanup_environment(self) -> None:
    """
    `cleanup_environment` is a method that clears the environment.
    """
    # Clear all the responses, credentials and filesystems
    self.__remote_devices__.clear()
    self.__local_filesystem__.file_system.clear()

get_local_file(path)

get_local_file is a method that retrieves a mock file from the local filesystem.

  • path: The path of the file.

Returns: The mock file.

Source code in src/paramiko_mock/mocked_env.py
240
241
242
243
244
245
246
247
248
249
def get_local_file(self, path: str) -> 'LocalFileMock':
    """
    `get_local_file` is a method that retrieves a mock file from the local
    filesystem.

    - path: The path of the file.

    Returns: The mock file.
    """
    return self.__local_filesystem__.get_file(path)

get_mock_file_for_host(host, port, path)

get_mock_file_for_host is a method that retrieves a mock file from the remote filesystem for a specific host.

  • host: The hostname of the remote device.
  • port: The port of the remote device.
  • path: The path of the file.

Returns: The mock file.

Source code in src/paramiko_mock/mocked_env.py
207
208
209
210
211
212
213
214
215
216
217
218
219
def get_mock_file_for_host(self, host: str, port: int, path: str) -> 'SFTPFileMock':
    """
    `get_mock_file_for_host` is a method that retrieves a mock file from the
    remote filesystem for a specific host.

    - host: The hostname of the remote device.
    - port: The port of the remote device.
    - path: The path of the file.

    Returns: The mock file.
    """
    device = self.get_remote_device(f'{host}:{port}')
    return device.filesystem.get_file(path)

get_remote_device(host)

get_remote_device is a method that retrieves a remote device from the environment.

  • host: The hostname of the remote device. Returns: The remote device.

Note: This method is protected and should not be used outside of the package. (We cannot guarantee that this method will not change in the future)

Source code in src/paramiko_mock/mocked_env.py
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def get_remote_device(self, host: str) -> 'MockRemoteDevice':
    """
    `get_remote_device` is a method that retrieves a remote device from the
    environment.

    - host: The hostname of the remote device.
    Returns: The remote device.

    _Note: This method is protected and should not be used outside of the
    package._ (We cannot guarantee that this method will not change in the
    future)
    """
    return self.__remote_devices__.get(host) or (_ for _ in ()).throw(
        BadSetupError(
            'Remote device not registered, did you forget to call '
            'add_responses_for_host?'
        )
    )

remove_local_file(path)

remove_local_file is a method that removes a mock file from the local filesystem.

  • path: The path of the file.
Source code in src/paramiko_mock/mocked_env.py
231
232
233
234
235
236
237
238
def remove_local_file(self, path: str) -> None:
    """
    `remove_local_file` is a method that removes a mock file from the local
    filesystem.

    - path: The path of the file.
    """
    self.__local_filesystem__.remove_file(path)

remove_mock_file_for_host(host, port, path)

remove_mock_file_for_host is a method that removes a mock file from the remote filesystem for a specific host.

  • host: The hostname of the remote device.
  • port: The port of the remote device.
  • path: The path of the file.
Source code in src/paramiko_mock/mocked_env.py
195
196
197
198
199
200
201
202
203
204
205
def remove_mock_file_for_host(self, host: str, port: int, path: str) -> None:
    """
    `remove_mock_file_for_host` is a method that removes a mock file from the
    remote filesystem for a specific host.

    - host: The hostname of the remote device.
    - port: The port of the remote device.
    - path: The path of the file.
    """
    device = self.get_remote_device(f'{host}:{port}')
    device.filesystem.remove_file(path)

setup_authentication_failure(host, port=22)

Set up an authentication failure for a host.

  • host: The hostname to fail authentication for
  • port: The port (default: 22)
Source code in src/paramiko_mock/mocked_env.py
332
333
334
335
336
337
338
339
340
341
342
def setup_authentication_failure(self, host: str, port: int = 22) -> None:
    """
    Set up an authentication failure for a host.

    - host: The hostname to fail authentication for
    - port: The port (default: 22)
    """
    self.add_responses_for_host(
        host, port, {},
        connection_failure=ConnectionFailureConfig.authentication_failure()
    )

setup_badhost_failure(host, port, custom_got_key=None, custom_pkey=None)

Set up a custom exception failure for a host.

  • host: The hostname to fail for
  • port: The port
  • exception: The custom exception to raise
Source code in src/paramiko_mock/mocked_env.py
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
def setup_badhost_failure(
    self,
    host: str,
    port: int,
    custom_got_key: PKey | None = None,
    custom_pkey: PKey | None = None
) -> None:
    """
    Set up a custom exception failure for a host.

    - host: The hostname to fail for
    - port: The port
    - exception: The custom exception to raise
    """
    self.add_responses_for_host(
        host, port, {},
        connection_failure=ConnectionFailureConfig.bad_host_exception(
            host, custom_got_key, custom_pkey
        )
    )

setup_connection_refused(host, port=22)

Set up a connection refused failure for a host.

  • host: The hostname to refuse connection for
  • port: The port (default: 22)
Source code in src/paramiko_mock/mocked_env.py
344
345
346
347
348
349
350
351
352
353
354
def setup_connection_refused(self, host: str, port: int = 22) -> None:
    """
    Set up a connection refused failure for a host.

    - host: The hostname to refuse connection for
    - port: The port (default: 22)
    """
    self.add_responses_for_host(
        host, port, {},
        connection_failure=ConnectionFailureConfig.connection_refused()
    )

setup_custom_failure(host, port, exception)

Set up a custom exception failure for a host.

  • host: The hostname to fail for
  • port: The port
  • exception: The custom exception to raise
Source code in src/paramiko_mock/mocked_env.py
356
357
358
359
360
361
362
363
364
365
366
367
def setup_custom_failure(self, host: str, port: int, exception: Exception) -> None:
    """
    Set up a custom exception failure for a host.

    - host: The hostname to fail for
    - port: The port
    - exception: The custom exception to raise
    """
    self.add_responses_for_host(
        host, port, {},
        connection_failure=ConnectionFailureConfig.custom_exception(exception)
    )

setup_dns_failure(host, port=22, hostname=None)

Set up a DNS resolution failure for a host.

  • host: The hostname to fail DNS resolution for
  • port: The port (default: 22)
  • hostname: Optional custom hostname for the error message
Source code in src/paramiko_mock/mocked_env.py
307
308
309
310
311
312
313
314
315
316
317
318
def setup_dns_failure(self, host: str, port: int = 22, hostname: str = None) -> None:
    """
    Set up a DNS resolution failure for a host.

    - host: The hostname to fail DNS resolution for
    - port: The port (default: 22)
    - hostname: Optional custom hostname for the error message
    """
    self.add_responses_for_host(
        host, port, {},
        connection_failure=ConnectionFailureConfig.dns_failure(hostname)
    )

setup_timeout_failure(host, port=22)

Set up a connection timeout failure for a host.

  • host: The hostname to timeout for
  • port: The port (default: 22)
Source code in src/paramiko_mock/mocked_env.py
320
321
322
323
324
325
326
327
328
329
330
def setup_timeout_failure(self, host: str, port: int = 22) -> None:
    """
    Set up a connection timeout failure for a host.

    - host: The hostname to timeout for
    - port: The port (default: 22)
    """
    self.add_responses_for_host(
        host, port, {},
        connection_failure=ConnectionFailureConfig.timeout_failure()
    )

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

SSHClientMock

The SSHClientMock is a class that mocks the paramiko.SSHClient class. This class is intended to be patched in place of the paramiko.SSHClient class.

Source code in src/paramiko_mock/ssh_mock.py
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 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
class SSHClientMock():
    """
    The SSHClientMock is a class that mocks the paramiko.SSHClient class.
    This class is intended to be patched in place of the paramiko.SSHClient class.
    """

    def __init__(self, *args: Any, **kwds: Any) -> None:
        self.device: MockRemoteDevice | None = None
        self.sftp_client_mock: SFTPClientMock | None = None

    def set_missing_host_key_policy(self, policy: Any) -> None:
        pass

    def open_sftp(self) -> SFTPClientMock:
        if self.device is None:
            raise NoValidConnectionsError('No valid connection')
        if self.sftp_client_mock is None:
            # Create a new SFTPClientMock instance with the filesystem for the
            # selected host
            self.sftp_client_mock = SFTPClientMock(
                self.device.filesystem,
                self.device.local_filesystem
            )
        return self.sftp_client_mock

    def set_log_channel(self, log_channel: str) -> None:
        pass

    def get_host_keys(self) -> None:
        pass

    def save_host_keys(self, filename: str) -> None:
        pass

    def load_host_keys(self, filename: str) -> None:
        pass

    def load_system_host_keys(self, filename: str | None = None) -> None:
        pass

    def connect(
        self,
        hostname: str,
        port: int = 22,
        username: str | None = None,
        password: str | None = None,
        **kwargs: Any
    ) -> None:
        self.selected_host = f'{hostname}:{port}'
        self.device = ParamikoMockEnviron().get_remote_device(
            self.selected_host
        )

        # Check for connection failure configuration
        if self.device.connection_failure is not None:
            raise self.device.connection_failure

        # Check authentication
        if self.device.authenticate(username, password) is False:
            raise AuthenticationException()

        self.last_connect_kwargs = kwargs
        self.device.clear()

    def exec_command(
        self,
        command: str,
        bufsize: int = -1,
        timeout: int | None = None,
        get_pty: bool = False,
        environment: dict[str, str] | None = None
    ) -> tuple[BytesIO, BytesIO, BytesIO]:
        if self.selected_host is None:
            raise NoValidConnectionsError('No valid connections')
        self.device.add_command_to_history(command)
        response = self.device.responses.get(command)
        if response is None:
            # check if there is a command that can be used as regexp
            for command_key in self.device.responses:
                if command_key.startswith('re(') and command_key.endswith(')'):
                    regexp_exp = command_key[3:-1]
                    if re.match(regexp_exp, command):
                        response = self.device.responses[command_key]
                        break
            if response is None:
                raise NotImplementedError('No valid response for this command')
        return response(self, command)

    def invoke_shell(
        self,
        term: str = 'vt100',
        width: int = 80,
        height: int = 24,
        width_pixels: int = 0,
        height_pixels: int = 0,
        environment: dict[str, str] | None = None
    ) -> None:
        pass

    def close(self) -> None:
        self.device = None

SSHCommandMock

Bases: SSHResponseMock

SSHCommandMock is a class that represents a response for a command. It's constructed with the stdin, stdout, and stderr that the command will return.

When called the instance of this class will return a tuple of BytesIO objects, where stderr is enhanced with channel functionality for exit status.

  • stdin: The stdin of the command.
  • stdout: The stdout of the command.
  • stderr: The stderr of the command.
  • exit_status: The exit status code for the command (default: 0).
Source code in src/paramiko_mock/ssh_mock.py
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
class SSHCommandMock(SSHResponseMock):
    """
    SSHCommandMock is a class that represents a response for a command.
    It's constructed with the stdin, stdout, and stderr that the command will
    return.

    When called the instance of this class will return a tuple of BytesIO
    objects, where stderr is enhanced with channel functionality for exit status.

    - stdin: The stdin of the command.
    - stdout: The stdout of the command.
    - stderr: The stderr of the command.
    - exit_status: The exit status code for the command (default: 0).
    """

    def __init__(
        self,
        stdin: str | BytesIO,
        stdout: str | BytesIO,
        stderr: str | BytesIO,
        str_encoding="utf-8",
        exit_status: int = 0
    ) -> None:
        if isinstance(stdin, str):
            stdin = BytesIO(stdin.encode(str_encoding))
        if isinstance(stdout, str):
            stdout = BytesIO(stdout.encode(str_encoding))
        if isinstance(stderr, str):
            stderr_bytes = stderr.encode(str_encoding)
        elif isinstance(stderr, BytesIO):
            stderr_bytes = stderr.getvalue()
        else:
            stderr_bytes = stderr

        self.stdin: BytesIO = stdin
        self.stdout: BytesIO = stdout
        self.stderr: StderrMock = StderrMock(stderr_bytes, exit_status)
        self.exit_status: int = exit_status

    def __call__(
        self,
        ssh_client_mock: SSHClientMock,
        command: str
    ) -> tuple[BytesIO, BytesIO, StderrMock]:
        return self.stdin, self.stdout, self.stderr

    def append_to_stdout(self, new_stdout: str) -> None:
        current_content = self.stdout.getvalue()
        self.stdout = BytesIO(current_content + new_stdout.encode('utf-8'))

    def remove_line_containing(self, line: str) -> None:
        current_content = self.stdout.getvalue().decode('utf-8')
        lines = current_content.split('\n')
        filtered_lines = [x for x in lines if line not in x]
        self.stdout = BytesIO('\n'.join(filtered_lines).encode('utf-8'))

    def set_exit_status(self, exit_status: int) -> None:
        """
        Set the exit status for this command.

        Args:
            exit_status: The exit status code to set
        """
        self.exit_status = exit_status
        self.stderr.set_exit_status(exit_status)

    def get_exit_status(self) -> int:
        """
        Get the current exit status.

        Returns:
            The current exit status code
        """
        return self.exit_status

get_exit_status()

Get the current exit status.

Returns:

Type Description
int

The current exit status code

Source code in src/paramiko_mock/ssh_mock.py
207
208
209
210
211
212
213
214
def get_exit_status(self) -> int:
    """
    Get the current exit status.

    Returns:
        The current exit status code
    """
    return self.exit_status

set_exit_status(exit_status)

Set the exit status for this command.

Parameters:

Name Type Description Default
exit_status int

The exit status code to set

required
Source code in src/paramiko_mock/ssh_mock.py
197
198
199
200
201
202
203
204
205
def set_exit_status(self, exit_status: int) -> None:
    """
    Set the exit status for this command.

    Args:
        exit_status: The exit status code to set
    """
    self.exit_status = exit_status
    self.stderr.set_exit_status(exit_status)

SSHResponseMock

Bases: ABC

The SSHResponseMock is a generic class that represents a response for a command. This can be used to create custom responses for commands that would invoke a callback.

Source code in src/paramiko_mock/ssh_mock.py
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
class SSHResponseMock(ABC):
    """
    The SSHResponseMock is a generic class that represents a response for a
    command. This can be used to create custom responses for commands that would
    invoke a callback.
    """

    @abstractmethod
    def __call__(
        self,
        ssh_client_mock: SSHClientMock,
        command: str
    ) -> tuple[BytesIO, BytesIO, BytesIO | StderrMock]:
        """
        A method that should be implemented by the subclasses.
        This method is called when the command is executed

        - ssh_client_mock: The SSHClientMock instance that is executing the
          command.
        - command: The command that is being executed.

        Returns: A tuple of (stdin, stdout, stderr) where stderr can be a BytesIO
                 or StderrMock with channel functionality.
        """
        pass

__call__(ssh_client_mock, command) abstractmethod

A method that should be implemented by the subclasses. This method is called when the command is executed

  • ssh_client_mock: The SSHClientMock instance that is executing the command.
  • command: The command that is being executed.

A tuple of (stdin, stdout, stderr) where stderr can be a BytesIO

Type Description
tuple[BytesIO, BytesIO, BytesIO | StderrMock]

or StderrMock with channel functionality.

Source code in src/paramiko_mock/ssh_mock.py
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
@abstractmethod
def __call__(
    self,
    ssh_client_mock: SSHClientMock,
    command: str
) -> tuple[BytesIO, BytesIO, BytesIO | StderrMock]:
    """
    A method that should be implemented by the subclasses.
    This method is called when the command is executed

    - ssh_client_mock: The SSHClientMock instance that is executing the
      command.
    - command: The command that is being executed.

    Returns: A tuple of (stdin, stdout, stderr) where stderr can be a BytesIO
             or StderrMock with channel functionality.
    """
    pass

StderrMock

Bases: BytesIO

Enhanced stderr mock that extends BytesIO and adds channel functionality. This simulates the behavior of paramiko's stderr which has both read() capabilities and a channel with recv_exit_status() method.

Source code in src/paramiko_mock/stderr_mock.py
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class StderrMock(BytesIO):
    """
    Enhanced stderr mock that extends BytesIO and adds channel functionality.
    This simulates the behavior of paramiko's stderr which has both read() capabilities
    and a channel with recv_exit_status() method.
    """

    def __init__(self, initial_bytes: bytes = b'', exit_status: int = 0):
        """
        Initialize the stderr mock with initial data and exit status.

        Args:
            initial_bytes: Initial bytes for the stderr content
            exit_status: The exit status code for the channel
        """
        super().__init__(initial_bytes)
        self.channel = ChannelMock(exit_status)

    def set_exit_status(self, exit_status: int) -> None:
        """
        Set the exit status for the channel.

        Args:
            exit_status: The exit status code to set
        """
        self.channel.set_exit_status(exit_status)

    def get_exit_status(self) -> int:
        """
        Get the current exit status.

        Returns:
            The current exit status code
        """
        return self.channel.recv_exit_status()

__init__(initial_bytes=b'', exit_status=0)

Initialize the stderr mock with initial data and exit status.

Parameters:

Name Type Description Default
initial_bytes bytes

Initial bytes for the stderr content

b''
exit_status int

The exit status code for the channel

0
Source code in src/paramiko_mock/stderr_mock.py
15
16
17
18
19
20
21
22
23
24
def __init__(self, initial_bytes: bytes = b'', exit_status: int = 0):
    """
    Initialize the stderr mock with initial data and exit status.

    Args:
        initial_bytes: Initial bytes for the stderr content
        exit_status: The exit status code for the channel
    """
    super().__init__(initial_bytes)
    self.channel = ChannelMock(exit_status)

get_exit_status()

Get the current exit status.

Returns:

Type Description
int

The current exit status code

Source code in src/paramiko_mock/stderr_mock.py
35
36
37
38
39
40
41
42
def get_exit_status(self) -> int:
    """
    Get the current exit status.

    Returns:
        The current exit status code
    """
    return self.channel.recv_exit_status()

set_exit_status(exit_status)

Set the exit status for the channel.

Parameters:

Name Type Description Default
exit_status int

The exit status code to set

required
Source code in src/paramiko_mock/stderr_mock.py
26
27
28
29
30
31
32
33
def set_exit_status(self, exit_status: int) -> None:
    """
    Set the exit status for the channel.

    Args:
        exit_status: The exit status code to set
    """
    self.channel.set_exit_status(exit_status)