Andros Fenollosa
bfd8b510b6
Clean architecture (core use cases + infra gateways), X25519/AES-GCM E2E encryption, tkinter GUI, and a pytest suite. 45 tests passing.
63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
"""Tests for decrypting incoming frames into messages."""
|
|
|
|
from neotalk.core.entities.constants import MessageKind
|
|
from neotalk.core.entities.responses import ResponseTypes
|
|
from neotalk.core.use_cases.conversation.compose_message import (
|
|
compose_message_use_case,
|
|
compose_typing_use_case,
|
|
)
|
|
from neotalk.core.use_cases.conversation.receive_frame import (
|
|
receive_frame_use_case,
|
|
)
|
|
from tests.fakes import ExplodingCrypto
|
|
|
|
|
|
def test_receive_frame_recovers_a_message(crypto, shared_key):
|
|
# Given a frame composed by the sender
|
|
sent = compose_message_use_case("hola", crypto, shared_key)["data"]["frame"]
|
|
|
|
# When the receiver opens it
|
|
response = receive_frame_use_case(sent, "bob", crypto, shared_key)
|
|
|
|
# Then it becomes an incoming message authored by the peer
|
|
assert response["type"] == ResponseTypes.SUCCESS
|
|
message = response["data"]["message"]
|
|
assert message.text == "hola"
|
|
assert message.author == "bob"
|
|
assert message.outgoing is False
|
|
assert message.kind == MessageKind.MESSAGE
|
|
|
|
|
|
def test_receive_frame_recovers_a_typing_signal(crypto, shared_key):
|
|
# Given a typing frame
|
|
sent = compose_typing_use_case(crypto, shared_key)["data"]["frame"]
|
|
|
|
# When received
|
|
response = receive_frame_use_case(sent, "bob", crypto, shared_key)
|
|
|
|
# Then it is a typing message with no text
|
|
message = response["data"]["message"]
|
|
assert message.kind == MessageKind.TYPING
|
|
assert message.text == ""
|
|
|
|
|
|
def test_receive_frame_rejects_a_non_secure_frame(crypto, shared_key):
|
|
# Given a frame of the wrong type
|
|
frame = {"type": "handshake"}
|
|
|
|
# When received
|
|
response = receive_frame_use_case(frame, "bob", crypto, shared_key)
|
|
|
|
# Then it is a parameters error
|
|
assert response["type"] == ResponseTypes.PARAMETERS_ERROR
|
|
|
|
|
|
def test_receive_frame_reports_a_decryption_failure(shared_key):
|
|
# Given a secure frame but crypto that cannot open it
|
|
frame = {"type": "secure", "body": "AAAA"}
|
|
|
|
# When received
|
|
response = receive_frame_use_case(frame, "bob", ExplodingCrypto(), shared_key)
|
|
|
|
# Then it is surfaced as a system error, never an exception
|
|
assert response["type"] == ResponseTypes.SYSTEM_ERROR
|