[文档] 代码-文档一致性整理 - README 同步 v3 全量(2.1网络/3模块表/4表结构/5 API含people-clips/6.1历史标注/8部署systemd+Vue3+config/12进度); docs/DEPLOY.md 重写为 v3; 删除过时 start_ui.sh(Streamlit); start_edge.sh 注明 systemd 为准; 补提交 test_motion_notifier

fam-core/tests/test_motion_notifier.py

@@ -0,0 +1,179 @@
+import time
+
+from fam_core.motion_notifier.motion_notifier import MotionNotifier
+
+
+def _cfg(**overrides):
+    base = {
+        "enabled": True,
+        "poll_enabled": False,
+        "dsm_host": "192.168.50.64",
+        "dsm_port": 5000,
+        "dsm_account": "ericwyuan",
+        "dsm_password": "secret",
+        "camera_name_to_id": {"Generic_ONVIF-001": 2},
+        "oracle_base_url": "http://oracle.example:5000",
+        "oracle_token": "tok123",
+        "heartbeat_interval_sec": 300,
+        "timeout_sec": 5,
+    }
+    base.update(overrides)
+    return {"motion_notifier": base}
+
+
+def _make_notifier(monkeypatch, **cfg_overrides):
+    monkeypatch.setattr(
+        "fam_core.motion_notifier.motion_notifier.load_config",
+        lambda: _cfg(**cfg_overrides))
+    return MotionNotifier()
+
+
+class _FakeResp:
+    def __init__(self, status_code=200, payload=None, text=""):
+        self.status_code = status_code
+        self._payload = payload or {}
+        self.text = text
+
+    def json(self):
+        return self._payload
+
+
+# ----------------------------------------------------------------------
+# build_event_from_webhook / 时间解析 / 摄像头映射
+# ----------------------------------------------------------------------
+
+def test_build_event_from_webhook_maps_known_camera(monkeypatch):
+    n = _make_notifier(monkeypatch)
+    ev = n.build_event_from_webhook({
+        "device_name": "Generic_ONVIF-001",
+        "event_time": "2026-08-22T10:35:00",
+        "thumbnail_url": "abc,123",
+    })
+    assert ev is not None
+    assert ev["camera_id"] == 2
+    assert ev["event_type"] == 10
+    assert ev["duration"] == 0
+    assert isinstance(ev["start_time"], int)
+
+
+def test_build_event_from_webhook_missing_fields_returns_none(monkeypatch):
+    n = _make_notifier(monkeypatch)
+    assert n.build_event_from_webhook({"device_name": "Generic_ONVIF-001"}) is None
+    assert n.build_event_from_webhook({"event_time": "2026-08-22T10:35:00"}) is None
+
+
+def test_build_event_from_webhook_unmapped_camera_gives_none_id(monkeypatch):
+    n = _make_notifier(monkeypatch)
+    ev = n.build_event_from_webhook({
+        "device_name": "某个从没见过的摄像头",
+        "event_time": "2026-08-22T10:35:00",
+    })
+    assert ev is not None
+    assert ev["camera_id"] is None
+
+
+def test_synth_event_id_stable_for_same_input(monkeypatch):
+    n = _make_notifier(monkeypatch)
+    a = n._synth_event_id("cam1", "2026-08-22T10:00:00", "thumb1")
+    b = n._synth_event_id("cam1", "2026-08-22T10:00:00", "thumb1")
+    assert a == b
+
+
+def test_synth_event_id_differs_for_different_thumbnail(monkeypatch):
+    """核心诉求: 同一分钟内的多次真实事件要能区分开，不能因为哈希碰撞被去重掉。"""
+    n = _make_notifier(monkeypatch)
+    a = n._synth_event_id("cam1", "2026-08-22T10:00:00", "thumb1")
+    b = n._synth_event_id("cam1", "2026-08-22T10:00:00", "thumb2")
+    assert a != b
+
+
+def test_parse_ss_time_to_epoch_iso_format(monkeypatch):
+    n = _make_notifier(monkeypatch)
+    assert n.parse_ss_time_to_epoch("2026-08-22T10:35:00") is not None
+
+
+def test_parse_ss_time_to_epoch_invalid_returns_none(monkeypatch):
+    n = _make_notifier(monkeypatch)
+    assert n.parse_ss_time_to_epoch("not a time") is None
+
+
+def test_parse_ss_time_to_epoch_empty_returns_none(monkeypatch):
+    n = _make_notifier(monkeypatch)
+    assert n.parse_ss_time_to_epoch("") is None
+
+
+# ----------------------------------------------------------------------
+# 心跳（跟轮询无关，enabled=true 就该跑）
+# ----------------------------------------------------------------------
+
+def test_send_heartbeat_success_updates_state(monkeypatch):
+    n = _make_notifier(monkeypatch)
+    calls = []
+
+    def fake_post(url, json=None, timeout=None):
+        calls.append((url, json))
+        return _FakeResp(200, {"status": "ok", "received": 0, "stored": 0})
+    monkeypatch.setattr("fam_core.motion_notifier.motion_notifier.requests.post", fake_post)
+    ok = n.send_heartbeat()
+    assert ok is True
+    assert n._last_heartbeat_at is not None
+    assert n._last_heartbeat_error is None
+    assert calls[0][0] == "http://oracle.example:5000/api/ss/motion"
+    assert calls[0][1]["events"] == []
+    assert calls[0][1]["token"] == "tok123"
+
+
+def test_send_heartbeat_http_error_records_failure(monkeypatch):
+    n = _make_notifier(monkeypatch)
+
+    def fake_post(url, json=None, timeout=None):
+        return _FakeResp(500, {}, text="boom")
+    monkeypatch.setattr("fam_core.motion_notifier.motion_notifier.requests.post", fake_post)
+    ok = n.send_heartbeat()
+    assert ok is False
+    assert n._last_heartbeat_error == "HTTP 500"
+
+
+def test_send_heartbeat_network_exception_records_failure(monkeypatch):
+    import requests as _requests
+    n = _make_notifier(monkeypatch)
+
+    def fake_post(url, json=None, timeout=None):
+        raise _requests.RequestException("connection refused")
+    monkeypatch.setattr("fam_core.motion_notifier.motion_notifier.requests.post", fake_post)
+    ok = n.send_heartbeat()
+    assert ok is False
+    assert "connection refused" in n._last_heartbeat_error
+
+
+def test_start_runs_heartbeat_thread_even_when_poll_disabled(monkeypatch):
+    """核心诉求: 心跳跟"是否轮询 SS"是两回事——poll_enabled=false 时轮询线程
+    不应该启动，但心跳线程必须照样跑，否则甲骨文永远收不到心跳，
+    has_motion_in_range_local() 会一直 fail-open，省配额的效果就没了。"""
+    n = _make_notifier(monkeypatch, poll_enabled=False, heartbeat_interval_sec=3600)
+    monkeypatch.setattr(
+        "fam_core.motion_notifier.motion_notifier.requests.post",
+        lambda url, json=None, timeout=None: _FakeResp(200, {"stored": 0}))
+    try:
+        n.start()
+        time.sleep(0.2)
+        assert n.is_alive() is False  # 轮询线程未启动
+        assert n._hb_thread is not None and n._hb_thread.is_alive()
+    finally:
+        n.stop()
+
+
+def test_start_does_nothing_when_disabled(monkeypatch):
+    n = _make_notifier(monkeypatch, enabled=False)
+    n.start()
+    time.sleep(0.1)
+    assert n._hb_thread is None
+    assert n.is_alive() is False
+
+
+def test_status_includes_heartbeat_fields(monkeypatch):
+    n = _make_notifier(monkeypatch)
+    st = n.status()
+    assert "heartbeat_running" in st
+    assert "last_heartbeat_at" in st
+    assert "last_heartbeat_error" in st