test_team_mail_recipient_contract_isolated.py
Rama main · réplica de lectura del repositorio original.
"""Recipient preconditions through the pure adapter and isolated Hub storage.
No server import, network transport, provider process, or production state.
Run with the guarded public_chat_share_safe_tests.py runner.
"""
from contextlib import closing
import json
from pathlib import Path
import tempfile
import time
import unittest
from unittest import mock
import uuid
from agentsdock_team_hub.secure_peer import PeerAuthorization, ProxyRequest
from agentsdock_team_hub.secure_peer_hub import SecurePeerHubAdapter
from agentsdock_team_hub.store import HubError, HubStore
class TeamMailRecipientContractTests(unittest.TestCase):
def setUp(self):
temporary = tempfile.TemporaryDirectory(prefix="team-recipient-contract-")
self.addCleanup(temporary.cleanup)
self.store = HubStore(Path(temporary.name) / "hub",
managed_host_identity="recipient-contract-host")
self.store.bootstrap_managed_network("Recipient contract fixture")
self.host = self.store.managed_server_claims()
self.team = self.host.team_id
self.adapter = SecurePeerHubAdapter(self.store)
self.sender, _ = self.add_peer("recipient-contract-sender")
self.recipient, self.node = self.add_peer("recipient-contract-target")
self.lifecycle = self.store.get_network_server(
self.host, self.team, self.node)["server"]["mail_route_lifecycle_id"]
def add_peer(self, identity):
peer = PeerAuthorization(
str(uuid.uuid4()), str(uuid.uuid4()), identity, self.team,
frozenset({"teamspace.read", "teamspace.write"}),
"sha256:" + "a" * 64, int(time.time()) + 3600, "Fixture peer")
self.adapter.provision_peer({"peer_id": peer.peer_id,
"peer_server_identity": identity, "team_id": self.team},
display_name=peer.peer_display_name)
self.adapter.record_peer_heartbeat(peer.peer_id, self.team)
claims = self.store.secure_peer_claims(peer_id=peer.peer_id,
peer_server_identity=identity, team_id=self.team, scopes=peer.scopes,
expires_at=peer.certificate_expires_at)
node = self.store.list_team_messages(claims, self.team, box="inbox")["address"]["id"]
return peer, node
def body(self, recipient=None):
return {"kind": "message", "body": "Synthetic recipient contract mail",
"recipients": [recipient if recipient is not None else {
"kind": "server", "id": self.node,
"mail_route_lifecycle_id": self.lifecycle}],
"idempotency_key": "recipient-contract-" + uuid.uuid4().hex}
def request(self, body):
return ProxyRequest("POST", f"/v1/teams/{self.team}/network/messages", "", (),
json.dumps(body).encode("utf-8"), self.sender)
def message_count(self):
with closing(self.store.connect()) as connection:
return connection.execute("SELECT COUNT(*) FROM team_messages WHERE team_id=?",
(self.team,)).fetchone()[0]
def assert_rejected(self, body, *, code, status):
before = self.message_count()
response = self.adapter.forward(self.request(body))
self.assertEqual(response.status, status)
self.assertEqual(json.loads(response.body)["error"]["code"], code)
self.assertEqual(self.message_count(), before, "rejection must not commit mail")
def test_adapter_preserves_exact_lifecycle_and_member_to_member_commit(self):
body = self.body()
self.assertEqual(self.adapter._team_message_body(self.request(body)), body)
with mock.patch.object(self.store, "create_team_message",
wraps=self.store.create_team_message) as create:
response = self.adapter.forward(self.request(body))
self.assertEqual(response.status, 200)
self.assertEqual(create.call_args.args[2]["recipients"], body["recipients"])
self.assertEqual(self.message_count(), 1)
result = json.loads(response.body)["message"]
self.assertEqual(result["body"], body["body"])
self.assertEqual(result["recipients"][0]["id"], self.node)
def test_unknown_recipient_fields_still_fail_before_store(self):
for field in ("unexpected", "server_identity", "lifecycle_id"):
with self.subTest(field=field):
body = self.body()
body["recipients"][0][field] = "not-authorized"
with mock.patch.object(self.store, "create_team_message") as create:
self.assert_rejected(body, code="invalid_request", status=422)
create.assert_not_called()
def test_original_unbound_and_explicit_null_recipients_still_work(self):
for recipient in ({"kind": "server", "id": self.node},
{"kind": "server", "id": self.node, "mail_route_lifecycle_id": None}):
with self.subTest(recipient=recipient):
body = self.body(recipient)
self.assertEqual(self.adapter._team_message_body(self.request(body)), body)
response = self.adapter.forward(self.request(body))
self.assertEqual(response.status, 200)
self.assertEqual(self.message_count(), 2)
def test_store_rejects_invalid_lifecycle_values_after_shape_validation(self):
for value in ("", "a" * 63, "A" * 64, "g" * 64, True, 123, [], {}):
with self.subTest(value=value):
body = self.body()
body["recipients"][0]["mail_route_lifecycle_id"] = value
parsed = self.adapter._team_message_body(self.request(body))
with self.assertRaises(HubError) as failure:
self.store.create_team_message(self.host, self.team, parsed)
self.assertEqual((failure.exception.code, failure.exception.status_code),
("invalid_request", 422))
self.assertEqual(self.message_count(), 0)
def test_lifecycle_precondition_cannot_authorize_other_recipient_kinds(self):
for kind, target in (("human", "fixture-human"), ("all", "all"),
("all_servers", "all_servers")):
with self.subTest(kind=kind):
self.assert_rejected(self.body({"kind": kind, "id": target,
"mail_route_lifecycle_id": self.lifecycle}), code="invalid_request", status=422)
def test_stale_lifecycle_is_not_dropped_by_adapter(self):
body = self.body()
body["recipients"][0]["mail_route_lifecycle_id"] = (
"b" if self.lifecycle != "b" * 64 else "c") * 64
self.assert_rejected(body, code="mail_route_changed", status=409)
def test_revoked_recipient_fails_and_rejoin_requires_fresh_lifecycle(self):
old_body = self.body()
self.store.revoke_secure_peer_service(peer_id=self.recipient.peer_id, team_id=self.team)
self.assert_rejected(old_body, code="mail_route_changed", status=409)
_, replacement_node = self.add_peer(self.recipient.peer_server_identity)
self.assertEqual(replacement_node, self.node)
current = self.store.get_network_server(
self.host, self.team, self.node)["server"]["mail_route_lifecycle_id"]
self.assertNotEqual(current, self.lifecycle)
self.assert_rejected(old_body, code="mail_route_changed", status=409)
fresh = self.body()
fresh["recipients"][0]["mail_route_lifecycle_id"] = current
self.assertEqual(self.adapter.forward(self.request(fresh)).status, 200)
self.assertEqual(self.message_count(), 1)
if __name__ == "__main__":
unittest.main()