conftest.py 1.57 KB
Newer Older
1 2
import pytest

3 4
from unittest.mock import patch

5 6
import msgpack

7 8 9 10 11 12
def args_from_string(s):
    args = ["prog"]
    args.extend(s.split(" "))

    return args

13

14 15 16 17 18 19 20
def build_msg(msg_type, payload):
    t = bytes(msg_type, encoding="UTF-8")
    p = msgpack.packb(payload, encoding="UTF-8")

    return (t, p)


21 22
@pytest.fixture
def in_only_args():
23
    return args_from_string("--resource in,bind,PULL,*,8801")
24 25 26 27


@pytest.fixture
def out_only_args():
28
    return args_from_string("--resource out,connect,PUSH,127.0.0.1,8802")
29 30 31 32


@pytest.fixture
def in_out_args():
33
    return args_from_string("--resource in,connect,PULL,127.0.0.1,8801 --resource out,connect,PUSH,127.0.0.1,8802")
34 35 36 37


@pytest.fixture
def bad_socket_args():
38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62
    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
63 64 65 66


@pytest.fixture
def good_msg():
67
    return build_msg("sentinel/test", { "foo": "bar" })
68 69 70 71 72 73 74 75 76 77 78 79


@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