"""Synthetic invitation, cookie, upload and submission ledger tests.""" import concurrent.futures from pathlib import Path import sqlite3 import tempfile import unittest from unittest import mock from interactive_chat_shares import InteractiveChatShareStore, Unavailable, Conflict, MAX_SHARE_UPLOAD_BYTES, token_hash class InteractiveShareStoreTests(unittest.TestCase): def setUp(self): temporary = tempfile.TemporaryDirectory(prefix="interactive-share-store-") self.addCleanup(temporary.cleanup) self.root = Path(temporary.name) / "shares" self.now = 1000 self.store = InteractiveChatShareStore(self.root, now=lambda: self.now) self.share = self.store.create_share("chat-one", expires_at=2000) def redeem(self, share=None): share = share or self.share return self.store.redeem(share["id"], share["invitation_token"]) def test_shared_token_admits_multiple_browsers_and_persists_hashes_only(self): def attempt(_): try: return self.redeem() except Unavailable: return None with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: tokens = [token for token in pool.map(attempt, range(2)) if token] self.assertEqual(len(tokens), 2) self.assertEqual(tokens[0], self.share["invitation_token"]) self.assertEqual(self.store.authenticate(self.share["id"], tokens[0])["session_id"], "chat-one") self.assertEqual(self.store.authenticate(self.share["id"], tokens[1])["session_id"], "chat-one") data = self.store.database_path.read_bytes() for token in (tokens[0], self.share["invitation_token"]): self.assertNotIn(token.encode(), data) listed = self.store.list_shares("chat-one") self.assertNotIn("invitation_token", listed[0]) self.assertNotIn("session_id", listed[0]) def test_public_origin_is_durable_and_does_not_expose_capabilities(self): origin = "http://192.0.2.42:8080" share = self.store.create_share("chat-one", public_origin=origin) reopened = InteractiveChatShareStore.open_existing(self.root, now=lambda: self.now) self.assertEqual(reopened.share_origin(share["id"]), origin) self.assertEqual(reopened.share_origin(self.share["id"]), "") listed = reopened.list_shares("chat-one") for row in listed: self.assertNotIn("invitation_token", row) self.assertNotIn("browser_hash", row) with self.assertRaises(Unavailable): reopened.share_origin("interactive_" + "0" * 32) def test_legacy_origin_migration_is_admin_only_and_preserves_active_and_revoked_grants(self): active_token = self.redeem() revoked = self.store.create_share("chat-one") revoked_token = self.redeem(revoked) self.store.revoke_share(revoked["id"], session_id="chat-one") with self.store._connection(write=True) as db: db.execute("ALTER TABLE interactive_shares DROP COLUMN public_origin") before = self.store.database_path.read_bytes() with mock.patch("public_chat_shares.sqlite3.connect", wraps=sqlite3.connect) as connect: legacy = InteractiveChatShareStore.open_existing(self.root, now=lambda: self.now) self.assertEqual(legacy.share_origin(self.share["id"]), "") self.assertEqual(legacy.authenticate(self.share["id"], active_token)["session_id"], "chat-one") with self.assertRaises(Unavailable): legacy.authenticate(revoked["id"], revoked_token) self.assertTrue(all(call.args[0].endswith("?mode=ro") for call in connect.call_args_list)) self.assertEqual(self.store.database_path.read_bytes(), before) upgraded = InteractiveChatShareStore(self.root, now=lambda: self.now) self.assertEqual(upgraded.share_origin(self.share["id"]), "") self.assertEqual(upgraded.authenticate(self.share["id"], active_token)["session_id"], "chat-one") with self.assertRaises(Unavailable): upgraded.authenticate(revoked["id"], revoked_token) fresh = upgraded.create_share("chat-one", public_origin="https://share.example.test") self.assertEqual(upgraded.share_origin(fresh["id"]), "https://share.example.test") with upgraded._connection() as db: self.assertEqual(db.execute("PRAGMA foreign_key_check").fetchall(), []) self.assertEqual(db.execute("SELECT count(*) FROM interactive_shares").fetchone()[0], 3) def test_expiry_revocation_and_exact_chat_management(self): token = self.redeem() self.assertFalse(self.store.revoke_share(self.share["id"], session_id="other-chat")) self.now = 2000 with self.assertRaises(Unavailable): self.store.authenticate(self.share["id"], token) self.now = 1001 self.assertTrue(self.store.revoke_share(self.share["id"], session_id="chat-one")) with self.assertRaises(Unavailable): self.store.authenticate(self.share["id"], token) def test_reusable_token_preserves_legacy_browser_cookie_and_revokes_every_browser(self): legacy_cookie = "L" * 43 with self.store._connection(write=True) as db: db.execute("UPDATE interactive_shares SET browser_hash=?,redeemed_at=? WHERE id=?", (token_hash(legacy_cookie), self.now, self.share["id"])) current_cookie = self.redeem() self.assertEqual(self.store.authenticate(self.share["id"], legacy_cookie)["session_id"], "chat-one") self.assertEqual(self.store.authenticate(self.share["id"], current_cookie)["session_id"], "chat-one") other = self.store.create_share("chat-one") with self.assertRaises(Unavailable): self.store.redeem(other["id"], self.share["invitation_token"]) self.store.revoke_share(self.share["id"], session_id="chat-one") for token in (legacy_cookie, current_cookie): with self.assertRaises(Unavailable): self.store.authenticate(self.share["id"], token) with self.assertRaises(Unavailable): self.redeem() def test_uploads_belong_to_exact_share_and_unknown_writes_remain_charged(self): token = self.redeem() upload = self.store.reserve_upload(self.share["id"], token, name="note.txt", media_type="text/plain", byte_size=5) self.store.complete_upload(self.share["id"], token, upload, "private-file-reference") self.assertEqual(self.store.upload_refs(self.share["id"], token, [upload]), ["private-file-reference"]) other = self.store.create_share("chat-one") with self.assertRaises(Unavailable): self.store.upload_refs(other["id"], self.redeem(other), [upload]) # Incomplete reservations stay durable across restart; no refund merely # because the save callback's outcome could not be observed. pending = self.store.reserve_upload(self.share["id"], token, name="pending.txt", media_type="text/plain", byte_size=7) reopened = InteractiveChatShareStore.open_existing(self.root, now=lambda: self.now) with reopened._connection() as db: row = db.execute("SELECT byte_size,private_ref FROM interactive_uploads WHERE id=?", (pending,)).fetchone() self.assertEqual(tuple(row), (7, None)) with self.assertRaises(Unavailable): reopened.upload_refs(self.share["id"], token, [pending]) def test_stable_request_replays_only_exact_accepted_receipt_across_restart(self): token = self.redeem() self.assertIsNone(self.store.reserve_submission(self.share["id"], token, "request-one", "Prompt", [])) reopened = InteractiveChatShareStore.open_existing(self.root, now=lambda: self.now) with self.assertRaisesRegex(Conflict, "indeterminate"): reopened.reserve_submission(self.share["id"], token, "request-one", "Prompt", []) receipt = {"accepted": True, "queued": True, "request_id": "request-one"} reopened.accept_submission(self.share["id"], "request-one", receipt) self.assertEqual(self.store.reserve_submission(self.share["id"], token, "request-one", "Prompt", []), receipt) with self.assertRaisesRegex(Conflict, "different"): reopened.reserve_submission(self.share["id"], token, "request-one", "Prompt", [], operation="control:turn.stop") with self.assertRaisesRegex(Conflict, "different"): reopened.reserve_submission(self.share["id"], token, "request-one", "Changed", []) if __name__ == "__main__": unittest.main()