tegwick
93b6ba2aae
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a11fd9-caf6-7a63-b578-9daaaa134ff2
131 lines
4.9 KiB
Python
131 lines
4.9 KiB
Python
"""The user-engine draft parses and matches today's call set. It is not deployed."""
|
|
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from email_connect.callers import ScopedTransactionalApplication, load_callers
|
|
from email_connect.transactional import SQLiteDeliveryStore
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
DRAFT = ROOT / "config" / "callers.user-engine.draft.json"
|
|
TOKEN = "synthetic-user-engine-token"
|
|
|
|
|
|
class Provider:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def send(self, *args):
|
|
self.calls.append(args)
|
|
return "synthetic-reference"
|
|
|
|
def check(self):
|
|
return None
|
|
|
|
|
|
def load_draft(tmp_path):
|
|
token = tmp_path / "token"
|
|
token.write_text(TOKEN + "\n")
|
|
document = json.loads(DRAFT.read_text(encoding="utf-8"))
|
|
document["callers"][0]["token_file"] = str(token)
|
|
path = tmp_path / "policy.json"
|
|
path.write_text(json.dumps(document))
|
|
return load_callers(path)
|
|
|
|
|
|
def invoke(app, payload, path):
|
|
raw = json.dumps(payload).encode()
|
|
response = {}
|
|
body = b"".join(
|
|
app(
|
|
{
|
|
"PATH_INFO": path,
|
|
"REQUEST_METHOD": "POST",
|
|
"CONTENT_LENGTH": str(len(raw)),
|
|
"wsgi.input": io.BytesIO(raw),
|
|
"HTTP_AUTHORIZATION": "Bearer " + TOKEN,
|
|
"HTTP_IDEMPOTENCY_KEY": payload.get("id", ""),
|
|
},
|
|
lambda status, headers: response.update(status=int(status.split()[0])),
|
|
)
|
|
)
|
|
return response["status"], json.loads(body)
|
|
|
|
|
|
def test_draft_admits_only_the_live_registration_pair(tmp_path):
|
|
policies = load_draft(tmp_path)
|
|
assert len(policies) == 1
|
|
policy = policies[0]
|
|
assert policy.caller_id == "user-engine"
|
|
assert policy.source == "user-engine"
|
|
assert policy.bindings == frozenset({("tenant:coulomb", "coulomb-social")})
|
|
assert policy.binding_patterns == frozenset({
|
|
("tenant:*-family", "coulomb-social"), ("tenant:family-*", "coulomb-social")
|
|
})
|
|
assert policy.operations == frozenset(
|
|
{"send", "delivery-status", "status", "verify", "consume", "cancel"}
|
|
)
|
|
assert "suppress" not in policy.operations
|
|
assert policy.templates == frozenset({"family_member.invited", "family_invitation.resent"})
|
|
assert (policy.rate_per_minute, policy.burst, policy.max_payload_bytes) == (10, 5, 8192)
|
|
assert policy.adopt_legacy_verifications is True
|
|
assert policy.adopt_legacy_invitations is True
|
|
assert policy.legacy_invitation_binding is None
|
|
assert policy.token == TOKEN
|
|
assert TOKEN not in DRAFT.read_text(encoding="utf-8")
|
|
assert "token" not in json.loads(DRAFT.read_text(encoding="utf-8"))["callers"][0]
|
|
|
|
|
|
def test_current_invitation_shape_is_denied_and_registration_pair_is_allowed(tmp_path):
|
|
app = ScopedTransactionalApplication(
|
|
SQLiteDeliveryStore(str(tmp_path / "ledger.db")),
|
|
Provider(),
|
|
"https://issuer.example.test",
|
|
load_draft(tmp_path),
|
|
)
|
|
invitation = {
|
|
"id": "evt-1",
|
|
"type": "family_member.invited",
|
|
"source": "user-engine",
|
|
"tenant": "tenant:family",
|
|
"data": {"primary_email": "member@example.test", "invitation_id": "invite-1"},
|
|
}
|
|
status, body = invoke(app, invitation, "/v1/send")
|
|
assert status == 400 and body["error"] == "caller_scope_denied"
|
|
assert app.apps["user-engine"].provider.calls == []
|
|
|
|
verification = {
|
|
"registration_id": "reg-1",
|
|
"normalized_email": "member@example.test",
|
|
"preferred_username": "member",
|
|
"client_id": "coulomb-social",
|
|
"tenant": "tenant:coulomb",
|
|
"correlation_id": "trace-1",
|
|
}
|
|
status, body = invoke(app, verification, "/v1/registration-verifications")
|
|
assert status == 202 and body["accepted"] is True
|
|
assert len(app.apps["user-engine"].provider.calls) == 1
|
|
|
|
status, body = invoke(app, {"recipient": "member@example.test"}, "/v1/suppressions")
|
|
assert status == 403 and body["error"] == "operation_not_allowed"
|
|
|
|
|
|
def test_transport_draft_matches_receiver_bindings_and_accepts_family_invitation(tmp_path):
|
|
transport = json.loads((ROOT / "config" / "mail-client-bindings.user-engine.draft.json").read_text())
|
|
policies = load_draft(tmp_path)
|
|
policy = policies[0]
|
|
assert {tuple(pair) for pair in transport} == policy.bindings | policy.binding_patterns
|
|
provider = Provider()
|
|
app = ScopedTransactionalApplication(
|
|
SQLiteDeliveryStore(str(tmp_path / "ledger.db")), provider,
|
|
"https://issuer.example.test", policies,
|
|
)
|
|
invitation = {
|
|
"id": "evt-compatible", "type": "family_member.invited", "source": "user-engine",
|
|
"tenant": "tenant:synthetic-family", "client_id": "coulomb-social",
|
|
"data": {"primary_email": "member@example.test", "invitation_id": "invite-compatible"},
|
|
}
|
|
status, body = invoke(app, invitation, "/v1/send")
|
|
assert status == 202 and body["status"] == "accepted"
|
|
assert len(provider.calls) == 1
|