Commit 4cad1b3a authored by Robin Obůrka's avatar Robin Obůrka Committed by Robin Obůrka

tests: Add tests for msgloop (incomplete)

parent 89352d42
import pytest
import msgpack
@pytest.fixture
def in_only_args():
return "--resource in,bind,PULL,*,8801".split(" ")
@pytest.fixture
def out_only_args():
return "--resource out,connect,PUSH,127.0.0.1,8802".split(" ")
@pytest.fixture
def in_out_args():
return "--resource in,connect,PULL,127.0.0.1,8801 --resource out,connect,PUSH,127.0.0.1,8802".split(" ")
@pytest.fixture
def bad_socket_args():
return "--resource bad_name,connect,PUSH,127.0.0.1,8802".split(" ")
@pytest.fixture
def good_msg():
t = bytes("sentinel/test", encoding="UTF-8")
p = msgpack.packb({ "foo": "bar" }, encoding="UTF-8")
return (t, p)
import pytest
from unittest.mock import patch
import sn
def test_empty_name():
with pytest.raises(TypeError):
sn.sn_main()
def test_at_least_one_socket(bad_socket_args):
with pytest.raises(sn.LoopError) as e:
sn.sn_main("test", args=bad_socket_args)
assert str(e.value) == "Neither input nor output socket provided"
def test_no_process_function(in_out_args):
with pytest.raises(sn.LoopError) as e:
sn.sn_main("test",
setup=lambda: None,
teardown=lambda: None,
args=in_out_args)
assert str(e.value) == "Missing 'process' callback"
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_generator_needed(out_only_args):
with pytest.raises(sn.LoopError) as e:
sn.sn_main("test",
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):
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_only_args)
assert str(e.value) == "Box generated output but there is any output socket. Bad configuration?"
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"
import pytest
from unittest.mock import patch
import sn
def test_passed_userdata(in_out_args, good_msg):
def setup():
return { "foo": "bar" }
def teardown(userdata):
assert userdata["foo"] == "bar"
def process(envdata, userdata, msg_type, payload):
assert userdata["foo"] == "bar"
raise Exception()
with patch("zmq.Socket.recv_multipart", return_value=good_msg):
sn.sn_main("test",
setup=setup,
teardown=teardown,
process=process,
args=in_out_args)
def test_passed_name(in_out_args, good_msg):
def process(envdata, userdata, msg_type, payload):
assert envdata.name == "test"
raise Exception()
with patch("zmq.Socket.recv_multipart", return_value=good_msg):
sn.sn_main("test",
process=process,
args=in_out_args)
def test_passed_logger(in_out_args, good_msg):
def process(envdata, userdata, msg_type, payload):
assert type(envdata.logger.getEffectiveLevel()) == int
raise Exception()
with patch("zmq.Socket.recv_multipart", return_value=good_msg):
sn.sn_main("test",
process=process,
args=in_out_args)
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