import threading import time import unittest from types import SimpleNamespace from unittest.mock import patch from app.byte_tracking import ByteTracker, TrackingSettings from app.contracts.frames import FrameEnvelope, DetectionObservation from app.frame_ingress import FrameIngress from app.det_result_path import DetResultPath from app.native_metadata import NativeDetection, NativeFrame from app.metadata_publisher import MetadataPublisher, MetadataSettings def wait_for(predicate): deadline = time.monotonic() + 3 while not predicate() and time.monotonic() < deadline: time.sleep(.005) assert predicate(), 'bounded result wait expired' def envelope(seq=0, run='run', generation=1): return FrameEnvelope('server', 'camera', 'stream', run, generation, seq, 0, 100, 1280, 720) def native(score=.9, empty=False): return NativeFrame(0, 0, 1280, 720, () if empty else (NativeDetection(0, score, (.1, .1, .3, .3)),), 640, 360) class Collector: def __init__(self): self.items = [] def add(self, *args, **kwargs): self.items.append(args) class Frame: frame_number = 0 buffer_pts = 0 source_width = 1280 source_height = 720 def __init__(self): self.meta = [] def user_meta_items(self, kind): return iter(self.meta) def append(self, value): self.meta.append(value) class Meta: def set_user_data_json(self, value, kind): self.value = value def get_user_data_json(self): return self.value class IdentityTest(unittest.TestCase): def setUp(self): self.patch = patch('app.frame_ingress.identity_meta_type', return_value=1234) self.patch.start() self.addCleanup(self.patch.stop) self.rtp_patch = patch('app.frame_ingress.rtp_meta_type', return_value=4321) self.rtp_patch.start() self.addCleanup(self.rtp_patch.stop) def append(batch, frame, value): meta = Meta() meta.value = value frame.append(meta) self.writer = patch('app.frame_ingress.append_identity', side_effect=append) self.writer.start() self.addCleanup(self.writer.stop) self.ingress = FrameIngress('server', 'camera', 'stream', 'run', 1, clock_ns=lambda: 123000000) def stamp(self, frame): self.ingress.stamp(SimpleNamespace(frame_items=[frame], acquire_user_meta=Meta)) def test_frame_zero_and_duplicate_pts_skip_identity(self): first, second = Frame(), Frame() self.stamp(first) self.stamp(second) a = self.ingress.copy(first) with self.assertRaises(ValueError): self.ingress.copy(second) self.assertEqual(a.frame_seq, 0) self.assertEqual(self.ingress._seq, 2) self.assertEqual(a.ingest_time_wall_ms, 123) self.assertIsNone(a.frame_time) first.meta.clear() self.assertEqual(a.frame_width, 1280) def test_missing_duplicate_foreign_and_changed_dimensions_rejected(self): f = Frame() with self.assertRaises(ValueError): self.ingress.copy(f) self.stamp(f) with self.assertRaises(ValueError): self.stamp(f) with self.assertRaises(ValueError): FrameIngress('server', 'camera', 'stream', 'new', 2).copy(f) f.source_width = 12 with self.assertRaises(ValueError): self.ingress.copy(f) def test_no_fake_install_identity(self): with self.assertRaises(ValueError): FrameIngress(None, 'camera', 'stream', 'run', 1) class TrackingTest(unittest.TestCase): def test_dependency_and_native_abi_failure_are_explicit(self): from app.frame_ingress import require_identity_binding with patch('app.byte_tracking.version', return_value='0.31.0'): with self.assertRaisesRegex(RuntimeError, 'validated supervision'): ByteTracker() with patch.dict('sys.modules', contextra_frame_meta=SimpleNamespace()): with self.assertRaisesRegex(RuntimeError, 'ABI'): require_identity_binding() def test_actual_bytetrack_low_score_id_and_class_isolation(self): tracker = ByteTracker() self.addCleanup(tracker.reset) def obs(score, cls=0): return DetectionObservation(envelope(), 'RFDETRMedium', cls, score, (.1, .1, .3, .3)) first = tracker.update((obs(.9), obs(.9, 1))) self.assertTrue(all(x is not None for x in first)) self.assertNotEqual(*first) low = tracker.update((obs(.2), obs(.2, 1))) self.assertEqual(first, low) self.assertEqual(first, tracker.update((obs(.9), obs(.9, 1)))) self.assertEqual([], tracker.update(())) tracker.reset() self.assertEqual(0, tracker.active_tracks()) def test_unmatched_candidate_preserved_and_parameters_are_not_publish_floor(self): tracker = ByteTracker() observations = tuple(DetectionObservation(envelope(), 'RFDETRMedium', 0, score, (i*.2, 0, i*.2+.1, .1)) for i, score in enumerate((.9, .05))) ids = tracker.update(observations) self.assertEqual(len(ids), 2) self.assertIsNone(ids[1]) with self.assertRaises(ValueError): TrackingSettings(frame_rate=0) class ResultPathTest(unittest.TestCase): def make_path(self, **kwargs): self.collector = Collector() self.errors = [] self.active = True path = DetResultPath(FrameIngress('server', 'camera', 'stream', 'run', 1), kwargs.pop('tracker', ByteTracker()), 'RFDETRMedium', .5, self.collector, lambda: self.active, lambda: self.errors.append(True), **kwargs) self.addCleanup(lambda: path.close(2)) return path def test_native_all_candidates_snapshot_publisher_empty_and_order(self): publisher = MetadataPublisher(MetadataSettings('not-connected'), sink=object()) path = self.make_path(publisher=publisher) path.start() self.assertTrue(path.submit(envelope(), native())) wait_for(lambda: path.status().get('processed') == 1) original = path.snapshot() self.assertEqual(original['frameSeq'], 0) tid = original['objects'][0]['trackId'] self.assertTrue(path.submit(envelope(1), native(.2))) wait_for(lambda: path.status().get('processed') == 2) low = path.snapshot()['objects'][0] self.assertEqual(low['trackId'], tid) self.assertFalse(low['visible']) self.assertEqual(self.collector.items[-1][3]['count'], 0) self.assertFalse(path.submit(envelope(1), native())) self.assertTrue(path.submit(envelope(2), native(empty=True))) wait_for(lambda: path.status().get('processed') == 3) self.assertEqual(path.snapshot()['objects'], []) self.assertEqual(path.status().get('publisher_rejected', 0), 0) original['objects'].clear() self.assertEqual(tid, low['trackId']) self.active = False self.assertFalse(path.submit(envelope(3), native())) with self.assertRaises(ValueError): path.submit(envelope(4, 'new', 2), native()) path.close(2) self.assertFalse(publisher._closed) # process-owned transport를 닫지 않는다. def test_queue_is_bounded_and_age_drop(self): clock = [0.] path = self.make_path(capacity=2, clock=lambda: clock[0]) for i in range(5): path.submit(envelope(i), native()) self.assertEqual(path.status()['queued'], 2) self.assertEqual(path.status()['queue_drop'], 3) clock[0] = 2. path.start() wait_for(lambda: path.status().get('stale') == 2) self.assertEqual(self.collector.items, []) def test_error_not_silent_and_worker_cleanup_failure_retains_handle(self): entered, release = threading.Event(), threading.Event() class BlockingTracker: def update(self, observations): entered.set() release.wait(3) raise ValueError('private detail') def reset(self): pass path = self.make_path(tracker=BlockingTracker()) path.start() path.submit(envelope(), native()) self.assertTrue(entered.wait(1)) try: with self.assertRaises(RuntimeError): path.close(.01) self.assertTrue(path._thread.is_alive()) finally: release.set() wait_for(lambda: bool(self.errors)) path.close(1) self.assertEqual(path.status()['error'], 'DET_RESULT_FAILED') self.assertNotIn('private', str(path.status())) def test_geometry_invalid_and_new_run_reset(self): path = self.make_path() wrong = NativeFrame(0, 0, 640, 360, (), 640, 360) with self.assertRaises(ValueError): path.submit(envelope(), wrong) path.start() path.submit(envelope(), native()) wait_for(lambda: path.status().get('processed') == 1) self.assertGreater(path.tracker.active_tracks(), 0) path.close(1) self.assertEqual(path.tracker.active_tracks(), 0) self.assertIsNone(path.snapshot()) class GroupConnectionTest(unittest.TestCase): def test_identity_validation_precedes_config_write_and_no_silent_fallback(self): from app.stream_manager import StreamGroup with self.assertRaises(ValueError): StreamGroup({'stream_id': 'stream', 'server_id': 'server'}) group = StreamGroup({'stream_id': 'stream'}) self.assertEqual(group.status()['metadata_path']['state'], 'NOT_CONNECTED') with self.assertRaises(ValueError): group.check_restart({'stream_id': 'stream', 'source_id': 'camera', 'server_id': 'server', 'tracking': {'frame_rate': 0}}) self.assertEqual(group.spec, {'stream_id': 'stream'}) def test_group_preparation_worker_cleanup_and_new_generation(self): from app.stream_manager import StreamGroup spec = dict(stream_id='stream', source_id='camera', server_id='server', store=False) group = StreamGroup(spec) profile = SimpleNamespace(roles=('det',), resolved_models={'det': 'RFDETRMedium'}, score_threshold=.5) runs = [] with patch('app.frame_ingress.require_identity_binding'), patch.object(group, '_begin'): for _ in range(2): generation = group.start() stop = group._run_stop_event path = group._prepare_det_results(profile, generation, stop, Collector()) runs.append(path.ingress.identity[3]) group.tracker = path.tracker frame = FrameEnvelope('server', 'camera', 'stream', runs[-1], generation, 0, 0, 100, 1280, 720) self.assertTrue(path.submit(frame, native())) wait_for(lambda: path.status().get('processed') == 1) stop.set() group.stop() self.assertTrue(group._cleanup_done.wait(2)) self.assertFalse(path._thread.is_alive()) self.assertIsNone(group.result_path) self.assertIsNone(group.tracker) self.assertIsNone(group._cleanup_failure) self.assertNotEqual(*runs) def test_multi_role_identity_not_claimed_as_join(self): from app.stream_manager import StreamGroup group = StreamGroup(dict(stream_id='stream', source_id='camera', server_id='server')) with self.assertRaisesRegex(ValueError, 'join'): group._prepare_det_results(SimpleNamespace(roles=('det','pose')), 1, threading.Event(), Collector()) if __name__ == '__main__': unittest.main()