mrtoth/pipe-io-server
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Implement test case for large data

Browse Source
This commit is contained in:
Kristóf Tóth 2018-12-16 22:49:23 +01:00
parent dd78a0af25
commit b7bc795289
1 changed files with 25 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
tests.py
View File

@ -1,5 +1,5 @@
# pylint: disable=redefined-outer-name # pylint: disable=redefined-outer-name
from os import stat from os import stat, urandom
from os.path import exists, dirname, realpath, join from os.path import exists, dirname, realpath, join
from stat import S_ISFIFO from stat import S_ISFIFO
from secrets import token_urlsafe from secrets import token_urlsafe
@ -43,9 +43,26 @@ def test_pipes_isfifo(pipe_io):
token_urlsafe(32) token_urlsafe(32)
] ]
) )
def test_echo_server(pipe_io, test_data): def test_io(pipe_io, test_data):
File(pipe_io.in_pipe).write(test_data) File(pipe_io.in_pipe, string=True).write(test_data)
assert File(pipe_io.out_pipe).read() == test_data assert File(pipe_io.out_pipe, string=True).read() == test_data
@pytest.mark.parametrize(
'test_data_size', [
1024,
1024*1024,
2*1024*1024,
4*1024*1024,
8*1024*1024,
16*1024*1024,
32*1024*1024
]
)
def test_io_large_data(pipe_io, test_data_size):
random_data = urandom(test_data_size)
File(pipe_io.in_pipe).write(random_data)
assert File(pipe_io.out_pipe).read() == random_data
def test_stop_removes_pipes(pipe_io): def test_stop_removes_pipes(pipe_io):
@ -55,13 +72,14 @@ def test_stop_removes_pipes(pipe_io):
class File: class File:
def __init__(self, path): def __init__(self, path, string=False):
self.path = path self.path = path
self._filemode = '' if string else 'b'
def write(self, what): def write(self, what):
with open(self.path, 'w') as ofile: with open(self.path, f'w{self._filemode}') as ofile:
ofile.write(what) ofile.write(what)
def read(self): def read(self):
with open(self.path, 'r') as ifile: with open(self.path, f'r{self._filemode}') as ifile:
return ifile.read() return ifile.read()