108 lines
4.6 KiB
Python
108 lines
4.6 KiB
Python
import io
|
|
import json
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from urllib.error import HTTPError
|
|
|
|
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"])
|
|
payload = json.loads(request.data)
|
|
self.assertEqual("person@example.test", payload["normalized_email"])
|
|
self.assertNotIn("email", payload)
|
|
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)
|
|
|
|
@patch("user_engine.adapters.registration_verification.urlopen")
|
|
def test_cancel_requires_cancellation_purpose(self, opener):
|
|
opener.return_value = Response(json.dumps({
|
|
"purpose": "public-registration-cancel",
|
|
"verification_id": "fvc_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.cancel("x" * 32)
|
|
self.assertEqual("reg_1", evidence.registration_id)
|
|
self.assertTrue(opener.call_args.args[0].full_url.endswith("/cancel"))
|
|
|
|
@patch("user_engine.adapters.registration_verification.urlopen")
|
|
def test_expired_and_replayed_handles_have_the_same_redacted_failure(self, opener):
|
|
adapter = HTTPRegistrationVerificationAdapter(
|
|
base_url="http://verification", bearer_token="secret"
|
|
)
|
|
messages = []
|
|
for status in (409, 410):
|
|
opener.side_effect = HTTPError(
|
|
"http://verification/consume", status, "provider detail", {},
|
|
io.BytesIO(b'{"error":"account-specific detail"}'),
|
|
)
|
|
with self.assertRaises(RuntimeError) as raised:
|
|
adapter.consume("x" * 32)
|
|
messages.append(str(raised.exception))
|
|
self.assertEqual(
|
|
["registration verification was rejected"] * 2, messages
|
|
)
|
|
self.assertNotIn("account-specific", repr(messages))
|