xiaoxia-saas/tests/unit/test_asset_segment_tracker.py

"""素材片段使用记录追踪 + 受控复用机制测试（asset_segment_tracker）.

覆盖：
- get_used_segments 聚合 metadata 中持久化的区间
- record_used_segments 追加新记录（use_count=1，保留原有 metadata 字段）
- record_used_segments 复用同一区间时累加 use_count / 刷新 last_used_at
- remove_used_segment 匹配删除（tolerance + plan_id，旧数据按时间匹配）
- reset_used_segments 清空（其他字段不动）
- find_reusable_range：选最久未用且 use_count<3 的区间；全部达上限返回 None
- make_reuse_callback：返回复用区间、累加 reused_tracker、DB 异常返回 None
- _calc_random_start_time：100 次避不开时调用复用回调返回历史区间（不再清空历史）
"""

from __future__ import annotations

import json
import os
import sys
from pathlib import Path

os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))

import pytest
from app.services import asset_segment_tracker as ast
from app.services.asset_segment_tracker import (
    MAX_RANGE_USE_COUNT,
    REUSE_RATIO_LIMIT,
    SEGMENT_EDGE_GAP,
    find_reusable_range,
    get_used_segments,
    make_reuse_callback,
    record_used_segments,
    remove_used_segment,
    reset_used_segments,
)

from packages.domain.plan_generator_utils import _calc_random_start_time


class FakeModel:
    """模拟 AssetModel：id + classification_result（JSON Text）+ updated_at。"""

    def __init__(self, asset_id: str, meta: dict | None = None):
        self.id = asset_id
        self.classification_result = json.dumps(meta, ensure_ascii=False) if meta else None
        self.updated_at = None

    def meta(self) -> dict:
        return json.loads(self.classification_result) if self.classification_result else {}


class _InExpr:
    def __init__(self, ids, models):
        self._ids = ids
        self._models = models

    def all(self):
        return [self._models[i] for i in self._ids if i in self._models]


class _EqExpr:
    def __init__(self, target_id, models):
        self._target_id = target_id
        self._models = models

    def with_for_update(self):
        # 模拟 SQLAlchemy Query.with_for_update() 链式返回自身
        return self

    def first(self):
        return self._models.get(self._target_id)


class FakeSession:
    """模拟 db：db.query(Model).filter(Model.id.in_(ids)).all() / .filter(Model.id == id).first()。"""

    class _Col:
        def __init__(self, models):
            self._models = models

        def in_(self, ids):
            return _InExpr(list(ids), self._models)

        def __eq__(self, other):
            return _EqExpr(other, self._models)

    def __init__(self, models: dict[str, FakeModel]):
        self._models = models
        self.commits = 0

    def query(self, _model):
        col = self._Col(self._models)

        class _Q:
            def filter(self_inner, expr):
                return expr

        q = _Q()
        _model.id = col
        return q

    def commit(self):
        self.commits += 1


@pytest.fixture
def patched_model(monkeypatch):
    monkeypatch.setattr(ast, "AssetModel", FakeModel)


def _db(models):
    return FakeSession(models)


def _ranges(db, aid="a1"):
    model = db._models[aid]
    return json.loads(model.classification_result)["used_time_ranges"]


# ── 配置常量 ──────────────────────────────────────────────────────────────────


def test_config_constants():
    assert MAX_RANGE_USE_COUNT == 2
    assert REUSE_RATIO_LIMIT == 0.10
    assert SEGMENT_EDGE_GAP == 1.5


# ── get_used_segments ─────────────────────────────────────────────────────────


def test_get_used_segments_aggregates_ranges(patched_model):
    models = {
        "a1": FakeModel(
            "a1",
            {
                "used_time_ranges": [
                    {"start": 1.0, "end": 5.0, "plan_id": "p1", "use_count": 2},
                    {"start": 9.0, "end": 12.0, "plan_id": "p2"},
                ]
            },
        ),
        "a2": FakeModel("a2", {"other": 1}),
        "a3": FakeModel("a3"),
    }
    db = _db(models)
    assert get_used_segments(db, ["a1", "a2", "a3", "missing"]) == {"a1": [(1.0, 5.0), (9.0, 12.0)]}


def test_get_used_segments_empty(patched_model):
    assert get_used_segments(_db({}), []) == {}


# ── record_used_segments ──────────────────────────────────────────────────────


def test_record_appends_new_range_with_use_count_one(patched_model):
    models = {"a1": FakeModel("a1", {"generation_use_count": 48, "review_status": "pending_review"})}
    db = _db(models)
    record_used_segments(db, "a1", 12.5, 20.3, "plan-x")
    meta = json.loads(models["a1"].classification_result)
    assert meta["generation_use_count"] == 48
    assert meta["review_status"] == "pending_review"
    ranges = meta["used_time_ranges"]
    assert len(ranges) == 1
    assert ranges[0]["start"] == 12.5 and ranges[0]["end"] == 20.3
    assert ranges[0]["plan_id"] == "plan-x"
    assert ranges[0]["use_count"] == 1
    assert "created_at" in ranges[0] and "last_used_at" in ranges[0]
    assert db.commits == 0  # 不自行 commit


