import json from pathlib import Path from types import SimpleNamespace import tempfile import unittest from unittest.mock import patch from app.training_store import TrainingStore, DEFAULTS, MARKER, RESERVE class TrainingTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.store = TrainingStore(self.root) self.cfg = dict(DEFAULTS, min_free_bytes=0) def test_cas_validation_noop_and_restart_interval(self): initial = self.store.settings() self.assertEqual(self.store.patch(initial['revision'],{'enabled':True}),initial) with self.assertRaises(ValueError): self.store.patch(initial['revision'],{'width':True}) changed = self.store.patch(initial['revision'],{'minimum_persons':9}) with self.assertRaises(RuntimeError):self.store.patch(initial['revision'],{'width':640}) self.assertEqual(changed,TrainingStore(self.root).settings()) job,error=self.store.reserve(3,self.cfg,now=1000,reserve=100) self.assertIsNone(error) self.assertEqual(self.store.reserve(3,self.cfg,now=1001,reserve=100)[1],'interval') self.assertIsNone(self.store.reserve(5,self.cfg,now=1001,reserve=100)[1]) with self.store.db() as db: db.execute("UPDATE jobs SET state='READY' WHERE id=?",(job,)) db.execute('UPDATE cameras SET saved=1000 WHERE camera=?',('3',)) self.assertEqual(TrainingStore(self.root).reserve(3,self.cfg,now=1005,reserve=100)[1],'interval') def test_early_interval_skips_candidates_and_db_but_observes_settings(self): from app.training_capture import TrainingCapture from tests.test_provider_process_group import spec with patch.dict('os.environ',CONTEXTRA_TRAINING_ROOT=str(self.root)): capture=TrainingCapture(spec()) frame=SimpleNamespace(source_id='3',pts_ns=123) result=SimpleNamespace(frame=frame) with capture.store.db() as db: db.execute('INSERT INTO cameras VALUES (?,?,?)',('3',1000.,1000.)) with patch('app.training_capture.time.monotonic',return_value=100.), \ patch('app.training_store.time.time',return_value=1001.), \ patch.object(capture,'candidates',return_value=()) as candidates, \ patch.object(capture.store,'admission_wait',wraps=capture.store.admission_wait) as reads: capture.consider(result) for _ in range(100):capture.consider(result) self.assertEqual(reads.call_count,1);candidates.assert_not_called() current=capture.store.settings() capture.store.patch(current['revision'],{'interval_seconds':1}) with patch('app.training_capture.time.monotonic',return_value=102.), \ patch('app.training_store.time.time',return_value=1002.), \ patch.object(capture,'candidates',return_value=()) as candidates: capture.consider(result);candidates.assert_called_once() current=capture.store.settings();capture.store.patch(current['revision'],{'enabled':False}) with patch('app.training_capture.time.monotonic',return_value=104.), \ patch.object(capture,'candidates') as candidates: capture.consider(result);candidates.assert_not_called() def test_pending_hint_and_authoritative_duplicate_guard(self): job,error=self.store.reserve('3',self.cfg,now=1000,reserve=100) self.assertIsNone(error) self.assertEqual(self.store.admission_wait('3',self.cfg,now=5000),1.) self.assertEqual(self.store.reserve('3',self.cfg,now=5000,reserve=100)[1],'busy') self.store.finish(job,'FIXTURE_FAILURE') self.assertEqual(self.store.admission_wait('3',self.cfg,now=5000),0.) def test_parent_sequence_is_monotonic_without_frame_relay(self): from app.process_metadata import ChildPublisher from app.metadata_publisher import MetadataSettings publisher=ChildPublisher(MetadataSettings('unused'),'3','run',1,'token',sink=object()) publisher.channel('3','run',1,'RUNNING') publisher.note_provider_sequence(100);publisher.note_provider_sequence(2) self.assertEqual(publisher.metrics()['lastSeq'],100) self.assertTrue(all(value[1] is None for value in publisher._pending.values())) publisher.channel('3','run',1,'STOPPED');publisher.note_provider_sequence(101) self.assertEqual(publisher.metrics()['lastSeq'],100) def test_queue_capacity_failure_and_symlink_protection(self): self.assertGreater(RESERVE,32*1024**2+2*3840*2160*3+1024**2) for camera in range(4):self.assertIsNone(self.store.reserve(camera,self.cfg,reserve=100)[1]) self.assertEqual(self.store.reserve(9,self.cfg,reserve=100)[1],'busy') (self.root/'link').symlink_to('/etc') with self.assertRaises(ValueError):self.store.usage() with self.assertRaises(ValueError):self.store.safe('link/passwd') with self.assertRaises(ValueError):self.store.safe('../outside') def test_retention_only_owned_completed_bundle(self): owned=self.root/'bundle';owned.mkdir() (owned/'metadata.json').write_text(json.dumps(dict(owner=MARKER,jobId='owned'))) (owned/'raw.jpg').write_bytes(b'x'*20000) unknown=self.root/'unknown';unknown.mkdir();(unknown/'raw.jpg').write_bytes(b'x') stage=self.root/'.staging'/'pending';stage.mkdir(parents=True);(stage/'capture.bin').write_bytes(b'x') with self.store.db() as db: db.execute('INSERT INTO jobs VALUES (?,?,?,?,?,?,?)',('owned','3','READY',1,2,'bundle',None)) used=self.store.usage() with self.store.lock(),self.store.db() as db: self.assertTrue(self.store._space(db,dict(self.cfg,max_bytes=used-100),100)) self.assertFalse(owned.exists());self.assertTrue(unknown.exists());self.assertTrue(stage.exists()) def test_other_owner_reservation_protects_shared_disk_floor(self): with patch.dict('os.environ',CONTEXTRA_TRAINING_EXTERNAL_RESERVE_BYTES='200'), \ patch('app.training_store.shutil.disk_usage',return_value=SimpleNamespace(free=299)): with self.store.lock(),self.store.db() as db: self.assertFalse(self.store._space(db,dict(self.cfg,min_free_bytes=100),0)) def test_score_union_ignores_roi_tracking_time_and_display(self): from app.training_capture import TrainingCapture from tests.test_provider_process_group import spec config=spec();config['fall']['person_score']=.7;config['fight']['person_score']=.6 config['intrusion']=dict(enabled=True,confidence=.5) with patch.dict('os.environ',CONTEXTRA_TRAINING_ROOT=str(self.root)): capture=TrainingCapture(config) capture.pose=SimpleNamespace(nms_enabled=False) def command(score,cls=0): return SimpleNamespace(detection=SimpleNamespace(score=score,class_id=cls,model_key=capture.provider.model), track_id=None,visible=False,scenario=None) a,b,c,d=command(.5),command(.55),command(.8),command(.9,1) selected=capture.candidates(SimpleNamespace(commands=(a,b,c,d))) self.assertEqual(selected,(b,c)) def test_roi_render_uses_event_scenario_only(self): from PIL import Image from app.artifact_media import draw_frame image=Image.new('RGB',(100,100)) snapshot=dict(objects=[],scenarioConfig={ 'FALL':dict(debug_roi=True,roi=[[.1,.1],[.4,.1],[.4,.4]]), 'INTRUSION':dict(debug_roi=True,roi=[[.6,.6],[.9,.6],[.9,.9]])}) result=draw_frame(image,snapshot,(100,100),scenario='FALL') self.assertNotEqual(result.getpixel((10,10)),(0,0,0)) self.assertEqual(result.getpixel((60,60)),(0,0,0)) self.assertEqual(draw_frame(image,snapshot,(100,100),scenario='MISSING').getpixel((10,10)),(0,0,0)) def test_authenticated_settings(self): from fastapi import FastAPI from fastapi.testclient import TestClient from app.secure_control import build_router app=FastAPI();app.state.training=SimpleNamespace(store=self.store,status=lambda:{}) app.include_router(build_router(SimpleNamespace(),'secret-token-123456','server')) client=TestClient(app) path='/internal/secure-control/training/settings' self.assertEqual(client.get(path).status_code,401) headers={'Authorization':'Bearer secret-token-123456'} current=client.get(path,headers=headers).json() self.assertEqual(client.patch(path,headers=headers,json=dict(expected_revision=current['revision'], changes={'interval_seconds':12})).status_code,200) self.assertEqual(client.patch(path,headers=headers,json=dict(expected_revision=current['revision'], changes={'interval_seconds':13})).status_code,409) if __name__=='__main__':unittest.main()