xiaoxia-saas/tests/unit/test_asset_availability.py

"""Task H 单测：素材余量四字段（used_duration/available_duration/used_ratio/usable）。

覆盖：
1. compute_asset_availability 纯函数各分支（无区间/未满/可复用/全达上限/非视频/无时长/区间合并/扩边判定）；
2. _asset_availability_fields 路由辅助（视频有值、非视频 None+usable=True、异常零影响）；
3. _to_asset_response 四字段注入；
4. smart_match_assets 结果层过滤 usable=false。
"""

import json
import os
import sys
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock

import pytest

REPO_ROOT = Path(__file__).resolve().parents[2]
if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))

from app.api.routes.assets import (  # noqa: E402
    _asset_availability_fields,
    _to_asset_response,
    smart_match_assets,
)
from app.schemas.asset import SmartMatchRequest  # noqa: E402
from app.services.asset_segment_tracker import (  # noqa: E402
    MAX_RANGE_USE_COUNT,
    SEGMENT_EDGE_GAP,
    compute_asset_availability,
)

VIDEO_DURATION = 60.0


def _make_asset(duration=VIDEO_DURATION, ranges=None, file_type="video", classification_result=None):
    """构造测试用 Asset-like 对象（领域实体形态：metadata 为 dict）。

    ranges: list of dicts（used_time_ranges 条目），同时写入 metadata dict
    （Asset 实体形态，repository 返回）与 classification_result JSON 字符串
    （ORM AssetModel 形态）；_read_meta 两种形态都必须能读到。
    """
    meta_dict = {"used_time_ranges": ranges} if ranges is not None else {}
    if classification_result is None and ranges is not None:
        classification_result = json.dumps(meta_dict)
    return SimpleNamespace(
        id="asset-test",
        project_id="proj-1",
        library_id="lib-1",
        name="测试素材",
        storage_key="key/test-asset.mp4",
        thumbnail_url=None,
        mime_type="video/mp4" if file_type == "video" else "audio/mpeg",
        file_size=1000,
        duration=duration,
        width=1080,
        height=1920,
        fps=30,
        codec="h264",
        status=SimpleNamespace(value="ready"),
        classification_status=SimpleNamespace(value="completed"),
        quality_score=90.0,
        created_at=__import__("datetime").datetime(2026, 8, 1, 12, 0, 0),
        uploaded_by_user_id="user-1",
        tag_ids=[],
        file_type=file_type,
        # ORM 形态
        classification_result=classification_result,
        # 领域实体形态（真实路由 repository 返回的 Asset）
        metadata=meta_dict,
    )


def _make_orm_style_asset(duration=VIDEO_DURATION, ranges=None):
    """ORM AssetModel 形态：只有 classification_result JSON 字符串，无 metadata 属性。"""
    a = _make_asset(duration=duration, ranges=ranges)
    del a.metadata
    return a


def _range(start, end, use_count=1):
    return {
        "start": start,
        "end": end,
        "plan_id": "plan-1",
        "created_at": "2026-08-29T10:00:00",
        "use_count": use_count,
        "last_used_at": "2026-08-29T10:00:00",
    }


# ── compute_asset_availability 纯函数 ─────────────────────────────────────────


