import tempfile import threading import time import unittest from types import SimpleNamespace from unittest.mock import patch from app.stream_manager import StreamGroup, StreamManager from app.stream_lifecycle import RoleRunHandle from app.stream_config_store import ConfigPersistenceError class AsyncRestartTest(unittest.TestCase): def running(self, group=None): g = group or StreamGroup({'stream_id': 'unit', 'store': False}) gen = g.lifecycle.begin() g.lifecycle.configure_roles(gen, ['det']) g.lifecycle.signal_frame(gen, 'det') release, called = threading.Event(), threading.Event() h = RoleRunHandle('det', gen) h.bind_pipeline(SimpleNamespace(stop=called.set)) t = threading.Thread(target=lambda: release.wait(3)) h.bind_thread(t) t.start() g.role_handles['det'] = h self.addCleanup(lambda: (release.set(), t.join(3), g._cleanup_done.wait(3))) return g, h, release, called @staticmethod def begin(g, generation, stop_event): g.lifecycle.configure_roles(generation, ['det']) g.lifecycle.signal_frame(generation, 'det') def test_pending_then_automatic_restart_no_overlap(self): g, h, release, called = self.running() old = g.lifecycle.snapshot()['run_id'] flushed = [] g.writer = SimpleNamespace(_flush=lambda: flushed.append(not h.is_alive())) with patch.object(StreamGroup, '_begin', self.begin): g.restart() self.assertTrue(called.wait(1)) self.assertEqual('stopping', g.state) self.assertTrue(g.status()['restart_pending']) g.restart() self.assertEqual(1, g.run_generation) release.set() self.assertTrue(g._cleanup_done.wait(2)) with g._teardown_lock: self.assertEqual(2, g.run_generation) self.assertNotEqual(old, g.lifecycle.snapshot()['run_id']) self.assertEqual([True], flushed) self.assertFalse(g.lifecycle.signal_frame(1, 'det')) g.stop() self.assertTrue(g._cleanup_done.wait(2)) self.assertEqual('stopped', g.state) def test_explicit_stop_cancels_pending_restart(self): g, _, release, called = self.running() with patch.object(StreamGroup, '_begin', side_effect=AssertionError('late start')): g.restart() self.assertTrue(called.wait(1)) g.stop() self.assertFalse(g.status()['restart_pending']) release.set() self.assertTrue(g._cleanup_done.wait(2)) self.assertEqual(1, g.run_generation) self.assertEqual('stopped', g.state) def test_deadline_failure_is_sticky_and_preserves_handle(self): g, h, release, called = self.running() g.CLEANUP_TIMEOUT_S = .04 g.restart() self.assertTrue(called.wait(1)) self.assertTrue(g._cleanup_done.wait(1)) self.assertEqual('error', g.state) self.assertIs(g.role_handles['det'], h) release.set() h.join(1) with self.assertRaises(RuntimeError): g.start() self.assertEqual(1, g.run_generation) def test_sdk_stop_block_does_not_block_status_or_timer(self): g = StreamGroup({'stream_id': 'unit'}) gen = g.lifecycle.begin() entered, release = threading.Event(), threading.Event() h = RoleRunHandle('det', gen) h.bind_pipeline(SimpleNamespace(stop=lambda: (entered.set(), release.wait(2)))) g.role_handles['det'] = h g.CLEANUP_TIMEOUT_S = .04 try: g.stop() self.assertTrue(entered.wait(1)) time.sleep(.08) self.assertEqual('error', g.status()['state']) self.assertTrue(g.status()['cleanup_pending']) finally: release.set() self.assertTrue(g._cleanup_done.wait(2)) def test_apply_storage_conflict_and_other_camera_commands(self): with tempfile.TemporaryDirectory() as root: manager = StreamManager(root+'/base.yaml', root+'/overlay.yaml') initial = {'stream_id': 'unit', 'source_id': 'camera', 'store': False} g = manager.add(initial) initial = dict(g.spec) g, _, release, called = self.running(g) target = {**initial, 'site_id': 'changed'} with patch.object(StreamGroup, '_begin', self.begin): with patch.object(manager.config_store, 'update', side_effect=ConfigPersistenceError()): with self.assertRaises(ConfigPersistenceError): manager.apply_and_start(target) self.assertFalse(called.is_set()) manager.apply_and_start(target) self.assertTrue(called.wait(1)) self.assertEqual(initial, g.spec) self.assertEqual(target, manager.effective_specs()[0]) with patch.object(manager.config_store, 'update', side_effect=AssertionError('duplicate write')): manager.apply_and_start(target) with self.assertRaises(RuntimeError): manager.apply_and_start({**target, 'site_id': 'conflict'}) with self.assertRaises(RuntimeError): manager.apply_and_start(initial) done = threading.Event() t = threading.Thread(target=lambda: (manager.add({'stream_id': 'other'}), done.set())) t.start() self.assertTrue(done.wait(.5)) t.join(1) release.set() self.assertTrue(g._cleanup_done.wait(2)) with g._teardown_lock: self.assertEqual(target, g.spec) self.assertEqual(2, g.run_generation) g.stop() self.assertTrue(g._cleanup_done.wait(2)) def test_drop_only_when_explicitly_enabled_and_stopped(self): from app.multi_pipeline import _DecodedFrameObserver h = RoleRunHandle('det', 1) probe = _DecodedFrameObserver(h, drop_on_stop=True) self.assertTrue(probe.handle_buffer(None)) h.stop() self.assertFalse(probe.handle_buffer(None)) self.assertTrue(_DecodedFrameObserver(h).handle_buffer(None)) def test_http_stop_accepted_then_status_stopped(self): from fastapi import FastAPI from fastapi.testclient import TestClient from app.api.streams import router g, _, release, called = self.running() app = FastAPI() app.include_router(router) service = SimpleNamespace(stop=lambda sid: (g.stop(), g)[1], get=lambda sid: g) with patch('app.api.streams._sm', return_value=(service, RuntimeError)), TestClient(app) as client: response = client.post('/api/v1/streams/unit/stop') self.assertEqual(202, response.status_code) self.assertEqual('stopping', response.json()['state']) self.assertTrue(called.wait(1)) self.assertTrue(client.get('/api/v1/streams/unit').json()['cleanup_pending']) release.set() self.assertTrue(g._cleanup_done.wait(2)) self.assertEqual('stopped', client.get('/api/v1/streams/unit').json()['state']) self.assertEqual(200, client.post('/api/v1/streams/unit/stop').status_code) def test_http_restart_accepted_and_failure_sanitized(self): from fastapi import FastAPI from fastapi.testclient import TestClient from app.api.streams import router g, _, release, called = self.running() app = FastAPI() app.include_router(router) service = SimpleNamespace(restart=lambda sid: (g.restart(), g)[1]) with patch.object(StreamGroup, '_begin', self.begin), patch('app.api.streams._sm', return_value=(service, RuntimeError)), TestClient(app) as client: self.assertEqual(202, client.post('/api/v1/streams/unit/restart').status_code) self.assertTrue(called.wait(1)) g.stop() release.set() self.assertTrue(g._cleanup_done.wait(2)) service.restart = lambda sid: (_ for _ in ()).throw(RuntimeError('private secret')) response = client.post('/api/v1/streams/unit/restart') self.assertEqual(409, response.status_code) self.assertNotIn('private', response.text) if __name__ == '__main__': unittest.main()