5
5
import os
6
6
import random
7
7
import string
8
- import uuid
9
8
10
9
import pytest
11
10
12
11
from pylibsshext .errors import LibsshSCPException
12
+ from pylibsshext .scp import SCP_MAX_CHUNK
13
13
14
14
15
15
@pytest .fixture
@@ -22,11 +22,22 @@ def ssh_scp(ssh_client_session):
22
22
del scp # noqa: WPS420
23
23
24
24
25
- @pytest .fixture
26
- def transmit_payload ():
27
- """Generate a binary test payload."""
28
- uuid_name = uuid .uuid4 ()
29
- return 'Hello, {name!s}' .format (name = uuid_name ).encode ()
25
+ @pytest .fixture (
26
+ params = (32 , SCP_MAX_CHUNK + 1 ),
27
+ ids = ('small-payload' , 'large-payload' ),
28
+ )
29
+ def transmit_payload (request : pytest .FixtureRequest ):
30
+ """Generate a binary test payload.
31
+
32
+ The choice 32 is arbitrary small value.
33
+
34
+ The choice SCP_CHUNK_SIZE + 1 (64kB + 1B) is meant to be 1B larger than the chunk
35
+ size used in :file:`scp.pyx` to make sure we excercise at least two rounds of
36
+ reading/writing.
37
+ """
38
+ payload_len = request .param
39
+ random_bytes = [ord (random .choice (string .printable )) for _ in range (payload_len )]
40
+ return bytes (random_bytes )
30
41
31
42
32
43
@pytest .fixture
@@ -91,38 +102,3 @@ def test_get_existing_local(pre_existing_file_path, src_path, ssh_scp, transmit_
91
102
"""Check that SCP file download works and overwrites local file if it exists."""
92
103
ssh_scp .get (str (src_path ), str (pre_existing_file_path ))
93
104
assert pre_existing_file_path .read_bytes () == transmit_payload
94
-
95
-
96
- @pytest .fixture
97
- def large_payload ():
98
- """Generate a large 65537 byte (64kB+1B) test payload."""
99
- random_char_kilobyte = [ord (random .choice (string .printable )) for _ in range (1024 )]
100
- full_bytes_number = 64
101
- a_64kB_chunk = bytes (random_char_kilobyte * full_bytes_number )
102
- the_last_byte = random .choice (random_char_kilobyte ).to_bytes (length = 1 , byteorder = 'big' )
103
- return a_64kB_chunk + the_last_byte
104
-
105
-
106
- @pytest .fixture
107
- def src_path_large (tmp_path , large_payload ):
108
- """Return a remote path that to a 65537 byte-sized file.
109
-
110
- Typical single-read chunk size is 64kB in ``libssh`` so
111
- the test needs a file that would overflow that to trigger
112
- the read loop.
113
- """
114
- path = tmp_path / 'large.txt'
115
- path .write_bytes (large_payload )
116
- return path
117
-
118
-
119
- def test_get_large (dst_path , src_path_large , ssh_scp , large_payload ):
120
- """Check that SCP file download gets over 64kB of data."""
121
- ssh_scp .get (str (src_path_large ), str (dst_path ))
122
- assert dst_path .read_bytes () == large_payload
123
-
124
-
125
- def test_put_large (dst_path , src_path_large , ssh_scp , large_payload ):
126
- """Check that SCP file download gets over 64kB of data."""
127
- ssh_scp .put (str (src_path_large ), str (dst_path ))
128
- assert dst_path .read_bytes () == large_payload
0 commit comments