sentinel-home-ai/fam-core/tests/test_db_layer.py

"""db_layer 直读 SQLite 的单测（2026-09-13 迁云重写后新增）。

重点验证从 MySQL 翻译过来的几处方言：json_each 取代 JSON_CONTAINS、
substr 取代 LEFT、以及历史脏数据（person_list_json 不是合法 JSON）不能
把整条查询搞崩——这在 MySQL 下 JSON_CONTAINS 会直接报错，SQLite 下
json_each 同样会，所以查询里加了 json_valid 前置判断。
"""
import sqlite3

import pytest

from fam_core import db_layer

# fam-edge 建表语句的最小子集（列名与 oracle_db.py 保持一致）
_SCHEMA = """
CREATE TABLE videos (
    id INTEGER PRIMARY KEY AUTOINCREMENT, filename TEXT UNIQUE, camera_name TEXT,
    duration_sec REAL, event_start_time TEXT, status TEXT DEFAULT 'pending',
    summary_json TEXT, events_json TEXT, people_json TEXT, compute_provider TEXT,
    created_at TEXT, updated_at TEXT, processed_at TEXT);
CREATE TABLE events (
    id INTEGER PRIMARY KEY AUTOINCREMENT, video_id INTEGER, ts TEXT, description TEXT,
    person_list_json TEXT, person_appearances_json TEXT, is_attention_event INTEGER DEFAULT 0);
CREATE TABLE people (
    id INTEGER PRIMARY KEY AUTOINCREMENT, label TEXT UNIQUE, canonical_name TEXT,
    first_seen TEXT, appearances INTEGER DEFAULT 0, source TEXT, features_json TEXT,
    display_uid TEXT, updated_at TEXT);
CREATE TABLE model_calls (
    id INTEGER PRIMARY KEY AUTOINCREMENT, provider TEXT, model TEXT, video_id INTEGER,
    filename TEXT, started_at TEXT, duration_sec REAL, success INTEGER, error TEXT,
    created_at TEXT);
"""


@pytest.fixture
def db(tmp_path, monkeypatch):
    path = tmp_path / "oracle.db"
    conn = sqlite3.connect(path)
    conn.executescript(_SCHEMA)
    conn.executescript("""
    INSERT INTO videos (id, filename, camera_name, event_start_time, status, processed_at)
      VALUES (1, 'motion_20260913_080000.mp4', '客厅', '2026-09-13 08:00:00', 'done', '2026-09-13 08:10:00'),
             (2, 'whole_20260912_090000.mp4',  '客厅', '2026-09-12 09:00:00', 'done', '2026-09-12 09:10:00'),
             (3, 'whole_20260911_090000.mp4',  '客厅', '2026-09-11 09:00:00', 'done', '2026-09-11 09:10:00'),
             (4, 'motion_20260910_070000.mp4', '客厅', '2026-09-10 07:00:00', 'pending', NULL);
    -- video 2 有事件（应展示），video 3 没有事件且不是 motion_（应被过滤掉）
    INSERT INTO events (video_id, ts, description, person_list_json, is_attention_event)
      VALUES (1, '00:00:05', '汤圆在客厅玩耍', '["汤圆"]', 0),
             (1, '00:00:20', '有人靠近门口',   '["人物B"]', 1),
             (2, '00:01:00', '汤圆和奶奶',     '["汤圆","奶奶"]', 0),
             (2, '00:02:00', '脏数据事件',     '不是合法JSON', 0);
    INSERT INTO people (id, label, canonical_name, appearances, source)
      VALUES (1, '人物A', '汤圆', 12, 'manual'),
             (2, '人物B', NULL, 3, 'llm'),
             (3, '汤圆',  '汤圆', 5, 'manual');
    INSERT INTO model_calls (provider, model, success, duration_sec, created_at)
      VALUES ('gemini', 'flash', 1, 10.0, '2026-09-13 08:10:00'),
             ('gemini', 'flash', 0, 20.0, '2026-09-13 08:20:00'),
             ('nvidia', 'vila',  1, 5.0,  '2026-09-12 08:10:00');
    """)
    conn.commit()
    conn.close()
    monkeypatch.setattr(db_layer, '_db_path', str(path))
    monkeypatch.setattr(db_layer, '_chat_schema_ready', False)
    return path


