Commit e8aaf9aa authored by Robin Obůrka's avatar Robin Obůrka

tests: Some more refactoring of msgloop tests

parent 0763ea38
......@@ -6,7 +6,7 @@ from unittest.mock import Mock
import sn
def test_passed_userdata(in_out_args_mock, recv_multipart_mock, send_multipart_mock, good_msg):
def test_regulraly_processed(in_out_args_mock, recv_multipart_mock, send_multipart_mock, good_msg):
def side_effect():
yield good_msg
yield good_msg
......@@ -24,61 +24,77 @@ def test_passed_userdata(in_out_args_mock, recv_multipart_mock, send_multipart_m
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_teardown.call_count == 1 # Should be called right one
assert mock_teardown.call_args[0][0].name == "box_name" # Passed context data
assert isinstance(mock_teardown.call_args[0][0].logger.getEffectiveLevel(), int)
assert mock_teardown.call_args[0][0].foo == "bar" # Passed user data
assert mock_process.called
assert mock_process.call_args[0][0].name == "box_name"
assert mock_process.call_args[0][0].foo == "bar"
assert mock_process.call_args[0][0].name == "box_name" # Passed context data
assert isinstance(mock_process.call_args[0][0].logger.getEffectiveLevel(), int)
assert mock_process.call_args[0][0].foo == "bar" # Passed user data
@pytest.mark.skip
def test_passed_name(in_out_args, good_msg):
def process(envdata, userdata, msg_type, payload):
assert envdata.name == "test"
raise Exception()
def test_processed_from_generator(out_only_args_mock, send_multipart_mock):
msg_num = 5
with patch("zmq.Socket.recv_multipart", return_value=good_msg):
sn.sn_main("test",
process=process,
args=in_out_args)
def process(userdata):
for i in range(msg_num):
yield "sentinel/test", { "foo": "bar" }
mock_setup = Mock(return_value={ "foo": "bar" })
mock_teardown = Mock()
sn.sn_main("box_name", process, setup=mock_setup, teardown=mock_teardown)
assert mock_setup.called
assert mock_setup.call_count == 1
assert mock_teardown.called
assert mock_teardown.call_count == 1 # Should be called right one
assert mock_teardown.call_args[0][0].name == "box_name" # Passed context data
assert isinstance(mock_teardown.call_args[0][0].logger.getEffectiveLevel(), int)
assert mock_teardown.call_args[0][0].foo == "bar" # Passed user data
@pytest.mark.skip
def test_passed_logger(in_out_args, good_msg):
def process(envdata, userdata, msg_type, payload):
assert type(envdata.logger.getEffectiveLevel()) == int
raise Exception()
assert send_multipart_mock.called
assert send_multipart_mock.call_count == msg_num
with patch("zmq.Socket.recv_multipart", return_value=good_msg):
sn.sn_main("test",
process=process,
args=in_out_args)
def test_many_errors_in_row(out_only_args_mock, send_multipart_mock):
def process(userdata):
while True:
yield "šentinel/test", { "foo": "bar" }
@pytest.mark.skip
def test_generator(out_only_args, good_msg):
def generate(envdata, userdata):
for i in range(5):
yield "sentinel/test", { "foo": "bar" }
mock_setup = Mock(return_value={ "foo": "bar" })
mock_teardown = Mock()
with pytest.raises(SystemExit) as se:
sn.sn_main("box_name", process, setup=mock_setup, teardown=mock_teardown)
assert se.type == SystemExit
assert se.value.code == 1
assert mock_setup.called
assert mock_setup.call_count == 1
assert mock_teardown.called
assert mock_teardown.call_count == 1 # Should be called right one
assert mock_teardown.call_args[0][0].name == "box_name" # Passed context data
assert isinstance(mock_teardown.call_args[0][0].logger.getEffectiveLevel(), int)
assert mock_teardown.call_args[0][0].foo == "bar" # Passed user data
assert not send_multipart_mock.called
# I'm not going to test setup and teardown anymore. I tested all available
# versions.
def test_resetable_error_counter(out_only_args_mock, send_multipart_mock):
def process(_):
for i in range(10):
yield "šentinel/test", { "foo": "bar"}
yield "sentinel/test", { "foo": "bar" }
sn.sn_main("box_name", process)
with patch("zmq.Socket.send_multipart", return_value=None) as send_function:
sn.sn_main("test",
process=generate,
args=out_only_args)
assert send_function.called
assert send_function.call_count == 5
args, _ = send_function.call_args
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"
assert send_multipart_mock.called
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