test_msgloop.py 3.35 KB
Newer Older
1
from .conftest import build_msg
2

3 4
import pytest
from unittest.mock import Mock
5 6 7 8

import sn


9
def test_regulraly_processed(in_out_args_mock, recv_multipart_mock, send_multipart_mock, good_msg):
10 11 12 13
    def side_effect():
        yield good_msg
        yield good_msg
        raise StopIteration()
14

15
    recv_multipart_mock.side_effect = side_effect()
16

17 18 19
    mock_setup = Mock(return_value={ "foo": "bar" })
    mock_process = Mock(return_value=("sentinel/test", { "foo": "bar" }))
    mock_teardown = Mock()
20

21
    sn.sn_main("box_name", mock_process, setup=mock_setup, teardown=mock_teardown)
22

23 24
    assert mock_setup.called
    assert mock_setup.call_count == 1
25

26
    assert mock_teardown.called
27 28 29 30
    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
31 32

    assert mock_process.called
33 34 35
    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
36 37


38 39
def test_processed_from_generator(out_only_args_mock, send_multipart_mock):
    msg_num = 5
40

41 42 43 44 45 46 47 48 49 50 51
    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
52

53 54 55 56 57
    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
58

59 60
    assert send_multipart_mock.called
    assert send_multipart_mock.call_count == msg_num
61

62

63 64 65 66
def test_many_errors_in_row(out_only_args_mock, send_multipart_mock):
    def process(userdata):
        while True:
            yield "šentinel/test", { "foo": "bar" }
67

68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98
    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)
99

100
    assert send_multipart_mock.called