155 lines
6.7 KiB
Python
155 lines
6.7 KiB
Python
"""Integration contracts for the V2R1 authenticated runtime composition root."""
|
|
from __future__ import annotations
|
|
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from datetime import datetime, timedelta, timezone
|
|
import hashlib
|
|
import hmac
|
|
from tempfile import TemporaryDirectory
|
|
import unittest
|
|
|
|
from aios_v2r1.runtime import RuntimeWiring
|
|
from aios_v2r1.tenant_context import canonical_json
|
|
|
|
|
|
class MutableClock:
|
|
def __init__(self, now: datetime) -> None:
|
|
self.now = now
|
|
|
|
def __call__(self) -> datetime:
|
|
return self.now
|
|
|
|
|
|
class RuntimeWiringIntegrationTests(unittest.TestCase):
|
|
internal_key = b"runtime-internal-context-key-is-32bytes"
|
|
ingress_key = b"runtime-external-ingress-key-is-32bytes"
|
|
|
|
def setUp(self) -> None:
|
|
self.directory = TemporaryDirectory()
|
|
self.addCleanup(self.directory.cleanup)
|
|
self.clock = MutableClock(datetime(2026, 7, 13, tzinfo=timezone.utc))
|
|
self.runtime = RuntimeWiring.create(
|
|
self.directory.name,
|
|
signing_key=self.internal_key,
|
|
ingress_hmac_key=self.ingress_key,
|
|
clock=self.clock,
|
|
)
|
|
self.runtime.router.register_tenant("tenant-a", "cell-a")
|
|
self.runtime.router.register_tenant("tenant-b", "cell-b")
|
|
self.runtime.router.bind_channel(
|
|
tenant_id="tenant-a", channel="weixin", channel_account_id="account-a"
|
|
)
|
|
self.runtime.router.bind_channel(
|
|
tenant_id="tenant-b", channel="weixin", channel_account_id="account-b"
|
|
)
|
|
self.runtime.router.bind_principal(
|
|
tenant_id="tenant-a", channel="weixin", channel_account_id="account-a",
|
|
principal_id="user-a",
|
|
)
|
|
self.runtime.router.bind_session(
|
|
tenant_id="tenant-a", session_id="session-a", cell_id="cell-a",
|
|
principal_id="user-a", ttl_seconds=60,
|
|
)
|
|
|
|
def message(
|
|
self, *, event_id: str = "event-a", message_id: str = "message-a",
|
|
body: str = "hello", **changes: object,
|
|
) -> dict[str, object]:
|
|
values: dict[str, object] = {
|
|
"tenant_id": "tenant-a", "cell_id": "cell-a", "principal_id": "user-a",
|
|
"principal_type": "authenticated_user", "origin_type": "channel_user",
|
|
"channel": "weixin", "channel_account_id": "account-a", "peer_id": "peer-a",
|
|
"conversation_id": "conversation-a", "session_id": "session-a", "task_id": None,
|
|
"event_id": event_id, "message_id": message_id, "reply_to_message_id": None,
|
|
"direction": "inbound", "command_capable": True,
|
|
"occurred_at": "2026-07-13T00:00:00+00:00",
|
|
"received_at": "2026-07-13T00:00:01+00:00",
|
|
"content": {"body": body, "quoted_body": None},
|
|
"adapter": {"name": "weixin", "version": "v2"},
|
|
}
|
|
values.update(changes)
|
|
return values
|
|
|
|
def request(self, **changes: object) -> dict[str, object]:
|
|
values: dict[str, object] = {
|
|
"principal_id": "user-a", "tenant_id": "tenant-a", "cell_id": "cell-a",
|
|
"channel": "weixin", "channel_account_id": "account-a", "session_id": "session-a",
|
|
"message": self.message(),
|
|
}
|
|
values.update(changes)
|
|
values["signature"] = hmac.new(
|
|
self.ingress_key, canonical_json(values), hashlib.sha256
|
|
).hexdigest()
|
|
return values
|
|
|
|
def ingest(self, **changes: object):
|
|
return self.runtime.ingest(request=self.request(**changes))
|
|
|
|
def test_real_ingress_appends_exact_record_and_reopens(self) -> None:
|
|
first = self.ingest()
|
|
self.assertEqual("ACCEPTED", first.status)
|
|
assert first.result is not None
|
|
self.assertEqual("message-a", first.result.record.message_id)
|
|
self.assertEqual("tenant-a", first.result.route.tenant_id)
|
|
self.assertEqual(1, self.runtime.ledger.count())
|
|
self.assertEqual(1, self.runtime.session_pool.active_count())
|
|
|
|
reopened = RuntimeWiring.create(
|
|
self.directory.name, signing_key=self.internal_key,
|
|
ingress_hmac_key=self.ingress_key, clock=self.clock,
|
|
)
|
|
repeated = reopened.ingest(request=self.request())
|
|
self.assertEqual("ACCEPTED", repeated.status)
|
|
assert repeated.result is not None
|
|
self.assertEqual(first.result.record, repeated.result.record)
|
|
self.assertEqual(1, reopened.ledger.count())
|
|
self.assertEqual(1, reopened.session_pool.active_count())
|
|
|
|
def test_authoritative_tenant_route_rejects_cross_tenant_ingress_without_mutation(self) -> None:
|
|
result = self.ingest(
|
|
tenant_id="tenant-b", cell_id="cell-b",
|
|
message=self.message(tenant_id="tenant-b", cell_id="cell-b"),
|
|
)
|
|
self.assertEqual("AUTH_FAILED", result.status)
|
|
self.assertIsNone(result.result)
|
|
self.assertEqual(0, self.runtime.ledger.count())
|
|
self.assertEqual(0, self.runtime.session_pool.active_count())
|
|
|
|
def test_identical_concurrent_ingress_is_one_exact_ledger_record_and_one_session(self) -> None:
|
|
with ThreadPoolExecutor(max_workers=24) as executor:
|
|
results = list(executor.map(lambda _: self.ingest(), range(24)))
|
|
self.assertEqual({"ACCEPTED"}, {result.status for result in results})
|
|
self.assertEqual(1, self.runtime.ledger.count())
|
|
self.assertEqual(1, self.runtime.session_pool.active_count())
|
|
|
|
def test_collision_is_sanitized_and_does_not_leave_partial_session_or_ledger_state(self) -> None:
|
|
self.assertEqual("ACCEPTED", self.ingest(
|
|
message=self.message(event_id="event-a", message_id="message-a", body="first")
|
|
).status)
|
|
collision = self.ingest(
|
|
message=self.message(event_id="event-a", message_id="message-b", body="second")
|
|
)
|
|
self.assertEqual("CONFLICT", collision.status)
|
|
self.assertIsNone(collision.result)
|
|
self.assertEqual(1, self.runtime.ledger.count())
|
|
self.assertEqual(1, self.runtime.session_pool.active_count())
|
|
|
|
def test_unknown_cross_and_expired_sessions_fail_closed_without_mutation(self) -> None:
|
|
unknown = self.ingest(
|
|
session_id="unknown-session", message=self.message(session_id="unknown-session")
|
|
)
|
|
self.assertEqual("AUTH_FAILED", unknown.status)
|
|
cross = self.ingest(
|
|
principal_id="user-b", message=self.message(principal_id="user-b")
|
|
)
|
|
self.assertEqual("AUTH_FAILED", cross.status)
|
|
self.clock.now += timedelta(seconds=61)
|
|
expired = self.ingest()
|
|
self.assertEqual("AUTH_FAILED", expired.status)
|
|
self.assertEqual(0, self.runtime.ledger.count())
|
|
self.assertEqual(0, self.runtime.session_pool.active_count())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|