def test_record_reuse_same_range_increments_use_count(patched_model):
    """新片段与历史区间高度重叠（复用）→ 累加 use_count，不新增记录。"""
    models = {
        "a1": FakeModel(
            "a1",
            {
                "used_time_ranges": [
                    {
                        "start": 10.0,
                        "end": 20.0,
                        "plan_id": "p1",
                        "use_count": 1,
                        "created_at": "2026-01-01T00:00:00+00:00",
                        "last_used_at": "2026-01-01T00:00:00+00:00",
                    },
                ]
            },
        )
    }
    db = _db(models)
    # 同一起点复用（find_reusable_range 返回的就是历史区间起点）
    record_used_segments(db, "a1", 10.0, 20.0, "p2")
    ranges = _ranges(db)
    assert len(ranges) == 1
    assert ranges[0]["use_count"] == 2
    assert ranges[0]["last_used_at"] != "2026-01-01T00:00:00+00:00"


def test_record_distinct_range_appends(patched_model):
    models = {
        "a1": FakeModel(
            "a1",
            {
                "used_time_ranges": [
                    {"start": 10.0, "end": 20.0, "plan_id": "p1", "use_count": 1},
                ]
            },
        )
    }
    db = _db(models)
    record_used_segments(db, "a1", 25.0, 35.0, "p2")
    ranges = _ranges(db)
    assert len(ranges) == 2
    assert ranges[1]["use_count"] == 1


def test_record_missing_asset_no_raise(patched_model):
    db = _db({})
    record_used_segments(db, "ghost", 1.0, 2.0, "p")  # 不抛异常


# ── remove_used_segment ───────────────────────────────────────────────────────


def test_remove_matching_range(patched_model):
    models = {
        "a1": FakeModel(
            "a1",
            {
                "used_time_ranges": [
                    {"start": 1.0, "end": 5.0, "plan_id": "p1"},
                    {"start": 9.0, "end": 12.0, "plan_id": "p2"},
                ]
            },
        )
    }
    db = _db(models)
    assert remove_used_segment(db, "a1", 1.0, 5.0, plan_id="p1") is True
    assert len(_ranges(db)) == 1
    assert _ranges(db)[0]["start"] == 9.0


def test_remove_plan_mismatch_keeps_range(patched_model):
    models = {"a1": FakeModel("a1", {"used_time_ranges": [{"start": 1.0, "end": 5.0, "plan_id": "p1"}]})}
    db = _db(models)
    assert remove_used_segment(db, "a1", 1.0, 5.0, plan_id="other") is False
    assert len(_ranges(db)) == 1


def test_remove_legacy_range_without_plan_id(patched_model):
    """旧数据记录缺 plan_id → 按时间匹配可删除。"""
    models = {
        "a1": FakeModel(
            "a1",
            {
                "used_time_ranges": [
                    {"start": 2.0, "end": 12.0, "created_at": "2026-01-01T00:00:00"},
                ]
            },
        )
    }
    db = _db(models)
    assert remove_used_segment(db, "a1", 2.0, 12.0, plan_id="plan-new") is True
    assert _ranges(db) == []


# ── reset_used_segments（仅运维/测试）─────────────────────────────────────────


def test_reset_clears_ranges_keeps_other_fields(patched_model):
    models = {
        "a1": FakeModel(
            "a1",
            {
                "generation_use_count": 3,
                "used_time_ranges": [
                    {"start": 1.0, "end": 5.0},
                ],
            },
        )
    }
    db = _db(models)
    reset_used_segments(db, "a1")
    meta = json.loads(models["a1"].classification_result)
    assert meta["used_time_ranges"] == []
    assert meta["generation_use_count"] == 3


# ── find_reusable_range：受控复用选择 ─────────────────────────────────────────


def test_find_reusable_prefers_oldest_unused(patched_model):
    """选 last_used_at 最老、use_count 未达上限的区间；能容纳 clip_duration。"""
    models = {
        "a1": FakeModel(
            "a1",
            {
                "used_time_ranges": [
                    {"start": 0.0, "end": 8.0, "use_count": 1, "last_used_at": "2026-08-01T00:00:00+00:00"},
                    {
                        "start": 10.0,
                        "end": 20.0,
                        "use_count": 1,
                        "last_used_at": "2026-01-01T00:00:00+00:00",
                    },  # 最久未用
                ]
            },
        )
    }
    db = _db(models)
    result = find_reusable_range(db, "a1", clip_duration=5.0, asset_total=30.0)
    assert result is not None
    start, end = result
    assert start == 10.0 and end == 15.0


def test_find_reusable_excludes_max_use_count(patched_model):
    """use_count 达到上限（2）的区间不再参与复用；全部达上限返回 None。"""
    models = {
        "a1": FakeModel(
            "a1",
            {
                "used_time_ranges": [
                    {"start": 0.0, "end": 10.0, "use_count": 2, "last_used_at": "2026-01-01T00:00:00"},
                ]
            },
        )
    }
    db = _db(models)
    assert find_reusable_range(db, "a1", 5.0, 30.0) is None


