"""Isolated PostgreSQL schema; never reads/writes production Event rows.""" from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager from dataclasses import replace import os from pathlib import Path import tempfile import time import unittest import uuid from app.event_artifact_contract import ArtifactConfig, RESERVATION from app.event_artifact_store import ArtifactStore from tests.test_event_artifact import event @unittest.skipUnless(os.environ.get('DATABASE_URL'), 'PostgreSQL fixture connection required') class LedgerTests(unittest.TestCase): def setUp(self): import psycopg2 self.schema = 'artifact_fixture_'+uuid.uuid4().hex self.admin = psycopg2.connect(os.environ['DATABASE_URL']) self.admin.autocommit = True with self.admin.cursor() as cur: cur.execute('CREATE SCHEMA '+self.schema) self.temp = tempfile.TemporaryDirectory() config = ArtifactConfig(Path(self.temp.name),os.environ['DATABASE_URL'], 'http://web.invalid/events','http://ai.invalid/artifacts', low_free=0,resume_free=0,delivery_cameras=('1','2')) schema = self.schema class Store(ArtifactStore): @contextmanager def transaction(self): with super().transaction() as cur: cur.execute('SET LOCAL search_path TO '+schema) yield cur self.store = Store(config) self.store.initialize() self.store.recover() def tearDown(self): self.store.close() with self.admin.cursor() as cur: cur.execute('DROP SCHEMA '+self.schema+' CASCADE') self.admin.close() self.temp.cleanup() def make(self, channel='channel-1', age=0): e = event() e.update(cameraId=1,streamId=channel,eventId=uuid.uuid4().hex, aiRecordedAtWallMs=int((time.time()-age)*1000)) return e def ready(self, e): self.assertEqual(self.store.accept(e),(True,None)) self.store.captured(e['eventId']) def complete(self, row): self.store.rendered(row,{'clip':True},None,42) self.store.release_claim(row,'render') def test_simultaneous_admission_global_cap_and_duplicate(self): events=[self.make('channel-'+str(i%16)) for i in range(64)] with ThreadPoolExecutor(max_workers=16) as pool: result=list(pool.map(self.store.accept,events)) self.assertTrue(all(x==(True,None) for x in result)) self.assertEqual(self.store.summary()['storage']['reserved_jobs'],64) self.assertEqual(self.store.accept(self.make('channel-new')),(True,'PENDING_CAP_PROTECTED')) with ThreadPoolExecutor(max_workers=8) as pool: self.assertTrue(all(x==(False,'DUPLICATE') for x in pool.map(self.store.accept,[events[0]]*8))) self.assertEqual(self.store.summary()['storage']['reserved_bytes'],64*RESERVATION) def test_channel_claim_serialization_and_stale_completion(self): for channel in ('a','a','b','b'): self.ready(self.make(channel)) with ThreadPoolExecutor(max_workers=8) as pool: rows=[r for r in pool.map(self.store.claim,['render']*8) if r] self.assertEqual(len(rows),2) self.assertEqual({r['channel_key'] for r in rows},{'a','b'}) self.complete(rows[0]) with self.assertRaises(RuntimeError): self.store.rendered(rows[0],{},None,0) self.assertEqual(self.store.summary()['storage']['reserved_jobs'],3) self.assertIsNotNone(self.store.claim('render')) def test_channel_limit_evicts_only_waiting_then_releases_once(self): old=self.make(age=40);self.ready(old) for _ in range(7):self.ready(self.make()) self.assertEqual(self.store.accept(self.make()),(True,None)) with self.store.transaction() as cur: cur.execute('SELECT artifact_state,reservation,error FROM contextra_event_artifacts WHERE event_id=%s',(old['eventId'],)) row=cur.fetchone() self.assertEqual((row['artifact_state'],row['reservation'],row['error']),('EXPIRED',0,'EVICTED_CAPACITY')) self.assertEqual(self.store.summary()['storage']['reserved_jobs'],8) def test_new_before_old_expiry_and_restart_attempts(self): self.ready(self.make('old',age=40));new=self.make('new');self.ready(new) row=self.store.claim('render');self.assertEqual(row['event_id'],new['eventId']) self.complete(row) row=self.store.claim('send') self.assertTrue(self.store.prepare_delivery(row,{'events':[{}]})) self.assertEqual(self.store.begin_attempt(row),1) self.store.recover() with self.store.transaction() as cur: cur.execute('UPDATE contextra_event_artifacts SET next_attempt_at=0 WHERE event_id=%s',(new['eventId'],)) row=self.store.claim('send') self.store.prepare_delivery(row,{'different':'not used'}) self.assertEqual(self.store.begin_attempt(row),2) self.store.finish_attempt(row,'FAILED',503,None,'TEST') self.store.release_claim(row,'send') self.store.recover() self.assertIsNone(self.store.claim('send')) def test_expired_pending_and_excluded_delivery_never_replay(self): e=self.make();self.ready(e) with self.store.transaction() as cur: cur.execute('UPDATE contextra_event_artifacts SET confirmed_at=%s WHERE event_id=%s',(time.time()-121,e['eventId'])) self.store.expire() row=self.store.claim('render');self.assertEqual(row['artifact_state'],'EXPIRED') self.store.cleanup_expired(row) self.assertEqual(self.store.summary()['storage']['reserved_jobs'],0) e=self.make('excluded');e['cameraId']=79;self.ready(e) self.complete(self.store.claim('render')) row=self.store.claim('send');self.assertIsNone(self.store.prepare_delivery(row,{'events':[{}]})) self.store.release_claim(row,'send') self.store.recover();self.assertIsNone(self.store.claim('send')) with self.store.transaction() as cur: cur.execute('SELECT delivery_state,attempts FROM contextra_event_artifacts WHERE event_id=%s',(e['eventId'],)) self.assertEqual(dict(cur.fetchone()),dict(delivery_state='SKIPPED',attempts=0)) def test_time_block_and_failed_artifact_release(self): e=self.make();self.store.accept(e);self.store.captured(e['eventId'],'NO_VIDEO') row=self.store.claim('render');self.store.rendered(row,{},'NO_VIDEO',0);self.store.release_claim(row,'render') self.assertEqual(self.store.summary()['storage']['reserved_jobs'],0) row=self.store.claim('send');self.assertIsNone(self.store.prepare_delivery(row,None)) self.store.release_claim(row,'send');self.assertIsNone(self.store.claim('send')) if __name__=='__main__':unittest.main()