# ---------------------------------------------------------------- videos
def test_get_videos_only_returns_content_sessions(db):
    """只展示运动片段或含事件的会话：分割 0 段的整段素材不该淹没时间轴。"""
    ids = [v['id'] for v in db_layer.get_videos()]
    assert ids == [1, 2]          # 3 无事件且非 motion_，4 未完成
    assert db_layer.get_videos()[0]['event_count'] == 2


def test_get_videos_date_filter_and_paging(db):
    assert [v['id'] for v in db_layer.get_videos(date_filter='2026-09-12')] == [2]
    assert [v['id'] for v in db_layer.get_videos(limit=1, offset=1)] == [2]


def test_get_video_and_events(db):
    assert db_layer.get_video(1)['filename'].startswith('motion_')
    assert db_layer.get_video(999) is None
    evs = db_layer.get_events_for_video(1)
    assert [e['ts'] for e in evs] == ['00:00:05', '00:00:20']


def test_get_attention_events(db):
    rows = db_layer.get_attention_events()
    assert len(rows) == 1 and rows[0]['ev_date'].startswith('2026-09-13')


# ---------------------------------------------------------------- 人物命中（JSON）
def test_people_clips_matches_label_and_its_aliases(db):
    """'人物A' 的规范名是"汤圆"，而事件里记的是"汤圆"——要能顺着别名找到。"""
    clips = db_layer.get_people_clips('人物A')
    assert {c['video_id'] for c in clips} == {1, 2}
    assert clips[0]['video_id'] == 1          # 按录制时间倒序
    assert clips[0]['first_ts'] == '00:00:05'
    assert clips[0]['clip_events'] == 1


def test_person_hit_is_exact_not_substring(db):
    """json_each 是精确匹配：查"人物"不该命中"人物B"。"""
    assert db_layer.get_people_clips('人物') == []
    assert {c['video_id'] for c in db_layer.get_people_clips('人物B')} == {1}


def test_malformed_person_json_does_not_break_queries(db):
    """video 2 里混了一条非 JSON 的脏数据，查询必须照常返回而不是整条报错。"""
    rows = db_layer.query_events_for_person_date('汤圆', '2026-09-12')
    assert len(rows) == 1 and rows[0]['ts'] == '00:01:00'


def test_query_events_for_person_date(db):
    assert db_layer.query_events_for_person_date('汤圆', '2026-09-13')[0]['description'] == '汤圆在客厅玩耍'
    assert db_layer.query_events_for_person_date('汤圆', '2026-09-01') == []


# ---------------------------------------------------------------- people / model_calls
def test_people_and_named_members(db):
    assert len(db_layer.get_people()) == 3
    assert db_layer.get_named_members() == ['汤圆']        # distinct 且非空
    ctx = db_layer.get_known_members_context()
    assert '- 汤圆（标识：人物A）' in ctx and '- 人物B' in ctx


def test_model_calls_stats(db):
    stats = {s['model']: s for s in db_layer.get_model_calls_stats()}
    assert stats['flash']['ok_cnt'] == 1 and stats['flash']['fail_cnt'] == 1
    assert stats['flash']['avg_duration'] == 15.0
    assert len(db_layer.get_model_calls(limit=2)) == 2


# ---------------------------------------------------------------- 统计
def test_stats_全量与按日(db):
    total = db_layer.get_stats()
    assert total['videos'] == 2 and total['events'] == 4 and total['attention'] == 1
    day = db_layer.get_stats('2026-09-13')
    assert day['videos'] == 1 and day['events'] == 2
    # 人物数：汤圆（含 label 人物A/汤圆两行都归一到"汤圆"）+ 人物B
    assert total['people'] >= 2


# ---------------------------------------------------------------- chat_history
def test_chat_history_roundtrip_creates_table_on_demand(db):
    """chat_history 是本模块唯一写的表，建表是懒加载的（库文件属于 fam-edge）。"""
    chat_id = db_layer.insert_chat_history('今天有人来吗', '有，08:00 有人靠近门口',
                                           '上下文', '2026-09-13', '汤圆')
    assert chat_id == 1
    rows = db_layer.get_chat_history(limit=10)
    assert rows[0]['user_question'] == '今天有人来吗'
    assert rows[0]['created_at']
    assert db_layer.get_chat_history(person_filter='不存在的人') == []
    assert len(db_layer.get_chat_history(date_filter='2026-09-13')) == 1