import unittest import threading from dataclasses import replace from unittest.mock import Mock from app.publisher_metrics import PublisherMetrics from app.metadata_publisher import MetadataPublisher, MetadataSettings, camera_key from test_metadata_publisher import frame, frame_snapshot, wait_until class MetricsTest(unittest.TestCase): def test_worker_lanes_isolate_slow_sink_and_keep_channel_owner(self): sinks=[Mock(),Mock()];entered,release=threading.Event(),threading.Event() def slow(*args): entered.set() if not release.wait(2):raise RuntimeError('fixture timeout') sinks[0].write.side_effect=slow p=MetadataPublisher(MetadataSettings('unused',workers=2),sink_factory=lambda i:sinks[i]) self.addCleanup(p.close) for cam in ('001','002'):p.channel(cam,'run1',1,'RUNNING') p.start();self.assertTrue(entered.wait(1)) try: self.assertTrue(p.submit(frame_snapshot(replace(frame(1),source_id='002'),'002'))) wait_until(lambda:p.metrics().get('sent_frames')==1) self.assertEqual(sinks[1].write.call_args.args[0],camera_key('002')) self.assertEqual(p._lanes[camera_key('001')],0) p.channel('002','run1',1,'STOPPED') p.channel('002','run2',2,'RUNNING') self.assertTrue(p.submit(frame_snapshot(replace(frame(2,2),source_id='002'),'002'))) wait_until(lambda:p.metrics().get('sent_frames')==2) self.assertEqual(p._lanes[camera_key('002')],1) self.assertEqual(sinks[1].write.call_args.args[1]['runGeneration'],2) self.assertFalse(p.submit(frame_snapshot(replace(frame(3),source_id='002'),'002'))) finally:release.set() self.assertTrue(p.close(1)) for sink in sinks:sink.close.assert_called_once() def test_multiworker_close_uses_one_deadline(self): sinks=[Mock(),Mock()];entered=[threading.Event(),threading.Event()];release=threading.Event() for i,sink in enumerate(sinks): def write(*args,index=i): entered[index].set();release.wait(2) sink.write.side_effect=write p=MetadataPublisher(MetadataSettings('unused',workers=2),sink_factory=lambda i:sinks[i]) self.addCleanup(p.close) for cam in ('001','002'):p.channel(cam,'run1',1,'RUNNING') p.start() try: self.assertTrue(all(e.wait(1) for e in entered)) self.assertFalse(p.close(.001)) finally:release.set() self.assertTrue(p.close(1));self.assertFalse(any(t.is_alive() for t in p._threads)) def test_worker_bounds_and_environment(self): from app.metadata_runtime import publisher_from_environment for workers in (0,9,True,1.5): with self.assertRaises(ValueError):MetadataSettings('unused',workers=workers) with self.assertRaises(ValueError):MetadataPublisher(MetadataSettings('unused',workers=2),Mock()) p=publisher_from_environment({'CONTEXTRA_METADATA_ENABLED':'true', 'CONTEXTRA_METADATA_REDIS_URL':'redis://user:fixture@localhost:1/0','CONTEXTRA_METADATA_WORKERS':'4'}) self.assertEqual(p.settings.workers,4);self.assertEqual(len(p._sinks),4);p.close() def test_timeline_bounds_and_inflight_correlation(self): now=[0.];m=PublisherMetrics(lambda:now[0]);m.configure(True,60) m.enqueue('a',1,'frame',None);trace=m.dequeue('a','frame') now[0]=.05;m.enqueue('b',2,'frame',None) now[0]=.06;m.enqueue('b',3,'frame',(.05,'payload',7)) m.complete(m.token(),trace,1,100_000_000,2,100_000_010) sample=m.snapshot(True)['channels']['b']['replacements'][0] self.assertEqual(sample['old_sequence'],2);self.assertEqual(sample['new_sequence'],3) self.assertEqual(sample['inflight']['sink_finished_ns'],100_000_000) sample['inflight']['sink_finished_ns']=0 self.assertEqual(m.snapshot(True)['channels']['b']['replacements'][0]['inflight']['sink_finished_ns'],100_000_000) for i in range(300):m.enqueue('b',i+4,'frame',(.05,'payload',7)) row=m.snapshot(True)['channels']['b'] self.assertEqual(len(row['timeline']),256);self.assertEqual(len(row['replacements']),32) self.assertNotIn('timeline',m.snapshot()['channels']['b']) now[0]=60.;m.enqueue('b',999,'frame',None) self.assertFalse(m.snapshot()['enabled']);m.configure(False) self.assertEqual(m.snapshot(True)['channels'],{}) def test_storage_100_is_per_stream_and_ttl_unchanged(self): from app.metadata_publisher import RedisSnapshotSink from app.metadata_runtime import publisher_from_environment client=Mock();sink=RedisSnapshotSink(MetadataSettings('unused'),client) for key in ('a','b'): sink.write(key,dict(runId='r',state='RUNNING'),'{}') calls=client.pipeline.return_value.xadd.call_args_list self.assertNotEqual(calls[0].args[0],calls[1].args[0]) self.assertTrue(all(c.kwargs['maxlen']==100 and c.kwargs['approximate'] is False for c in calls)) self.assertTrue(all(c.args[1]==10 for c in client.pipeline.return_value.expire.call_args_list)) p=publisher_from_environment({'CONTEXTRA_METADATA_ENABLED':'true', 'CONTEXTRA_METADATA_REDIS_URL':'redis://user:fixture@localhost:1/0', 'CONTEXTRA_METADATA_STREAM_LENGTH':'100'}) self.assertEqual(p.settings.stream_length,100) self.assertEqual(p.settings.stream_ttl,10);p.close() def test_disabled_bounded_expiry_revision(self): now=[0.];m=PublisherMetrics(lambda:now[0]) m.count('key','ignored');self.assertEqual(m.snapshot()['channels'],{}) m.configure(True,1);token=m.token() for _ in range(300):m.duration(token,'key','write',.01) stat=m.snapshot()['channels']['key']['timings']['write'] self.assertEqual(stat['count'],300);self.assertEqual(stat['recent_samples'],256) now[0]=1.;self.assertFalse(m.snapshot()['enabled']) m.count('key','late');self.assertEqual(m.snapshot()['channels']['key']['counts'],{}) m.configure(True,1);m.duration(token,'key','old',1) self.assertEqual(m.snapshot()['channels'],{}) m.configure(False);self.assertFalse(m.snapshot()['enabled']) def test_frame_and_state_replacement_distinguished(self): p=MetadataPublisher(MetadataSettings('unused'),Mock());self.addCleanup(p.close) p.configure_metrics(True,10) p.channel('001','run1',1,'RUNNING') p.submit(frame_snapshot(frame(),'001'));p.submit(frame_snapshot(frame(1),'001')) p.channel('001','run1',1,'STOPPED') row=p.metrics()['profile']['channels'][camera_key('001')]['counts'] self.assertEqual(row['replaced_by_frame'],1);self.assertEqual(row['replaced_by_state'],1) p.start();wait_until(lambda:'status_write_wall' in p.metrics()['profile']['channels'][camera_key('001')]['timings']) self.assertIn('status_write_wall',p.metrics()['profile']['channels'][camera_key('001')]['timings']) p.configure_metrics(False) self.assertEqual(p.metrics()['profile']['channels'],{}) def test_running_noop_preserves_payload_age_and_sequence(self): p=MetadataPublisher(MetadataSettings('unused'),Mock());self.addCleanup(p.close) p.channel('001','run1',1,'RUNNING') p.submit(frame_snapshot(frame(),'001')) key=camera_key('001');pending=p._pending[key];before=p.metrics()['queuedBytes'] for _ in range(10):self.assertTrue(p.channel('001','run1',1,'RUNNING')) self.assertEqual(p._pending[key],pending) self.assertEqual(p.metrics()['queuedBytes'],before) self.assertEqual(p._channels[key]['lastSeq'],0) self.assertEqual(p.metrics().get('replaced',0),0) p.start();wait_until(lambda:p.metrics().get('sent_frames')==1) self.assertIn('"frameSeq":0',p.sink.write.call_args.args[2]) def test_real_stop_and_new_run_still_replace_old_frame(self): p=MetadataPublisher(MetadataSettings('unused'),Mock());self.addCleanup(p.close) p.channel('001','run1',1,'RUNNING');p.submit(frame_snapshot(frame(),'001')) p.channel('001','run1',1,'STOPPED') self.assertIsNone(p._pending[camera_key('001')][1]) self.assertFalse(p.channel('001','run1',1,'RUNNING')) self.assertTrue(p.channel('001','run2',2,'RUNNING')) self.assertFalse(p.submit(frame_snapshot(frame(1),'001'))) self.assertTrue(p.submit(frame_snapshot(frame(1,2),'001'))) def test_noop_does_not_disable_heartbeat(self): now=[0.];sink=Mock();p=MetadataPublisher(MetadataSettings('unused'),sink,clock=lambda:now[0]) self.addCleanup(p.close) p.channel('001','run1',1,'RUNNING');p.start() wait_until(lambda:p.metrics().get('sent_status')==1) p.channel('001','run1',1,'RUNNING') now[0]=4. with p._cv:p._cv.notify() # Use a real pending channel to wake wait_for without changing channel001 state. p.channel('002','run1',1,'RUNNING') wait_until(lambda:p.metrics().get('sent_status',0)>=3) self.assertTrue(any(c.args[0]==camera_key('001') for c in sink.write.call_args_list[1:])) def test_concurrent_channels_keep_latest_and_own_sequence(self): sink=Mock();p=MetadataPublisher(MetadataSettings('unused'),sink);self.addCleanup(p.close) entered,release=threading.Event(),threading.Event();errors=[] def write(*args): entered.set() if not release.wait(2):raise RuntimeError('fixture timeout') sink.write.side_effect=write for cam in ('001','002'):p.channel(cam,'run1',1,'RUNNING') p.start();self.assertTrue(entered.wait(1)) def submit(cam): try: for i in range(30): p.channel(cam,'run1',1,'RUNNING') self.assertTrue(p.submit(frame_snapshot(replace(frame(i),source_id=cam),cam))) except Exception as error:errors.append(error) threads=[threading.Thread(target=submit,args=(cam,)) for cam in ('001','002')] try: for t in threads:t.start() for t in threads:t.join(1) self.assertFalse(any(t.is_alive() for t in threads));self.assertEqual(errors,[]) self.assertEqual(p.metrics()['queued'],2) for cam in ('001','002'): self.assertEqual(p._channels[camera_key(cam)]['lastSeq'],29) self.assertIn('"frameSeq":29',p._pending[camera_key(cam)][1]) finally:release.set() wait_until(lambda:p.metrics().get('sent_frames')==2)