xiaoxia-saas/tests/unit/test_default_project_idempotent_1775.py

"""默认项目/默认素材库幂等化测试（Issue #1775）。

覆盖：
- get_or_create_default_project：同用户幂等、不同用户独立、重复调用返回同一个
- get_or_create_default_library：同项目同 kind 幂等、IntegrityError 后重查
- Project.is_default 字段传递
- ensure-default-context 组合逻辑（用内存仓储）
"""

from __future__ import annotations

from unittest.mock import MagicMock

import pytest

from packages.adapters.in_memory.asset_library_repository import InMemoryAssetLibraryRepository
from packages.adapters.in_memory.project_repository import InMemoryProjectRepository
from packages.adapters.sqlalchemy_impl.asset_library_repository import (
    SQLAlchemyAssetLibraryRepository,
)
from packages.adapters.sqlalchemy_impl.project_repository import SQLAlchemyProjectRepository
from packages.domain import AssetLibrary, AssetLibraryKind, Project


class TestDefaultProjectIdempotent:
    """默认项目幂等。"""

    def test_first_call_creates_default(self):
        repo = InMemoryProjectRepository()
        project = repo.get_or_create_default_project("user-1")
        assert project.owner_user_id == "user-1"
        assert project.is_default is True
        assert project.name == "默认项目"

    def test_second_call_returns_same_project(self):
        repo = InMemoryProjectRepository()
        p1 = repo.get_or_create_default_project("user-1")
        p2 = repo.get_or_create_default_project("user-1")
        assert p1.id == p2.id

    def test_different_users_independent(self):
        repo = InMemoryProjectRepository()
        p1 = repo.get_or_create_default_project("user-1")
        p2 = repo.get_or_create_default_project("user-2")
        assert p1.id != p2.id
        assert p1.owner_user_id == "user-1"
        assert p2.owner_user_id == "user-2"

    def test_find_default_by_owner(self):
        repo = InMemoryProjectRepository()
        created = repo.get_or_create_default_project("user-1")
        found = repo.find_default_by_owner("user-1")
        assert found is not None
        assert found.id == created.id

    def test_find_default_returns_none_when_no_default(self):
        repo = InMemoryProjectRepository()
        # 手动建一个非默认项目
        normal = Project.create(owner_user_id="user-1", name="普通项目")
        normal.is_default = False
        repo.save(normal)
        assert repo.find_default_by_owner("user-1") is None

    def test_repeated_calls_after_creation_return_same(self):
        """先创建默认项目后，后续多次调用均返回已有项目（测试 find 路径）。
        注：真正的并发保护依赖 PostgreSQL partial unique index，
        InMemory 仓储不做并发测试（无 DB 约束），并发场景由
        TestSqlRepoIntegrityErrorRecovery 通过 SQLAlchemy + SQLite 验证。
        """
        repo = InMemoryProjectRepository()
        first = repo.get_or_create_default_project("user-repeat")
        for _ in range(9):
            again = repo.get_or_create_default_project("user-repeat")
            assert again.id == first.id
        defaults = [p for p in repo.find_by_owner_user_id("user-repeat") if p.is_default]
        assert len(defaults) == 1


class TestDefaultLibraryIdempotent:
    """默认素材库幂等。"""

    def test_first_call_creates(self):
        repo = InMemoryAssetLibraryRepository()
        lib = repo.get_or_create_default_library("proj-1", AssetLibraryKind.VIDEO)
        assert lib.project_id == "proj-1"
        assert lib.kind == AssetLibraryKind.VIDEO

    def test_second_call_returns_same(self):
        repo = InMemoryAssetLibraryRepository()
        l1 = repo.get_or_create_default_library("proj-1", AssetLibraryKind.VIDEO)
        l2 = repo.get_or_create_default_library("proj-1", AssetLibraryKind.VIDEO)
        assert l1.id == l2.id

    def test_different_kinds_independent(self):
        repo = InMemoryAssetLibraryRepository()
        v = repo.get_or_create_default_library("proj-1", AssetLibraryKind.VIDEO)
        a = repo.get_or_create_default_library("proj-1", AssetLibraryKind.VOICE)
        i = repo.get_or_create_default_library("proj-1", AssetLibraryKind.IMAGE)
        assert len({v.id, a.id, i.id}) == 3

    def test_different_projects_independent(self):
        repo = InMemoryAssetLibraryRepository()
        l1 = repo.get_or_create_default_library("proj-1", AssetLibraryKind.VIDEO)
        l2 = repo.get_or_create_default_library("proj-2", AssetLibraryKind.VIDEO)
        assert l1.id != l2.id

    def test_repeated_calls_after_creation_return_same(self):
        """先创建后多次调用均返回同一素材库（测试 find 路径）。"""
        repo = InMemoryAssetLibraryRepository()
        first = repo.get_or_create_default_library("proj-repeat", AssetLibraryKind.VOICE)
        for _ in range(9):
            again = repo.get_or_create_default_library("proj-repeat", AssetLibraryKind.VOICE)
            assert again.id == first.id


class TestProjectIsDefaultField:
    """Project.is_default 字段语义。"""

    def test_create_non_default_by_default(self):
        p = Project.create(owner_user_id="u", name="普通项目")
        assert p.is_default is False

    def test_create_default(self):
        p = Project.create(owner_user_id="u", name="默认项目", is_default=True)
        assert p.is_default is True


class TestSqlRepoIntegrityErrorRecovery:
    """SQLAlchemy 仓储：唯一约束冲突时回滚重查，返回已有记录（不报 500）。"""

    def test_project_integrity_error_returns_existing(self):
        from sqlalchemy.exc import IntegrityError

        session = MagicMock()
        # commit 第一次抛 IntegrityError（并发冲突），回滚后查询返回已有项目
        existing_model = MagicMock()
        existing_model.id = "existing-id"
        existing_model.owner_user_id = "user-1"
        existing_model.name = "默认项目"
        existing_model.description = ""
        existing_model.shared_users = []
        existing_model.is_default = True
        existing_model.created_at = None

        session.commit.side_effect = [IntegrityError("stmt", {}, Exception("dup")), None]
        # 第一次 query（快速路径 find_default）返回 None；rollback 后第二次返回 existing
        session.query.return_value.filter.return_value.first.side_effect = [None, existing_model]

        repo = SQLAlchemyProjectRepository(session)
        result = repo.get_or_create_default_project("user-1")

        assert result.id == "existing-id"
        session.rollback.assert_called_once()

    def test_library_integrity_error_returns_existing(self):
        from sqlalchemy.exc import IntegrityError

        session = MagicMock()
        existing_model = MagicMock()
        existing_model.id = "lib-existing"
        existing_model.project_id = "proj-1"
        existing_model.name = "视频素材库"
        existing_model.kind = "video"
        existing_model.asset_count = 0
        existing_model.total_size = 0
        existing_model.created_at = None
        existing_model.updated_at = None

        session.commit.side_effect = [IntegrityError("stmt", {}, Exception("dup")), None]
        session.query.return_value.filter.return_value.first.side_effect = [None, existing_model]

        repo = SQLAlchemyAssetLibraryRepository(session)
        result = repo.get_or_create_default_library("proj-1", AssetLibraryKind.VIDEO)

        assert result.id == "lib-existing"
        assert result.kind == AssetLibraryKind.VIDEO
        session.rollback.assert_called_once()