"""Tests for ONVIF state that must not depend on the autotracking config. Regression coverage for a camera that is initialized while autotracking is off and has it enabled later, which is the normal wizard flow: set the camera up first, configure autotracking afterwards. get_camera_status skips its re-init branch when init is True, so everything it reads must exist whether or not autotracking was enabled at init time. Also covers the inverse direction: the ptz movement timestamps must not be written for a camera that has autotracking off, because nothing clears them back out. """ import asyncio import threading import unittest from unittest.mock import AsyncMock, MagicMock from frigate.camera import PTZMetrics from frigate.config import FrigateConfig from frigate.ptz.autotrack import ptz_moving_at_frame_time from frigate.ptz.onvif import OnvifController CAMERA = "ptz_cam" def _config(autotracking_enabled: bool) -> FrigateConfig: return FrigateConfig( **{ "mqtt": {"enabled": False}, "cameras": { CAMERA: { "ffmpeg": { "inputs": [ {"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]} ] }, "detect": {"width": 1920, "height": 1080}, "zones": {"zone": {"coordinates": "0,0,1,0,1,1,0,1"}}, "onvif": { "host": "10.0.0.1", "autotracking": { "enabled": autotracking_enabled, "required_zones": ["zone"], }, }, } }, } ) def _make_profile() -> MagicMock: profile = MagicMock() profile.token = "profile_1" profile.Name = "MainStream" profile.VideoEncoderConfiguration = MagicMock() ptz_config = MagicMock() ptz_config.token = "ptz_config_1" ptz_config.DefaultContinuousPanTiltVelocitySpace = "space" ptz_config.DefaultContinuousZoomVelocitySpace = "space" profile.PTZConfiguration = ptz_config return profile def _make_onvif_camera() -> MagicMock: """A camera that supports PTZ but nothing optional, so init takes the simplest path through the feature detection below.""" onvif = MagicMock() onvif.update_xaddrs = AsyncMock() video_source = MagicMock() video_source.token = "video_source_1" media = MagicMock() media.GetProfiles = AsyncMock(return_value=[_make_profile()]) media.GetVideoSources = AsyncMock(return_value=[video_source]) onvif.create_media_service = AsyncMock(return_value=media) onvif.get_definition = MagicMock(return_value={"ptz": "definition"}) ptz = MagicMock() # create_type is a local WSDL lookup, so tag the result to assert on it later ptz.create_type = MagicMock(side_effect=lambda name: MagicMock(request_type=name)) ptz.GetConfigurationOptions = AsyncMock(side_effect=Exception("not supported")) onvif.create_ptz_service = AsyncMock(return_value=ptz) onvif.create_imaging_service = AsyncMock(side_effect=Exception("not supported")) return onvif def _make_controller(autotracking_enabled: bool) -> OnvifController: """Build a controller without invoking __init__, which would start an event loop thread and reach out to the camera.""" config = _config(autotracking_enabled) controller = OnvifController.__new__(OnvifController) controller.config = config controller.cams = {CAMERA: {"onvif": _make_onvif_camera(), "init": False}} controller.failed_cams = {} controller.ptz_metrics = {CAMERA: MagicMock()} return controller def _make_move_controller(autotracking_enabled: bool) -> OnvifController: """Build an already initialized controller for a camera that supports relative FOV movement, with real metrics so the timestamp writes can be asserted on.""" config = _config(autotracking_enabled) controller = OnvifController.__new__(OnvifController) controller.config = config controller.failed_cams = {} ptz = MagicMock() ptz.RelativeMove = AsyncMock() controller.cams = { CAMERA: { "init": True, "active": False, "ptz": ptz, "features": ["pt", "pt-r-fov"], "relative_move_request": MagicMock(), "relative_fov_range": { "XRange": {"Min": -1.0, "Max": 1.0}, "YRange": {"Min": -1.0, "Max": 1.0}, }, } } controller.ptz_metrics = {CAMERA: PTZMetrics()} return controller class TestOnvifInitRequests(unittest.IsolatedAsyncioTestCase): async def test_camera_status_independent_of_autotracking_at_init(self) -> None: # the wizard flow: onvif configured first, autotracking enabled later for autotracking_enabled in (True, False): with self.subTest(autotracking_enabled=autotracking_enabled): controller = _make_controller(autotracking_enabled) controller.status_locks = {CAMERA: asyncio.Lock()} self.assertTrue(await controller._init_onvif(CAMERA)) status = MagicMock() status.MoveStatus.PanTilt = "IDLE" status.MoveStatus.Zoom = "IDLE" ptz = controller.cams[CAMERA]["ptz"] ptz.GetStatus = AsyncMock(return_value=status) await controller.get_camera_status(CAMERA) ptz.GetStatus.assert_awaited_once_with({"ProfileToken": "profile_1"}) self.assertFalse(controller.cams[CAMERA]["active"]) async def test_requests_built_without_contacting_camera(self) -> None: # create_type is a local WSDL lookup; cameras that do not implement # GetServiceCapabilities must not be asked about it during init controller = _make_controller(autotracking_enabled=False) await controller._init_onvif(CAMERA) ptz = controller.cams[CAMERA]["ptz"] ptz.GetServiceCapabilities.assert_not_called() ptz.GetStatus.assert_not_called() class TestManualRelativeMoveMetrics(unittest.IsolatedAsyncioTestCase): """A manual move from the UI (click to move, drag to zoom) sends move_relative for any camera that advertises pt-r-fov, autotracking or not.""" async def test_metrics_untouched_when_autotracking_disabled(self) -> None: # only camera_maintenance polls get_camera_status, and only for autotracking # cameras, so a manual move that starts the clock here is never stopped controller = _make_move_controller(autotracking_enabled=False) metrics = controller.ptz_metrics[CAMERA] metrics.frame_time.value = 1000.0 await controller._move_relative(CAMERA, 0.25, -0.25, 0, 1) controller.cams[CAMERA]["ptz"].RelativeMove.assert_awaited_once() self.assertEqual(metrics.start_time.value, 0) self.assertEqual(metrics.stop_time.value, 0) self.assertTrue(metrics.motor_stopped.is_set()) async def test_detection_regions_not_suppressed_after_manual_move(self) -> None: # the symptom of the bug: object detection stops entirely because motion # boxes are never promoted to detection regions again controller = _make_move_controller(autotracking_enabled=False) metrics = controller.ptz_metrics[CAMERA] metrics.frame_time.value = 1000.0 await controller._move_relative(CAMERA, 0.25, -0.25, 0, 1) for later_frame_time in (1001.0, 1060.0, 4600.0): with self.subTest(frame_time=later_frame_time): self.assertFalse( ptz_moving_at_frame_time( later_frame_time, metrics.start_time.value, metrics.stop_time.value, ) ) async def test_metrics_written_when_autotracking_enabled(self) -> None: # get_camera_status resets stop_time once the camera reports IDLE, so the # autotracking path keeps its motion estimation timestamps controller = _make_move_controller(autotracking_enabled=True) metrics = controller.ptz_metrics[CAMERA] metrics.frame_time.value = 1000.0 await controller._move_relative(CAMERA, 0.25, -0.25, 0, 1) self.assertEqual(metrics.start_time.value, 1000.0) self.assertEqual(metrics.stop_time.value, 0) self.assertFalse(metrics.motor_stopped.is_set()) self.assertTrue( ptz_moving_at_frame_time( 1001.0, metrics.start_time.value, metrics.stop_time.value ) ) class TestOnvifClose(unittest.TestCase): """close() must release everything on the loop, since whatever it leaves is garbage collected during interpreter shutdown, where the resulting warnings fail to log and fill the shutdown output with logging errors.""" def setUp(self) -> None: self.controller = _make_controller(autotracking_enabled=False) self.onvif = self.controller.cams[CAMERA]["onvif"] self.onvif.close = AsyncMock() self.controller.config_subscriber = MagicMock() self.controller.loop = asyncio.new_event_loop() self.controller.loop_thread = threading.Thread( target=self.controller._run_event_loop, daemon=True ) self.controller.loop_thread.start() self.addCleanup(self.controller.loop.close) def test_close_closes_camera_sessions(self) -> None: self.controller.close() self.onvif.close.assert_awaited_once() def test_close_cancels_tasks_left_on_the_loop(self) -> None: async def forever() -> None: while True: await asyncio.sleep(1) poll = asyncio.run_coroutine_threadsafe(forever(), self.controller.loop) self.controller.close() self.assertTrue(poll.cancelled()) self.assertFalse(self.controller.loop_thread.is_alive()) if __name__ == "__main__": unittest.main()