class TestComputeAssetAvailability:
    def test_no_ranges_fully_usable(self):
        """无历史区间：used=0, ratio=0, usable=True。"""
        info = compute_asset_availability(_make_asset(ranges=[]))
        assert info is not None
        assert info["used_duration"] == 0.0
        assert info["available_duration"] == VIDEO_DURATION
        assert info["used_ratio"] == 0.0
        assert info["usable"] is True

    def test_none_model_returns_none(self):
        assert compute_asset_availability(None) is None

    def test_non_video_returns_none(self):
        """非视频（音频）返回 None，路由层按可用处理。"""
        info = compute_asset_availability(_make_asset(file_type="audio"))
        assert info is None

    def test_zero_duration_returns_none(self):
        info = compute_asset_availability(_make_asset(duration=0.0))
        assert info is None

    def test_partial_usage_usable(self):
        """使用 10s，剩余 50s 空闲（≥3s），usable=True。"""
        info = compute_asset_availability(_make_asset(ranges=[_range(5.0, 15.0)]))
        assert info["used_duration"] == pytest.approx(10.0, abs=0.01)
        assert info["available_duration"] == pytest.approx(50.0, abs=0.01)
        assert info["used_ratio"] == pytest.approx(10.0 / 60.0, abs=0.001)
        assert info["usable"] is True

    def test_overlapping_ranges_merged(self):
        """重叠区间合并后计算 used_duration，不重复计时。"""
        info = compute_asset_availability(_make_asset(ranges=[_range(0.0, 10.0), _range(5.0, 20.0)]))
        # 合并后 [0,20] → 20s
        assert info["used_duration"] == pytest.approx(20.0, abs=0.01)
        assert info["used_ratio"] == pytest.approx(20.0 / 60.0, abs=0.001)

    def test_full_coverage_but_reusable(self):
        """区间铺满全片（无空闲段），但 use_count 未达上限 → usable=True（受控复用）。"""
        info = compute_asset_availability(
            _make_asset(
                duration=10.0,
                ranges=[_range(0.0, 10.0, use_count=1)],
            )
        )
        assert info["used_duration"] == pytest.approx(10.0, abs=0.01)
        assert info["available_duration"] == 0.0
        assert info["usable"] is True

    def test_exhausted_not_usable(self):
        """无空闲段 且 所有区间 use_count 达上限 → usable=False。"""
        info = compute_asset_availability(
            _make_asset(
                duration=10.0,
                ranges=[_range(0.0, 10.0, use_count=MAX_RANGE_USE_COUNT)],
            )
        )
        assert info["usable"] is False
        assert info["available_duration"] == 0.0
        assert info["used_ratio"] == pytest.approx(1.0, abs=0.001)

    def test_exhausted_multiple_ranges_all_capped(self):
        """多个区间铺满、全部达上限 → usable=False；任一未满即 usable=True。"""
        info_capped = compute_asset_availability(
            _make_asset(
                duration=20.0,
                ranges=[
                    _range(0.0, 10.0, use_count=MAX_RANGE_USE_COUNT),
                    _range(10.0, 20.0, use_count=MAX_RANGE_USE_COUNT),
                ],
            )
        )
        assert info_capped["usable"] is False

        info_partial = compute_asset_availability(
            _make_asset(
                duration=20.0,
                ranges=[
                    _range(0.0, 10.0, use_count=MAX_RANGE_USE_COUNT),
                    _range(10.0, 20.0, use_count=MAX_RANGE_USE_COUNT - 1),
                ],
            )
        )
        assert info_partial["usable"] is True

    def test_edge_gap_consumed_not_usable(self):
        """区间未物理铺满，但扩边（+0.3s）后空闲段 <3s → 视为无空闲段；
        区间 use_count 均达上限 → usable=False。"""
        # 10s 素材：[0, 4.0] 与 [4.6, 10]，物理空闲 [4.0,4.6] 仅 0.6s，
        # 扩边后左区间延至 4.3、右区间起于 4.3，空闲被吃掉
        info = compute_asset_availability(
            _make_asset(
                duration=10.0,
                ranges=[
                    _range(0.0, 4.0, use_count=MAX_RANGE_USE_COUNT),
                    _range(4.6, 10.0, use_count=MAX_RANGE_USE_COUNT),
                ],
            )
        )
        assert info["usable"] is False

    def test_large_gap_remains_usable(self):
        """区间之间留有 ≥3s 空闲段（扩边后仍 ≥3s）→ usable=True。"""
        # [0,2] 扩边到 [0,2.3]，[5.3,10] 扩边前为 [5,10] 扩边起 4.7；空闲 [2.3,4.7]=2.4s <3
        # 改用更大间隙：[0,2] 与 [6,10]，扩边后空闲 [2.3,5.7]=3.4s ≥3
        info = compute_asset_availability(
            _make_asset(
                duration=10.0,
                ranges=[
                    _range(0.0, 2.0, use_count=MAX_RANGE_USE_COUNT),
                    _range(6.0, 10.0, use_count=MAX_RANGE_USE_COUNT),
                ],
            )
        )
        assert info["usable"] is True

    def test_invalid_ranges_skipped(self):
        """脏数据（缺 start/end、end<=start、use_count 非法）不崩溃，合法区间照常计算。"""
        info = compute_asset_availability(
            _make_asset(
                duration=30.0,
                ranges=[
                    {"start": "bad"},
                    {"start": 5.0, "end": 3.0},
                    "junk",
                    _range(0.0, 10.0, use_count="not-a-number"),
                ],
            )
        )
        assert info is not None
        assert info["used_duration"] == pytest.approx(10.0, abs=0.01)
        # use_count 非法按 1 处理 → 未达上限，且空闲段充足
        assert info["usable"] is True

    def test_broken_classification_json_treated_as_unused(self):
        """classification_result 是非法 JSON 时按无历史区间处理。"""
        info = compute_asset_availability(_make_asset(classification_result="not-json{{{"))
        assert info is not None
        assert info["used_duration"] == 0.0
        assert info["usable"] is True

    def test_segment_edge_gap_constant(self):
        """边缘间隙常量为 0.3s（与 MediaKit 冲突检测同口径）。"""
        assert SEGMENT_EDGE_GAP == 0.3

    def test_domain_entity_metadata_dict_form(self):
        """领域实体形态（metadata 为 dict，无 classification_result）也能读到区间。

        真实路由 repository 返回 Asset 实体，区间记录在 metadata dict 里
        （repository 与 ORM classification_result JSON 互转）。
        """
        a = _make_asset(duration=30.0, ranges=[_range(0.0, 12.0)])
        del a.classification_result  # 实体没有该列
        info = compute_asset_availability(a)
        assert info is not None
        assert info["used_duration"] == pytest.approx(12.0, abs=0.01)
        assert info["usable"] is True

    def test_orm_model_classification_result_form(self):
        """ORM AssetModel 形态（只有 classification_result JSON 字符串）正常。"""
        a = _make_orm_style_asset(duration=30.0, ranges=[_range(0.0, 12.0)])
        assert not hasattr(a, "metadata")
        info = compute_asset_availability(a)
        assert info is not None
        assert info["used_duration"] == pytest.approx(12.0, abs=0.01)


