Merge pull request #115 from muzzammilshahid/interop-session

Add interoperablity tests for session messages

tests/interoptests/messages/abort_test.py

@@ -0,0 +1,68 @@
+import pytest
+
+from tests.interoptests.helpers import run_command
+from wampproto import messages
+from wampproto.serializers import JSONSerializer, CBORSerializer, MsgPackSerializer
+
+
+def is_equal(msg1: messages.Abort, msg2: messages.Abort) -> bool:
+    return (
+        msg1.reason == msg2.reason
+        and msg1.details == msg2.details
+        and msg1.args == msg2.args
+        and msg1.kwargs == msg2.kwargs
+    )
+
+
+@pytest.mark.asyncio
+async def test_json_serializer():
+    msg = messages.Abort(messages.AbortFields(details={}, reason="crash"))
+    command = f"wampproto message abort {msg.reason} --serializer json"
+
+    output = await run_command(command)
+
+    serializer = JSONSerializer()
+    message = serializer.deserialize(output)
+
+    assert is_equal(msg, message)
+
+
+@pytest.mark.asyncio
+async def test_cbor_serializer():
+    msg = messages.Abort(messages.AbortFields(details={}, reason="crash"))
+    command = f"wampproto message abort {msg.reason} --serializer cbor --output hex"
+
+    output = await run_command(command)
+    output_bytes = bytes.fromhex(output)
+
+    serializer = CBORSerializer()
+    message = serializer.deserialize(output_bytes)
+
+    assert is_equal(msg, message)
+
+
+@pytest.mark.asyncio
+async def test_msgpack_serializer():
+    msg = messages.Abort(messages.AbortFields(details={}, reason="crash"))
+    command = f"wampproto message abort {msg.reason} --serializer msgpack --output hex"
+
+    output = await run_command(command)
+    output_bytes = bytes.fromhex(output)
+
+    serializer = MsgPackSerializer()
+    message = serializer.deserialize(output_bytes)
+
+    assert is_equal(msg, message)
+
+
+@pytest.mark.asyncio
+async def test_with_args_kwargs_details():
+    msg = messages.Abort(messages.AbortFields(details={"a": "b"}, reason="crash", args=["foo"], kwargs={"a": "b"}))
+    command = f"wampproto message abort {msg.reason} foo -d a=b -k a=b --serializer json"
+
+    output = await run_command(command)
+
+    serializer = JSONSerializer()
+    message = serializer.deserialize(output)
+
+    assert is_equal(msg, message)

tests/interoptests/messages/authenticate_test.py

@@ -0,0 +1,50 @@
+import pytest
+
+from tests.interoptests.helpers import run_command
+from wampproto import messages
+from wampproto.serializers import JSONSerializer, CBORSerializer, MsgPackSerializer
+
+
+def is_equal(msg1: messages.Authenticate, msg2: messages.Authenticate) -> bool:
+    return msg1.signature == msg2.signature and msg1.extra == msg2.extra
+
+
+@pytest.mark.asyncio
+async def test_json_serializer():
+    msg = messages.Authenticate(messages.AuthenticateFields("signature", extra={"ticket": "test"}))
+    command = f"wampproto message authenticate {msg.signature} -e ticket=test --serializer json"
+
+    output = await run_command(command)
+
+    serializer = JSONSerializer()
+    message = serializer.deserialize(output)
+
+    assert is_equal(msg, message)
+
+
+@pytest.mark.asyncio
+async def test_cbor_serializer():
+    msg = messages.Authenticate(messages.AuthenticateFields("signature"))
+    command = f"wampproto message authenticate {msg.signature} --serializer cbor --output hex"
+
+    output = await run_command(command)
+    output_bytes = bytes.fromhex(output)
+
+    serializer = CBORSerializer()
+    message = serializer.deserialize(output_bytes)
+
+    assert is_equal(msg, message)
+
+
+@pytest.mark.asyncio
+async def test_msgpack_serializer():
+    msg = messages.Authenticate(messages.AuthenticateFields("signature", extra={"secret": "test"}))
+    command = f"wampproto message authenticate {msg.signature} -e secret=test --serializer msgpack --output hex"
+
+    output = await run_command(command)
+    output_bytes = bytes.fromhex(output)
+
+    serializer = MsgPackSerializer()
+    message = serializer.deserialize(output_bytes)
+
+    assert is_equal(msg, message)

tests/interoptests/messages/cancel_test.py

