from concurrent.futures import ThreadPoolExecutor from dataclasses import replace import json import time from unittest.mock import patch from app.artifact_completion import ArtifactCompletion from app.event_artifact_store import StaleArtifactClaim from tests.test_event_artifact_ledger import LedgerTests class CompletionTests(LedgerTests): def test_trace_claim_identity_survives_secret_masking(self): from contextra_trace.sanitize import copy_safe event=self.make();self.ready(event);row=self.store.claim('render') with patch('contextra_trace.emit') as emit: ArtifactCompletion(self.store).submit('rendered',row,{},'NO_VIDEO',0) record=emit.call_args.kwargs clean,_=copy_safe(record) self.assertEqual(clean['claimId'],ArtifactCompletion.key(row)) self.assertNotIn(row['claim_token'],json.dumps(clean)) def row(self, event): with self.store.transaction() as cur: cur.execute('SELECT * FROM contextra_event_artifacts WHERE event_id=%s',(event['eventId'],)) return dict(cur.fetchone()) def test_render_finish_releases_and_settles_atomically(self): event=self.make();self.ready(event);row=self.store.claim('render') original=self.store._release def fail(cur,*args): original(cur,*args) raise RuntimeError('injected before commit') with patch.object(self.store,'_release',side_effect=fail): with self.assertRaises(RuntimeError):self.store.rendered(row,{'clip':True},None,12) state=self.row(event) self.assertEqual((state['artifact_state'],state['reservation'],state['claim_token']), ('RENDERING',134217728,row['claim_token'])) self.store.rendered(row,{'clip':True},None,12) state=self.row(event) self.assertEqual((state['artifact_state'],state['reservation'],state['claim_token']),('READY',0,None)) self.assertEqual(self.store.summary()['storage']['reserved_jobs'],0) def test_http_success_db_retry_does_not_resend_and_receipt_survives(self): import psycopg2 event=self.make();self.ready(event);self.complete(self.store.claim('render')) row=self.store.claim('send');self.store.prepare_delivery(row,{'events':[{}]}) self.assertEqual(self.store.begin_attempt(row),1) completion=ArtifactCompletion(self.store) with patch.object(self.store,'finish_attempt',side_effect=psycopg2.OperationalError('db unavailable')): self.assertFalse(completion.submit('finish_attempt',row,'SENT',200,{'accepted':1,'duplicated':0},None)) self.assertEqual(self.row(event)['delivery_state'],'SENDING') self.assertEqual(len(list(completion.root.glob('*.json'))),1) # Restore runs before generic restart recovery and makes no HTTP request. restored=ArtifactCompletion(self.store);restored.restore() state=self.row(event) self.assertEqual((state['delivery_state'],state['attempts'],state['claim_token']),('SENT',1,None)) self.assertEqual(restored.status()['pending'],0) self.assertFalse(list(completion.root.glob('*.json'))) def test_connection_outcome_ambiguity_is_fenced_without_second_effect(self): import psycopg2 event=self.make();self.ready(event);row=self.store.claim('render') real=self.store.rendered;calls=[] def committed_then_lost(*args,**kwargs): calls.append(1);real(*args,**kwargs) raise psycopg2.OperationalError('commit response lost') completion=ArtifactCompletion(self.store) with patch.object(self.store,'rendered',side_effect=committed_then_lost): completion.submit('rendered',row,{'clip':True},None,123) ArtifactCompletion(self.store).restore() self.assertEqual(len(calls),1) self.assertEqual(self.store.summary()['storage']['observed_bytes'],123) self.assertEqual(self.store.summary()['storage']['reserved_jobs'],0) def test_run_and_token_fencing_and_concurrent_finalization(self): event=self.make();self.ready(event);row=self.store.claim('render') wrong=dict(row,payload=dict(row['payload'],runId='foreign-run')) with self.assertRaises(StaleArtifactClaim):self.store.rendered(wrong,{},None,0) wrong=dict(row,claim_token='foreign-token') with self.assertRaises(StaleArtifactClaim):self.store.rendered(wrong,{},None,0) def finish(_): try:self.store.rendered(row,{'clip':True},None,123);return True except StaleArtifactClaim:return False with ThreadPoolExecutor(max_workers=2) as pool: self.assertEqual(sum(pool.map(finish,range(2))),1) self.assertEqual(self.store.summary()['storage']['reserved_jobs'],0) def test_live_claim_never_recovered_by_completion_manager(self): event=self.make();self.ready(event);row=self.store.claim('render') completion=ArtifactCompletion(self.store);completion.retry();completion.restore() self.assertEqual(self.row(event)['claim_token'],row['claim_token']) def test_real_database_lock_timeout_then_bounded_completion(self): import psycopg2 event=self.make();self.ready(event);row=self.store.claim('render') blocker=psycopg2.connect(self.store.config.database) try: with blocker.cursor() as cur: cur.execute('SET search_path TO '+self.schema) cur.execute('SELECT id FROM contextra_artifact_storage WHERE id=1 FOR UPDATE') completion=ArtifactCompletion(self.store) self.assertFalse(completion.submit('rendered',row,{},'NO_VIDEO',0)) self.assertEqual(self.row(event)['claim_token'],row['claim_token']) blocker.rollback() for item in completion.pending.values():item['next_at']=0 completion.retry() self.assertEqual(completion.status()['pending'],0) self.assertIsNone(self.row(event)['claim_token']) finally:blocker.close() def test_retry_budget_is_finite_and_observable(self): event=self.make();self.ready(event);row=self.store.claim('render') completion=ArtifactCompletion(self.store) with patch.object(self.store,'rendered',side_effect=RuntimeError('unavailable')) as call: completion.submit('rendered',row,{},None,0) for _ in range(20): for item in completion.pending.values():item['next_at']=0 completion.retry() self.assertEqual(call.call_count,completion.ATTEMPTS) self.assertEqual(completion.status()['retry_exhausted'],1) self.assertEqual(self.row(event)['claim_token'],row['claim_token'])