89 lines
2.9 KiB
Python
89 lines
2.9 KiB
Python
"""Gate 1 contract tests for the Hermes shared agent fabric."""
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from tempfile import TemporaryDirectory
|
|
import unittest
|
|
|
|
from aios_v2r1.hermes_fabric import HermesFabric
|
|
from aios_v2r1.tenant_context import CanonicalMessage, ConflictError, ScopeError
|
|
|
|
|
|
def authenticated_command(
|
|
event_id: str,
|
|
*,
|
|
command_capable: bool = True,
|
|
origin_type: str = "channel_user",
|
|
) -> CanonicalMessage:
|
|
return CanonicalMessage(
|
|
tenant_id="tenant-a",
|
|
cell_id="cell-a",
|
|
principal_id="user-a",
|
|
principal_type="authenticated_user",
|
|
origin_type=origin_type,
|
|
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="1234567890123456789",
|
|
reply_to_message_id=None,
|
|
direction="inbound",
|
|
command_capable=command_capable,
|
|
occurred_at="2026-07-13T00:00:00+00:00",
|
|
received_at="2026-07-13T00:00:01+00:00",
|
|
body="/run",
|
|
quoted_body=None,
|
|
adapter_name="weixin",
|
|
adapter_version="v2",
|
|
)
|
|
|
|
|
|
class HermesFabricContractTests(unittest.TestCase):
|
|
def test_duplicate_event_returns_sanitized_conflict(self) -> None:
|
|
with TemporaryDirectory() as directory:
|
|
fabric = HermesFabric(f"{directory}/fabric.sqlite3")
|
|
message = authenticated_command("event-a")
|
|
|
|
first = fabric.enqueue(message=message, profile_id="profile-a")
|
|
self.assertEqual(first.event_id, "event-a")
|
|
|
|
with self.assertRaises(ConflictError) as raised:
|
|
fabric.enqueue(message=message, profile_id="profile-a")
|
|
|
|
error = raised.exception
|
|
self.assertNotIsInstance(error, sqlite3.IntegrityError)
|
|
self.assertNotIn("sqlite", str(error).lower())
|
|
self.assertIsNone(error.__cause__)
|
|
self.assertIsNone(error.__context__)
|
|
|
|
def test_non_command_event_cannot_enqueue(self) -> None:
|
|
with TemporaryDirectory() as directory:
|
|
fabric = HermesFabric(f"{directory}/fabric.sqlite3")
|
|
|
|
with self.assertRaises(ScopeError):
|
|
fabric.enqueue(
|
|
message=authenticated_command(
|
|
"event-non-command", command_capable=False
|
|
),
|
|
profile_id="profile-a",
|
|
)
|
|
|
|
def test_heartbeat_provenance_cannot_enqueue(self) -> None:
|
|
with TemporaryDirectory() as directory:
|
|
fabric = HermesFabric(f"{directory}/fabric.sqlite3")
|
|
|
|
with self.assertRaises(ScopeError):
|
|
fabric.enqueue(
|
|
message=authenticated_command(
|
|
"event-heartbeat", origin_type="heartbeat"
|
|
),
|
|
profile_id="profile-a",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|