Commit 59dda586 authored by Robin Obůrka's avatar Robin Obůrka

tests: Refactoring old tests to not need a special parameter in SN

parent 43ee1650
import pytest
from unittest.mock import patch
import sn
def args_from_string(s):
args = ["prog"]
args.extend(s.split(" "))
return args
@pytest.fixture
def required_type_arg():
return "--resource res,connect,PUSH,127.0.0.1,8800".split(" ")
def one_resource_mock():
with patch("sys.argv", args_from_string("--resource res,connect,PUSH,127.0.0.1,8800")) as m:
yield m
@pytest.fixture
def empty_args_mock():
with patch("sys.argv", args_from_string("")) as m:
yield m
@pytest.fixture(params=[
"--resource",
"--resource BAD",
"--resource res,connect,PUSH,,",
"--resource res,connect,PUSH,sentinel.cz,",
"--resource res,connect,PULL,,7700",
"--resource ,connect,PUSH,sentinel.cz,7700",
"--resource connect,PUSH,sentinel.cz,7700",
"--resource res,connect,PUSh,sentinel.cz,7700",
"--resource res,connect,FOO,sentinel.cz,7700",
"--resource res,connect,FOO,sentinel.cz,7700",
"--resource res,connect,PUSH,sentinel.cz,0",
"--resource res,connect,PUSH,*,8800",
"--resource res,conn,PUSH,127.0.0.1,8800",
"--resource res,connect,PUSH,localhost,8800"
" --resource res,connect,PUSH,localhost,8800",
"--resource res,bind,PULL,localhost,8800"
" --resource res,bind,PULL,localhost,8800",
])
def bad_resources_mock(request):
with patch("sys.argv", args_from_string(request.param)) as m:
yield m
@pytest.fixture(params=[
"--resource res,connect,PUSH,127.0.0.1,8800",
"--resource res,connect,PUSH,setinel.turris.cz,8800",
])
def connect_resources_mock(request):
with patch("sys.argv", args_from_string(request.param)) as m:
yield m
@pytest.fixture(params=[
"--resource res,bind,PUSH,*,8800",
"--resource res,bind,PULL,*,8801",
"--resource res,bind,PULL,127.0.0.1,8802",
])
def bind_resources_mock(request):
with patch("sys.argv", args_from_string(request.param)) as m:
yield m
@pytest.fixture(params=[
"--resource res1,connect,PUSH,localhost,8800"
" --resource res2,connect,PUSH,localhost,8800",
"--resource res1,connect,PUSH,localhost,8800"
" --resource res1,connect,PUSH,localhost,8801"
" --resource res2,connect,PUB,localhost,8802",
"--resource res1,connect,PUB,localhost,8800"
" --resource res1,connect,PUB,localhost,8801"
" --resource res2,connect,PUB,localhost,8802",
])
def multisock_resources_mock(request):
with patch("sys.argv", args_from_string(request.param)) as m:
yield m
......@@ -5,91 +5,48 @@ import zmq
import sn
def test_empty_args(arg_parser):
with pytest.raises(SystemExit):
arg_parser.parse_args([])
def test_empty_args_native(zmq_context):
def test_empty_args(zmq_context, empty_args_mock):
with pytest.raises(SystemExit):
ctx = sn.SN(zmq_context)
def test_undefined_resource(zmq_context):
def test_undefined_resource(zmq_context, one_resource_mock):
with pytest.raises(sn.UndefinedSocketError):
ctx = sn.SN(zmq_context, args="--resource out,connect,PUSH,sentinel.cz,7700".split(" "))
ctx = sn.SN(zmq_context)
assert ctx.get_socket("in")
@pytest.mark.parametrize("bad_arg", [
"--resource",
"--resource BAD",
"--resource res,connect,PUSH,,",
"--resource res,connect,PUSH,sentinel.cz,",
"--resource res,connect,PULL,,7700",
"--resource ,connect,PUSH,sentinel.cz,7700",
"--resource connect,PUSH,sentinel.cz,7700",
"--resource res,connect,PUSh,sentinel.cz,7700",
"--resource res,connect,FOO,sentinel.cz,7700",
"--resource res,connect,FOO,sentinel.cz,7700",
"--resource res,connect,PUSH,sentinel.cz,0",
"--resource res,connect,PUSH,*,8800",
"--resource res,conn,PUSH,127.0.0.1,8800",
"--resource res,connect,PUSH,localhost,8800"
" --resource res,connect,PUSH,localhost,8800",
"--resource res,bind,PULL,localhost,8800"
" --resource res,bind,PULL,localhost,8800",
])
def test_bad_args(zmq_context, arg_parser, bad_arg):
def test_bad_args(zmq_context, bad_resources_mock):
with pytest.raises((SystemExit, sn.SockConfigError, zmq.error.ZMQError)):
ctx = sn.SN(zmq_context, arg_parser, args=bad_arg.split(" "))
ctx = sn.SN(zmq_context)
assert ctx.get_socket("res")
@pytest.mark.parametrize("conn_arg", [
"--resource res,connect,PUSH,127.0.0.1,8800",
"--resource res,connect,PUSH,setinel.turris.cz,8800",
])
def test_connect_args(zmq_context, arg_parser, conn_arg):
ctx = sn.SN(zmq_context, arg_parser, args=conn_arg.split(" "))
def test_connect_args(zmq_context, connect_resources_mock):
ctx = sn.SN(zmq_context)
assert ctx.get_socket("res")
@pytest.mark.parametrize("bind_arg", [
"--resource res,bind,PUSH,*,8800",
"--resource res,bind,PULL,*,8801",
"--resource res,bind,PULL,127.0.0.1,8802",
])
def test_bind_args(zmq_context, arg_parser, bind_arg):
ctx = sn.SN(zmq_context, arg_parser, args=bind_arg.split(" "))
def test_bind_args(zmq_context, bind_resources_mock):
ctx = sn.SN(zmq_context)
assert ctx.get_socket("res")
@pytest.mark.parametrize("multisock_arg", [
"--resource res1,connect,PUSH,localhost,8800"
" --resource res2,connect,PUSH,localhost,8800",
"--resource res1,connect,PUSH,localhost,8800"
" --resource res1,connect,PUSH,localhost,8801"
" --resource res2,connect,PUB,localhost,8802",
"--resource res1,connect,PUB,localhost,8800"
" --resource res1,connect,PUB,localhost,8801"
" --resource res2,connect,PUB,localhost,8802",
])
def test_multisock_args(zmq_context, arg_parser, multisock_arg):
ctx = sn.SN(zmq_context, arg_parser, args=multisock_arg.split(" "))
def test_multisock_args(zmq_context, multisock_resources_mock):
ctx = sn.SN(zmq_context)
assert ctx
r1, r2 = ctx.get_socket("res1", "res2")
assert r1
assert r2
def test_required_type_accept(zmq_context, arg_parser, required_type_arg):
ctx = sn.SN(zmq_context, arg_parser, args=required_type_arg)
def test_required_type_accept(zmq_context, one_resource_mock):
ctx = sn.SN(zmq_context)
assert ctx
assert ctx.get_socket(("res", "PUSH"))
def test_required_type_decline(zmq_context, arg_parser, required_type_arg):
def test_required_type_decline(zmq_context, one_resource_mock):
with pytest.raises(sn.SockConfigError):
ctx = sn.SN(zmq_context, arg_parser, args=required_type_arg)
ctx = sn.SN(zmq_context)
assert ctx
assert ctx.get_socket(("res", "PUB"))
......@@ -10,8 +10,3 @@ import sn
@pytest.fixture
def zmq_context():
return zmq.Context.instance()
@pytest.fixture
def arg_parser():
return sn.get_arg_parser()
import pytest
from unittest.mock import patch
import sn
def test_send(zmq_context, arg_parser, socket_binded):
msg = b"ping"
ctx = sn.SN(zmq_context, arg_parser, args=["--resource", "res,connect,PUSH,127.0.0.1,8800"])
s = ctx.get_socket("res")
assert s
s.send(msg)
m = socket_binded.recv()
assert m == msg
s.close()
ctx.context.destroy()
def test_send(zmq_context, socket_binded):
with patch("sys.argv", ["prog", "--resource", "res,connect,PUSH,127.0.0.1,8800"]):
msg = b"ping"
ctx = sn.SN(zmq_context)
s = ctx.get_socket("res")
assert s
s.send(msg)
m = socket_binded.recv()
assert m == msg
s.close()
ctx.context.destroy()
def test_recv(zmq_context, arg_parser, socket_connected):
msg = b"ping"
ctx = sn.SN(zmq_context, arg_parser, args=["--resource", "res,bind,PULL,127.0.0.1,8800"])
s = ctx.get_socket("res")
assert s
socket_connected.send(msg)
m = s.recv()
assert m == msg
s.close()
ctx.context.destroy()
def test_recv(zmq_context, socket_connected):
with patch("sys.argv", ["prog", "--resource", "res,bind,PULL,127.0.0.1,8800"]):
msg = b"ping"
ctx = sn.SN(zmq_context)
s = ctx.get_socket("res")
assert s
socket_connected.send(msg)
m = s.recv()
assert m == msg
s.close()
ctx.context.destroy()
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