xiaoxia-saas/tests/unit/test_asset_count_sync_1776.py

"""Issue #1776: asset_libraries.asset_count 同步维护测试。

覆盖场景:
1. 素材创建 → count +1
2. 素材硬删除 → count -1
3. 素材软删除（batch_delete）→ count -N
4. 幂等上传（prepare 占位 + complete 复用）→ 不重复计数
5. 重试场景（complete 重试）→ 不重复计数
6. recount 方法修正计数
"""

import uuid
from datetime import datetime, timezone

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl.asset_library_repository import SQLAlchemyAssetLibraryRepository
from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
from packages.adapters.sqlalchemy_impl.models import AssetLibraryModel, AssetModel, Base
from packages.domain import Asset, AssetLibrary, AssetLibraryKind, AssetStatus


@pytest.fixture
def db_session():
    """创建测试用 SQLite 内存数据库。"""
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    session = Session(engine)
    yield session
    session.close()


@pytest.fixture
def asset_repo(db_session):
    return SQLAlchemyAssetRepository(db_session)


@pytest.fixture
def library_repo(db_session):
    return SQLAlchemyAssetLibraryRepository(db_session)


def _make_library(project_id: str, kind: AssetLibraryKind = AssetLibraryKind.VIDEO) -> AssetLibrary:
    return AssetLibrary.create(project_id=project_id, name=f"测试{kind.value}库", kind=kind)


def _make_asset(
    library_id: str,
    project_id: str,
    *,
    status: AssetStatus = AssetStatus.PROCESSING,
    file_size: int = 1024,
    file_hash: str = "",
    client_upload_id: str = "",
) -> Asset:
    return Asset.create(
        project_id=project_id,
        library_id=library_id,
        name=f"test_{uuid.uuid4().hex[:8]}.mp4",
        storage_key=f"uploads/{uuid.uuid4().hex[:8]}/test.mp4",
        mime_type="video/mp4",
        status=status,
        uploaded_by_user_id="test-user",
        file_hash=file_hash,
        client_upload_id=client_upload_id,
        file_size=file_size,
    )


class TestAssetCountOnCreate:
    """素材创建时计数递增。"""

    def test_create_asset_increments_count(self, library_repo, asset_repo, db_session):
        """创建一个素材 → count 从 0 变 1。"""
        library = library_repo.create(_make_library("project-1"))
        assert library.asset_count == 0

        asset = _make_asset(library.id, "project-1")
        asset_repo.create(asset)

        # 重新查询验证计数
        updated_library = library_repo.get(library.id)
        assert updated_library.asset_count == 1
        assert updated_library.total_size == 1024

    def test_create_multiple_assets_increments_count(self, library_repo, asset_repo, db_session):
        """创建多个素材 → count 累加。"""
        library = library_repo.create(_make_library("project-1"))

        for _i in range(3):
            asset = _make_asset(library.id, "project-1", file_size=100 * (_i + 1))
            asset_repo.create(asset)

        updated_library = library_repo.get(library.id)
        assert updated_library.asset_count == 3
        assert updated_library.total_size == 100 + 200 + 300


class TestAssetCountOnDelete:
    """素材删除时计数递减。"""

    def test_hard_delete_decrements_count(self, library_repo, asset_repo, db_session):
        """硬删除素材 → count -1。"""
        library = library_repo.create(_make_library("project-1"))
        asset = asset_repo.create(_make_asset(library.id, "project-1"))
        assert library_repo.get(library.id).asset_count == 1

        asset_repo.delete(asset.id)

        assert library_repo.get(library.id).asset_count == 0

    def test_hard_delete_already_deleted_no_change(self, library_repo, asset_repo, db_session):
        """删除已删除的素材 → count 不变。"""
        library = library_repo.create(_make_library("project-1"))
        asset = asset_repo.create(_make_asset(library.id, "project-1"))
        # 先软删除（count 已经 -1）
        asset_repo.batch_delete([asset.id])
        assert library_repo.get(library.id).asset_count == 0

        # 再硬删除（不应再 -1）
        asset_repo.delete(asset.id)
        assert library_repo.get(library.id).asset_count == 0

    def test_batch_delete_decrements_count(self, library_repo, asset_repo, db_session):
        """批量软删除 → count -N。"""
        library = library_repo.create(_make_library("project-1"))
        asset_ids = []
        for _i in range(5):
            asset = asset_repo.create(_make_asset(library.id, "project-1", file_size=200))
            asset_ids.append(asset.id)
        assert library_repo.get(library.id).asset_count == 5

        # 删除 3 个
        deleted_count = asset_repo.batch_delete(asset_ids[:3])
        assert deleted_count == 3
        assert library_repo.get(library.id).asset_count == 2
        assert library_repo.get(library.id).total_size == 200 * 2

    def test_batch_delete_skips_already_deleted(self, library_repo, asset_repo, db_session):
        """批量删除已删除的素材 → count 不变。"""
        library = library_repo.create(_make_library("project-1"))
        asset_ids = []
        for _i in range(3):
            asset = asset_repo.create(_make_asset(library.id, "project-1"))
            asset_ids.append(asset.id)
        assert library_repo.get(library.id).asset_count == 3

        # 先删除 2 个
        asset_repo.batch_delete(asset_ids[:2])
        assert library_repo.get(library.id).asset_count == 1

        # 再删除同样的 2 个（应被跳过）
        deleted_count = asset_repo.batch_delete(asset_ids[:2])
        assert deleted_count == 0
        assert library_repo.get(library.id).asset_count == 1

    def test_count_never_negative(self, library_repo, asset_repo, db_session):
        """计数下限为 0，不会出现负数。"""
        library = library_repo.create(_make_library("project-1"))
        # 手动设置计数为 0
        library.asset_count = 0
        library_repo.update(library)

        # 尝试递减（通过直接调用 decrement）
        library_repo.decrement_asset_count(library.id, count_delta=5)
        db_session.commit()

        updated = library_repo.get(library.id)
        assert updated.asset_count == 0


