115 lines
5.5 KiB
Python
115 lines
5.5 KiB
Python
"""Executable completeness contract for the tenant-partitioned Hermes fabric."""
|
|
from __future__ import annotations
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import pytest
|
|
from aios_v2r1.hermes_fabric import HermesFabric
|
|
from aios_v2r1.tenant_context import CanonicalMessage, ConflictError, ScopeError
|
|
|
|
def authenticated_command(event_id: str, **changes: object) -> CanonicalMessage:
|
|
value = CanonicalMessage(
|
|
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=f"message-{event_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", body="/run",
|
|
quoted_body=None, adapter_name="weixin", adapter_version="v2")
|
|
return replace(value, **changes)
|
|
|
|
@pytest.fixture
|
|
def fabric(tmp_path: Path) -> HermesFabric:
|
|
return HermesFabric(tmp_path / "fabric.sqlite3")
|
|
|
|
def test_duplicate_event_returns_sanitized_conflict(fabric: HermesFabric) -> None:
|
|
command = authenticated_command("event-a")
|
|
assert fabric.enqueue(message=command, profile_id="profile-a").event_id == "event-a"
|
|
with pytest.raises(ConflictError) as raised:
|
|
fabric.enqueue(message=command, profile_id="profile-a")
|
|
assert not isinstance(raised.value, sqlite3.IntegrityError)
|
|
assert "sqlite" not in str(raised.value).lower()
|
|
assert raised.value.__cause__ is None
|
|
assert raised.value.__context__ is None
|
|
|
|
@pytest.mark.parametrize("changes", [
|
|
{"command_capable": False}, {"origin_type": "heartbeat"},
|
|
{"principal_type": "service"}, {"direction": "internal"}])
|
|
def test_non_claimable_provenance_cannot_enqueue(
|
|
fabric: HermesFabric, changes: dict[str, object]
|
|
) -> None:
|
|
with pytest.raises(ScopeError):
|
|
fabric.enqueue(message=authenticated_command("event-denied", **changes),
|
|
profile_id="profile-a")
|
|
|
|
def test_claim_returns_running_and_matches_durable_state(fabric: HermesFabric) -> None:
|
|
fabric.enqueue(message=authenticated_command("event-a"), profile_id="profile-a")
|
|
claimed = fabric.claim("tenant-a", "cell-a", "profile-a")
|
|
assert claimed is not None
|
|
assert claimed.status == "running"
|
|
with sqlite3.connect(fabric.database) as db:
|
|
assert db.execute("SELECT status FROM fabric_jobs WHERE job_id=?",
|
|
(claimed.job_id,)).fetchone()[0] == "running"
|
|
|
|
@pytest.mark.parametrize(("tenant_id", "cell_id", "profile_id"), [
|
|
("tenant-b", "cell-a", "profile-a"), ("tenant-a", "cell-b", "profile-a"),
|
|
("tenant-a", "cell-a", "profile-b")])
|
|
def test_claim_scope_is_exact(
|
|
fabric: HermesFabric, tenant_id: str, cell_id: str, profile_id: str
|
|
) -> None:
|
|
fabric.enqueue(message=authenticated_command("event-a"), profile_id="profile-a")
|
|
assert fabric.claim(tenant_id, cell_id, profile_id) is None
|
|
assert fabric.claim("tenant-a", "cell-a", "profile-a") is not None
|
|
|
|
def test_only_one_concurrent_claimant_wins(fabric: HermesFabric) -> None:
|
|
fabric.enqueue(message=authenticated_command("event-a"), profile_id="profile-a")
|
|
with ThreadPoolExecutor(max_workers=16) as executor:
|
|
claimed = list(executor.map(
|
|
lambda _: fabric.claim("tenant-a", "cell-a", "profile-a"), range(16)))
|
|
winners = [job for job in claimed if job is not None]
|
|
assert len(winners) == 1
|
|
assert winners[0].status == "running"
|
|
|
|
@pytest.mark.parametrize("status", ["running", "completed", "failed", "cancelled"])
|
|
def test_nonqueued_states_are_not_claimable(
|
|
fabric: HermesFabric, status: str
|
|
) -> None:
|
|
job = fabric.enqueue(message=authenticated_command("event-a"), profile_id="profile-a")
|
|
with sqlite3.connect(fabric.database) as db:
|
|
db.execute("UPDATE fabric_jobs SET status=? WHERE job_id=?", (status, job.job_id))
|
|
assert fabric.claim("tenant-a", "cell-a", "profile-a") is None
|
|
|
|
def test_restart_preserves_queue_and_claims_once(tmp_path: Path) -> None:
|
|
database = tmp_path / "fabric.sqlite3"
|
|
HermesFabric(database).enqueue(message=authenticated_command("event-a"),
|
|
profile_id="profile-a")
|
|
reopened = HermesFabric(database)
|
|
claimed = reopened.claim("tenant-a", "cell-a", "profile-a")
|
|
assert claimed is not None and claimed.status == "running"
|
|
assert reopened.claim("tenant-a", "cell-a", "profile-a") is None
|
|
|
|
def test_concurrent_duplicate_enqueue_has_one_winner(fabric: HermesFabric) -> None:
|
|
command = authenticated_command("event-a")
|
|
def enqueue(_: int) -> str:
|
|
try:
|
|
fabric.enqueue(message=command, profile_id="profile-a")
|
|
return "queued"
|
|
except ConflictError:
|
|
return "conflict"
|
|
with ThreadPoolExecutor(max_workers=16) as executor:
|
|
results = list(executor.map(enqueue, range(16)))
|
|
assert results.count("queued") == 1
|
|
assert results.count("conflict") == 15
|
|
|
|
def test_sqlite_failures_are_sanitized(fabric: HermesFabric) -> None:
|
|
Path(fabric.database).write_bytes(b"not-a-sqlite-database")
|
|
with pytest.raises(ConflictError, match="fabric storage unavailable") as raised:
|
|
fabric.health()
|
|
assert not isinstance(raised.value, sqlite3.Error)
|
|
assert "sqlite" not in str(raised.value).lower()
|
|
assert raised.value.__cause__ is None
|
|
assert raised.value.__context__ is None
|