@@ -0,0 +1,63 @@
+import pytest
+
+from tests.interoptests.helpers import run_command
+from wampproto import messages
+from wampproto.serializers import JSONSerializer, CBORSerializer, MsgPackSerializer
+
+
+def is_equal(msg1: messages.Cancel, msg2: messages.Cancel) -> bool:
+    return msg1.request_id == msg2.request_id and msg1.options == msg2.options
+
+
+@pytest.mark.asyncio
+async def test_json_serializer():
+    msg = messages.Cancel(messages.CancelFields(1))
+    command = f"wampproto message cancel {msg.request_id} --serializer json"
+
+    output = await run_command(command)
+
+    serializer = JSONSerializer()
+    message = serializer.deserialize(output)
+
+    assert is_equal(msg, message)
+
+
+@pytest.mark.asyncio
+async def test_cbor_serializer():
+    msg = messages.Cancel(messages.CancelFields(1))
+    command = f"wampproto message cancel {msg.request_id} --serializer cbor --output hex"
+
+    output = await run_command(command)
+    output_bytes = bytes.fromhex(output)
+
+    serializer = CBORSerializer()
+    message = serializer.deserialize(output_bytes)
+
+    assert is_equal(msg, message)
+
+
+@pytest.mark.asyncio
+async def test_msgpack_serializer():
+    msg = messages.Cancel(messages.CancelFields(1))
+    command = f"wampproto message cancel {msg.request_id} --serializer msgpack --output hex"
+
+    output = await run_command(command)
+    output_bytes = bytes.fromhex(output)
+
+    serializer = MsgPackSerializer()
+    message = serializer.deserialize(output_bytes)
+
+    assert is_equal(msg, message)
+
+
+@pytest.mark.asyncio
+async def test_with_options():
+    msg = messages.Cancel(messages.CancelFields(1, options={"a": "b"}))
+    command = f"wampproto message cancel {msg.request_id} -o a=b --serializer json"
+
+    output = await run_command(command)
+
+    serializer = JSONSerializer()
+    message = serializer.deserialize(output)
+
+    assert is_equal(msg, message)

tests/interoptests/messages/challenge_test.py

@@ -0,0 +1,50 @@
+import pytest
+
+from tests.interoptests.helpers import run_command
+from wampproto import messages
+from wampproto.serializers import JSONSerializer, CBORSerializer, MsgPackSerializer
+
+
+def is_equal(msg1: messages.Challenge, msg2: messages.Challenge) -> bool:
+    return msg1.authmethod == msg2.authmethod and msg1.extra == msg2.extra
+
+
+@pytest.mark.asyncio
+async def test_json_serializer():
+    msg = messages.Challenge(messages.ChallengeFields("ticket", extra={"ticket": "test"}))
+    command = f"wampproto message challenge {msg.authmethod} -e ticket=test --serializer json"
+
+    output = await run_command(command)
+
+    serializer = JSONSerializer()
+    message = serializer.deserialize(output)
+
+    assert is_equal(msg, message)
+
+
+@pytest.mark.asyncio
+async def test_cbor_serializer():
+    msg = messages.Challenge(messages.ChallengeFields("anonymous"))
+    command = f"wampproto message challenge {msg.authmethod} --serializer cbor --output hex"
+
+    output = await run_command(command)
+    output_bytes = bytes.fromhex(output)
+
+    serializer = CBORSerializer()
+    message = serializer.deserialize(output_bytes)
+
+    assert is_equal(msg, message)
+
+
+@pytest.mark.asyncio
+async def test_msgpack_serializer():
+    msg = messages.Challenge(messages.ChallengeFields("wampcra", extra={"secret": "test"}))
+    command = f"wampproto message challenge {msg.authmethod} -e secret=test --serializer msgpack --output hex"
+
+    output = await run_command(command)
+    output_bytes = bytes.fromhex(output)
+
+    serializer = MsgPackSerializer()
+    message = serializer.deserialize(output_bytes)
+
+    assert is_equal(msg, message)

tests/interoptests/messages/error_test.py

