pipe-io-server/tests.py

93 lines
2.2 KiB
Python
Raw Normal View History

# pylint: disable=redefined-outer-name
2018-12-16 21:49:23 +00:00
from os import stat, urandom
from os.path import exists, dirname, realpath, join
2018-12-14 14:40:11 +00:00
from stat import S_ISFIFO
2018-12-16 20:42:54 +00:00
from secrets import token_urlsafe
2018-12-16 21:54:15 +00:00
from random import randint
2018-12-14 14:40:11 +00:00
import pytest
from echo_server import EchoPipeIOServer
@pytest.fixture
def pipe_io():
here = dirname(realpath(__file__))
pipe_server = EchoPipeIOServer(
join(here, 'in_pipe_tests'),
join(here, 'out_pipe_tests')
)
2018-12-14 14:40:11 +00:00
pipe_server.run()
yield pipe_server
pipe_server.stop()
def test_pipes_exist(pipe_io):
for path in (pipe_io.in_pipe, pipe_io.out_pipe):
assert exists(path)
def test_pipes_isfifo(pipe_io):
for path in (pipe_io.in_pipe, pipe_io.out_pipe):
assert S_ISFIFO(stat(path).st_mode)
@pytest.mark.parametrize(
'test_data', [
2018-12-16 20:42:54 +00:00
'Cats and cheese',
2018-12-14 14:40:11 +00:00
'You ever wonder why we are here?',
'Lorem ipsum dolor sit amet',
2018-12-16 20:42:54 +00:00
'You always have a plan, Dutch!',
token_urlsafe(32),
token_urlsafe(32),
token_urlsafe(32),
token_urlsafe(32)
2018-12-14 14:40:11 +00:00
]
)
2018-12-16 21:49:23 +00:00
def test_io(pipe_io, test_data):
2018-12-16 22:26:51 +00:00
File(pipe_io.in_pipe).write(test_data.encode())
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()
2018-12-16 21:49:23 +00:00
@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):
2018-12-16 22:26:51 +00:00
test_data = urandom(test_data_size)
File(pipe_io.in_pipe).write(test_data)
assert File(pipe_io.out_pipe).read() == test_data
2018-12-14 14:40:11 +00:00
2018-12-16 21:54:15 +00:00
def test_io_stress(pipe_io):
for _ in range(2222):
2018-12-16 22:26:51 +00:00
test_data = urandom(randint(1, 1024))
File(pipe_io.in_pipe).write(test_data)
assert File(pipe_io.out_pipe).read() == test_data
2018-12-16 21:54:15 +00:00
def test_stop_removes_pipes(pipe_io):
pipe_io.stop()
for path in (pipe_io.in_pipe, pipe_io.out_pipe):
assert not exists(path)