diff --git a/frigate/test/test_ptz_autotrack.py b/frigate/test/test_ptz_autotrack.py new file mode 100644 index 0000000000..1abcc780d0 --- /dev/null +++ b/frigate/test/test_ptz_autotrack.py @@ -0,0 +1,130 @@ +"""Tests for autotracker state that must survive runtime config changes. + +Regression coverage for a family of bugs where per-camera autotracker state was +built once at startup and never revisited. A camera that is added or enabled +after startup, or has autotracking enabled from the UI, would either raise a +KeyError on the autotracker thread or silently keep the wrong state: + +- autotracker_init only got an entry for cameras enabled when PtzAutoTracker was + constructed, so runtime-enabled cameras raised KeyError on lookup. +- ptz_metrics autotracker_enabled is what the camera processes read, but nothing + updated it when autotracking was enabled through a config save, so it stayed + False and the tracker never built a motion estimator. +""" + +import unittest +from unittest.mock import MagicMock + +from frigate.camera import PTZMetrics +from frigate.config import FrigateConfig +from frigate.ptz.autotrack import PtzAutoTracker + +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_tracker(autotracking_enabled: bool = True) -> PtzAutoTracker: + """Build a PtzAutoTracker without invoking __init__, which would try to set up + onvif over the network. Only the config/metrics state is relevant here.""" + tracker = PtzAutoTracker.__new__(PtzAutoTracker) + tracker.config = _config(autotracking_enabled) + tracker.ptz_metrics = {CAMERA: PTZMetrics(autotracker_enabled=False)} + tracker.onvif = MagicMock() + tracker.config_subscriber = MagicMock() + tracker.autotracker_init = {} + tracker.calibrating = {} + tracker.tracked_object = {} + return tracker + + +class TestAutotrackerInitGuards(unittest.IsolatedAsyncioTestCase): + async def test_camera_maintenance_returns_early_when_not_initialized(self) -> None: + # a camera enabled at runtime has no autotracker_init entry, which used to + # raise KeyError and kill the autotracker thread for every camera + tracker = _make_tracker() + self.assertNotIn(CAMERA, tracker.autotracker_init) + + await tracker.camera_maintenance(CAMERA) + + tracker.onvif.get_camera_status.assert_not_called() + + async def test_camera_maintenance_returns_early_when_init_incomplete(self) -> None: + # autotracker_init is seeded False for enabled cameras before setup runs + tracker = _make_tracker() + tracker.autotracker_init[CAMERA] = False + + await tracker.camera_maintenance(CAMERA) + + tracker.onvif.get_camera_status.assert_not_called() + + +class TestAutotrackerMetricSync(unittest.TestCase): + def test_metric_follows_config_when_enabled_by_update(self) -> None: + # autotracking enabled via a config save: the metric was seeded False when + # the camera was added and nothing else updates it + tracker = _make_tracker(autotracking_enabled=True) + metrics = tracker.ptz_metrics[CAMERA] + self.assertFalse(metrics.autotracker_enabled.value) + + tracker.config_subscriber.check_for_updates.return_value = {"onvif": [CAMERA]} + tracker.check_for_updates() + + self.assertTrue(metrics.autotracker_enabled.value) + + def test_metric_follows_config_when_disabled_by_update(self) -> None: + tracker = _make_tracker(autotracking_enabled=False) + metrics = tracker.ptz_metrics[CAMERA] + metrics.autotracker_enabled.value = True + + tracker.config_subscriber.check_for_updates.return_value = { + "autotracking": [CAMERA] + } + tracker.check_for_updates() + + self.assertFalse(metrics.autotracker_enabled.value) + + def test_metric_sync_skips_camera_without_metrics(self) -> None: + # `add` reaches the maintainer and the autotracker on separate threads with + # no ordering guarantee, so the metrics may not exist yet + tracker = _make_tracker() + tracker.ptz_metrics = {} + tracker.config_subscriber.check_for_updates.return_value = {"add": [CAMERA]} + + tracker.check_for_updates() + + def test_metric_sync_skips_unknown_camera(self) -> None: + tracker = _make_tracker() + tracker.config_subscriber.check_for_updates.return_value = { + "add": ["not_in_config"] + } + + tracker.check_for_updates() + + +if __name__ == "__main__": + unittest.main() diff --git a/frigate/test/test_ptz_onvif.py b/frigate/test/test_ptz_onvif.py new file mode 100644 index 0000000000..d2dd2d046e --- /dev/null +++ b/frigate/test/test_ptz_onvif.py @@ -0,0 +1,147 @@ +"""Tests for ONVIF init 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. The autotracking-only request objects used to +be created only when autotracking was enabled at init time, so the camera was left +with init=True but no status_request. get_camera_status skips its re-init branch +when init is True, so it went straight to the missing key and raised KeyError on +the tracking thread. + +The request objects are built from the locally parsed WSDL and cost no network, so +they are always created and init=True now implies they exist. +""" + +import unittest +from unittest.mock import AsyncMock, MagicMock + +from frigate.config import FrigateConfig +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.camera_configs = {CAMERA: config.cameras[CAMERA]} + controller.ptz_metrics = {CAMERA: MagicMock()} + return controller + + +class TestOnvifInitRequests(unittest.IsolatedAsyncioTestCase): + async def test_status_request_created_when_autotracking_disabled(self) -> None: + # the wizard flow: onvif configured first, autotracking enabled later + controller = _make_controller(autotracking_enabled=False) + + self.assertTrue(await controller._init_onvif(CAMERA)) + + cam = controller.cams[CAMERA] + self.assertTrue(cam["init"]) + self.assertIn("status_request", cam) + self.assertIn("service_capabilities_request", cam) + + async def test_status_request_created_when_autotracking_enabled(self) -> None: + controller = _make_controller(autotracking_enabled=True) + + self.assertTrue(await controller._init_onvif(CAMERA)) + + cam = controller.cams[CAMERA] + self.assertIn("status_request", cam) + self.assertIn("service_capabilities_request", cam) + + async def test_init_implies_status_request_exists(self) -> None: + # the invariant get_camera_status relies on: it skips re-init when init is + # True and then reads status_request without guarding + for autotracking_enabled in (True, False): + with self.subTest(autotracking_enabled=autotracking_enabled): + controller = _make_controller(autotracking_enabled) + + await controller._init_onvif(CAMERA) + + cam = controller.cams[CAMERA] + if cam["init"]: + self.assertEqual(cam["status_request"].request_type, "GetStatus") + + 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() + + +if __name__ == "__main__": + unittest.main()