# ── 路由层辅助：_asset_availability_fields / _to_asset_response ──────────────


class TestAssetAvailabilityFields:
    def test_video_asset_returns_values(self):
        fields = _asset_availability_fields(_make_asset(ranges=[_range(0.0, 10.0)]))
        assert fields["usable"] is True
        assert fields["used_duration"] == pytest.approx(10.0, abs=0.01)
        assert fields["available_duration"] == pytest.approx(50.0, abs=0.01)
        assert fields["used_ratio"] is not None

    def test_non_video_returns_none_fields_usable_true(self):
        fields = _asset_availability_fields(_make_asset(file_type="audio"))
        assert fields["used_duration"] is None
        assert fields["available_duration"] is None
        assert fields["used_ratio"] is None
        assert fields["usable"] is True

    def test_exception_falls_back_to_zero_impact(self, monkeypatch):
        """compute 抛异常时路由层兜底：None 字段 + usable=True，不影响响应。"""
        import app.api.routes.assets as assets_module

        def _boom(_model):
            raise RuntimeError("unexpected")

        monkeypatch.setattr(assets_module, "compute_asset_availability", _boom)
        fields = _asset_availability_fields(_make_asset())
        assert fields["used_duration"] is None
        assert fields["usable"] is True


class TestToAssetResponseInjectsFields:
    def _storage_stub(self):
        svc = MagicMock()
        svc.get_download_url.return_value = "https://example.com/signed"
        return svc

    def test_video_response_carries_availability_fields(self):
        asset = _make_asset(ranges=[_range(0.0, 12.0)])
        resp = _to_asset_response(asset, storage_service=self._storage_stub())
        assert resp.usable is True
        assert resp.used_duration == pytest.approx(12.0, abs=0.01)
        assert resp.available_duration == pytest.approx(48.0, abs=0.01)
        assert resp.used_ratio == pytest.approx(0.2, abs=0.01)

    def test_exhausted_asset_response_usable_false(self):
        asset = _make_asset(
            duration=10.0,
            ranges=[_range(0.0, 10.0, use_count=MAX_RANGE_USE_COUNT)],
        )
        resp = _to_asset_response(asset, storage_service=self._storage_stub())
        assert resp.usable is False
        assert resp.used_ratio == pytest.approx(1.0, abs=0.001)

    def test_non_video_response_fields_none_usable_true(self):
        asset = _make_asset(file_type="audio")
        resp = _to_asset_response(asset, storage_service=self._storage_stub())
        assert resp.used_duration is None
        assert resp.available_duration is None
        assert resp.used_ratio is None
        assert resp.usable is True


# ── smart_match_assets 结果层过滤 ────────────────────────────────────────────


def _exhausted_asset(asset_id):
    """构造一个 usable=false 的视频素材：10s 铺满、区间 use_count 均达上限。"""
    a = _make_asset(
        duration=10.0,
        ranges=[_range(0.0, 10.0, use_count=MAX_RANGE_USE_COUNT)],
    )
    a.id = asset_id
    a.name = f"exhausted-{asset_id}"
    return a


def _fresh_asset(asset_id, duration=60.0):
    a = _make_asset(duration=duration, ranges=[])
    a.id = asset_id
    a.name = f"fresh-{asset_id}"
    return a


