"""Admission and consumption failures leave recoverable, atomic identified decisions.""" import base64 from contextlib import closing, contextmanager import json import random import struct import zlib import sqlite3 import urllib.error from harness import exclusive, operation_id from image_limits_test import png from operations_test import InputScenario class OperationLifecycleTests(InputScenario): @contextmanager def database(self): with ( closing(sqlite3.connect(self.app.home / "albedo.sqlite", timeout=10)) as db, db, ): yield db def create(self, session_id, intent): with self.app.api( f"/sessions/{session_id}", intent, method="PUT", headers={"If-None-Match": "*"}, ) as response: return json.load(response) @exclusive def test_creation_failure_rolls_back_session_and_receipt(self): session_id = operation_id() intent = { "kind": "new", "workspace": str(self.app.workspace), "provider_profile": self.app.profile, } with self.database() as database: before = database.execute("SELECT count(*) FROM sessions").fetchone()[0] database.execute( "CREATE TRIGGER fail_creation BEFORE INSERT ON sessions BEGIN SELECT RAISE(ABORT,'fixture'); END" ) with self.assertRaises(urllib.error.HTTPError): self.create(session_id, intent) with self.database() as database: self.assertEqual( database.execute("SELECT count(*) FROM sessions").fetchone()[0], before ) self.assertEqual( database.execute( "SELECT count(*) FROM operations WHERE id=?", (session_id,) ).fetchone()[0], 0, ) database.execute("DROP TRIGGER fail_creation") created = self.create(session_id, intent) self.assertEqual(created["id"], session_id) with self.app.api(f"/sessions/{session_id}") as response: self.assertEqual(json.load(response)["creation"], created["creation"]) @exclusive def test_admission_failure_can_retry_original_id(self): session = self.app.session() input_id = operation_id() intent = {"kind": "message", "text": "durable input"} with self.database() as database: database.execute( "CREATE TRIGGER fail_admission BEFORE INSERT ON pending_inputs BEGIN SELECT RAISE(ABORT,'fixture'); END" ) with self.assertRaises(urllib.error.HTTPError): self.put_input(session, input_id, intent) with self.database() as database: self.assertEqual( database.execute( "SELECT count(*) FROM operations WHERE id=?", (input_id,) ).fetchone()[0], 0, ) self.assertEqual( database.execute( "SELECT count(*) FROM transcript WHERE session=?", (session,) ).fetchone()[0], 0, ) database.execute("DROP TRIGGER fail_admission") accepted = self.put_input(session, input_id, intent) self.committed(session, input_id) self.app.idle(session) self.assertEqual( self.put_input(session, input_id, intent)["acceptance_order"], accepted["acceptance_order"], ) self.assertEqual( [entry["input_id"] for entry in self.users(session)], [input_id] ) @exclusive def test_consumption_failure_keeps_pending_input_until_storage_recovers(self): session = self.app.session() input_id = operation_id() intent = { "kind": "message", "text": "recover after storage repair", "images": [ { "mime_type": "image/png", "data": base64.b64encode(png(2, 2)).decode(), } ], } with self.database() as database: database.execute( "CREATE TRIGGER fail_consumption BEFORE INSERT ON submission_events BEGIN SELECT RAISE(ABORT,'fixture'); END" ) accepted = self.put_input(session, input_id, intent) self.assertEqual(accepted["admission"], "accepted") self.assertEqual(self.receipt(session, input_id)["delivery"], "pending") with self.database() as database: self.assertEqual( database.execute( "SELECT count(*) FROM pending_inputs WHERE operation_id=?", (input_id,), ).fetchone()[0], 1, ) self.assertEqual( database.execute( "SELECT count(*) FROM transcript WHERE session=?", (session,) ).fetchone()[0], 0, ) self.assertEqual( database.execute("SELECT count(*) FROM images").fetchone()[0], 0 ) self.assertEqual( database.execute("SELECT count(*) FROM submission_events").fetchone()[ 0 ], 0, ) self.app.restart(crash=True) self.assertEqual( self.put_input(session, input_id, intent)["acceptance_order"], accepted["acceptance_order"], ) self.assertEqual(self.receipt(session, input_id)["delivery"], "pending") with self.database() as database: database.execute("DROP TRIGGER fail_consumption") self.committed(session, input_id) self.app.idle(session) self.app.restart(crash=True) self.assertEqual(self.receipt(session, input_id)["delivery"], "committed") self.conflict(session, input_id, {**intent, "text": "changed after restart"}) self.assertEqual( [entry["input_id"] for entry in self.users(session)], [input_id] ) @exclusive def test_repeated_image_commits_grow_only_compact_metadata(self): session = self.app.session() width, height = 256, 256 def chunk(kind, data): body = kind + data return ( struct.pack(">I", len(data)) + body + struct.pack(">I", zlib.crc32(body)) ) pixels = random.Random(1).randbytes(width * height * 3) rows = b"".join( b"\x00" + pixels[start : start + width * 3] for start in range(0, len(pixels), width * 3) ) image = ( b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0)) + chunk(b"IDAT", zlib.compress(rows)) + chunk(b"IEND", b"") ) attachment = { "mime_type": "image/png", "data": base64.b64encode(image).decode(), } def durable_bytes(): with self.database() as database: return sum( database.execute( f"SELECT coalesce(sum(length({column})),0) FROM {table}" ).fetchone()[0] for table, column in ( ("images", "data"), ("transcript", "payload"), ("submission_events", "payload"), ("continuation_markers", "payload"), ("pending_inputs", "payload"), ("operations", "response"), ) ) inputs, sizes = [], [] for index in range(3): input_id = operation_id() inputs.append(input_id) self.put_input( session, input_id, { "kind": "message", "text": "same image", "client_id": f"image-echo-{index}", "images": [attachment], }, ) self.committed(session, input_id) self.app.idle(session) sizes.append(durable_bytes()) self.assertLess(sizes[2] - sizes[0], len(image)) self.app.restart(crash=True) users = self.users(session) self.assertEqual([entry["input_id"] for entry in users], inputs) for index, input_id in enumerate(inputs): self.assertEqual( self.receipt(session, input_id)["client_id"], f"image-echo-{index}" ) metadata = [ part["image"] for entry in users for part in entry["content"] if part["kind"] == "image" ] self.assertEqual( [ (item["width"], item["height"], item["original_bytes"]) for item in metadata ], [(width, height, len(image))] * 3, ) @exclusive def test_rejected_creation_replays_after_workspace_is_repaired(self): session_id = operation_id() workspace = self.app.workspace / "missing" intent = { "kind": "new", "workspace": str(workspace), "provider_profile": self.app.profile, } with self.assertRaises(urllib.error.HTTPError) as original: self.create(session_id, intent) problem = json.load(original.exception) with self.assertRaises(urllib.error.HTTPError) as lookup: self.app.api(f"/sessions/{session_id}") self.assertEqual( json.load(lookup.exception)["decision"]["admission"], "rejected" ) workspace.mkdir() self.app.restart(crash=True) with self.assertRaises(urllib.error.HTTPError) as repeated: self.create(session_id, intent) self.assertEqual(repeated.exception.code, original.exception.code) self.assertEqual(json.load(repeated.exception), problem) with self.database() as database: self.assertEqual( database.execute("SELECT count(*) FROM sessions").fetchone()[0], 0 ) @exclusive def test_failed_cancellation_does_not_discard_pending_input(self): session = self.app.session() input_id = operation_id() with self.database() as database: database.execute( "CREATE TRIGGER hold_consumption BEFORE INSERT ON transcript WHEN NEW.row_class='user' BEGIN SELECT RAISE(ABORT,'fixture'); END" ) self.put_input( session, input_id, {"kind": "message", "text": "keep until cancellation commits"}, ) with self.database() as database: database.execute( "CREATE TRIGGER fail_cancellation BEFORE UPDATE ON operations WHEN NEW.delivery='cancelled' BEGIN SELECT RAISE(ABORT,'fixture'); END" ) resource = f"/sessions/{session}/inputs/{input_id}/cancel" with self.assertRaises(urllib.error.HTTPError): self.app.api(resource, {}) self.assertEqual(self.receipt(session, input_id)["delivery"], "pending") with self.database() as database: database.execute("DROP TRIGGER fail_cancellation") with self.app.api(resource, {}) as response: self.assertEqual(json.load(response)["result"], "cancelled") self.app.restart(crash=True) self.assertEqual(self.receipt(session, input_id)["delivery"], "cancelled") with self.database() as database: self.assertEqual( database.execute( "SELECT count(*) FROM pending_inputs WHERE operation_id=?", (input_id,), ).fetchone()[0], 0, )