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 # ---------------------------------------------------------------------- # 人物对应关系表（video_id, raw_uid) -> canonical_name # ---------------------------------------------------------------------- def _seed_video_with_events(db, filename="motion_1_1000.mp4"): vid = db.ensure_video(filename, f"/tmp/{filename}", event_start_time="2026-08-22 10:00:00") events = [ {"timestamp": "10:00:01", "description": "在客厅走动", "people": ["人物A"], "person_appearances": [{"uid": "人物A", "features": {"gender": "男"}, "action": "走动"}]}, {"timestamp": "10:00:05", "description": "坐下", "people": ["人物A", "人物B"], "person_appearances": [ {"uid": "人物A", "features": {"gender": "男"}, "action": "坐下"}, {"uid": "人物B", "features": {"gender": "女"}, "action": "站立"}]}, ] db.mark_video_processed(vid, "摘要", events, ["人物A", "人物B"], "gemini") return vid def test_set_identity_mapping_inserts_new_row(tmp_path): db = _db(tmp_path) assert db.set_identity_mapping(1, "人物A", "爷爷", source="auto_id") is True assert db.get_identity_map_for_video(1) == {"人物A": "爷爷"} def test_set_identity_mapping_updates_existing_non_manual_row(tmp_path): db = _db(tmp_path) db.set_identity_mapping(1, "人物A", "爷爷", source="auto_id") assert db.set_identity_mapping(1, "人物A", "爸爸", source="auto_id") is True assert db.get_identity_map_for_video(1) == {"人物A": "爸爸"} def test_set_identity_mapping_manual_protected_from_auto_overwrite(tmp_path): """核心诉求: 人工纠正过的映射不能被后续自动识别悄悄改回去。""" db = _db(tmp_path) db.set_identity_mapping(1, "人物A", "爸爸", source="manual") changed = db.set_identity_mapping(1, "人物A", "爷爷", source="auto_id") assert changed is False assert db.get_identity_map_for_video(1) == {"人物A": "爸爸"} def test_set_identity_mapping_manual_can_override_manual(tmp_path): db = _db(tmp_path) db.set_identity_mapping(1, "人物A", "爸爸", source="manual") changed = db.set_identity_mapping(1, "人物A", "爷爷", source="manual") assert changed is True assert db.get_identity_map_for_video(1) == {"人物A": "爷爷"} def test_set_identity_mapping_no_change_returns_false(tmp_path): db = _db(tmp_path) db.set_identity_mapping(1, "人物A", "爷爷", source="auto_id") changed = db.set_identity_mapping(1, "人物A", "爷爷", source="auto_id") assert changed is False def test_get_identity_map_for_video_scoped_per_video(tmp_path): """核心诉求: 同一个 raw_uid 字符串在不同视频里可能是不同真人，映射必须按 video_id 隔离，不能串。""" db = _db(tmp_path) db.set_identity_mapping(1, "人物A", "爷爷", source="auto_id") db.set_identity_mapping(2, "人物A", "爸爸", source="auto_id") assert db.get_identity_map_for_video(1) == {"人物A": "爷爷"} assert db.get_identity_map_for_video(2) == {"人物A": "爸爸"} def test_rewrite_event_person_names_updates_events_and_video(tmp_path): db = _db(tmp_path) vid = _seed_video_with_events(db) db.rewrite_event_person_names(vid, {"人物A": "爷爷", "人物B": "媳妇"}) rows = db._conn.execute( "SELECT person_list_json, person_appearances_json FROM events " "WHERE video_id=? ORDER BY id", (vid,)).fetchall() assert json.loads(rows[0]["person_list_json"]) == ["爷爷"] pa0 = json.loads(rows[0]["person_appearances_json"]) assert pa0[0]["uid"] == "爷爷" assert json.loads(rows[1]["person_list_json"]) == ["爷爷", "媳妇"] pa1 = json.loads(rows[1]["person_appearances_json"]) assert {p["uid"] for p in pa1} == {"爷爷", "媳妇"} vrow = db._conn.execute("SELECT people_json FROM videos WHERE id=?", (vid,)).fetchone() assert set(json.loads(vrow["people_json"])) == {"爷爷", "媳妇"} def test_rewrite_event_person_names_rewrites_description_text(tmp_path): """核心诉求: description 是大模型写的自然语言句子，"人物A/人物B"这类 uid 会直接以文字形式嵌在句子里，只改 person_list_json/person_appearances_json 这些结构化字段的话，事件卡片上方徽章显示对了，描述文字里还是旧 uid，两处 对不上——description 也要做文本替换。""" db = _db(tmp_path) vid = db.ensure_video("motion_2_2000.mp4", "/tmp/motion_2_2000.mp4", event_start_time="2026-08-22 13:00:00") events = [ {"timestamp": "13:29:24", "description": "人物B双手叉腰站在客厅中央；人物A在远处厨房；儿童已离开画面。", "people": ["人物A", "人物B"], "person_appearances": [ {"uid": "人物A", "features": {"gender": "男"}, "action": "站立"}, {"uid": "人物B", "features": {"gender": "女"}, "action": "叉腰"}]}, ] db.mark_video_processed(vid, "人物A和人物B都在客厅活动。", events, ["人物A", "人物B"], "gemini") db.rewrite_event_person_names(vid, {"人物A": "爸爸", "人物B": "媳妇"}) ev_row = db._conn.execute( "SELECT description FROM events WHERE video_id=?", (vid,)).fetchone() assert ev_row["description"] == "媳妇双手叉腰站在客厅中央；爸爸在远处厨房；儿童已离开画面。" v_row = db._conn.execute( "SELECT summary_json FROM videos WHERE id=?", (vid,)).fetchone() assert v_row["summary_json"] == "爸爸和媳妇都在客厅活动。" def test_rewrite_event_person_names_longer_labels_replaced_before_shorter(tmp_path): """核心诉求: uid 可能带 "#2"/"#3" 这类同名冲突后缀，"人物A" 是 "人物A#2" 的 前缀——如果先替换短的 "人物A"，"人物A#2" 会被错误地部分命中变成"爷爷#2"， 而不是走它自己在 rename_map 里对应的正确目标。必须长的先替换。""" db = _db(tmp_path) vid = db.ensure_video("motion_3_3000.mp4", "/tmp/motion_3_3000.mp4", event_start_time="2026-08-22 13:00:00") events = [ {"timestamp": "13:00:01", "description": "人物A和人物A#2一起在客厅。", "people": ["人物A", "人物A#2"], "person_appearances": [ {"uid": "人物A", "features": {"gender": "男"}, "action": "站立"}, {"uid": "人物A#2", "features": {"gender": "女"}, "action": "站立"}]}, ] db.mark_video_processed(vid, "摘要", events, ["人物A", "人物A#2"], "gemini") db.rewrite_event_person_names(vid, {"人物A": "爷爷", "人物A#2": "媳妇"}) ev_row = db._conn.execute( "SELECT description FROM events WHERE video_id=?", (vid,)).fetchone() assert ev_row["description"] == "爷爷和媳妇一起在客厅。" def test_rewrite_event_person_names_noop_on_empty_map(tmp_path): db = _db(tmp_path) vid = _seed_video_with_events(db) before = db._conn.execute( "SELECT person_list_json FROM events WHERE video_id=?", (vid,)).fetchall() db.rewrite_event_person_names(vid, {}) after = db._conn.execute( "SELECT person_list_json FROM events WHERE video_id=?", (vid,)).fetchall() assert [r["person_list_json"] for r in before] == [r["person_list_json"] for r in after] def test_correct_video_identity_end_to_end(tmp_path): """核心诉求: 纠错入口应该找到当前展示名对应的映射行，改写映射 + 立即重写 展示数据，且标记为 manual（受保护）。""" db = _db(tmp_path) vid = _seed_video_with_events(db) db.set_identity_mapping(vid, "人物A", "爷爷", source="auto_id") db.rewrite_event_person_names(vid, {"人物A": "爷爷"}) db.correct_video_identity(vid, current_name="爷爷", new_name="爸爸") assert db.get_identity_map_for_video(vid) == {"人物A": "爸爸"} rows = db._conn.execute( "SELECT person_list_json FROM events WHERE video_id=? ORDER BY id", (vid,)).fetchall() assert json.loads(rows[0]["person_list_json"]) == ["爸爸"] # manual 之后不能被自动识别覆盖回去 changed = db.set_identity_mapping(vid, "人物A", "爷爷", source="auto_id") assert changed is False def test_correct_video_identity_without_prior_mapping_uses_current_name_as_raw_uid(tmp_path): """核心诉求: 老流水线时代产出的数据从没跑过闭集识别，映射表里没有记录—— 纠错依然要能生效，把 current_name 本身当 raw_uid 存一条新映射。""" db = _db(tmp_path) vid = db.ensure_video("motion_2_2000.mp4", "/tmp/x.mp4", event_start_time="2026-08-22 10:00:00") events = [{"timestamp": "10:00:01", "description": "走动", "people": ["爷爷"], "person_appearances": [{"uid": "爷爷", "features": {"gender": "男"}, "action": "走动"}]}] db.mark_video_processed(vid, "摘要", events, ["爷爷"], "gemini") db.correct_video_identity(vid, current_name="爷爷", new_name="爸爸") assert db.get_identity_map_for_video(vid) == {"爷爷": "爸爸"} rows = db._conn.execute( "SELECT person_list_json FROM events WHERE video_id=?", (vid,)).fetchall() assert json.loads(rows[0]["person_list_json"]) == ["爸爸"] def test_delete_video_removes_video_and_events_rows(tmp_path): db = _db(tmp_path) vid = _seed_video_with_events(db) local_path = db.delete_video(vid) assert local_path == f"/tmp/motion_1_1000.mp4" assert db._conn.execute("SELECT * FROM videos WHERE id=?", (vid,)).fetchone() is None assert db._conn.execute("SELECT * FROM events WHERE video_id=?", (vid,)).fetchall() == [] def test_delete_video_removes_disk_file(tmp_path): db = _db(tmp_path) clip_path = tmp_path / "motion_9_2000.mp4" clip_path.write_bytes(b"fake mp4 bytes") vid = db.ensure_video("motion_9_2000.mp4", str(clip_path), event_start_time="2026-08-22 10:00:00") db.mark_video_processed(vid, "摘要", [], [], "gemini") db.delete_video(vid) assert not clip_path.exists() def test_delete_video_missing_file_on_disk_does_not_raise(tmp_path): """核心诉求: local_path 指向的文件已经不存在（比如手动清理过）时，删除记录 本身不能因为 os.remove 报错而失败——文件缺失不是数据库操作的错误。""" db = _db(tmp_path) vid = db.ensure_video("motion_9_2000.mp4", str(tmp_path / "already_gone.mp4"), event_start_time="2026-08-22 10:00:00") db.mark_video_processed(vid, "摘要", [], [], "gemini") local_path = db.delete_video(vid) assert local_path == str(tmp_path / "already_gone.mp4") assert db._conn.execute("SELECT * FROM videos WHERE id=?", (vid,)).fetchone() is None def test_delete_video_nonexistent_returns_none(tmp_path): db = _db(tmp_path) assert db.delete_video(99999) is None def test_get_oldest_purgeable_material_none_when_empty(tmp_path): db = _db(tmp_path) assert db.get_oldest_purgeable_material() is None def test_get_oldest_purgeable_material_ignores_motion_clips(tmp_path): """核心诉求: 运动片段（motion_ 前缀）是独立的分析产物，事件时间轴/人物 头像都依赖它，磁盘清理绝不能碰它，只能清理原始整段素材。""" db = _db(tmp_path) vid = db.ensure_video("motion_1_1000.mp4", "/tmp/motion_1_1000.mp4", event_start_time="2026-08-22 10:00:00") db.mark_video_processed(vid, "摘要", [], [], "gemini") assert db.get_oldest_purgeable_material() is None def test_get_oldest_purgeable_material_ignores_non_done_status(tmp_path): """核心诉求: 还在 pending/processing 的素材不能被清理，避免删掉还没 来得及处理的数据。""" db = _db(tmp_path) db.ensure_video("Generic_ONVIF-001-20260815-000000.mp4", "/tmp/Generic_ONVIF-001-20260815-000000.mp4") assert db.get_oldest_purgeable_material() is None def test_get_oldest_purgeable_material_returns_oldest_done_material(tmp_path): db = _db(tmp_path) vid1 = db.ensure_video("Generic_ONVIF-001-20260815-000000.mp4", "/tmp/Generic_ONVIF-001-20260815-000000.mp4") db.mark_video_processed(vid1, "（整段素材已分割 0 段运动片段）", [], [], 'motion_segment') vid2 = db.ensure_video("Generic_ONVIF-001-20260816-000000.mp4", "/tmp/Generic_ONVIF-001-20260816-000000.mp4") db.mark_video_processed(vid2, "（整段素材已分割 0 段运动片段）", [], [], 'motion_segment') candidate = db.get_oldest_purgeable_material() assert candidate["id"] == vid1 assert candidate["local_path"] == "/tmp/Generic_ONVIF-001-20260815-000000.mp4" def test_delete_video_does_not_touch_ss_motion_events(tmp_path): """核心诉求: ss_motion_events 是运动侦测源事件，跟切出来的视频片段生命周期 独立，删视频不该连带删掉源事件（否则分割逻辑的幂等判断会被破坏）。""" db = _db(tmp_path) db.record_motion_events([ {"event_id": 555, "camera_id": 2, "event_type": 10, "start_time": 1700000000, "duration": 10, "thumbnail_url": ""}, ]) vid = db.ensure_video("motion_555_1700000000.mp4", "/tmp/motion_555_1700000000.mp4", event_start_time="2026-08-22 10:00:00", motion_event_id=555) db.mark_video_processed(vid, "摘要", [], [], "gemini") db.delete_video(vid) assert db.get_video_by_motion_event_id(555) is None row = db._conn.execute("SELECT * FROM ss_motion_events WHERE event_id=?", (555,)).fetchone() assert row is not None