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

Refactor tests

Browse Source
This commit is contained in:
Kristóf Tóth 2018-12-16 23:26:51 +01:00
parent 929502a586
commit 512247163c
1 changed files with 21 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
tests.py
View File

@ -45,8 +45,21 @@ def test_pipes_isfifo(pipe_io):
] ]
) )
def test_io(pipe_io, test_data): def test_io(pipe_io, test_data):
File(pipe_io.in_pipe, string=True).write(test_data) File(pipe_io.in_pipe).write(test_data.encode())
assert File(pipe_io.out_pipe, string=True).read() == test_data assert File(pipe_io.out_pipe).read().decode() == test_data
class File:
def __init__(self, path):
self.path = path
def write(self, what):
with open(self.path, 'wb') as ofile:
ofile.write(what)
def read(self):
with open(self.path, 'rb') as ifile:
return ifile.read()
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -61,33 +74,19 @@ def test_io(pipe_io, test_data):
] ]
) )
def test_io_large_data(pipe_io, test_data_size): def test_io_large_data(pipe_io, test_data_size):
random_data = urandom(test_data_size) test_data = urandom(test_data_size)
File(pipe_io.in_pipe).write(random_data) File(pipe_io.in_pipe).write(test_data)
assert File(pipe_io.out_pipe).read() == random_data assert File(pipe_io.out_pipe).read() == test_data
def test_io_stress(pipe_io): def test_io_stress(pipe_io):
for _ in range(2222): for _ in range(2222):
random_data = urandom(randint(1, 1024)) test_data = urandom(randint(1, 1024))
File(pipe_io.in_pipe).write(random_data) File(pipe_io.in_pipe).write(test_data)
assert File(pipe_io.out_pipe).read() == random_data assert File(pipe_io.out_pipe).read() == test_data
def test_stop_removes_pipes(pipe_io): def test_stop_removes_pipes(pipe_io):
pipe_io.stop() pipe_io.stop()
for path in (pipe_io.in_pipe, pipe_io.out_pipe): for path in (pipe_io.in_pipe, pipe_io.out_pipe):
assert not exists(path) assert not exists(path)
class File:
def __init__(self, path, string=False):
self.path = path
self._filemode = '' if string else 'b'
def write(self, what):
with open(self.path, f'w{self._filemode}') as ofile:
ofile.write(what)
def read(self):
with open(self.path, f'r{self._filemode}') as ifile:
return ifile.read()