import json from fam_edge.oracle_db import OracleDB def _db(tmp_path): return OracleDB(str(tmp_path / "oracle.db")) def _set_heartbeat_age(db, age_sec): """把心跳时间戳直接改写成"距现在 age_sec 秒前",用于测试新鲜度阈值边界。""" from datetime import datetime, timedelta, timezone ts = (datetime.now(timezone(timedelta(hours=8))) - timedelta(seconds=age_sec)) db.set_cursor('motion_heartbeat_at', ts.strftime('%Y-%m-%d %H:%M:%S')) def test_upsert_person_same_gender_merges_into_one_row(tmp_path): db = _db(tmp_path) db.upsert_person("人物A", features={"gender": "男", "hair": "短发黑色"}) db.upsert_person("人物A", features={"gender": "男", "clothing": "蓝色T恤"}) rows = db._conn.execute("SELECT * FROM people").fetchall() assert len(rows) == 1 assert rows[0]["appearances"] == 2 feats = json.loads(rows[0]["features_json"]) assert feats["hair"] == "短发黑色" assert feats["clothing"] == "蓝色T恤" def test_upsert_person_gender_conflict_splits_into_new_label(tmp_path): """核心诉求: 大模型给的 人物A/B/C 这类 uid 只在单次视频分析内稳定,不同视频 独立编号,同一字符串完全可能撞到不同真人(实测 "人物A" 混了男女两个人)。 性别冲突时不能直接合并覆盖,要拆成新 label,避免两个人的特征越merge越乱。""" db = _db(tmp_path) db.upsert_person("人物A", features={"gender": "男", "clothing": "蓝色Polo衫"}) db.upsert_person("人物A", features={"gender": "女", "clothing": "白色上衣"}) rows = {r["label"]: r for r in db._conn.execute("SELECT * FROM people").fetchall()} assert set(rows.keys()) == {"人物A", "人物A#2"} assert rows["人物A"]["appearances"] == 1 assert json.loads(rows["人物A"]["features_json"])["gender"] == "男" assert rows["人物A#2"]["appearances"] == 1 assert json.loads(rows["人物A#2"]["features_json"])["gender"] == "女" # 派生行的 display_uid 仍然记录原始大模型 uid,方便追溯来源 assert rows["人物A#2"]["display_uid"] == "人物A" def test_upsert_person_gender_conflict_allocates_next_free_suffix(tmp_path): db = _db(tmp_path) db.upsert_person("人物A", features={"gender": "男"}) db.upsert_person("人物A", features={"gender": "女"}) # -> 人物A#2 db.upsert_person("人物A", features={"gender": "unknown"}) # unknown 不冲突,合并回 人物A db.upsert_person("人物A", features={"gender": "男", "hair": "光头"}) # 冲突,人物A 有 gender 男了不冲突;应仍合并 labels = {r["label"] for r in db._conn.execute("SELECT label FROM people").fetchall()} assert labels == {"人物A", "人物A#2"} # 再来一次性别冲突(对着 人物A#2,性别女)应该分配 人物A#3,而不是复用 人物A#2 db.upsert_person("人物A", features={"gender": "男"}) db.upsert_person("人物A", features={"gender": "女"}) # 这次新的女性冲突会先撞到 人物A(此时是男),分裂出下一个空闲后缀 labels = {r["label"] for r in db._conn.execute("SELECT label FROM people").fetchall()} assert "人物A" in labels assert len(labels) >= 2 def test_upsert_person_unknown_gender_never_triggers_split(tmp_path): db = _db(tmp_path) db.upsert_person("人物A", features={"gender": "男"}) db.upsert_person("人物A", features={"gender": "unknown"}) db.upsert_person("人物A", features={"gender": "未知"}) rows = db._conn.execute("SELECT * FROM people").fetchall() assert len(rows) == 1 assert rows[0]["appearances"] == 3 def test_upsert_person_no_features_never_triggers_split(tmp_path): db = _db(tmp_path) db.upsert_person("人物A", features={"gender": "男"}) db.upsert_person("人物A") # 无 features(source 更新等场景) rows = db._conn.execute("SELECT * FROM people").fetchall() assert len(rows) == 1 assert rows[0]["appearances"] == 2 # ---------------------------------------------------------------------- # 运动侦测事件(NAS 推送) # ---------------------------------------------------------------------- def test_record_motion_events_upserts_by_event_id(tmp_path): db = _db(tmp_path) n = db.record_motion_events([ {"event_id": 1, "camera_id": 2, "event_type": 10, "start_time": 1000, "duration": 5}, {"event_id": 2, "camera_id": 2, "event_type": 10, "start_time": 2000, "duration": 3}, ]) assert n == 2 rows = db._conn.execute("SELECT * FROM ss_motion_events ORDER BY event_id").fetchall() assert len(rows) == 2 # 重复推送同一个 event_id(幂等)应该更新而不是新增一行 db.record_motion_events( [{"event_id": 1, "camera_id": 2, "event_type": 10, "start_time": 1000, "duration": 99}]) rows = db._conn.execute("SELECT * FROM ss_motion_events").fetchall() assert len(rows) == 2 updated = db._conn.execute( "SELECT duration FROM ss_motion_events WHERE event_id=1").fetchone() assert updated['duration'] == 99 def test_record_motion_events_skips_missing_event_id(tmp_path): db = _db(tmp_path) n = db.record_motion_events([{"camera_id": 2, "start_time": 1000}]) assert n == 0 def test_heartbeat_age_none_when_never_recorded(tmp_path): db = _db(tmp_path) assert db.get_motion_heartbeat_age_sec() is None def test_heartbeat_age_near_zero_right_after_recording(tmp_path): db = _db(tmp_path) db.record_motion_heartbeat() age = db.get_motion_heartbeat_age_sec() assert age is not None and age < 5 def test_has_motion_in_range_local_fails_open_without_heartbeat(tmp_path): """核心诉求: 从未收到过心跳(冷启动,NAS 推送链路还没接上)必须 fail-open, 不能因为本地表是空的就悄悄跳过分析。""" db = _db(tmp_path) assert db.has_motion_in_range_local(1000, 2000) is None def test_has_motion_in_range_local_fails_open_when_heartbeat_stale(tmp_path): """核心诉求: 表里有大量历史运动事件(曾经推送链路是健康的),但心跳已经 过期太久(NAS 服务挂了/网络断了/DSM Webhook 规则被误关)——这时候不能信任 "查询结果是 0 条 = 确认无运动",必须当作链路已死,fail-open。""" db = _db(tmp_path) db.record_motion_events( [{"event_id": 1, "camera_id": 2, "event_type": 10, "start_time": 500, "duration": 10}]) _set_heartbeat_age(db, 1000) # 超过默认阈值 900s assert db.has_motion_in_range_local(2000, 3000, max_heartbeat_age_sec=900) is None def test_has_motion_in_range_local_trusts_result_when_heartbeat_fresh(tmp_path): db = _db(tmp_path) db.record_motion_heartbeat() assert db.has_motion_in_range_local(2000, 3000, max_heartbeat_age_sec=900) is False db.record_motion_events( [{"event_id": 1, "camera_id": 2, "event_type": 10, "start_time": 2500, "duration": 5}]) assert db.has_motion_in_range_local(2000, 3000, max_heartbeat_age_sec=900) is True def test_has_motion_in_range_local_respects_heartbeat_boundary(tmp_path): db = _db(tmp_path) _set_heartbeat_age(db, 899) assert db.has_motion_in_range_local(2000, 3000, max_heartbeat_age_sec=900) is not None _set_heartbeat_age(db, 901) assert db.has_motion_in_range_local(2000, 3000, max_heartbeat_age_sec=900) is None def test_has_motion_in_range_local_overlap_semantics(tmp_path): """事件区间 [start_time, start_time+duration] 只要和查询窗口有重叠就算命中, 不要求事件完全落在窗口内部(也不要求窗口完全覆盖事件)。""" db = _db(tmp_path) db.record_motion_heartbeat() # 事件在窗口开始之前就开始,但持续到窗口内 -> 应该命中 db.record_motion_events( [{"event_id": 1, "camera_id": 2, "event_type": 10, "start_time": 1990, "duration": 20}]) assert db.has_motion_in_range_local(2000, 3000) is True def test_has_motion_in_range_local_ignores_non_motion_event_type(tmp_path): db = _db(tmp_path) db.record_motion_heartbeat() db.record_motion_events( [{"event_id": 1, "camera_id": 2, "event_type": 99, "start_time": 2500, "duration": 5}]) assert db.has_motion_in_range_local(2000, 3000) is False def test_has_motion_in_range_local_filters_by_camera_id(tmp_path): db = _db(tmp_path) db.record_motion_heartbeat() db.record_motion_events( [{"event_id": 1, "camera_id": 99, "event_type": 10, "start_time": 2500, "duration": 5}]) assert db.has_motion_in_range_local(2000, 3000, camera_id=2) is False assert db.has_motion_in_range_local(2000, 3000, camera_id=99) is True