import contextlib import hashlib import json import signal import sqlite3 import tempfile import time import unittest from pathlib import Path from types import SimpleNamespace as S from unittest.mock import patch import full_collect as f class RecoveryTests(unittest.TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory();self.root=Path(self.tmp.name) self.patches=[patch.object(f,'ROOT',self.root),patch.object(f,'STATE',self.root/'state'),patch.object(f.old,'OUT_ROOT',self.root/'competitions'),patch.object(f,'RECEIPT_DB',self.root/'receipts.sqlite'),patch.object(f,'room',return_value=True),patch.object(f,'metadata_competition_lock',side_effect=lambda slug:contextlib.nullcontext())] for p in self.patches:p.start() f.STATE.mkdir();f.old.OUT_ROOT.mkdir() def tearDown(self): for p in reversed(self.patches):p.stop() self.tmp.cleanup() def test_requests_without_timeout_get_bounded_and_explicit_timeout_survives(self): calls=[] def original(self,method,url,**kw):calls.append(kw);return kw with patch.object(f.requests.sessions.Session,'request',original): f.install_metadata_transport();f.install_metadata_transport() session=f.requests.Session() session.get('https://example.org/a',timeout=None) session.get('https://example.org/b',timeout=(3,4)) self.assertEqual(calls[0]['timeout'],(15,75));self.assertEqual(calls[1]['timeout'],(3,4)) def test_trickling_attachment_is_deferred_with_partial_and_timer_restored(self): folder=f.old.OUT_ROOT/'x/data/assets';url='https://example.org/slow.tar' class Response: headers={'Content-Length':'100000','ETag':'same-object'} def __enter__(self):return self def __exit__(self,*args):self.close() def raise_for_status(self):pass def close(self):pass def iter_content(self,n): while True:time.sleep(.02);yield b'x' old_handler=signal.getsignal(signal.SIGALRM) with patch.object(f,'METADATA_ATTACHMENT_SECONDS',.11),patch.object(f,'room',return_value=True),patch.object(f.requests,'get',return_value=Response()): with f.metadata_deadline(dict(done=0,total=1,current_slug='x'),seconds=1,heartbeat=.03): items=f.page_assets(url,'https://example.org',folder) self.assertIsNone(f.METADATA_ACTIVE['attachment_deadline']) partial=folder/'.repair-download'/(hashlib.sha256(url.encode()).hexdigest()[:24]+'.tar') self.assertEqual(items[0]['status'],'deferred_timeout');self.assertGreater(partial.stat().st_size,0) self.assertEqual(f.read(partial.with_name(partial.name+'.resume.json'))['validator'],'same-object') self.assertEqual(signal.getitimer(signal.ITIMER_REAL)[0],0);self.assertEqual(signal.getsignal(signal.SIGALRM),old_handler) def test_total_deadline_escapes_broad_exception_handlers_preserves_record_then_continues(self): row=dict(id='1',slug='slow');p=f.old.OUT_ROOT/'slow/ADDITIONS.json' f.atomic(p,dict(version=2,status='partial',data={'status':'ok','local_path':'original.md'})) def stuck(*args): try:time.sleep(10) except Exception:raise AssertionError('wall-clock timeout was swallowed') with patch.object(f,'METADATA_JOB_SECONDS',.08),patch.object(f,'enrich',side_effect=stuck): out=f.bounded_enrich(row,{},None,None,dict(done=0,total=2,current_slug='slow')) self.assertEqual(out['status'],'partial');self.assertEqual(out['data']['local_path'],'original.md') self.assertTrue(list((f.STATE/'history').rglob('*.previous'))) with patch.object(f,'enrich',return_value={'status':'ok'}): self.assertEqual(f.bounded_enrich(dict(id='2',slug='next'),{},None,None,dict(done=1,total=2,current_slug='next'))['status'],'ok') self.assertEqual(signal.getitimer(signal.ITIMER_REAL)[0],0) def test_outer_deadline_wins_over_long_attachment_budget(self): with patch.object(f,'METADATA_ATTACHMENT_SECONDS',5): with self.assertRaises(f.MetadataTimeout): with f.metadata_deadline(dict(done=0,total=1),seconds=.06,heartbeat=.02): with f.attachment_deadline():time.sleep(10) self.assertIsNone(f.METADATA_ACTIVE);self.assertEqual(signal.getitimer(signal.ITIMER_REAL)[0],0) def test_resume_checks_catalog_and_retries_inflight_item(self): rows=[dict(id=str(i),slug=str(i)) for i in range(3)] self.assertEqual(f.metadata_resume_index(rows,dict(phase='live_backfill',current_slug='1'),{}),1) cp=dict(catalog_digest=f.catalog_digest(rows),next_index=2) self.assertEqual(f.metadata_resume_index(rows,{},cp),2) self.assertEqual(f.metadata_resume_index(list(reversed(rows)),{},cp),0) def test_worker_resumes_tail_without_replaying_offline_bootstrap(self): rows=[dict(id=str(i),slug=str(i)) for i in range(3)] f.atomic(f.STATE/'metadata-status.json',dict(phase='live_backfill',current_slug='1',done=1)) calls=[] def enrich(row,*args): calls.append(row['slug']) if row['slug']=='2':raise StopIteration('simulated interruption') return {'status':'partial'} module=S(KaggleApi=lambda:S(authenticate=lambda:None)) with patch.dict('sys.modules',{'kaggle.api.kaggle_api_extended':module}),patch.object(f,'install_metadata_transport'),patch.object(f,'metadata_rows',return_value={}),patch.object(f.old,'session_from_kaggle_json'),patch.object(f,'bootstrap',side_effect=AssertionError('bootstrap already finished')),patch.object(f,'bounded_enrich',side_effect=enrich): with self.assertRaises(StopIteration):f.metadata_worker(rows) self.assertEqual(calls,['1','2']);self.assertEqual(f.read(f.STATE/'metadata-checkpoint.json')['next_index'],2) def test_verified_move_receipt_prevents_attachment_redownload(self): url='https://example.org/moved.pdf';folder=f.old.OUT_ROOT/'x/data/assets' final=folder/(hashlib.sha256(url.encode()).hexdigest()[:24]+'.pdf') with sqlite3.connect(f.RECEIPT_DB) as db: db.execute('create table receipts(id text,dataset text,path text,status text,note text,bytes integer)') db.execute('insert into receipts values(?,?,?,?,?,?)',('receipt','kaggle',str(final.relative_to(f.ROOT)),'removed','verified_and_unlinked',123)) with patch.object(f.requests,'get',side_effect=AssertionError('delivered resource must not be downloaded')): items=f.page_assets(url,'https://example.org',folder) self.assertEqual(items[0]['status'],'delivered_external');self.assertEqual(items[0]['receipt_id'],'receipt');self.assertFalse(final.exists()) def test_complete_record_never_calls_network_or_waits_on_collection_lock(self): row=dict(id='1',slug='complete');saved=dict(version=2,status='ok',data={'status':'ok'}) f.atomic(f.old.OUT_ROOT/'complete/ADDITIONS.json',saved) with patch.object(f,'enrich',side_effect=AssertionError('duplicate collection')): self.assertEqual(f.bounded_enrich(row,{},None,None,{}),saved) if __name__=='__main__':unittest.main()