import unittest from dataclasses import FrozenInstanceError from types import SimpleNamespace as NS from unittest.mock import patch from app.native_metadata import copy_detection_frame, failure_diagnostic, NativeMetadataError def frame(objects=None): objects = [] if objects is None else objects completion = dict(schema=1, status='completed', component=7, frame=0, pts=0, width=640, height=360, count=sum(int(o.unique_component_id) == 7 for o in objects)) return NS(frame_number=0, buffer_pts=0, source_width=1280, source_height=720, pipeline_width=640, pipeline_height=360, object_items=objects, user_meta_items=lambda _type: [NS(get_user_data_json=lambda: completion)]) def detection(component=7, score=.001): return NS(unique_component_id=component, class_id=0, confidence=score, rect_params=NS(left=64., top=36., width=128., height=72.)) class NativeMetadataTest(unittest.TestCase): def setUp(self): patcher = patch('app.native_metadata.completion_meta_type', return_value=9001) patcher.start() self.addCleanup(patcher.stop) def test_missing_wrong_or_failed_completion_is_not_zero_detection(self): for value in (None, dict(component=8), dict(component=7, schema=1, status='failed'), dict(component=7, schema=1, status='completed', frame=99, pts=0, count=0)): f = frame() f.user_meta_items = lambda _type: [] if value is None else [NS(get_user_data_json=lambda: value)] with self.assertRaises(NativeMetadataError): copy_detection_frame(f, 7) def test_copy_and_coordinates_are_detached(self): obj = detection() result = copy_detection_frame(frame([obj]), 7) obj.rect_params.left = 999 obj.class_id = 5 item = result.detections[0] self.assertEqual((0, 0, 1280, 720), (result.frame_number, result.pts, result.width, result.height)) self.assertEqual((.1, .1, .3, .3), item.bbox_xyxy) self.assertEqual(0, item.class_id) self.assertEqual(.001, item.score) # no display threshold or 1px filter for actual, expected in zip(item.downstream()['box'], (.2, .2, .2, .2)): self.assertAlmostEqual(expected, actual) with self.assertRaises(FrozenInstanceError): item.score = .9 def test_component_filter_and_zero_detections(self): self.assertEqual((), copy_detection_frame(frame([detection(9)]), 7).detections) self.assertEqual((), copy_detection_frame(frame(), 7).detections) def test_invalid_contract_fails(self): for attr, value in [('confidence', float('nan')), ('class_id', 80)]: obj = detection() setattr(obj, attr, value) with self.assertRaises(ValueError): copy_detection_frame(frame([obj]), 7) bad = frame() bad.pipeline_width = 320 with self.assertRaises(ValueError): copy_detection_frame(bad, 7) with self.assertRaises(ValueError): copy_detection_frame(frame(), None) def test_clipping_and_subpixel_box(self): obj = detection() obj.rect_params = NS(left=-.1, top=0., width=.2, height=.2) item = copy_detection_frame(frame([obj]), 7).detections[0] self.assertEqual(0, item.bbox_xyxy[0]) self.assertGreater(item.bbox_xyxy[2], 0) def test_missing_source_dimensions_remain_failure_with_bounded_reason(self): f = frame() f.source_width = f.source_height = 0 with self.assertRaises(NativeMetadataError) as caught: copy_detection_frame(f, 7) self.assertEqual({'reason': 'DIMENSIONS_UNAVAILABLE', 'dimensions': { 'source_width': 0, 'source_height': 0, 'pipeline_width': 640, 'pipeline_height': 360}}, failure_diagnostic(caught.exception)) def test_copied_frame_uses_native_basis_not_missing_sdk_fields(self): f = frame([detection()]) f.pipeline_width = f.pipeline_height = 0 result = copy_detection_frame(f, 7) self.assertEqual((640, 360), (result.bbox_width, result.bbox_height)) self.assertEqual((.1, .1, .3, .3), result.detections[0].bbox_xyxy) def test_sdk_error_text_is_not_exposed(self): result = failure_diagnostic(AttributeError('sensitive-value-must-not-appear')) self.assertEqual({'reason': 'SDK_ACCESS_OR_CONVERSION_FAILED', 'error_type': 'AttributeError'}, result) class NativeRetrieverTest(unittest.TestCase): def setUp(self): patcher = patch('app.native_metadata.completion_meta_type', return_value=9001) patcher.start() self.addCleanup(patcher.stop) @classmethod def setUpClass(cls): try: from app import multi_pipeline except ModuleNotFoundError as error: if error.name in ('numpy', 'gi', 'pyservicemaker'): raise unittest.SkipTest('SDK image required') from error raise cls.mp = multi_pipeline def run_frame(self, objects, active=lambda: True, handle=None, threshold=.5): collector = self.mp.ResultCollector(required_roles=['det']) f = frame(objects) # Det must never read tensor_items, including when no object exists. class SDKFrame: def __getattr__(self, name): if name == 'tensor_items': raise AssertionError('Det tensor parser accessed') return getattr(f, name) retriever = self.mp._make_retriever('det', 'unused', 'unused', collector, 0, {'n': 0}, component_id=7, is_current=active, role_handle=handle, score_threshold=threshold)() with patch.object(self.mp, 'parse_boxes_labels', side_effect=AssertionError('raw parser called')): retriever.consume(NS(batch_meta=NS(frame_items=[SDKFrame()]))) return collector def test_real_consume_uses_native_without_tensors_or_native_floor(self): result = self.run_frame([detection()], threshold=.0005) self.assertEqual(1, result.per_task['det']) self.assertEqual(.001, result.frames[0]['det']['items'][0]['score']) def test_existing_consumer_profile_filter_is_preserved(self): result = self.run_frame([detection(score=.5), detection(score=.9)], threshold=.5) self.assertEqual(1, result.frames[0]['det']['count']) self.assertEqual(.9, result.frames[0]['det']['items'][0]['score']) def test_zero_detection_is_first_frame(self): from app.stream_lifecycle import StreamLifecycle, RoleRunHandle lifecycle = StreamLifecycle() generation = lifecycle.begin() lifecycle.configure_roles(generation, ['det']) def update(gen, role, event, payload): if event == 'frame': return lifecycle.signal_frame(gen, role, payload) return True handle = RoleRunHandle('det', generation, update) result = self.run_frame([], handle=handle) self.assertEqual(0, result.frames[0]['det']['count']) self.assertEqual('running', lifecycle.snapshot()['observed_state']) def test_stale_and_stopped_do_not_publish(self): from app.stream_lifecycle import RoleRunHandle handle = RoleRunHandle('det', 1) handle.stop() self.assertFalse(self.run_frame([detection()], handle=handle).frames) self.assertFalse(self.run_frame([detection()], active=lambda: False).frames) calls = iter([True, True, False]) self.assertFalse(self.run_frame([detection()], active=lambda: next(calls)).frames) def test_diagnostic_is_once_per_retriever_and_no_raw_error(self): collector = self.mp.ResultCollector(required_roles=['det']) state = {'n': 0} receiver = self.mp._make_retriever('det', 'unused', 'unused', collector, 0, state, component_id=7)() buffer = NS(batch_meta=NS(frame_items=[frame()])) with patch.object(self.mp, 'copy_detection_frame', side_effect=AttributeError('private-value')), \ patch('builtins.print') as output: receiver.consume(buffer) receiver.consume(buffer) lines = [str(call.args[0]) for call in output.call_args_list] self.assertEqual(1, sum('[NATIVE_METADATA_DIAGNOSTIC]' in line for line in lines)) self.assertNotIn('private-value', ''.join(lines)) self.assertFalse(collector.frames) def test_missing_completion_does_not_publish_or_mark_first_frame(self): from app.stream_lifecycle import RoleRunHandle handle = RoleRunHandle('det', 1) f = frame() f.user_meta_items = lambda _type: [] collector = self.mp.ResultCollector(required_roles=['det']) receiver = self.mp._make_retriever('det', 'unused', 'unused', collector, 0, {'n': 0}, component_id=7, role_handle=handle)() receiver.consume(NS(batch_meta=NS(frame_items=[f]))) self.assertFalse(handle.first_frame) self.assertFalse(collector.frames) self.assertEqual('native_metadata', handle.failure.failure_stage) def test_bus_message_uses_only_public_kind(self): from app.stream_lifecycle import RoleRunHandle def forbidden(_self): raise AssertionError('raw message accessed') eos = type('EOSMessage', (), {'__str__': forbidden})() handle = RoleRunHandle('det', 1) self.mp._handle_pipeline_message(handle, eos) self.assertEqual('bus_eos', handle.failure.failure_stage) stopped = RoleRunHandle('det', 1) stopped.stop() self.mp._handle_pipeline_message(stopped, eos) self.assertIsNone(stopped.failure)