from __future__ import annotations import argparse from contextlib import redirect_stderr import io import json from pathlib import Path import subprocess import tempfile import unittest from unittest import mock import interactive_calibrator as wizard import scan_calibrator as core def completed(returncode: int, stdout: str = "", stderr: str = ""): return subprocess.CompletedProcess(["docker"], returncode, stdout, stderr) class ApiRecoveryTests(unittest.TestCase): def setUp(self) -> None: core._HELPER_READY = False @mock.patch("scan_calibrator.wait_for_scan_http", return_value=True) @mock.patch("scan_calibrator.subprocess.run") def test_transient_curl_failure_waits_and_retries(self, run_mock, wait_mock): core._HELPER_READY = True run_mock.side_effect = [ completed(52, stderr="curl: (52) Empty reply from server"), completed(0, stdout='{"ptz": {"pan": 1, "tilt": 2, "zoom": 3}}'), ] payload = core.api("GET", "/api/v1/ptz", timeout=5) self.assertEqual(payload["ptz"]["zoom"], 3) self.assertEqual(run_mock.call_count, 2) wait_mock.assert_called_once_with() self.assertFalse(core._HELPER_READY) @mock.patch("scan_calibrator.wait_for_scan_http", return_value=True) @mock.patch("scan_calibrator.subprocess.run") def test_connection_refused_is_retried(self, run_mock, wait_mock): run_mock.side_effect = [ completed(7, stderr="curl: (7) Failed to connect to 127.0.0.1"), completed(0, stdout='{"scan": {"state": "idle"}}'), ] payload = core.api("GET", "/api/v1/scan/status", timeout=5) self.assertEqual(payload["scan"]["state"], "idle") wait_mock.assert_called_once_with() def test_move_not_reached_error_is_explained_for_operator(self): result = completed( 22, stdout=json.dumps({ "error": { "code": "sdk_error", "message": ( "ERR move_not_reached pan_error=-25 tilt_error=56 " "zoom_error=0 target=25892,2044,4620 " "actual=25917,1988,4620" ), } }), stderr="curl: (22) The requested URL returned error: 500", ) detail = core.api_error_detail(result) self.assertIn("没有完全到位", detail) self.assertIn("tilt=2044 (20.44°)", detail) self.assertIn("实际", detail) self.assertIn("现场处理", detail) @mock.patch("scan_calibrator.wait_for_scan_http") @mock.patch("scan_calibrator.subprocess.run") def test_non_transient_http_error_is_not_retried(self, run_mock, wait_mock): run_mock.return_value = completed( 22, stderr="curl: (22) The requested URL returned error: 404" ) with self.assertRaisesRegex(SystemExit, "404"): core.api("GET", "/missing", timeout=5) self.assertEqual(run_mock.call_count, 1) wait_mock.assert_not_called() @mock.patch("scan_calibrator.time.sleep") @mock.patch("scan_calibrator.scan_http_ready", side_effect=[False, True]) @mock.patch("scan_calibrator.time.monotonic", side_effect=[0.0, 1.0, 2.0]) def test_health_wait_polls_until_service_recovers( self, _monotonic_mock, ready_mock, sleep_mock ): self.assertTrue(core.wait_for_scan_http(timeout=10.0)) self.assertEqual(ready_mock.call_count, 2) sleep_mock.assert_called_once_with(core.API_RECOVERY_POLL_INTERVAL) @mock.patch("scan_calibrator.run") @mock.patch("scan_calibrator.subprocess.run") def test_overlap_helper_compiles_only_once_per_session( self, subprocess_run_mock, checked_run_mock ): subprocess_run_mock.return_value = completed(0) core.ensure_helper() core.ensure_helper() self.assertEqual(checked_run_mock.call_count, 2) subprocess_run_mock.assert_called_once_with( ["docker", "exec", "smart-cam", "test", "-x", core.HELPER_CONTAINER_BIN], text=True, capture_output=True, timeout=10, ) @mock.patch("scan_calibrator.time.sleep") @mock.patch("scan_calibrator.measure_images") @mock.patch("scan_calibrator.snapshot", return_value="candidate.jpg") @mock.patch("scan_calibrator.move") def test_restore_failure_does_not_mask_measurement_failure( self, move_mock, _snapshot_mock, measure_mock, _sleep_mock ): origin = {"pan": 100, "tilt": 200, "zoom": 300} move_mock.side_effect = [dict(origin), SystemExit("恢复起点失败")] measure_mock.side_effect = SystemExit("原始测量失败") stderr = io.StringIO() with redirect_stderr(stderr): with self.assertRaisesRegex(SystemExit, "原始测量失败"): core.measure_delta( "pan", 50, 1, 0.0, baseline="baseline.jpg", origin=origin ) self.assertIn("恢复起点失败", stderr.getvalue()) self.assertEqual(move_mock.call_count, 2) @mock.patch("scan_calibrator.api") def test_calibration_ptz_requests_identify_their_control_source(self, api_mock): api_mock.return_value = {"ptz": {"pan": 1, "tilt": 2, "zoom": 3}} self.assertEqual(core.get_ptz(), {"pan": 1, "tilt": 2, "zoom": 3}) api_mock.assert_called_with( "GET", "/api/v1/ptz?source=scan_calibrator", timeout=50 ) core.move({"pan": 4, "tilt": 5, "zoom": 6}) api_mock.assert_called_with( "POST", "/api/v1/ptz/absolute", {"pan": 4, "tilt": 5, "zoom": 6, "source": "scan_calibrator"}, timeout=50, ) def test_hikvision_control_contract_is_supported(self): result = core.validate_control_status({ "vendor": "hikvision", "driver_ready": True, "capabilities": ["get_position", "move_absolute", "jog"], "camera": {"sdk_host": "192.168.2.10", "sdk_port": 8000}, }) self.assertEqual(result["vendor"], "hikvision") self.assertIn("move_absolute", result["capabilities"]) def test_tiandy_control_contract_is_supported(self): result = core.validate_control_status({ "vendor": "tiandy", "driver_ready": True, "capabilities": ["get_position", "move_absolute", "presets"], "camera": {"sdk_host": "192.168.100.64", "sdk_port": 3000}, }) self.assertEqual(result["vendor"], "tiandy") self.assertIn("get_position", result["capabilities"]) def test_control_contract_rejects_unready_driver(self): with self.assertRaisesRegex(SystemExit, "login failed"): core.validate_control_status({ "vendor": "tiandy", "driver_ready": False, "driver_error": "login failed", "capabilities": [], }) def test_control_contract_rejects_missing_absolute_move(self): with self.assertRaisesRegex(SystemExit, "move_absolute"): core.validate_control_status({ "vendor": "hikvision", "driver_ready": True, "capabilities": ["get_position", "jog"], }) def test_ptz_response_must_use_normalized_contract(self): with self.assertRaisesRegex(SystemExit, "pan超出"): core.parse_ptz_payload({ "ptz": {"pan": 40000, "tilt": 100, "zoom": 200} }) class InteractiveWizardTests(unittest.TestCase): def test_guided_steps_use_actual_motion_not_requested_motion(self): self.assertEqual( wizard.fov_guided_deltas(62, 0.802, "tilt"), [56, 88, 125, 163, 204], ) def test_guided_steps_wait_for_measurable_image_shift(self): self.assertEqual(wizard.fov_guided_deltas(33, 0.952, "tilt"), []) def test_probe_cap_uses_camera_reported_motion(self): self.assertEqual(wizard.safe_probe_cap(33, 0.952), 584) self.assertEqual(wizard.safe_probe_cap(62, 0.802), 266) @mock.patch("interactive_calibrator.core.get_ptz") @mock.patch("interactive_calibrator.core.api") def test_calibration_pauses_active_patrol_before_ptz_barrier( self, api_mock, get_ptz_mock ): status = {"control": {"scan_active": False, "patrol_active": True, "patrol_paused_by_operator": False}} camera = {"vendor": "tiandy", "camera": {}, "capabilities": set()} api_mock.return_value = {"success": True} with mock.patch( "interactive_calibrator.core.get_control_status", return_value=(status, camera), ): self.assertTrue(wizard.acquire_calibration_control()) api_mock.assert_called_once_with( "POST", "/api/v1/patrol/stop", {"source": "scan_calibrator"}, timeout=15, ) get_ptz_mock.assert_called_once_with() def test_manual_ptz_accepts_annotated_display_format(self): self.assertEqual( wizard.parse_ptz_input( "pan=21720 (217.20°), tilt=1268 (12.68°), " "zoom=4580 (45.80×)" ), {"pan": 21720, "tilt": 1268, "zoom": 4580}, ) def test_manual_ptz_rejects_incomplete_input(self): with self.assertRaisesRegex(ValueError, "pan=21720"): wizard.parse_ptz_input("pan=21720, tilt=1268") @mock.patch("interactive_calibrator.core.print_ptz") @mock.patch("interactive_calibrator.camera_ptz") @mock.patch("interactive_calibrator.read_zoom_after_move") def test_unreachable_average_uses_nearest_higher_zoom( self, move_mock, camera_mock, _print_mock ): origin = {"pan": 23052, "tilt": 1298, "zoom": 4510} camera_mock.return_value = origin def quantized(_origin, requested): return { **origin, "zoom": 4550 if requested <= 4565 else 4580, } move_mock.side_effect = quantized actual, average = wizard.move_to_scan_zoom({ "top-left": {**origin, "zoom": 4550}, "bottom-right": {**origin, "zoom": 4580}, }) self.assertEqual(average, 4565) self.assertEqual(actual["zoom"], 4580) self.assertEqual(move_mock.call_args_list[0].args[1], 4565) self.assertEqual(move_mock.call_args_list[-1].args[1], 4580) @mock.patch("interactive_calibrator.core.print_ptz") @mock.patch("interactive_calibrator.camera_ptz") @mock.patch("interactive_calibrator.read_zoom_after_move") def test_step_four_always_sends_zoom_even_when_already_equal( self, move_mock, camera_mock, _print_mock ): origin = {"pan": 100, "tilt": 200, "zoom": 4580} camera_mock.return_value = origin move_mock.return_value = origin actual, _average = wizard.move_to_scan_zoom({ "top-left": origin, "bottom-right": origin, }) self.assertEqual(actual["zoom"], 4580) move_mock.assert_called_once_with(origin, 4580) def test_partial_session_is_saved_to_history(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) history = root / "history" state_file = root / "calibration.json" with ( mock.patch.object(wizard, "HISTORY_DIR", history), mock.patch.object(wizard, "ACTIVE_HISTORY_FILE", None), mock.patch.object(core, "STATE_FILE", state_file), ): state = wizard.start_history_session("hanging", None) state["points"]["top-left"] = { "pan": 21720, "tilt": 1268, "zoom": 4580, } wizard.save_working_state(state) records = wizard.load_history_records() self.assertEqual(len(records), 1) self.assertTrue( wizard.state_item_exists(records[0][1], "points", "top-left") ) self.assertFalse( wizard.state_item_exists(records[0][1], "points", "bottom-right") ) class GenerateZoomTests(unittest.TestCase): def test_generate_uses_camera_reachable_scan_zoom(self): state = { "points": { "top-left": {"pan": 1000, "tilt": 200, "zoom": 4550}, "bottom-right": {"pan": 900, "tilt": 100, "zoom": 4580}, }, "steps": { "pan": { "value": 50, "overlap": 0.04, "target_overlap": 0.04, "overlap_method": "axis_shift_v1_retargeted", "inliers": 50, "visual_direction": "decreasing", }, "tilt": { "value": 50, "overlap": 0.04, "target_overlap": 0.04, "overlap_method": "axis_shift_v1_retargeted", "inliers": 50, }, }, "scan_zoom": 4580, } with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) with ( mock.patch.object(core, "ROOT", root), mock.patch.object(core, "load_state", return_value=state), ): with mock.patch("sys.stdout", new=io.StringIO()): core.command_generate(argparse.Namespace(mount="desktop")) result = next((root / "result").glob("scan_config_*.json")) generated = json.loads(result.read_text(encoding="utf-8")) self.assertEqual(generated["scan"]["zoom"], 4580) if __name__ == "__main__": unittest.main()