def test_find_reusable_third_use_rejected(patched_model):
    """同区间复用第 3 次被拒绝：use_count=1 的可复用，use_count=2 的不可复用。"""
    models = {
        "a1": FakeModel(
            "a1",
            {
                "used_time_ranges": [
                    {"start": 0.0, "end": 10.0, "use_count": 1, "last_used_at": "2026-03-01T00:00:00"},
                    {"start": 10.0, "end": 20.0, "use_count": 2, "last_used_at": "2026-01-01T00:00:00"},
                ]
            },
        )
    }
    db = _db(models)
    result = find_reusable_range(db, "a1", 5.0, 30.0)
    # 只能选 use_count=1 的区间（start=0），不能选 use_count=2 的（虽然它更老）
    assert result is not None and result[0] == 0.0


def test_find_reusable_clamps_to_asset_bounds(patched_model):
    """历史区间起点 + clip_duration 会越素材末尾时，起点钳制到 max_start。"""
    models = {
        "a1": FakeModel(
            "a1",
            {
                "used_time_ranges": [
                    {"start": 25.0, "end": 30.0, "use_count": 1, "last_used_at": "2026-01-01T00:00:00"},
                ]
            },
        )
    }
    db = _db(models)
    result = find_reusable_range(db, "a1", clip_duration=10.0, asset_total=30.0)
    assert result is not None
    start, end = result
    assert end <= 30.0 + 1e-6 and start >= 0.0


def test_find_reusable_no_ranges_returns_none(patched_model):
    models = {"a1": FakeModel("a1", {"other": 1})}
    db = _db(models)
    assert find_reusable_range(db, "a1", 5.0, 30.0) is None


# ── make_reuse_callback ───────────────────────────────────────────────────────


def test_reuse_callback_returns_range_and_tracks_duration(patched_model):
    models = {
        "a1": FakeModel(
            "a1",
            {
                "used_time_ranges": [
                    {"start": 10.0, "end": 20.0, "use_count": 1, "last_used_at": "2026-01-01T00:00:00"},
                ]
            },
        )
    }
    db = _db(models)
    reused: dict[str, float] = {}
    cb = make_reuse_callback(db, {"a1": 30.0}, reused)
    result = cb("a1", 8.0)
    assert result is not None and result[0] == 10.0
    assert reused["a1"] == 8.0  # 复用时长累加


def test_reuse_callback_db_error_returns_none(patched_model):
    class BoomSession:
        def query(self, _m):
            raise RuntimeError("db down")

    reused: dict[str, float] = {}
    cb = make_reuse_callback(BoomSession(), {"a1": 30.0}, reused)
    assert cb("a1", 8.0) is None  # 异常被吞，返回 None
    assert reused == {}


# ── _calc_random_start_time 与受控回调集成 ────────────────────────────────────


def test_calc_random_start_uses_reuse_callback_when_exhausted(monkeypatch):
    """素材区间被占满、100 次随机找不到空位时，调用复用回调返回历史区间。"""
    import packages.domain.plan_generator_utils as pgu

    monkeypatch.setattr(pgu.random, "uniform", lambda a, b: 0.5)  # 固定候选点必撞区间

    durations = {"a1": 30.0}
    used = {"a1": [(0.0, 30.0)]}  # 全占满
    calls = []

    def reuse_cb(asset_id, clip_duration):
        calls.append((asset_id, clip_duration))
        return (10.0, 18.0)

    result = _calc_random_start_time("a1", 8.0, durations, used, on_exhausted=reuse_cb)
    assert calls == [("a1", 8.0)]
    assert result == 10.0


def test_calc_random_start_reuse_callback_none_returns_none(monkeypatch):
    """复用回调返回 None（区间全部达上限/复用占比超闸门）→ calc 返回 None。

    新机制下不做末尾/0.0 重叠降级（那会把片段放回已用过的画面），
    由调用方轮询下一个素材或报 400；历史记录不被清空。
    """
    import packages.domain.plan_generator_utils as pgu

    monkeypatch.setattr(pgu.random, "uniform", lambda a, b: 0.5)

    durations = {"a1": 30.0}
    used = {"a1": [(0.0, 30.0)]}
    used_before = list(used["a1"])
    result = _calc_random_start_time("a1", 8.0, durations, used, on_exhausted=lambda aid, d: None)
    assert result is None
    assert used["a1"] == used_before  # 历史记录未被清空


def test_calc_random_start_no_callback_backward_compatible(monkeypatch):
    """不传 on_exhausted 时行为与旧版兼容（100 次失败走降级）。"""
    import packages.domain.plan_generator_utils as pgu

    monkeypatch.setattr(pgu.random, "uniform", lambda a, b: 0.5)
    result = _calc_random_start_time("a1", 8.0, {"a1": 30.0}, {"a1": [(0.0, 30.0)]})
    assert result is not None