class TestIdempotentUpload:
    """幂等上传场景：不重复计数。"""

    def test_prepare_then_complete_no_double_count(self, library_repo, asset_repo, db_session):
        """prepare 创建占位 + complete 复用占位 → count 只 +1。"""
        library = library_repo.create(_make_library("project-1"))

        # prepare 阶段：创建占位
        placeholder = _make_asset(
            library.id,
            "project-1",
            status=AssetStatus.PROCESSING,
            client_upload_id="upload-token-123",
        )
        asset_repo.create(placeholder)
        assert library_repo.get(library.id).asset_count == 1

        # complete 阶段：查找已有占位并复用（通过 client_upload_id）
        existing = asset_repo.find_by_library_and_client_upload_id(
            library_id=library.id,
            client_upload_id="upload-token-123",
        )
        assert existing is not None
        # 复用占位，不创建新记录 → count 不变
        assert library_repo.get(library.id).asset_count == 1

    def test_complete_retry_no_double_count(self, library_repo, asset_repo, db_session):
        """complete 重试（通过 file_hash 去重）→ count 只 +1。"""
        library = library_repo.create(_make_library("project-1"))

        # 第一次 complete：创建素材
        asset1 = _make_asset(
            library.id,
            "project-1",
            file_hash="hash-abc-123",
        )
        asset_repo.create(asset1)
        assert library_repo.get(library.id).asset_count == 1

        # 重试 complete：通过 file_hash 查找已有
        existing = asset_repo.find_by_library_and_file_hash(
            library_id=library.id,
            file_hash="hash-abc-123",
        )
        assert existing is not None
        assert existing.id == asset1.id
        # 不创建新记录 → count 不变
        assert library_repo.get(library.id).asset_count == 1


class TestRecountAssets:
    """recount_assets 方法修正计数。"""

    def test_recount_fixes_drift(self, library_repo, asset_repo, db_session):
        """计数漂移后，recount 能修正。"""
        library = library_repo.create(_make_library("project-1"))
        # 创建 3 个素材
        for _ in range(3):
            asset_repo.create(_make_asset(library.id, "project-1"))
        assert library_repo.get(library.id).asset_count == 3

        # 手动破坏计数（模拟历史数据问题）
        db_session.query(AssetLibraryModel).filter(AssetLibraryModel.id == library.id).update(
            {AssetLibraryModel.asset_count: 999, AssetLibraryModel.total_size: 999999}
        )
        db_session.commit()
        assert library_repo.get(library.id).asset_count == 999

        # recount 修正
        actual = library_repo.recount_assets(library.id)
        db_session.commit()

        assert actual == 3
        assert library_repo.get(library.id).asset_count == 3
        assert library_repo.get(library.id).total_size == 1024 * 3

    def test_recount_excludes_deleted(self, library_repo, asset_repo, db_session):
        """recount 排除已删除素材。"""
        library = library_repo.create(_make_library("project-1"))
        assets = []
        for _ in range(5):
            asset = asset_repo.create(_make_asset(library.id, "project-1"))
            assets.append(asset)
        assert library_repo.get(library.id).asset_count == 5

        # 软删除 2 个
        asset_repo.batch_delete([assets[0].id, assets[1].id])
        assert library_repo.get(library.id).asset_count == 3

        # 破坏计数
        db_session.query(AssetLibraryModel).filter(AssetLibraryModel.id == library.id).update(
            {AssetLibraryModel.asset_count: 100}
        )
        db_session.commit()

        # recount 应排除 deleted
        actual = library_repo.recount_assets(library.id)
        db_session.commit()

        assert actual == 3
        assert library_repo.get(library.id).asset_count == 3


class TestConcurrentSafety:
    """并发安全测试（SQLite 模拟有限并发）。"""

    def test_increment_is_atomic(self, library_repo, db_session):
        """increment_asset_count 使用 SQL 级 UPDATE，并发安全。"""
        library = library_repo.create(_make_library("project-1"))

        # 多次递增
        for _ in range(10):
            library_repo.increment_asset_count(library.id, count_delta=1, size_delta=100)
        db_session.commit()

        updated = library_repo.get(library.id)
        assert updated.asset_count == 10
        assert updated.total_size == 1000

    def test_decrement_with_floor_zero(self, library_repo, db_session):
        """decrement_asset_count 下限为 0。"""
        library = library_repo.create(_make_library("project-1"))
        library.asset_count = 3
        library_repo.update(library)

        # 尝试递减 10 次
        for _ in range(10):
            library_repo.decrement_asset_count(library.id, count_delta=1)
        db_session.commit()

        updated = library_repo.get(library.id)
        assert updated.asset_count == 0