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 对象。

    ranges: list of dicts（used_time_ranges 条目），会自动写入 classification_result JSON。
    """
    if classification_result is None and ranges is not None:
        classification_result = json.dumps({"used_time_ranges": ranges})
    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",
        metadata={},
        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,
        classification_result=classification_result,
    )


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


# ── 路由层辅助：_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.asset.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.asset.usable for item in resp.items)

    def test_all_exhausted_returns_empty(self):
        """全部素材已用尽时返回空列表（不报错，前端显示空结果）。"""
        assets = [_exhausted_asset("a-ex-1"), _exhausted_asset("a-ex-2")]
        resp = self._call(assets)
        assert resp.items == []
        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.asset.usable for item in resp.items)