112 lines
4.2 KiB
Python
112 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
from contextlib import redirect_stderr
|
|
import io
|
|
import subprocess
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
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()
|
|
|
|
@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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|