Verify full and essentials recovery and implement bounded retention tooling

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883

tests/test_backup_retention_plan.py

@@ -0,0 +1,47 @@
+from datetime import datetime,timedelta,timezone
+from pathlib import Path
+import sys
+import unittest
+sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts'))
+from backup_retention_plan import plan
+
+class RetentionPlan(unittest.TestCase):
+    def objects(self):
+        now=datetime(2026,9,6,tzinfo=timezone.utc)
+        return now,[{'name':'forgejo-essentials-'+(now-timedelta(days=i)).strftime('%Y%m%dT%H%M%SZ')+'.zip.age',
+                     'verified':True,'etag':'"v'+str(i)+'"','sha256':'a'*64,'bytes':100} for i in range(30)]
+    def test_pinned_recovery_survives_daily_weekly_expiration(self):
+        now,items=self.objects();protected=items[-1]['name']
+        result=plan(items,[protected],100,3*1024**3,now)
+        self.assertIn(protected,result['keep']);self.assertIn(items[0]['name'],result['keep'])
+        self.assertGreater(len(result['delete_candidates']),0)
+        self.assertNotIn(protected,[i['name'] for i in result['delete_candidates']])
+    def test_unknown_full_and_unverified_objects_are_untouched(self):
+        now,items=self.objects();unknown={'name':'forgejo-dump-full.zip.age'}
+        items.append(unknown);items[-2]['verified']=False
+        result=plan(items,[items[0]['name']],100,3*1024**3,now)
+        self.assertIn(unknown['name'],result['ignored']);self.assertIn(items[-2]['name'],result['ignored'])
+    def test_missing_recovery_anchor_refuses(self):
+        now,items=self.objects()
+        with self.assertRaises(ValueError): plan(items,['missing'],100,3*1024**3,now)
+    def test_low_space_does_not_assume_deletion_frees_upload_room(self):
+        now,items=self.objects();result=plan(items,[items[0]['name']],100,0,now)
+        self.assertEqual(result['status'],'insufficient_upload_headroom');self.assertFalse(result['provider_mutations'])
+
+class OwnerExpiration(unittest.TestCase):
+    def fixture(self):
+        now,objects=RetentionPlan().objects()
+        p=plan(objects,[objects[0]['name']],100,3*1024**3,now)
+        return p,{'status':'restored_essentials','cleanup':True,'offsite_artifact':objects[0]['name']}
+    def test_pinned_anchor_cannot_be_expired(self):
+        from execute_nextcloud_retention import validate
+        p,r=self.fixture();p['delete_candidates'].append({'name':r['offsite_artifact'],'etag':'"v"','bytes':100})
+        with self.assertRaises(Exception):validate(p,r)
+    def test_full_archive_is_outside_executor_scope(self):
+        from execute_nextcloud_retention import validate
+        p,r=self.fixture();p['delete_candidates'].append({'name':'forgejo-dump-full.zip.age','etag':'"v"','bytes':100})
+        with self.assertRaises(Exception):validate(p,r)
+    def test_missing_recovery_proof_refuses(self):
+        from execute_nextcloud_retention import validate
+        p,r=self.fixture();r['status']='failed'
+        with self.assertRaises(Exception):validate(p,r)