Commit 2bca6237 authored by Robin Obůrka's avatar Robin Obůrka

test: First version of msgloop refactoring

parent 8196280a
import pytest import pytest
from unittest.mock import patch
import msgpack import msgpack
def args_from_string(s):
args = ["prog"]
args.extend(s.split(" "))
return args
def build_msg(msg_type, payload): def build_msg(msg_type, payload):
t = bytes(msg_type, encoding="UTF-8") t = bytes(msg_type, encoding="UTF-8")
...@@ -12,24 +20,60 @@ def build_msg(msg_type, payload): ...@@ -12,24 +20,60 @@ def build_msg(msg_type, payload):
@pytest.fixture @pytest.fixture
def in_only_args(): def in_only_args():
return "--resource in,bind,PULL,*,8801".split(" ") return args_from_string("--resource in,bind,PULL,*,8801")
@pytest.fixture @pytest.fixture
def out_only_args(): def out_only_args():
return "--resource out,connect,PUSH,127.0.0.1,8802".split(" ") return args_from_string("--resource out,connect,PUSH,127.0.0.1,8802")
@pytest.fixture @pytest.fixture
def in_out_args(): def in_out_args():
return "--resource in,connect,PULL,127.0.0.1,8801 --resource out,connect,PUSH,127.0.0.1,8802".split(" ") return args_from_string("--resource in,connect,PULL,127.0.0.1,8801 --resource out,connect,PUSH,127.0.0.1,8802")
@pytest.fixture @pytest.fixture
def bad_socket_args(): def bad_socket_args():
return "--resource bad_name,connect,PUSH,127.0.0.1,8802".split(" ") return args_from_string("--resource bad_name,connect,PUSH,127.0.0.1,8802")
@pytest.fixture
def in_only_args_mock(in_only_args):
with patch("sys.argv", in_only_args) as m:
yield m
@pytest.fixture
def out_only_args_mock(out_only_args):
with patch("sys.argv", out_only_args) as m:
yield m
@pytest.fixture
def in_out_args_mock(in_out_args):
with patch("sys.argv", in_out_args) as m:
yield m
@pytest.fixture
def bad_socket_args_mock(bad_socket_args):
with patch("sys.argv", bad_socket_args) as m:
yield m
@pytest.fixture @pytest.fixture
def good_msg(): def good_msg():
return build_msg("sentinel/test", { "foo": "bar" }) return build_msg("sentinel/test", { "foo": "bar" })
@pytest.fixture
def recv_multipart_mock(good_msg):
with patch("zmq.Socket.recv_multipart", return_value=good_msg) as m:
yield m
@pytest.fixture
def send_multipart_mock():
with patch("zmq.Socket.send_multipart") as m:
yield m
import pytest import pytest
from unittest.mock import patch
import sn import sn
def test_empty_name(): def test_empty_name():
...@@ -9,51 +7,45 @@ def test_empty_name(): ...@@ -9,51 +7,45 @@ def test_empty_name():
sn.sn_main() sn.sn_main()
def test_at_least_one_socket(bad_socket_args): def test_no_process():
with pytest.raises(sn.LoopError) as e: with pytest.raises(TypeError):
sn.sn_main("test", args=bad_socket_args) sn.sn_main("test")
def test_at_least_one_socket(bad_socket_args_mock):
with pytest.raises(sn.SetupError) as e:
sn.sn_main("test", lambda: None)
assert str(e.value) == "Neither input nor output socket provided" assert str(e.value) == "Neither input nor output socket provided"
def test_no_process_function(in_out_args): def test_generator_needed(out_only_args_mock):
with pytest.raises(sn.LoopError) as e: def process(u, t, p):
sn.sn_main("test", return t, p
setup=lambda: None,
teardown=lambda: None,
args=in_out_args)
assert str(e.value) == "Missing 'process' callback"
with pytest.raises(sn.SetupError) as e:
sn.sn_main("test", process)
assert str(e.value) == "Generator is expected for output-only box"
def test_no_setup_without_teardown(in_out_args):
with pytest.raises(sn.LoopError) as e:
sn.sn_main("test",
teardown=lambda: None,
process=lambda: None,
args=in_out_args)
assert str(e.value) == "There is teardown callback without setup"
def test_setup_dictionary(out_only_args_mock):
def setup():
return 42
def test_generator_needed(out_only_args): with pytest.raises(sn.SetupError) as e:
with pytest.raises(sn.LoopError) as e: sn.sn_main("test", lambda: None, setup=setup)
sn.sn_main("test", assert str(e.value) == "Setup function didn't return a dictionary"
process=lambda: None,
args=out_only_args)
assert str(e.value) == "Generator is expected for output-only box"
def test_unnecessary_output(in_only_args, good_msg): def test_setup_reserved_word(out_only_args_mock):
with pytest.raises(sn.LoopError) as e: def setup():
with patch("zmq.Socket.recv_multipart", return_value=good_msg): return { "name": "foo" }
sn.sn_main("test",
process=lambda e, u, t, p: ("foo", "bar"),
args=in_only_args)
assert str(e.value) == "Box generated output but there is any output socket. Bad configuration?"
with pytest.raises(sn.SetupError) as e:
sn.sn_main("test", lambda: None, setup=setup)
assert "Used reserved word in user_data" in str(e.value)
def test_generated_broken_message(in_out_args, good_msg):
with pytest.raises(sn.LoopError) as e: def test_unnecessary_output(in_only_args_mock, recv_multipart_mock):
with patch("zmq.Socket.recv_multipart", return_value=good_msg): with pytest.raises(sn.SetupError) as e:
sn.sn_main("test", sn.sn_main("test", lambda u, t, p: ("foo", "bar"))
process=lambda e, u, t, p: ("foo", "bar"), assert str(e.value) == "Box generated output but there is any output socket. Bad configuration?"
args=in_out_args)
assert str(e.value) == "Box generates broken messages"
from .conftest import build_msg from .conftest import build_msg
from unittest.mock import patch import pytest
from unittest.mock import Mock
import sn import sn
def test_passed_userdata(in_out_args, good_msg): def test_passed_userdata(in_out_args_mock, recv_multipart_mock, send_multipart_mock, good_msg):
def setup(): def side_effect():
return { "foo": "bar" } yield good_msg
yield good_msg
raise StopIteration()
def teardown(userdata): recv_multipart_mock.side_effect = side_effect()
assert userdata["foo"] == "bar"
def process(envdata, userdata, msg_type, payload): mock_setup = Mock(return_value={ "foo": "bar" })
assert userdata["foo"] == "bar" mock_process = Mock(return_value=("sentinel/test", { "foo": "bar" }))
raise Exception() mock_teardown = Mock()
with patch("zmq.Socket.recv_multipart", return_value=good_msg): sn.sn_main("box_name", mock_process, setup=mock_setup, teardown=mock_teardown)
sn.sn_main("test",
setup=setup,
teardown=teardown,
process=process,
args=in_out_args)
assert mock_setup.called
assert mock_setup.call_count == 1
assert mock_teardown.called
assert mock_teardown.call_count == 1
assert mock_teardown.call_args[0][0].name == "box_name"
assert mock_teardown.call_args[0][0].foo == "bar"
assert mock_process.called
assert mock_process.call_args[0][0].name == "box_name"
assert mock_process.call_args[0][0].foo == "bar"
@pytest.mark.skip
def test_passed_name(in_out_args, good_msg): def test_passed_name(in_out_args, good_msg):
def process(envdata, userdata, msg_type, payload): def process(envdata, userdata, msg_type, payload):
assert envdata.name == "test" assert envdata.name == "test"
...@@ -35,6 +45,7 @@ def test_passed_name(in_out_args, good_msg): ...@@ -35,6 +45,7 @@ def test_passed_name(in_out_args, good_msg):
args=in_out_args) args=in_out_args)
@pytest.mark.skip
def test_passed_logger(in_out_args, good_msg): def test_passed_logger(in_out_args, good_msg):
def process(envdata, userdata, msg_type, payload): def process(envdata, userdata, msg_type, payload):
assert type(envdata.logger.getEffectiveLevel()) == int assert type(envdata.logger.getEffectiveLevel()) == int
...@@ -46,6 +57,7 @@ def test_passed_logger(in_out_args, good_msg): ...@@ -46,6 +57,7 @@ def test_passed_logger(in_out_args, good_msg):
args=in_out_args) args=in_out_args)
@pytest.mark.skip
def test_generator(out_only_args, good_msg): def test_generator(out_only_args, good_msg):
def generate(envdata, userdata): def generate(envdata, userdata):
for i in range(5): for i in range(5):
...@@ -60,3 +72,13 @@ def test_generator(out_only_args, good_msg): ...@@ -60,3 +72,13 @@ def test_generator(out_only_args, good_msg):
assert send_function.call_count == 5 assert send_function.call_count == 5
args, _ = send_function.call_args args, _ = send_function.call_args
assert args[0] == build_msg("sentinel/test", { "foo": "bar" }) assert args[0] == build_msg("sentinel/test", { "foo": "bar" })
@pytest.mark.skip
def test_generated_broken_message(in_out_args, good_msg):
with pytest.raises(sn.LoopError) as e:
with patch("zmq.Socket.recv_multipart", return_value=good_msg):
sn.sn_main("test",
process=lambda e, u, t, p: ("foo", "bar"),
args=in_out_args)
assert str(e.value) == "Box generates broken messages"
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment