import asyncio from io import BytesIO import json from pathlib import Path import tempfile import threading import unittest from PIL import Image import httpx from halloween.controller import Controller from halloween.detection import DEFAULT_CONFIG, DetectionService, LineTracker, NanoDet, validate_config from halloween.event_store import EventStore from halloween.web import create_app def person(x=0.5, y=0.4): return {"box": [x-0.03, y-0.1, x+0.03, y], "score": 0.9} class TrackerTest(unittest.TestCase): def setUp(self): self.tracker = LineTracker() self.config = json.loads(json.dumps(DEFAULT_CONFIG)) self.time = 0 def step(self, people): self.time += 0.2 return self.tracker.update(people, self.config, self.time)[1] def test_forward_and_reverse(self): for _ in range(3): self.assertFalse(self.step([person()])) self.assertTrue(self.step([person(y=0.65)])) self.assertFalse(self.step([person(y=0.4)])) self.assertFalse(self.step([person(y=0.4)])) def test_reverse_direction_configuration(self): self.config['direction'] = -1 for _ in range(3): self.assertFalse(self.step([person(y=0.65)])) self.assertTrue(self.step([person(y=0.4)])) def test_appearance_behind_line(self): for _ in range(5): self.assertFalse(self.step([person(y=0.65)])) def test_jitter_and_parallel_motion(self): for y in [0.54, 0.56, 0.54, 0.56, 0.54, 0.56]: self.assertFalse(self.step([person(y=y)])) for x in [0.45, 0.5, 0.55, 0.6]: self.assertFalse(self.step([person(x=x, y=0.4)])) def test_flicker_and_outside_segment(self): for _ in range(3): self.step([person()]) self.step([]) self.assertFalse(self.step([person(y=0.65)])) self.setUp() for _ in range(3): self.step([person(x=0.15)]) self.assertFalse(self.step([person(x=0.15, y=0.65)])) class ServiceTest(unittest.TestCase): def setUp(self): self.folder = tempfile.TemporaryDirectory() self.addCleanup(self.folder.cleanup) self.directory = Path(self.folder.name) self.time = 0 self.controller = Controller(self.directory, clock=lambda: self.time) self.events = EventStore(self.directory) self.camera = type('Camera', (), {'error': None})() self.service = DetectionService(self.camera, self.controller, self.events, self.directory, clock=lambda: self.time) self.service.configure({**DEFAULT_CONFIG, 'enabled': True}) buffer = BytesIO() Image.new('RGB', (640, 480)).save(buffer, format='JPEG') self.jpeg = buffer.getvalue() self.number = 0 def step(self, people=(), dt=0.2): self.time += dt self.number += 1 self.service.process(list(people), self.number, self.jpeg, self.time, self.service.revision) def free(self): for _ in range(12): self.step() def cross(self, x=0.5): for _ in range(3): self.step([person(x=x)]) self.step([person(x=x, y=0.65)]) def test_group_cooldown_and_new_group(self): self.free() self.assertTrue(self.service.status()['armed']) self.cross() self.assertEqual(len(self.events.recent()), 1) event = self.events.recent()[0] self.assertEqual(event['kind'], 'person_line_crossing') self.assertIsNone(event['profile_id']) self.cross(x=0.65) self.assertEqual(len(self.events.recent()), 1) for _ in range(60): self.step([person(y=0.65)]) self.cross(x=0.65) self.assertEqual(len(self.events.recent()), 1) self.free() self.cross() self.assertEqual(len(self.events.recent()), 2) def test_no_delayed_crossing(self): self.free() self.cross() self.free() self.cross(x=0.65) self.assertEqual(len(self.events.recent()), 1) for _ in range(60): self.step([person(x=0.65, y=0.65)]) self.assertEqual(len(self.events.recent()), 1) def test_start_stop_reset_and_configuration(self): self.cross() self.assertEqual(self.events.recent(), []) self.free() self.controller.stop() self.cross() self.assertEqual(self.events.recent(), []) self.controller.reset() self.service.reset() self.cross() self.assertEqual(self.events.recent(), []) self.free() old_revision = self.service.revision self.service.configure(self.service.get_config()) self.service.process([], 99, self.jpeg, self.time, old_revision) self.assertIsNone(self.service.last_valid) self.cross() self.assertEqual(self.events.recent(), []) self.free() self.cross() self.assertEqual(len(self.events.recent()), 1) def test_watchdog_and_manual_bypass(self): self.free() self.time += 2.01 self.service.watchdog() self.assertTrue(self.controller.stopped) self.assertTrue(self.service.status()['reset_required']) self.controller.zone_clear() self.assertFalse(self.controller.trigger()[0]) self.free() self.assertTrue(self.service.status()['reset_required']) self.controller.reset() self.service.reset() self.free() self.cross() self.assertEqual(len(self.events.recent()), 1) def test_camera_error_and_stale_results(self): self.free() self.camera.error = 'camera disconnected' self.service.watchdog() self.assertTrue(self.service.fault) self.assertTrue(self.controller.stopped) self.time += 3 result = self.service.snapshot() self.assertEqual(result['people'], []) self.assertIsNone(result['image_base64']) def test_disabled_detection_does_not_stop_manual_tests(self): self.service.configure(DEFAULT_CONFIG) self.service.fail('missing model') self.assertFalse(self.controller.stopped) self.controller.zone_clear() self.assertTrue(self.controller.trigger()[0]) def test_inference_stall_is_stopped_by_independent_watchdog(self): entered, release = threading.Event(), threading.Event() class BlockingDetector: def detect(inner, jpeg): entered.set() release.wait(3) return [] self.camera.latest_timed = lambda: (1, self.jpeg, self.time) self.service.factory = lambda path: BlockingDetector() self.service.start() try: self.assertTrue(entered.wait(1)) self.time += 2.1 self.service.watchdog() self.assertTrue(self.controller.stopped) self.assertTrue(self.service.fault) finally: release.set() self.service.stop() def test_database_stall_does_not_block_watchdog(self): entered, release = threading.Event(), threading.Event() def blocked_record(mode): entered.set() release.wait(3) self.events.record_detection_trigger = blocked_record self.free() for _ in range(3): self.step([person()]) worker = threading.Thread(target=lambda: self.step([person(y=0.65)])) worker.start() try: self.assertTrue(entered.wait(1)) self.time += 2.1 self.service.watchdog() self.assertTrue(self.controller.stopped) finally: release.set() worker.join(timeout=1) self.assertFalse(worker.is_alive()) def test_database_failure_stops_simulation(self): def broken_record(mode): raise OSError('disk full') self.events.record_detection_trigger = broken_record self.free() self.cross() self.assertTrue(self.controller.stopped) self.assertTrue(self.service.fault) self.assertIn('disk full', self.service.status()['error']) def test_shutdown_drops_inflight_frame(self): self.free() version = self.service.revision self.service.stop() self.service.process([], 99, self.jpeg, self.time, version) self.assertTrue(self.controller.stopped) self.assertEqual(self.events.recent(), []) def test_persistence_and_corrupt_settings(self): other = DetectionService(self.camera, self.controller, self.events, self.directory) self.assertTrue(other.get_config()['enabled']) self.assertFalse(other.status()['armed']) self.service.path.write_text('{') other = DetectionService(self.camera, self.controller, self.events, self.directory) self.assertFalse(other.get_config()['enabled']) self.assertIsNotNone(other.status()['error']) def test_unconfirmed_person_prevents_clear(self): for _ in range(30): self.step([person()]) self.assertFalse(self.service.status()['armed']) def test_snapshot_matches_number_and_image(self): self.step([person()]) snapshot = self.service.snapshot() self.assertEqual(snapshot['frame_number'], self.number) self.assertTrue(snapshot['image_base64']) self.assertEqual(len(snapshot['people']), 1) class ApiTest(unittest.TestCase): def test_config_validation_persistence_and_missing_model(self): async def scenario(): with tempfile.TemporaryDirectory() as folder: app = create_app(demo=True, state_dir=Path(folder)) async with app.router.lifespan_context(app): async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url='http://test') as client: headers = {'X-Halloween-UI': '1'} config = (await client.get('/api/detection/config')).json() self.assertFalse(config['enabled']) self.assertEqual((await client.post('/api/detection/config', json=config)).status_code, 403) bad = {**config, 'region': [0.8,0.2,0.3,0.9]} self.assertEqual((await client.post('/api/detection/config', json=bad, headers=headers)).status_code, 422) config['enabled'] = True self.assertEqual((await client.post('/api/detection/config', json=config, headers=headers)).status_code, 200) for _ in range(30): await asyncio.sleep(0.05) status = (await client.get('/api/status')).json() if status['detection']['reset_required']: break self.assertTrue(status['detection']['reset_required']) self.assertTrue(status['controller']['stopped']) self.assertIn('NanoDet fehlt', status['detection']['error']) result = (await client.get('/api/detection')).json() self.assertIsNone(result['image_base64']) self.assertTrue(create_app(demo=True, state_dir=Path(folder)).state.detection.get_config()['enabled']) asyncio.run(scenario()) def test_invalid_values(self): for config in [ {**DEFAULT_CONFIG, 'direction': True}, {**DEFAULT_CONFIG, 'line': [[0.5,0.5],[0.5,0.5]]}, {**DEFAULT_CONFIG, 'region': [0,0,float('nan'),1]}, {**DEFAULT_CONFIG, 'enabled': 'true'}]: with self.assertRaises(ValueError): validate_config(config) if __name__ == '__main__': unittest.main()