test_team_notification_hints_isolated.py
Rama main · réplica de lectura del repositorio original.
"""Quiet v2 notifications under the guarded temporary-state test runner."""
from dataclasses import replace
import json
import sqlite3
import threading
import unittest
from unittest import mock
from agentsdock_team_hub.mail_hints import MailArrival, MailHintClosed
from agentsdock_team_hub.notification_hints import (
BulletinChange, NotificationBroker, NotificationCursor, NotificationLease,
)
from agentsdock_team_hub.store import HubError
from agentsdock_team_hub.database import MIGRATIONS, _statements
import tests.test_team_mail_hints_isolated as fixtures
class NotificationPrimitiveTests(unittest.TestCase):
def cursor(self, sequence=0):
return NotificationCursor(MailArrival("team_test", "server_test", 0, None),
BulletinChange("team_test", sequence, f"bchg_{sequence:032x}" if sequence else None,
f"tmsg_{sequence:032x}" if sequence else None, "created" if sequence else None,
1 if sequence else None))
def test_exact_shape_preserves_independent_reset_flags_and_rejects_leaks(self):
wire = replace(self.cursor(1), bulletin_reset=True).as_dict()
self.assertEqual(NotificationCursor.from_dict(wire).as_dict(), wire)
self.assertFalse(wire["mail"]["reset"])
self.assertTrue(wire["bulletin"]["reset"])
self.assertFalse(NotificationCursor.from_dict(wire).as_dict(reset=False)["bulletin"]["reset"])
for changes in ({"body": "leak"}, {"through_sequence": True}, {"change_id": "bchg_bad"},
{"message_id": "another message"}, {"message_version": 0}, {"reset": 1},
{"team_id": "team_other"}, {"change_kind": "unknown"}, {"change_kind": ["created"]}):
with self.subTest(changes=changes), self.assertRaises(ValueError):
NotificationCursor.from_dict({**wire, "bulletin": {**wire["bulletin"], **changes}})
def test_bounded_pair_coalesces_without_cross_recipient_mail(self):
broker = NotificationBroker()
self.addCleanup(broker.close)
mine = broker.subscribe("team_test", "server_test")
mine.seed(self.cursor())
other = broker.subscribe("team_test", "server_other")
other.seed(NotificationCursor(MailArrival("team_test", "server_other", 0, None), BulletinChange("team_test")))
for sequence in range(1, 1001):
broker.publish_bulletin(self.cursor(sequence).bulletin)
broker.publish_mail(MailArrival("team_test", "server_test", sequence, f"tmsg_{sequence:032x}"))
delivered = mine.take(0)
self.assertEqual(delivered.mail.through_sequence, 1000)
self.assertEqual(delivered.bulletin.through_sequence, 1000)
self.assertEqual(other.take(0).mail.through_sequence, 0)
self.assertIsNone(mine.take(0))
self.assertIsNone(other.take(0))
def test_seed_preserves_arrival_during_snapshot_and_drops_already_included(self):
broker = NotificationBroker()
self.addCleanup(broker.close)
sub = broker.subscribe("team_test", "server_test")
broker.publish_bulletin(self.cursor(2).bulletin)
sub.seed(self.cursor(1))
self.assertEqual(sub.take(0).bulletin.through_sequence, 2)
second = broker.subscribe("team_test", "server_test")
broker.publish_bulletin(self.cursor(3).bulletin)
second.seed(self.cursor(3))
self.assertIsNone(second.take(0))
def test_conflicting_history_closes_only_affected_team(self):
broker = NotificationBroker()
self.addCleanup(broker.close)
sub = broker.subscribe("team_test", "server_test")
sub.seed(self.cursor(1))
other = broker.subscribe("team_other", "server_other")
other.seed(NotificationCursor(MailArrival("team_other", "server_other", 0, None), BulletinChange("team_other")))
broker.publish_bulletin(replace(self.cursor(1).bulletin, change_id="bchg_" + "f" * 32))
self.assertTrue(sub.closed)
self.assertFalse(other.closed)
def test_idle_lease_never_queries_authority_and_cancel_wakes(self):
broker = NotificationBroker()
self.addCleanup(broker.close)
sub = broker.subscribe("team_test", "server_test")
authorize = mock.Mock()
lease = NotificationLease(sub, self.cursor().as_dict(reset=True), hub_id="hub_test",
authorize=authorize, expires_at=None)
self.addCleanup(lease.close)
done = threading.Event()
waiter = threading.Thread(target=lambda: (lease.take(), done.set()), daemon=True)
waiter.start()
self.assertFalse(done.wait(.02))
authorize.assert_not_called()
lease.cancel()
self.assertTrue(done.wait(1))
waiter.join(1)
self.assertFalse(waiter.is_alive())
class NotificationStoreTests(unittest.TestCase):
def setUp(self):
self.fixture = fixtures.TeamMailArrivalStoreTests()
self.fixture.setUp()
self.addCleanup(self.fixture.doCleanups)
self.store, self.team = self.fixture.store, self.fixture.team
self.addCleanup(self.store.notification_broker.close)
def subscribe(self, caller=None):
sub, snapshot = self.store.subscribe_team_notifications(caller or self.fixture.peer, self.team)
self.addCleanup(sub.close)
return sub, snapshot
def bulletin(self, **extra):
return self.fixture.send(recipients=[{"kind": "all"}], **extra)
def edit(self, message, version=1, **extra):
return self.store.revise_team_message(self.fixture.host, self.team, message["id"], {
"body": "Updated bulletin body", "body_format": "markdown", "expected_version": version,
"idempotency_key": "revision-test-key", **extra})
def test_create_edit_delete_are_distinct_committed_hints_not_mail(self):
sub, snapshot = self.subscribe()
legacy, _ = self.fixture.subscribe()
observed = []
publish = self.store.notification_broker.publish_bulletin
def committed(change):
with self.store.connect() as db:
db.execute("BEGIN IMMEDIATE")
row = db.execute("SELECT change_kind FROM team_bulletin_changes WHERE id=?", (change.change_id,)).fetchone()
self.assertEqual(row[0], change.change_kind)
db.rollback()
observed.append(change)
publish(change)
with mock.patch.object(self.store.notification_broker, "publish_bulletin", side_effect=committed):
message = self.bulletin(title="Title remains", body="Original bulletin body")
created = sub.take(0)
self.assertEqual(created.bulletin.change_kind, "created")
self.assertEqual(created.bulletin.message_id, message["id"])
self.edit(message)
edited = sub.take(0)
self.assertEqual((edited.bulletin.change_kind, edited.bulletin.message_version), ("revised", 2))
self.fixture.delete(message)
deleted = sub.take(0)
self.assertEqual((deleted.bulletin.change_kind, deleted.bulletin.message_version), ("deleted", 2))
self.assertEqual(len(observed), 3)
for cursor in (created, edited, deleted):
self.assertEqual(cursor.mail.as_dict(reset=True), snapshot["mail"])
self.assertNotIn("body", json.dumps(cursor.as_dict()))
self.assertNotIn("Title remains", json.dumps(cursor.as_dict()))
self.assertIsNone(legacy.take(0))
def test_retry_failed_cas_unauthorized_edit_and_rollback_do_not_notify(self):
sub, _ = self.subscribe()
message = self.bulletin(idempotency_key="original-post-key")
sub.take(0)
self.assertEqual(self.bulletin(idempotency_key="original-post-key")["id"], message["id"])
self.assertIsNone(sub.take(0))
self.edit(message)
version_two = sub.take(0)
self.edit(message)
self.assertIsNone(sub.take(0))
with self.assertRaises(HubError):
self.edit(message, idempotency_key="stale-edit-key")
with self.assertRaises(HubError):
self.store.revise_team_message(self.fixture.peer, self.team, message["id"], {
"body": "Other server cannot edit", "expected_version": 2, "idempotency_key": "foreign-edit-key"})
with mock.patch.object(self.store, "_audit", side_effect=RuntimeError("transaction rollback")):
with self.assertRaises(RuntimeError):
self.bulletin()
with self.assertRaises(RuntimeError):
self.edit(message, version=2, body="Rolled back revision", idempotency_key="rollback-revision")
with self.assertRaises(RuntimeError):
self.fixture.delete(message)
self.assertIsNone(sub.take(0))
current = self.store.team_notification_snapshot(self.fixture.peer, self.team)
self.assertEqual(current["bulletin"], version_two.bulletin.as_dict(reset=True))
def test_failed_notification_does_not_fail_committed_message_and_reconnect_recovers(self):
sub, before = self.subscribe()
with mock.patch.object(self.store.notification_broker, "publish_bulletin", side_effect=RuntimeError("hint failure")):
message = self.bulletin()
self.assertTrue(sub.closed)
after = self.store.team_notification_snapshot(self.fixture.peer, self.team, previous_cursor=before)
self.assertEqual(after["bulletin"]["message_id"], message["id"])
self.assertFalse(after["bulletin"]["reset"])
self.assertEqual(self.store.get_team_message(self.fixture.peer, self.team, message["id"])["message"]["id"], message["id"])
def test_hint_preparation_failure_cannot_rollback_create_edit_or_delete(self):
sub, _ = self.subscribe()
with mock.patch.object(self.store, "_team_bulletin_change", side_effect=ValueError("notification metadata unavailable")):
message = self.bulletin()
self.assertEqual(self.edit(message)["message"]["revision"]["version"], 2)
self.assertTrue(self.fixture.delete(message)["deleted"])
self.assertTrue(sub.closed)
snapshot = self.store.team_notification_snapshot(self.fixture.peer, self.team)
self.assertEqual(snapshot["bulletin"]["change_kind"], "deleted")
self.assertEqual(snapshot["bulletin"]["message_version"], 2)
with self.store.connect() as db:
self.assertEqual(db.execute("SELECT count(*) FROM team_bulletin_changes WHERE message_id=?", (message["id"],)).fetchone()[0], 3)
def test_mail_recipient_isolation_all_servers_and_skills_have_separate_heads(self):
sub, _ = self.subscribe()
other, _ = self.subscribe(self.fixture.other)
direct = self.fixture.send()
self.assertEqual(sub.take(0).mail.arrival_id, direct["id"])
self.assertIsNone(other.take(0))
all_mail = self.fixture.send(recipients=[{"kind": "all_servers"}])
self.assertEqual(sub.take(0).mail.arrival_id, all_mail["id"])
self.assertEqual(other.take(0).mail.arrival_id, all_mail["id"])
skill = self.bulletin(kind="skill", title="Example", skill={"slug": "example", "summary": "Skill summary", "tags": []})
for recipient in (sub, other):
cursor = recipient.take(0)
self.assertEqual(cursor.bulletin.message_id, skill["id"])
self.assertEqual(cursor.mail.arrival_id, all_mail["id"])
def test_independent_restore_anchor_and_foreign_recipient_fencing(self):
message = self.bulletin()
self.fixture.send()
previous = self.store.team_notification_snapshot(self.fixture.peer, self.team)
restored = {**previous, "bulletin": {**previous["bulletin"], "change_id": "bchg_" + "f" * 32}}
snapshot = self.store.team_notification_snapshot(self.fixture.peer, self.team, previous_cursor=restored)
self.assertFalse(snapshot["mail"]["reset"])
self.assertTrue(snapshot["bulletin"]["reset"])
self.fixture.delete(message)
snapshot = self.store.team_notification_snapshot(self.fixture.peer, self.team, previous_cursor=previous)
self.assertFalse(snapshot["bulletin"]["reset"])
with self.assertRaises(HubError) as error:
self.store.team_notification_snapshot(self.fixture.other, self.team, previous_cursor=previous)
self.assertEqual(error.exception.status_code, 403)
def test_subscribe_snapshot_race_preserves_post_committed_after_snapshot(self):
original = self.store.team_notification_snapshot
calls = 0
created = []
def snapshot(*args, **kwargs):
nonlocal calls
result = original(*args, **kwargs)
calls += 1
if calls == 2:
created.append(self.bulletin())
return result
with mock.patch.object(self.store, "team_notification_snapshot", side_effect=snapshot):
sub, baseline = self.subscribe()
self.assertEqual(baseline["bulletin"]["through_sequence"], 0)
self.assertEqual(sub.take(0).bulletin.message_id, created[0]["id"])
def test_migration_backfills_old_posts_and_keeps_original_bodies(self):
message = self.bulletin(body="Original before migration")
self.edit(message)
legacy = sqlite3.connect(":memory:")
self.addCleanup(legacy.close)
source = self.store.connect()
try:
source.backup(legacy)
finally:
source.close()
# Reconstruct the prior schema in this in-memory copy only.
for trigger in ("team_bulletin_created", "team_bulletin_revised", "team_bulletin_deleted",
"team_bulletin_changes_immutable", "team_bulletin_changes_retained"):
legacy.execute("DROP TRIGGER " + trigger)
legacy.execute("DROP TABLE team_bulletin_changes")
for statement in _statements(next(item.source for item in MIGRATIONS if item.version == 22)):
legacy.execute(statement)
self.assertEqual(legacy.execute("SELECT message_id,change_kind,message_version FROM team_bulletin_changes").fetchall(),
[(message["id"], "created", 2)])
self.assertEqual(legacy.execute("SELECT body FROM team_messages WHERE id=?", (message["id"],)).fetchone()[0],
"Original before migration")
self.assertEqual(legacy.execute("PRAGMA foreign_key_check").fetchall(), [])
def test_snapshot_queries_metadata_only_and_journal_is_immutable(self):
self.bulletin()
original_connect = self.store.connect
statements = []
def traced():
connection = original_connect()
connection.set_trace_callback(statements.append)
return connection
with mock.patch.object(self.store, "connect", side_effect=traced):
snapshot = self.store.team_notification_snapshot(self.fixture.peer, self.team)
self.assertFalse(any("current_body" in sql or "COUNT(" in sql.upper() for sql in statements))
with self.store.connect() as db:
plans = db.execute("EXPLAIN QUERY PLAN SELECT * FROM team_bulletin_changes WHERE team_id=? ORDER BY sequence DESC LIMIT 1", (self.team,)).fetchall()
self.assertTrue(any("team_bulletin_changes_by_team" in row[3] for row in plans))
for sql in ("UPDATE team_bulletin_changes SET message_version=9", "DELETE FROM team_bulletin_changes"):
with self.assertRaises(sqlite3.IntegrityError):
db.execute(sql)
self.assertGreater(snapshot["bulletin"]["through_sequence"], 0)