@@ -0,0 +1,72 @@
+import pytest
+
+from tests.interoptests.helpers import run_command
+from wampproto import messages
+from wampproto.serializers import JSONSerializer, CBORSerializer, MsgPackSerializer
+
+
+def is_equal(msg1: messages.Error, msg2: messages.Error) -> bool:
+    return (
+        msg1.message_type == msg2.message_type
+        and msg1.request_id == msg2.request_id
+        and msg1.uri == msg2.uri
+        and msg1.details == msg2.details
+        and msg1.args == msg2.args
+        and msg1.kwargs == msg2.kwargs
+    )
+
+
+@pytest.mark.asyncio
+async def test_json_serializer():
+    msg = messages.Error(messages.ErrorFields(1, 1, "wamp.error"))
+    command = f"wampproto message error {msg.message_type} {msg.request_id} {msg.uri} --serializer json"
+
+    output = await run_command(command)
+
+    serializer = JSONSerializer()
+    message = serializer.deserialize(output)
+
+    assert is_equal(msg, message)
+
+
+@pytest.mark.asyncio
+async def test_cbor_serializer():
+    msg = messages.Error(messages.ErrorFields(1, 1, "wamp.error"))
+    command = f"wampproto message error {msg.message_type} {msg.request_id} {msg.uri} --serializer cbor --output hex"
+
+    output = await run_command(command)
+    output_bytes = bytes.fromhex(output)
+
+    serializer = CBORSerializer()
+    message = serializer.deserialize(output_bytes)
+
+    assert is_equal(msg, message)
+
+
+@pytest.mark.asyncio
+async def test_msgpack_serializer():
+    msg = messages.Error(messages.ErrorFields(1, 1, "wamp.error"))
+    command = f"wampproto message error {msg.message_type} {msg.request_id} {msg.uri} --serializer msgpack --output hex"
+
+    output = await run_command(command)
+    output_bytes = bytes.fromhex(output)
+
+    serializer = MsgPackSerializer()
+    message = serializer.deserialize(output_bytes)
+
+    assert is_equal(msg, message)
+
+
+@pytest.mark.asyncio
+async def test_with_args_kwargs_details():
+    msg = messages.Error(messages.ErrorFields(1, 1, "wamp.error", details={"a": "b"}, args=["foo"], kwargs={"a": "b"}))
+    command = (
+        f"wampproto message error {msg.message_type} {msg.request_id} {msg.uri} foo -d a=b -k a=b --serializer json"
+    )
+
+    output = await run_command(command)
+
+    serializer = JSONSerializer()
+    message = serializer.deserialize(output)
+
+    assert is_equal(msg, message)

tests/interoptests/messages/goodbye_test.py

@@ -0,0 +1,63 @@
+import pytest
+
+from tests.interoptests.helpers import run_command
+from wampproto import messages
+from wampproto.serializers import JSONSerializer, CBORSerializer, MsgPackSerializer
+
+
+def is_equal(msg1: messages.Goodbye, msg2: messages.Goodbye) -> bool:
+    return msg1.reason == msg2.reason and msg1.details == msg2.details
+
+
+@pytest.mark.asyncio
+async def test_json_serializer():
+    msg = messages.Goodbye(messages.GoodbyeFields(details={}, reason="crash"))
+    command = f"wampproto message goodbye {msg.reason} --serializer json"
+
+    output = await run_command(command)
+
+    serializer = JSONSerializer()
+    message = serializer.deserialize(output)
+
+    assert is_equal(msg, message)
+
+
+@pytest.mark.asyncio
+async def test_cbor_serializer():
+    msg = messages.Goodbye(messages.GoodbyeFields(details={}, reason="crash"))
+    command = f"wampproto message goodbye {msg.reason} --serializer cbor --output hex"
+
+    output = await run_command(command)
+    output_bytes = bytes.fromhex(output)
+
+    serializer = CBORSerializer()
+    message = serializer.deserialize(output_bytes)
+
+    assert is_equal(msg, message)
+
+
+@pytest.mark.asyncio
+async def test_msgpack_serializer():
+    msg = messages.Goodbye(messages.GoodbyeFields(details={}, reason="crash"))
+    command = f"wampproto message goodbye {msg.reason} --serializer msgpack --output hex"
+
+    output = await run_command(command)
+    output_bytes = bytes.fromhex(output)
+
+    serializer = MsgPackSerializer()
+    message = serializer.deserialize(output_bytes)
+
+    assert is_equal(msg, message)
+
+
+@pytest.mark.asyncio
+async def test_with_details():
+    msg = messages.Goodbye(messages.GoodbyeFields(details={"a": "b"}, reason="crash"))
+    command = f"wampproto message goodbye {msg.reason} -d a=b --serializer json"
+
+    output = await run_command(command)
+
+    serializer = JSONSerializer()
+    message = serializer.deserialize(output)
+
+    assert is_equal(msg, message)