class TestSmartMatchFiltersExhausted:
    def _call(self, assets):
        lib_repo = MagicMock()
        lib_repo.get.return_value = SimpleNamespace(project_id="proj-1")
        asset_repo = MagicMock()
        asset_repo.find_by_library_and_file_type.return_value = assets
        project_repo = MagicMock()
        project = MagicMock()
        project.can_access.return_value = True
        project_repo.find_by_id.return_value = project

        user = SimpleNamespace(id="user-1")
        auth_user = SimpleNamespace(user=user)

        # storage_service 在 _to_asset_response 内 get_storage_service()，patch 掉
        import app.api.routes.assets as assets_module

        svc = MagicMock()
        svc.get_download_url.return_value = "https://example.com/signed"
        original_get_storage = assets_module.get_storage_service
        assets_module.get_storage_service = lambda: svc
        try:
            resp = smart_match_assets(
                SmartMatchRequest(library_id="lib-1", kind="video"),
                authenticated_user=auth_user,
                asset_repository=asset_repo,
                asset_library_repository=lib_repo,
                project_repository=project_repo,
            )
        finally:
            assets_module.get_storage_service = original_get_storage
        return resp

    def test_exhausted_assets_excluded(self):
        """smart-match 结果中 usable=false 的素材被剔除，新鲜素材保留。"""
        assets = [
            _exhausted_asset("a-exhausted-1"),
            _exhausted_asset("a-exhausted-2"),
            _fresh_asset("a-fresh-1"),
        ]
        resp = self._call(assets)
        returned_ids = {item.id for item in resp.items}
        assert "a-fresh-1" in returned_ids
        assert "a-exhausted-1" not in returned_ids
        assert "a-exhausted-2" not in returned_ids
        # total_candidates 是过滤前的候选总数
        assert resp.total_candidates == 3
        # 返回的素材全部 usable=True
        assert all(item.usable for item in resp.items)

    def test_all_exhausted_falls_back_to_all(self):
        """全部素材已用尽时回退保留全部（不返回空——空结果会让前端回退全选，
        反而绕过评分排序；耗尽素材仍可走复用区间）。"""
        assets = [_exhausted_asset("a-ex-1"), _exhausted_asset("a-ex-2")]
        resp = self._call(assets)
        returned_ids = {item.id for item in resp.items}
        assert returned_ids == {"a-ex-1", "a-ex-2"}
        assert resp.total_candidates == 2

    def test_fresh_assets_all_returned(self):
        assets = [_fresh_asset("a-1"), _fresh_asset("a-2")]
        resp = self._call(assets)
        assert len(resp.items) == 2
        assert all(item.usable for item in resp.items)


class TestSmartMatchFlatStructure:
    """P0 回归：smart-match 响应必须扁平——item 顶层直接可读素材字段，
    前端 items.map(a => a.id) 不能再拿到 undefined（此前 item.asset 嵌套包装
    导致 GET /assets/undefined 404 + from-assets 422，自动模式全链路断裂）。"""

    def test_item_id_at_top_level(self):
        """item.id 直接在顶层可读，不存在 item.asset 包装层。"""
        assets = [_fresh_asset("a-flat-1"), _fresh_asset("a-flat-2")]
        resp = TestSmartMatchFiltersExhausted()._call(assets)
        ids = [item.id for item in resp.items]
        assert ids == ["a-flat-1", "a-flat-2"]
        # 嵌套 asset 字段已移除
        assert all(not hasattr(item, "asset") for item in resp.items)

    def test_item_is_asset_response_superset(self):
        """条目携带 AssetResponse 全部关键字段 + usable/余量，前端可直接渲染卡片。"""
        assets = [_fresh_asset("a-fields-1")]
        resp = TestSmartMatchFiltersExhausted()._call(assets)
        item = resp.items[0]
        assert item.id == "a-fields-1"
        assert item.name == "fresh-a-fields-1"
        assert item.storage_key == "key/test-asset.mp4"
        assert item.mime_type == "video/mp4"
        assert item.duration == 60.0
        assert item.status == "ready"
        assert item.thumbnail_url is None or item.thumbnail_url.startswith("http")
        # 余量/可用性字段顶层可读（isAssetUsable 依赖）
        assert item.usable is True
        assert item.used_duration == 0.0
        assert item.available_duration == 60.0
        assert item.used_ratio == 0.0
        # 评分字段保留
        assert 0.0 <= item.score <= 100.0
        assert isinstance(item.breakdown, dict)

    def test_score_and_breakdown_preserved(self):
        """扁平化后评分字段不丢失。"""
        assets = [_fresh_asset("a-score-1")]
        resp = TestSmartMatchFiltersExhausted()._call(assets)
        item = resp.items[0]
        assert isinstance(item.score, float)
        assert item.score > 0
        assert isinstance(item.breakdown, dict) and item.breakdown