from dataclasses import replace import json import os from pathlib import Path import time import unittest from unittest.mock import patch from app.artifact_retention import maintain, _remove from app.event_artifact_contract import RESERVATION, ArtifactConfig, relative_directory from tests.test_event_artifact_ledger import LedgerTests class RetentionTests(LedgerTests): def bundle(self, *, terminal=True, age=0): event=self.make('channel-'+str(time.time_ns()),age=age) self.ready(event) row=self.store.claim('render') folder=self.store.config.root/relative_directory(event) folder.mkdir(parents=True) (folder/'clip.mp4').write_bytes(b'x'*2048) self.store.rendered(row,{'clip':True},None,2048) self.store.release_claim(row,'render') if terminal: with self.store.transaction() as cur: cur.execute("UPDATE contextra_event_artifacts SET delivery_state='SENT',attempts=1 WHERE event_id=%s",(event['eventId'],)) return event,folder def pressure(self): self.store.scan_storage() self.store.config=replace(self.store.config,retention_enabled=True,cap=RESERVATION+1024) def test_oldest_terminal_only_and_ledger_preserved(self): a,pa=self.bundle(age=20);b,pb=self.bundle(terminal=False);c,pc=self.bundle() self.pressure() self.assertEqual(maintain(self.store,max_jobs=1)['deleted'],1) self.assertFalse(pa.exists());self.assertTrue(pb.exists());self.assertTrue(pc.exists()) with self.store.transaction() as cur: cur.execute('SELECT * FROM contextra_event_artifacts WHERE event_id=%s',(a['eventId'],)) row=cur.fetchone() self.assertEqual(row['payload'],a) self.assertEqual((row['attempts'],row['delivery_state'],row['artifact_state'],row['media_retention_state']),(1,'SENT','READY','DELETED')) def test_resume_after_rename_and_scan_does_not_double_subtract(self): e,p=self.bundle();self.pressure() with patch('app.artifact_retention.shutil.rmtree',side_effect=RuntimeError('INTERRUPTED')) as delete: delete.avoids_symlink_attacks=True with self.assertRaises(RuntimeError):maintain(self.store) self.assertFalse(p.exists()) tomb=self.store.config.root/'.retiring'/e['eventId'];self.assertTrue(tomb.exists()) self.store.recover() self.assertEqual(maintain(self.store)['deleted'],1) self.assertFalse(tomb.exists()) self.assertEqual(self.store.summary()['storage']['observed_bytes'],0) self.assertEqual(maintain(self.store)['deleted'],0) def test_symlink_and_ownership_mismatch_refused(self): e,p=self.bundle();outside=self.store.config.root/'unowned';outside.write_text('keep') (p/'link').symlink_to(outside);self.pressure() self.assertEqual(maintain(self.store)['blocked'],1) self.assertTrue(outside.exists());self.assertTrue(p.exists()) self.store.config=replace(self.store.config,cap=100*1024**3) e2,p2=self.bundle() with self.store.transaction() as cur: cur.execute("UPDATE contextra_event_artifacts SET relative_path='unowned' WHERE event_id=%s",(e2['eventId'],)) self.pressure() self.assertEqual(maintain(self.store)['blocked'],1) self.assertTrue(outside.exists());self.assertTrue(p2.exists()) def test_active_claim_and_default_off_are_protected(self): e,p=self.bundle();self.assertEqual(maintain(self.store)['deleted'],0) self.pressure() with self.store.transaction() as cur: cur.execute("UPDATE contextra_event_artifacts SET claim_token='busy' WHERE event_id=%s",(e['eventId'],)) self.assertEqual(maintain(self.store)['deleted'],0);self.assertTrue(p.exists()) def test_external_pending_storage_is_reserved(self): from types import SimpleNamespace self.store.config=replace(self.store.config,low_free=100,external_reserve=200) with patch('app.event_artifact_store.shutil.disk_usage',return_value=SimpleNamespace(free=RESERVATION+299)): self.assertEqual(self.store.accept(self.make()),(True,'STORAGE_FREE_LOW')) def test_high_water_starts_below_cap_and_stops_at_target(self): a,pa=self.bundle(age=20);b,pb=self.bundle() self.store.scan_storage() # Include the next-job reservation in the capacity calculation. used=self.store.summary()['storage']['observed_bytes']+RESERVATION cap=used*100//95 self.store.config=replace(self.store.config,retention_enabled=True,cap=cap, cleanup_target=used-2048) self.assertLess(used,cap) self.assertEqual(maintain(self.store)['deleted'],1) self.assertFalse(pa.exists());self.assertTrue(pb.exists()) self.assertEqual(maintain(self.store)['deleted'],0) def test_owned_failed_staging_is_cleaned_but_unknown_is_preserved(self): e,p=self.bundle();stage=self.store.config.root/'.staging'/e['eventId'] stage.mkdir(parents=True);(stage/'capture.bin').write_bytes(b'cache') self.pressure() self.assertEqual(maintain(self.store)['blocked'],1) self.assertTrue(stage.exists());self.assertTrue(p.exists()) (stage/'failure.json').write_text(json.dumps(dict(event=e,error='NO_VIDEO'))) with self.store.transaction() as cur: cur.execute("UPDATE contextra_event_artifacts SET media_retention_state='PRESENT' WHERE event_id=%s",(e['eventId'],)) self.assertEqual(maintain(self.store)['deleted'],1) self.assertFalse(stage.exists());self.assertFalse(p.exists()) class RetentionConfigTests(unittest.TestCase): def test_site_bytes_and_legacy_default(self): env=dict(CONTEXTRA_EVENT_ARTIFACT_ENABLED='1',DATABASE_URL='fixture',CONTEXTRA_EVENT_ENDPOINT='http://web.invalid/events',CONTEXTRA_ARTIFACT_PUBLIC_BASE='http://ai.invalid/artifacts') with patch.dict(os.environ,env,clear=True): self.assertFalse(ArtifactConfig.from_environment().retention_enabled) with patch.dict(os.environ,CONTEXTRA_ARTIFACT_CAP_BYTES='4000000000000',CONTEXTRA_ARTIFACT_MIN_FREE_BYTES='100000000000',CONTEXTRA_ARTIFACT_RETENTION_ENABLED='true'): c=ArtifactConfig.from_environment();self.assertEqual(c.cap,4_000_000_000_000) self.assertEqual(c.low_free,100_000_000_000);self.assertTrue(c.retention_enabled) with patch.dict(os.environ,CONTEXTRA_ARTIFACT_CAP_BYTES='nan'): with self.assertRaises(ValueError):ArtifactConfig.from_environment() if __name__=='__main__':unittest.main()