68 lines
2.7 KiB
Python
68 lines
2.7 KiB
Python
import io
|
|
import json
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from user_engine.adapters.registration_verification import (
|
|
HTTPRegistrationVerificationAdapter,
|
|
)
|
|
from user_engine.ports import RegistrationVerificationRequest
|
|
|
|
|
|
class Response(io.BytesIO):
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): return None
|
|
|
|
|
|
class RegistrationVerificationAdapterTests(unittest.TestCase):
|
|
@patch("user_engine.adapters.registration_verification.urlopen")
|
|
def test_request_sends_bound_context_and_returns_only_opaque_receipt(self, opener):
|
|
opener.return_value = Response(
|
|
json.dumps({"request_id": "vrq_opaque", "accepted": True}).encode()
|
|
)
|
|
adapter = HTTPRegistrationVerificationAdapter(
|
|
base_url="http://verification", bearer_token="secret\n"
|
|
)
|
|
receipt = adapter.request(RegistrationVerificationRequest(
|
|
registration_id="reg_1",
|
|
normalized_email="person@example.test",
|
|
preferred_username="person",
|
|
client_id="coulomb-social",
|
|
tenant="tenant:coulomb",
|
|
correlation_id="corr_1",
|
|
))
|
|
self.assertEqual("vrq_opaque", receipt.request_id)
|
|
request = opener.call_args.args[0]
|
|
self.assertEqual("Bearer secret", request.headers["Authorization"])
|
|
self.assertNotIn(b"return_to", request.data)
|
|
|
|
@patch("user_engine.adapters.registration_verification.urlopen")
|
|
def test_consume_requires_registration_purpose_and_normalizes_email(self, opener):
|
|
opener.return_value = Response(json.dumps({
|
|
"purpose": "public-registration",
|
|
"verification_id": "fvr_1",
|
|
"registration_id": "reg_1",
|
|
"email": "Person@Example.Test ",
|
|
"preferred_username": "person",
|
|
"client_id": "coulomb-social",
|
|
"tenant": "tenant:coulomb",
|
|
"source_system": "mail-verifier",
|
|
"assurance": {"mailbox_control": True},
|
|
}).encode())
|
|
adapter = HTTPRegistrationVerificationAdapter(
|
|
base_url="http://verification", bearer_token="secret"
|
|
)
|
|
evidence = adapter.consume("x" * 32)
|
|
self.assertEqual("person@example.test", evidence.normalized_email)
|
|
self.assertTrue(evidence.assurance["mailbox_control"])
|
|
|
|
@patch("user_engine.adapters.registration_verification.urlopen")
|
|
def test_consume_rejects_wrong_purpose(self, opener):
|
|
opener.return_value = Response(json.dumps({
|
|
"purpose": "password-reset"
|
|
}).encode())
|
|
adapter = HTTPRegistrationVerificationAdapter(
|
|
base_url="http://verification", bearer_token="secret"
|
|
)
|
|
with self.assertRaisesRegex(RuntimeError, "wrong purpose"):
|
|
adapter.consume("x" * 32)
|