2
2
3
3
"""Tests suite for sftp."""
4
4
5
+ import random
6
+ import string
5
7
import uuid
6
8
7
9
import pytest
@@ -18,11 +20,21 @@ def sftp_session(ssh_client_session):
18
20
del sftp_sess # noqa: WPS420
19
21
20
22
21
- @pytest .fixture
22
- def transmit_payload ():
23
- """Generate a binary test payload."""
24
- uuid_name = uuid .uuid4 ()
25
- return 'Hello, {name!s}' .format (name = uuid_name ).encode ()
23
+ @pytest .fixture (
24
+ params = (32 , 1024 + 1 ),
25
+ ids = ('small-payload' , 'large-payload' ),
26
+ )
27
+ def transmit_payload (request : pytest .FixtureRequest ) -> bytes :
28
+ """Generate binary test payloads of assorted sizes.
29
+
30
+ The choice 32 is arbitrary small value.
31
+
32
+ The choice 1024 + 1 is meant to be 1B larger than the chunk size used in
33
+ :file:`sftp.pyx` to make sure we excercise at least two rounds of reading/writing.
34
+ """
35
+ payload_len = request .param
36
+ random_bytes = [ord (random .choice (string .printable )) for _ in range (payload_len )]
37
+ return bytes (random_bytes )
26
38
27
39
28
40
@pytest .fixture
@@ -48,6 +60,21 @@ def dst_path(file_paths_pair):
48
60
return path
49
61
50
62
63
+ @pytest .fixture
64
+ def other_payload ():
65
+ """Generate a binary test payload."""
66
+ uuid_name = uuid .uuid4 ()
67
+ return 'Original content: {name!s}' .format (name = uuid_name ).encode ()
68
+
69
+
70
+ @pytest .fixture
71
+ def pre_existing_dst_path (dst_path , other_payload ):
72
+ """Return a data destination path."""
73
+ dst_path .write_bytes (other_payload )
74
+ assert dst_path .exists ()
75
+ return dst_path
76
+
77
+
51
78
def test_make_sftp (sftp_session ):
52
79
"""Smoke-test SFTP instance creation."""
53
80
assert sftp_session
@@ -63,3 +90,15 @@ def test_get(dst_path, src_path, sftp_session, transmit_payload):
63
90
"""Check that SFTP file download works."""
64
91
sftp_session .get (str (src_path ), str (dst_path ))
65
92
assert dst_path .read_bytes () == transmit_payload
93
+
94
+
95
+ def test_get_existing (pre_existing_dst_path , src_path , sftp_session , transmit_payload ):
96
+ """Check that SFTP file download works when target file exists."""
97
+ sftp_session .get (str (src_path ), str (pre_existing_dst_path ))
98
+ assert pre_existing_dst_path .read_bytes () == transmit_payload
99
+
100
+
101
+ def test_put_existing (pre_existing_dst_path , src_path , sftp_session , transmit_payload ):
102
+ """Check that SFTP file upload works when target file exists."""
103
+ sftp_session .put (str (src_path ), str (pre_existing_dst_path ))
104
+ assert pre_existing_dst_path .read_bytes () == transmit_payload
0 commit comments