Commit ce1c9d74 authored by Robin Obůrka's avatar Robin Obůrka Committed by Robin Obůrka

test: Rewrite test_msgloop for new SNBox

parent f731092a
from .conftest import build_msg
import pytest
from unittest.mock import Mock
from unittest.mock import Mock, patch
import sn
def test_regulraly_processed(in_out_args_mock, recv_multipart_mock, send_multipart_mock, good_msg):
def test_context_data_passed(in_out_args_mock, recv_multipart_mock, send_multipart_mock, good_msg):
def side_effect():
yield good_msg
yield good_msg
......@@ -14,87 +14,100 @@ def test_regulraly_processed(in_out_args_mock, recv_multipart_mock, send_multipa
recv_multipart_mock.side_effect = side_effect()
mock_setup = Mock(return_value={ "foo": "bar" })
mock_process = Mock(return_value=("sentinel/test", { "foo": "bar" }))
mock_teardown = Mock()
class TestBox(sn.SNPipelineBox):
def setup(self):
return {
"foo": "bar",
}
def teardown(self):
pass
def process(self, msg_type, payload):
return msg_type, payload
tb = TestBox("test")
tb.setup = mock_setup = Mock(return_value={"foo": "bar"})
tb.teardown = mock_teardown = Mock()
tb.process = mock_process = Mock(return_value=("sentinel/test", {"foo": "bar"}))
sn.sn_main("box_name", mock_process, setup=mock_setup, teardown=mock_teardown)
tb.run()
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 mock_teardown.call_count == 1
assert mock_process.called
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
assert mock_process.call_count == 2
assert send_multipart_mock.called
assert send_multipart_mock.call_count == 2
def test_processed_from_generator(out_only_args_mock, send_multipart_mock):
msg_num = 5
assert tb.name == "test"
assert isinstance(tb.logger.getEffectiveLevel(), int)
assert tb.ctx.foo == "bar"
def process(userdata):
for i in range(msg_num):
yield "sentinel/test", { "foo": "bar" }
mock_setup = Mock(return_value={ "foo": "bar" })
mock_teardown = Mock()
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
raise StopIteration()
sn.sn_main("box_name", process, setup=mock_setup, teardown=mock_teardown)
recv_multipart_mock.side_effect = side_effect()
assert mock_setup.called
assert mock_setup.call_count == 1
class TestBox(sn.SNPipelineBox):
def process(self, msg_type, payload):
return "processed", payload
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
TestBox("test").run()
assert send_multipart_mock.called
assert send_multipart_mock.call_count == 2
assert send_multipart_mock.call_args[0][0][0] == b"processed"
def test_processed_from_generator(out_only_args_mock, send_multipart_mock):
msg_num = 5
class TestBox(sn.SNGeneratorBox):
def process(self):
for i in range(msg_num):
yield "sentinel/test", { "foo": "bar" }
TestBox("test").run()
assert send_multipart_mock.called
assert send_multipart_mock.call_count == msg_num
assert send_multipart_mock.call_count == msg_num
assert send_multipart_mock.call_args[0][0][0] == b"sentinel/test"
def test_many_errors_in_row(out_only_args_mock, send_multipart_mock):
def process(userdata):
while True:
yield "šentinel/test", { "foo": "bar" }
class TestBox(sn.SNGeneratorBox):
def process(self):
while True:
yield "šentinel/test", { "foo": "bar" }
mock_setup = Mock(return_value={ "foo": "bar" })
mock_teardown = Mock()
tb = TestBox("test")
with pytest.raises(SystemExit) as se:
sn.sn_main("box_name", process, setup=mock_setup, teardown=mock_teardown)
tb.run()
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" }
class TestBox(sn.SNGeneratorBox):
def process(self):
for i in range(10):
yield "šentinel/test", { "foo": "bar" }
yield "sentinel/test", { "foo": "bar" }
sn.sn_main("box_name", process)
TestBox("test").run()
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