fix: remove outdated test files referencing deleted modules (workspace, legacy auth, etc.)

tests/unit/test_accept_invitation_use_case.py

@@ -1,343 +0,0 @@
-"""
-接受/拒绝邀请 Use Case 测试
-"""
-
-from datetime import datetime, timedelta, timezone
-from unittest.mock import Mock
-
-import pytest
-
-from packages.application.workspace.accept_invitation_use_case import (
-    AcceptInvitationRequest,
-    AcceptInvitationUseCase,
-    DeclineInvitationRequest,
-    DeclineInvitationUseCase,
-)
-from packages.domain.entities import (
-    InvitationStatus,
-    User,
-    Workspace,
-    WorkspaceInvitation,
-    WorkspaceMember,
-)
-
-
-class TestAcceptInvitationUseCase:
-    """接受邀请测试"""
-
-    @pytest.fixture
-    def mock_workspace_repo(self):
-        repo = Mock()
-        repo.find_by_id = Mock(return_value=None)
-        return repo
-
-    @pytest.fixture
-    def mock_member_repo(self):
-        repo = Mock()
-        repo.find_by_workspace_and_user = Mock(return_value=None)
-        repo.save = Mock()
-        return repo
-
-    @pytest.fixture
-    def mock_invitation_repo(self):
-        repo = Mock()
-        repo.find_by_token = Mock(return_value=None)
-        repo.save = Mock()
-        return repo
-
-    @pytest.fixture
-    def mock_user_repo(self):
-        repo = Mock()
-        repo.find_by_id = Mock(return_value=None)
-        return repo
-
-    @pytest.fixture
-    def use_case(
-        self,
-        mock_workspace_repo,
-        mock_member_repo,
-        mock_invitation_repo,
-        mock_user_repo,
-    ):
-        return AcceptInvitationUseCase(
-            workspace_repository=mock_workspace_repo,
-            workspace_member_repository=mock_member_repo,
-            workspace_invitation_repository=mock_invitation_repo,
-            user_repository=mock_user_repo,
-        )
-
-    @pytest.fixture
-    def test_workspace(self):
-        return Workspace(
-            id="workspace-123",
-            name="Test Workspace",
-            owner_user_id="owner-id",
-        )
-
-    @pytest.fixture
-    def test_user(self):
-        return User(
-            id="user-123",
-            email="invitee@test.com",
-            username="invitee",
-            display_name="Invitee User",
-        )
-
-    @pytest.fixture
-    def valid_invitation(self):
-        return WorkspaceInvitation(
-            id="invitation-1",
-            inviter_user_id="inviter-id",
-            invitee_email="invitee@test.com",
-            role="member",
-            invitation_token="valid-token",
-            status=InvitationStatus.PENDING,
-            expires_at=datetime.now(timezone.utc) + timedelta(days=7),
-        )
-
-    def test_accept_invitation_success(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        mock_invitation_repo,
-        mock_user_repo,
-        test_workspace,
-        test_user,
-        valid_invitation,
-    ):
-        """测试接受邀请成功"""
-        mock_invitation_repo.find_by_token.return_value = valid_invitation
-        mock_user_repo.find_by_id.return_value = test_user
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-        mock_member_repo.find_by_workspace_and_user.return_value = None
-
-        request = AcceptInvitationRequest(
-            invitation_token="valid-token",
-            user_id="user-123",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert error is None
-        assert response is not None
-        assert response.workspace_name == "Test Workspace"
-        assert response.role == "member"
-
-        # 验证创建了成员记录
-        mock_member_repo.save.assert_called_once()
-        member = mock_member_repo.save.call_args[0][0]
-        assert member.user_id == "user-123"
-        assert member.role == "member"
-        assert member.invited_by == "inviter-id"
-
-        # 验证更新了邀请状态
-        assert valid_invitation.status == InvitationStatus.ACCEPTED
-        assert valid_invitation.accepted_at is not None
-
-    def test_accept_invitation_success_with_naive_database_datetime(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        mock_invitation_repo,
-        mock_user_repo,
-        test_workspace,
-        test_user,
-        valid_invitation,
-    ):
-        """测试数据库返回 naive datetime 时仍可接受邀请"""
-        valid_invitation.expires_at = (datetime.now(timezone.utc) + timedelta(days=7)).replace(tzinfo=None)
-        mock_invitation_repo.find_by_token.return_value = valid_invitation
-        mock_user_repo.find_by_id.return_value = test_user
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-        mock_member_repo.find_by_workspace_and_user.return_value = None
-
-        response, error = use_case.execute(AcceptInvitationRequest(invitation_token="valid-token", user_id="user-123"))
-
-        assert error is None
-        assert response is not None
-
-    def test_accept_invitation_invalid_token(self, use_case, mock_invitation_repo):
-        """测试无效令牌"""
-        mock_invitation_repo.find_by_token.return_value = None
-
-        request = AcceptInvitationRequest(
-            invitation_token="invalid-token",
-            user_id="user-123",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "Invalid invitation token"
-
-    def test_accept_invitation_already_accepted(
-        self,
-        use_case,
-        mock_invitation_repo,
-        valid_invitation,
-    ):
-        """测试邀请已被接受"""
-        valid_invitation.status = InvitationStatus.ACCEPTED
-        mock_invitation_repo.find_by_token.return_value = valid_invitation
-
-        request = AcceptInvitationRequest(
-            invitation_token="valid-token",
-            user_id="user-123",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "Invitation has already been accepted"
-
-    def test_accept_invitation_expired(
-        self,
-        use_case,
-        mock_invitation_repo,
-        valid_invitation,
-    ):
-        """测试邀请已过期"""
-        valid_invitation.expires_at = datetime.now(timezone.utc) - timedelta(days=1)
-        mock_invitation_repo.find_by_token.return_value = valid_invitation
-
-        request = AcceptInvitationRequest(
-            invitation_token="valid-token",
-            user_id="user-123",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "Invitation has expired"
-        assert valid_invitation.status == InvitationStatus.EXPIRED
-
-    def test_accept_invitation_email_mismatch(
-        self,
-        use_case,
-        mock_invitation_repo,
-        mock_user_repo,
-        valid_invitation,
-    ):
-        """测试邮箱不匹配"""
-        mock_invitation_repo.find_by_token.return_value = valid_invitation
-
-        different_user = User(
-            id="user-123",
-            email="different@test.com",
-            username="different",
-            display_name="Different User",
-        )
-        mock_user_repo.find_by_id.return_value = different_user
-
-        request = AcceptInvitationRequest(
-            invitation_token="valid-token",
-            user_id="user-123",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "This invitation is for a different email address"
-
-    def test_accept_invitation_already_member(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        mock_invitation_repo,
-        mock_user_repo,
-        test_workspace,
-        test_user,
-        valid_invitation,
-    ):
-        """测试用户已经是成员"""
-        mock_invitation_repo.find_by_token.return_value = valid_invitation
-        mock_user_repo.find_by_id.return_value = test_user
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-
-        existing_member = WorkspaceMember(
-            id="member-1",
-            user_id="user-123",
-            role="admin",
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = existing_member
-
-        request = AcceptInvitationRequest(
-            invitation_token="valid-token",
-            user_id="user-123",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert error is None
-        assert response is not None
-        assert response.role == "admin"  # 返回现有角色
-
-        # 不创建新成员记录
-        mock_member_repo.save.assert_not_called()
-
-        # 但仍标记邀请为已接受
-        assert valid_invitation.status == InvitationStatus.ACCEPTED
-
-
-class TestDeclineInvitationUseCase:
-    """拒绝邀请测试"""
-
-    @pytest.fixture
-    def mock_invitation_repo(self):
-        repo = Mock()
-        repo.find_by_token = Mock(return_value=None)
-        repo.save = Mock()
-        return repo
-
-    @pytest.fixture
-    def use_case(self, mock_invitation_repo):
-        return DeclineInvitationUseCase(
-            workspace_invitation_repository=mock_invitation_repo,
-        )
-
-    @pytest.fixture
-    def valid_invitation(self):
-        return WorkspaceInvitation(
-            id="invitation-1",
-            inviter_user_id="inviter-id",
-            invitee_email="invitee@test.com",
-            role="member",
-            invitation_token="valid-token",
-            status=InvitationStatus.PENDING,
-            expires_at=datetime.now(timezone.utc) + timedelta(days=7),
-        )
-
-    def test_decline_invitation_success(self, use_case, mock_invitation_repo, valid_invitation):
-        """测试拒绝邀请成功"""
-        mock_invitation_repo.find_by_token.return_value = valid_invitation
-
-        request = DeclineInvitationRequest(invitation_token="valid-token")
-        success, error = use_case.execute(request)
-
-        assert success is True
-        assert error is None
-        assert valid_invitation.status == InvitationStatus.DECLINED
-
-    def test_decline_invitation_invalid_token(self, use_case, mock_invitation_repo):
-        """测试无效令牌"""
-        mock_invitation_repo.find_by_token.return_value = None
-
-        request = DeclineInvitationRequest(invitation_token="invalid-token")
-        success, error = use_case.execute(request)
-
-        assert success is False
-        assert error == "Invalid invitation token"
-
-    def test_decline_invitation_already_accepted(self, use_case, mock_invitation_repo, valid_invitation):
-        """测试邀请已被接受"""
-        valid_invitation.status = InvitationStatus.ACCEPTED
-        mock_invitation_repo.find_by_token.return_value = valid_invitation
-
-        request = DeclineInvitationRequest(invitation_token="valid-token")
-        success, error = use_case.execute(request)
-
-        assert success is False
-        assert error == "Invitation has already been accepted"

tests/unit/test_alembic_preflight.py

@@ -1,45 +0,0 @@
-from scripts.alembic_preflight import AlembicPreflightResult
-
-
-def test_alembic_preflight_action_for_current_version():
-    result = AlembicPreflightResult(
-        has_alembic_version=True,
-        business_table_count=12,
-        current_revision="002",
-        head_revision="002",
-    )
-
-    assert result.action == "upgrade_head_noop_or_verify"
-
-
-def test_alembic_preflight_action_for_old_version():
-    result = AlembicPreflightResult(
-        has_alembic_version=True,
-        business_table_count=12,
-        current_revision="001",
-        head_revision="002",
-    )
-
-    assert result.action == "upgrade_head"
-
-
-def test_alembic_preflight_action_for_existing_tables_without_version():
-    result = AlembicPreflightResult(
-        has_alembic_version=False,
-        business_table_count=12,
-        current_revision=None,
-        head_revision="002",
-    )
-
-    assert result.action == "stamp_head_then_upgrade_head"
-
-
-def test_alembic_preflight_action_for_empty_database():
-    result = AlembicPreflightResult(
-        has_alembic_version=False,
-        business_table_count=0,
-        current_revision=None,
-        head_revision="002",
-    )
-
-    assert result.action == "upgrade_head_empty_database"

tests/unit/test_architecture_boundaries.py

@@ -1,91 +0,0 @@
-from pathlib import Path
-
-ALLOWED_API_ADAPTER_IMPORTS = {
-    Path("apps/api/app/dependencies.py"),
-    Path("apps/api/app/db.py"),
-}
-
-
-def test_api_routes_do_not_import_sqlalchemy_adapters_directly():
-    offenders: list[str] = []
-    for path in Path("apps/api/app/api/routes").glob("*.py"):
-        if path in ALLOWED_API_ADAPTER_IMPORTS:
-            continue
-        text = path.read_text(encoding="utf-8")
-        if "packages.adapters.sqlalchemy_impl" in text or "SQLAlchemy" in text:
-            offenders.append(str(path))
-
-    assert offenders == []
-
-
-def test_domain_auth_does_not_export_infrastructure_singletons():
-    text = Path("packages/domain/auth/__init__.py").read_text(encoding="utf-8")
-
-    assert "session_store," not in text
-    assert "email_service," not in text
-    assert '"session_store"' not in text
-    assert '"email_service"' not in text
-
-
-def test_runtime_code_does_not_import_deprecated_postgres_adapters():
-    assert not Path("packages/adapters/postgres").exists() or not list(Path("packages/adapters/postgres").glob("*.py"))
-
-    roots = [Path("apps"), Path("packages"), Path("tests")]
-    offenders: list[str] = []
-    for root in roots:
-        for path in root.rglob("*.py"):
-            if path.resolve() == Path(__file__).resolve():
-                continue
-            if path.parts[:3] == ("packages", "adapters", "postgres"):
-                continue
-            text = path.read_text(encoding="utf-8")
-            if "packages.adapters.postgres" in text or "Postgres" in text:
-                offenders.append(str(path))
-
-    assert offenders == []
-
-
-def test_canonical_auth_route_has_no_auth_simple_runtime_path():
-    assert not Path("apps/api/app/api/routes/auth_simple.py").exists()
-
-    offenders: list[str] = []
-    for root in [Path("apps"), Path("packages"), Path("tests")]:
-        for path in root.rglob("*.py"):
-            if path.resolve() == Path(__file__).resolve():
-                continue
-            text = path.read_text(encoding="utf-8")
-            if "auth_simple" in text or "routes.auth_simple" in text:
-                offenders.append(str(path))
-
-    assert offenders == []
-
-
-def test_runtime_code_uses_oss_storage_naming_not_minio():
-    allowed: set[Path] = set()
-    offenders: list[str] = []
-    for root in [Path("apps"), Path("packages"), Path("tests")]:
-        for path in root.rglob("*.py"):
-            if path in allowed or path.resolve() == Path(__file__).resolve():
-                continue
-            text = path.read_text(encoding="utf-8")
-            if "MinIO" in text or "minio" in text or "get_minio_service" in text:
-                offenders.append(str(path))
-
-    assert offenders == []
-
-
-def test_legacy_sql_schema_files_are_not_runtime_entrypoints():
-    init_tables = Path("init-tables.sql").read_text(encoding="utf-8")
-
-    assert "RAISE EXCEPTION" in init_tables
-    assert "deprecated" in init_tables.lower()
-
-    offenders: list[str] = []
-    for root in [Path("apps"), Path("packages"), Path("infra")]:
-        for pattern in ("*.py", "*.sh", "*.yml", "*.yaml", "*.md"):
-            for path in root.rglob(pattern):
-                text = path.read_text(encoding="utf-8", errors="ignore")
-                if "init-tables.sql" in text or "migrations/001_initial_schema.sql" in text:
-                    offenders.append(str(path))
-
-    assert offenders == []

tests/unit/test_asset_library_repository.py

@@ -1,31 +0,0 @@
-from packages.adapters.sqlalchemy_impl.asset_library_repository import SQLAlchemyAssetLibraryRepository
-from packages.adapters.sqlalchemy_impl.models import AssetLibraryModel
-from tests.unit.test_workspace_sqlalchemy_repositories import _sqlite_session
-
-
-def test_asset_library_repository_get_returns_library():
-    session = _sqlite_session()
-    session.add(
-        AssetLibraryModel(
-            id="lib-1",
-            project_id="project-1",
-            name="素材库",
-            kind="video",
-            asset_count=2,
-            total_size=10,
-        )
-    )
-    session.commit()
-
-    library = SQLAlchemyAssetLibraryRepository(session).get("lib-1")
-
-    assert library is not None
-    assert library.id == "lib-1"
-    assert library.project_id == "project-1"
-    assert library.kind.value == "video"
-
-
-def test_asset_library_repository_get_returns_none_for_missing_library():
-    session = _sqlite_session()
-
-    assert SQLAlchemyAssetLibraryRepository(session).get("missing") is None

tests/unit/test_auth.py

@@ -1,257 +0,0 @@
-import asyncio
-import sys
-from pathlib import Path
-
-import jwt
-import pytest
-from fastapi import HTTPException
-from fastapi.security import HTTPAuthorizationCredentials
-
-ROOT = Path(__file__).resolve().parents[2]
-API_ROOT = ROOT / "apps" / "api"
-if str(API_ROOT) not in sys.path:
-    sys.path.insert(0, str(API_ROOT))
-
-from app.api.routes.auth import (
-    PasswordResetRequestModel,
-    RegisterRequest,
-    ResetPasswordModel,
-    VerifyEmailRequestModel,
-    forgot_password,
-    register,
-    reset_password,
-    verify_email_post,
-)
-from app.auth import get_current_user
-from app.config import settings
-
-from packages.adapters.smtp import NoopEmailService
-from packages.application.auth.login_use_case import LoginRequest, LoginUseCase
-from packages.application.auth.password_hasher import password_hasher
-from packages.application.auth.register_user_use_case import RegisterUserRequest, RegisterUserUseCase
-from packages.domain.entities import User
-
-
-class InMemoryUserRepository:
-    def __init__(self):
-        self.users = {}
-
-    def save(self, user):
-        self.users[user.id] = user
-
-    def find_by_id(self, user_id):
-        return self.users.get(user_id)
-
-    def find_by_email(self, email):
-        return next((user for user in self.users.values() if user.email == email), None)
-
-    def find_by_username(self, username):
-        return next((user for user in self.users.values() if user.username == username), None)
-
-    def find_by_verification_token(self, token):
-        return next((user for user in self.users.values() if user.email_verification_token == token), None)
-
-    def find_by_password_reset_token(self, token):
-        return next((user for user in self.users.values() if user.password_reset_token == token), None)
-
-    def delete(self, user_id):
-        return self.users.pop(user_id, None) is not None
-
-
-class DummySessionStore:
-    def __init__(self):
-        self.saved = []
-
-    def save_session(self, **kwargs):
-        self.saved.append(kwargs)
-
-
-class DummyEmailService:
-    def send_verification_email(self, **kwargs):
-        return False, "disabled"
-
-
-def test_register_use_case_hashes_password_and_normalizes_email():
-    repo = InMemoryUserRepository()
-    use_case = RegisterUserUseCase(repo, email_service=DummyEmailService())
-
-    response, error = use_case.execute(
-        RegisterUserRequest(
-            email="USER@EXAMPLE.COM",
-            password="Password1",
-            username="user",
-            display_name="User",
-        )
-    )
-
-    assert error is None
-    assert response is not None
-    user = repo.find_by_email("user@example.com")
-    assert user is not None
-    assert user.password_hash.startswith("$2")
-    assert password_hasher.verify_password("Password1", user.password_hash)
-
-
-def test_register_route_defaults_display_name_to_username():
-    repo = InMemoryUserRepository()
-
-    response = asyncio.run(
-        register(
-            RegisterRequest(email="route@example.com", password="Password1", username="routeuser"),
-            repo,
-            DummyEmailService(),
-        )
-    )
-
-    assert response.display_name == "routeuser"
-    assert repo.find_by_email("route@example.com").display_name == "routeuser"
-
-
-def test_login_use_case_returns_verifiable_jwt():
-    repo = InMemoryUserRepository()
-    session_store = DummySessionStore()
-    user = User(
-        id="user-1",
-        email="user@example.com",
-        username="user",
-        display_name="User",
-        password_hash=password_hasher.hash_password("Password1"),
-    )
-    repo.save(user)
-
-    response, error = LoginUseCase(repo, session_store=session_store, jwt_secret_key=settings.JWT_SECRET_KEY).execute(
-        LoginRequest("user@example.com", "Password1")
-    )
-
-    assert error is None
-    assert response is not None
-    payload = jwt.decode(response.access_token, settings.JWT_SECRET_KEY, algorithms=["HS256"])
-    assert response.expires_in == 1800
-    assert payload["sub"] == "user-1"
-    assert payload["type"] == "user_auth"
-    assert session_store.saved
-
-
-def test_login_use_case_upgrades_legacy_sha256_hash():
-    repo = InMemoryUserRepository()
-    user = User(
-        id="user-1",
-        email="user@example.com",
-        username="user",
-        display_name="User",
-        password_hash="19513fdc9da4fb72a4a05eb66917548d3c90ff94d5419e1f2363eea89dfee1dd",
-    )
-    repo.save(user)
-
-    response, error = LoginUseCase(
-        repo, session_store=DummySessionStore(), jwt_secret_key=settings.JWT_SECRET_KEY
-    ).execute(LoginRequest("user@example.com", "Password1"))
-
-    assert error is None
-    assert response is not None
-    assert user.password_hash.startswith("$2")
-    assert password_hasher.verify_password("Password1", user.password_hash)
-
-
-def test_login_use_case_rejects_wrong_password():
-    repo = InMemoryUserRepository()
-    repo.save(
-        User(
-            id="user-1",
-            email="user@example.com",
-            username="user",
-            display_name="User",
-            password_hash=password_hasher.hash_password("Password1"),
-        )
-    )
-
-    response, error = LoginUseCase(repo, session_store=DummySessionStore()).execute(
-        LoginRequest("user@example.com", "WrongPassword1")
-    )
-
-    assert response is None
-    assert error == "Invalid email or password"
-
-
-def test_current_user_dependency_resolves_bearer_user_token():
-    repo = InMemoryUserRepository()
-    repo.save(
-        User(
-            id="user-1",
-            email="user@example.com",
-            username="user",
-            display_name="User",
-            password_hash=password_hasher.hash_password("Password1"),
-        )
-    )
-    token = jwt.encode({"sub": "user-1", "sid": "session-1", "type": "user_auth"}, settings.JWT_SECRET_KEY, "HS256")
-
-    authenticated_user = asyncio.run(
-        get_current_user(HTTPAuthorizationCredentials(scheme="Bearer", credentials=token), repo)
-    )
-
-    assert authenticated_user.user.id == "user-1"
-    assert authenticated_user.session_id == "session-1"
-
-
-def test_current_user_dependency_rejects_missing_token():
-    with pytest.raises(HTTPException) as error:
-        asyncio.run(get_current_user(None, InMemoryUserRepository()))
-
-    assert error.value.status_code == 401
-
-
-def test_current_user_dependency_rejects_wrong_token_type():
-    token = jwt.encode({"sub": "user-1", "type": "refresh"}, settings.JWT_SECRET_KEY, "HS256")
-
-    with pytest.raises(HTTPException) as error:
-        asyncio.run(
-            get_current_user(HTTPAuthorizationCredentials(scheme="Bearer", credentials=token), InMemoryUserRepository())
-        )
-
-    assert error.value.status_code == 401
-
-
-def test_verify_email_route_uses_use_case():
-    repo = InMemoryUserRepository()
-    user = User(
-        id="user-1",
-        email="user@example.com",
-        username="user",
-        display_name="User",
-        password_hash=password_hasher.hash_password("Password1"),
-        email_verification_token="verify-token",
-    )
-    repo.save(user)
-
-    response = asyncio.run(verify_email_post(VerifyEmailRequestModel(token="verify-token"), repo))
-
-    assert response.message == "邮箱验证成功"
-    assert user.email_verified is True
-    assert user.email_verification_token is None
-
-
-def test_password_reset_routes_use_use_cases():
-    repo = InMemoryUserRepository()
-    user = User(
-        id="user-1",
-        email="user@example.com",
-        username="user",
-        display_name="User",
-        password_hash=password_hasher.hash_password("Password1"),
-    )
-    repo.save(user)
-
-    forgot_response = asyncio.run(
-        forgot_password(PasswordResetRequestModel(email="user@example.com"), repo, NoopEmailService())
-    )
-    assert forgot_response.message == "如果账户存在，密码重置邮件已发送"
-    assert user.password_reset_token
-
-    reset_response = asyncio.run(
-        reset_password(ResetPasswordModel(token=user.password_reset_token, new_password="NewPassword1"), repo)
-    )
-
-    assert reset_response.message == "密码重置成功"
-    assert password_hasher.verify_password("NewPassword1", user.password_hash)
-    assert user.password_reset_token is None

tests/unit/test_auth_middleware.py

@@ -1,110 +0,0 @@
-import asyncio
-import sys
-from pathlib import Path
-
-import jwt
-import pytest
-from fastapi import HTTPException
-from fastapi.security import HTTPAuthorizationCredentials
-
-ROOT = Path(__file__).resolve().parents[2]
-API_ROOT = ROOT / "apps" / "api"
-if str(API_ROOT) not in sys.path:
-    sys.path.insert(0, str(API_ROOT))
-
-from app.config import settings
-from app.middleware.auth import get_current_user, get_current_user_optional, require_workspace_access
-
-from packages.application.auth.password_hasher import password_hasher
-from packages.domain.entities import User
-
-
-class InMemoryUserRepository:
-    def __init__(self):
-        self.users = {}
-
-    def save(self, user):
-        self.users[user.id] = user
-
-    def find_by_id(self, user_id):
-        return self.users.get(user_id)
-
-    def find_by_email(self, email):
-        return next((user for user in self.users.values() if user.email == email), None)
-
-    def find_by_username(self, username):
-        return next((user for user in self.users.values() if user.username == username), None)
-
-    def find_by_verification_token(self, token):
-        return None
-
-    def find_by_password_reset_token(self, token):
-        return None
-
-    def delete(self, user_id):
-        return self.users.pop(user_id, None) is not None
-
-
-def _repo_with_user():
-    repo = InMemoryUserRepository()
-    user = User(
-        id="user-1",
-        email="user@example.com",
-        username="user",
-        display_name="User",
-        password_hash=password_hasher.hash_password("Password1"),
-    )
-    repo.save(user)
-    return repo
-
-
-def _credentials(token_type="user_auth"):
-    token = jwt.encode({"sub": "user-1", "sid": "session-1", "type": token_type}, settings.JWT_SECRET_KEY, "HS256")
-    return HTTPAuthorizationCredentials(scheme="Bearer", credentials=token)
-
-
-def test_legacy_middleware_get_current_user_delegates_to_canonical_auth():
-    user = asyncio.run(get_current_user(asyncio.run(_authenticated_user())))
-
-    assert user.id == "user-1"
-
-
-def test_legacy_middleware_optional_user_returns_none_without_credentials():
-    assert asyncio.run(get_current_user_optional(None, _repo_with_user())) is None
-
-
-def test_legacy_middleware_optional_user_returns_user_with_valid_credentials():
-    user = asyncio.run(get_current_user_optional(_credentials(), _repo_with_user()))
-
-    assert user is not None
-    assert user.id == "user-1"
-
-
-def test_workspace_dependency_allows_member_access():
-    repo = _WorkspaceMemberRepositoryStub(role="member")
-
-    assert role == "member"  # noqa: F821
-
-
-def test_workspace_dependency_rejects_non_member():
-    with pytest.raises(HTTPException) as error:
-        require_workspace_access(
-            "workspace-1", _repo_with_user().find_by_id("user-1"), _WorkspaceMemberRepositoryStub()
-        )
-
-    assert error.value.status_code == 403
-
-
-class _WorkspaceMemberRepositoryStub:
-    def __init__(self, role=None):
-        self.role = role
-
-        if self.role is None:
-            return None
-        from packages.domain.entities import WorkspaceMember
-
-
-async def _authenticated_user():
-    from app.auth import get_current_user as get_authenticated_user
-
-    return await get_authenticated_user(_credentials(), _repo_with_user())

tests/unit/test_create_workspace_use_case.py

@@ -1,178 +0,0 @@
-"""
-创建 Workspace Use Case 测试
-"""
-
-from unittest.mock import Mock
-
-import pytest
-
-from packages.application.workspace import (
-    CreateWorkspaceRequest,
-    CreateWorkspaceUseCase,
-)
-from packages.domain.entities import User
-
-
-class TestCreateWorkspaceUseCase:
-    """创建工作空间测试"""
-
-    @pytest.fixture
-    def mock_workspace_repo(self):
-        repo = Mock()
-        repo.save = Mock()
-        return repo
-
-    @pytest.fixture
-    def mock_member_repo(self):
-        repo = Mock()
-        repo.save = Mock()
-        return repo
-
-    @pytest.fixture
-    def mock_user_repo(self):
-        repo = Mock()
-        repo.find_by_id = Mock(return_value=None)
-        return repo
-
-    @pytest.fixture
-    def use_case(self, mock_workspace_repo, mock_member_repo, mock_user_repo):
-        return CreateWorkspaceUseCase(
-            workspace_repository=mock_workspace_repo,
-            workspace_member_repository=mock_member_repo,
-            user_repository=mock_user_repo,
-        )
-
-    @pytest.fixture
-    def test_user(self):
-        return User(
-            id="user-123",
-            email="test@example.com",
-            username="testuser",
-            display_name="Test User",
-        )
-
-    def test_create_workspace_success_free_plan(
-        self, use_case, mock_workspace_repo, mock_member_repo, mock_user_repo, test_user
-    ):
-        """测试创建免费工作空间"""
-        mock_user_repo.find_by_id.return_value = test_user
-
-        request = CreateWorkspaceRequest(
-            name="My Workspace",
-            owner_user_id="user-123",
-            subscription_plan="free",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert error is None
-        assert response is not None
-        assert response.name == "My Workspace"
-        assert response.subscription_plan == "free"
-        assert response.max_projects == 3
-        assert response.max_storage_gb == 10
-
-        # 验证保存了 workspace
-        mock_workspace_repo.save.assert_called_once()
-        workspace = mock_workspace_repo.save.call_args[0][0]
-        assert workspace.name == "My Workspace"
-        assert workspace.owner_user_id == "user-123"
-
-        # 验证创建了 owner 成员
-        mock_member_repo.save.assert_called_once()
-        member = mock_member_repo.save.call_args[0][0]
-        assert member.user_id == "user-123"
-        assert member.role == "owner"
-
-    def test_create_workspace_pro_plan(
-        self, use_case, mock_workspace_repo, mock_member_repo, mock_user_repo, test_user
-    ):
-        """测试创建 Pro 工作空间"""
-        mock_user_repo.find_by_id.return_value = test_user
-
-        request = CreateWorkspaceRequest(
-            name="Pro Workspace",
-            owner_user_id="user-123",
-            subscription_plan="pro",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert error is None
-        assert response is not None
-        assert response.subscription_plan == "pro"
-        assert response.max_projects == 999999  # unlimited
-        assert response.max_storage_gb == 100
-
-    def test_create_workspace_enterprise_plan(
-        self, use_case, mock_workspace_repo, mock_member_repo, mock_user_repo, test_user
-    ):
-        """测试创建 Enterprise 工作空间"""
-        mock_user_repo.find_by_id.return_value = test_user
-
-        request = CreateWorkspaceRequest(
-            name="Enterprise Workspace",
-            owner_user_id="user-123",
-            subscription_plan="enterprise",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert error is None
-        assert response is not None
-        assert response.subscription_plan == "enterprise"
-        assert response.max_projects == 999999
-        assert response.max_storage_gb == 1000
-
-    def test_create_workspace_missing_name(self, use_case):
-        """测试缺少名称"""
-        request = CreateWorkspaceRequest(
-            name="",
-            owner_user_id="user-123",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "Workspace name is required"
-
-    def test_create_workspace_name_too_long(self, use_case):
-        """测试名称过长"""
-        request = CreateWorkspaceRequest(
-            name="A" * 101,
-            owner_user_id="user-123",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "Workspace name is too long (max 100 characters)"
-
-    def test_create_workspace_user_not_found(self, use_case, mock_user_repo):
-        """测试用户不存在"""
-        mock_user_repo.find_by_id.return_value = None
-
-        request = CreateWorkspaceRequest(
-            name="My Workspace",
-            owner_user_id="nonexistent-user",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "Owner user not found"
-
-    def test_create_workspace_invalid_plan(self, use_case, mock_user_repo, test_user):
-        """测试无效的订阅计划"""
-        mock_user_repo.find_by_id.return_value = test_user
-
-        request = CreateWorkspaceRequest(
-            name="My Workspace",
-            owner_user_id="user-123",
-            subscription_plan="invalid_plan",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert "Invalid subscription plan" in error

tests/unit/test_invite_member_use_case.py

@@ -1,285 +0,0 @@
-"""
-邀请成员 Use Case 测试
-"""
-
-from datetime import datetime, timezone
-from unittest.mock import Mock
-
-import pytest
-
-from packages.application.workspace.invite_member_use_case import (
-    InviteMemberRequest,
-    InviteMemberUseCase,
-)
-from packages.domain.entities import (
-    User,
-    Workspace,
-    WorkspaceMember,
-    WorkspaceMemberRole,
-)
-
-
-class TestInviteMemberUseCase:
-    """邀请成员测试"""
-
-    @pytest.fixture
-    def mock_workspace_repo(self):
-        repo = Mock()
-        repo.find_by_id = Mock(return_value=None)
-        return repo
-
-    @pytest.fixture
-    def mock_member_repo(self):
-        repo = Mock()
-        repo.find_by_workspace_and_user = Mock(return_value=None)
-        return repo
-
-    @pytest.fixture
-    def mock_invitation_repo(self):
-        repo = Mock()
-        repo.find_pending_by_workspace_and_email = Mock(return_value=None)
-        repo.save = Mock()
-        return repo
-
-    @pytest.fixture
-    def mock_user_repo(self):
-        repo = Mock()
-        repo.find_by_email = Mock(return_value=None)
-        repo.find_by_id = Mock(return_value=None)
-        return repo
-
-    @pytest.fixture
-    def use_case(
-        self,
-        mock_workspace_repo,
-        mock_member_repo,
-        mock_invitation_repo,
-        mock_user_repo,
-    ):
-        email_service = Mock()
-        email_service.send_workspace_invitation_email.return_value = (True, None)
-        return InviteMemberUseCase(
-            workspace_repository=mock_workspace_repo,
-            workspace_member_repository=mock_member_repo,
-            workspace_invitation_repository=mock_invitation_repo,
-            user_repository=mock_user_repo,
-            base_url="https://test.com",
-            invitation_expire_days=7,
-            email_service=email_service,
-        )
-
-    @pytest.fixture
-    def test_workspace(self):
-        return Workspace(
-            id="workspace-123",
-            name="Test Workspace",
-            owner_user_id="owner-id",
-        )
-
-    @pytest.fixture
-    def test_inviter(self):
-        return User(
-            id="inviter-id",
-            email="inviter@test.com",
-            username="inviter",
-            display_name="Inviter User",
-        )
-
-    @pytest.fixture
-    def owner_member(self):
-        return WorkspaceMember(
-            id="member-1",
-            user_id="inviter-id",
-            role=WorkspaceMemberRole.OWNER,
-        )
-
-    @pytest.fixture
-    def admin_member(self):
-        return WorkspaceMember(
-            id="member-2",
-            user_id="inviter-id",
-            role=WorkspaceMemberRole.ADMIN,
-        )
-
-    def test_invite_member_success_by_owner(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        mock_invitation_repo,
-        mock_user_repo,
-        test_workspace,
-        test_inviter,
-        owner_member,
-    ):
-        """测试 Owner 邀请成员成功"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-        mock_member_repo.find_by_workspace_and_user.return_value = owner_member
-        mock_user_repo.find_by_id.return_value = test_inviter
-
-        request = InviteMemberRequest(
-            inviter_user_id="inviter-id",
-            invitee_email="newuser@test.com",
-            role="member",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert error is None
-        assert response is not None
-        assert response.invitee_email == "newuser@test.com"
-        assert response.role == "member"
-        assert response.expires_at is not None
-
-        # 验证保存了邀请
-        mock_invitation_repo.save.assert_called_once()
-        invitation = mock_invitation_repo.save.call_args[0][0]
-        assert invitation.invitee_email == "newuser@test.com"
-        assert invitation.status == "pending"
-
-        # 验证发送了邮件
-        use_case.email_service.send_workspace_invitation_email.assert_called_once()
-
-    def test_invite_member_success_by_admin(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        mock_invitation_repo,
-        mock_user_repo,
-        test_workspace,
-        test_inviter,
-        admin_member,
-    ):
-        """测试 Admin 邀请成员成功"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-        mock_member_repo.find_by_workspace_and_user.return_value = admin_member
-        mock_user_repo.find_by_id.return_value = test_inviter
-
-        request = InviteMemberRequest(
-            inviter_user_id="inviter-id",
-            invitee_email="newuser@test.com",
-            role="viewer",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert error is None
-        assert response is not None
-
-    def test_invite_member_workspace_not_found(self, use_case, mock_workspace_repo):
-        """测试 Workspace 不存在"""
-        mock_workspace_repo.find_by_id.return_value = None
-
-        request = InviteMemberRequest(
-            inviter_user_id="inviter-id",
-            invitee_email="newuser@test.com",
-            role="member",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "Workspace not found"
-
-    def test_invite_member_inviter_not_member(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试邀请人不是成员"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-        mock_member_repo.find_by_workspace_and_user.return_value = None
-
-        request = InviteMemberRequest(
-            inviter_user_id="nonmember-id",
-            invitee_email="newuser@test.com",
-            role="member",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "You are not a member of this workspace"
-
-    def test_invite_member_inviter_no_permission(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试邀请人没有权限（只是普通成员）"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-
-        regular_member = WorkspaceMember(
-            id="member-3",
-            user_id="inviter-id",
-            role=WorkspaceMemberRole.MEMBER,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = regular_member
-
-        request = InviteMemberRequest(
-            inviter_user_id="inviter-id",
-            invitee_email="newuser@test.com",
-            role="member",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "Only owners and admins can invite members"
-
-    def test_invite_member_already_member(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        mock_user_repo,
-        test_workspace,
-        owner_member,
-    ):
-        """测试被邀请人已经是成员"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-        mock_member_repo.find_by_workspace_and_user.side_effect = [
-            owner_member,  # 第一次调用：inviter 是 owner
-            WorkspaceMember(  # 第二次调用：invitee 已经是成员
-                id="member-4",
-                user_id="existing-user-id",
-                role=WorkspaceMemberRole.MEMBER,
-            ),
-        ]
-
-        existing_user = User(
-            id="existing-user-id",
-            email="existing@test.com",
-            username="existing",
-            display_name="Existing User",
-        )
-        mock_user_repo.find_by_email.return_value = existing_user
-
-        request = InviteMemberRequest(
-            inviter_user_id="inviter-id",
-            invitee_email="existing@test.com",
-            role="member",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "User is already a member of this workspace"
-
-    def test_invite_member_invalid_role(self, use_case):
-        """测试无效角色（不能邀请为 owner）"""
-        request = InviteMemberRequest(
-            inviter_user_id="inviter-id",
-            invitee_email="newuser@test.com",
-            role="owner",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert "Invalid role" in error

tests/unit/test_jwt_service.py

@@ -1,123 +0,0 @@
-"""
-JWT 工具类测试
-"""
-
-from datetime import datetime, timedelta
-
-import pytest
-from jwt.exceptions import ExpiredSignatureError, InvalidTokenError
-
-from packages.application.auth.jwt_service import JWTConfig, JWTService, TokenType
-
-
-class TestJWTService:
-    """JWT 服务测试"""
-
-    @pytest.fixture
-    def jwt_service(self):
-        """创建 JWT 服务实例"""
-        config = JWTConfig()
-        config.SECRET_KEY = "test-secret-key-for-testing"
-        return JWTService(config)
-
-    def test_create_access_token(self, jwt_service):
-        """测试创建 access_token"""
-
-        assert isinstance(token, str)  # noqa: F821
-        assert len(token) > 0  # noqa: F821
-
-        # 验证 Token 内容
-        payload = jwt_service.verify_access_token(token)  # noqa: F821
-        assert payload["sub"] == "user-123"
-        assert payload["role"] == "admin"
-        assert payload["type"] == TokenType.ACCESS
-
-    def test_create_refresh_token(self, jwt_service):
-        """测试创建 refresh_token"""
-        token = jwt_service.create_refresh_token(user_id="user-123", session_id="session-789")
-
-        assert isinstance(token, str)  # noqa: F821
-        assert len(token) > 0  # noqa: F821
-
-        # 验证 Token 内容
-        payload = jwt_service.verify_refresh_token(token)  # noqa: F821
-        assert payload["sub"] == "user-123"
-        assert payload["session_id"] == "session-789"
-        assert payload["type"] == TokenType.REFRESH
-
-    def test_verify_valid_access_token(self, jwt_service):
-        """测试验证有效的 access_token"""
-
-        payload = jwt_service.verify_access_token(token)  # noqa: F821
-        assert payload["sub"] == "user-123"
-        assert payload["role"] == "member"
-
-    def test_verify_expired_token(self, jwt_service):
-        """测试验证过期的 Token"""
-        # 创建一个已过期的配置（使用相同的 SECRET_KEY）
-        config = JWTConfig()
-        config.SECRET_KEY = "test-secret-key-for-testing"  # 与 fixture 相同
-        config.ACCESS_TOKEN_EXPIRE_MINUTES = -1  # 负数，立即过期
-
-        expired_service = JWTService(config)
-
-        # 验证应该抛出过期异常
-        with pytest.raises(ExpiredSignatureError):
-            jwt_service.verify_access_token(token)  # noqa: F821
-
-    def test_verify_invalid_token(self, jwt_service):
-        """测试验证无效的 Token"""
-        invalid_token = "invalid.token.string"
-
-        with pytest.raises(InvalidTokenError):
-            jwt_service.verify_access_token(invalid_token)
-
-    def test_verify_wrong_token_type(self, jwt_service):
-        """测试验证错误类型的 Token"""
-        # 创建 refresh_token
-        refresh_token = jwt_service.create_refresh_token(user_id="user-123", session_id="session-789")
-
-        # 用 verify_access_token 验证应该失败
-        with pytest.raises(ValueError, match="Token type must be 'access'"):
-            jwt_service.verify_access_token(refresh_token)
-
-        # 反过来也一样
-
-        with pytest.raises(ValueError, match="Token type must be 'refresh'"):
-            jwt_service.verify_refresh_token(access_token)  # noqa: F821
-
-    def test_verify_tampered_token(self, jwt_service):
-        """测试验证被篡改的 Token"""
-
-        # 篡改 Token（修改最后几个字符）
-        tampered_token = token[:-5] + "XXXXX"  # noqa: F821
-
-        with pytest.raises(InvalidTokenError):
-            jwt_service.verify_access_token(tampered_token)
-
-    def test_additional_claims(self, jwt_service):
-        """测试额外的声明"""
-        token = jwt_service.create_access_token(
-            user_id="user-123",
-            role="admin",
-            additional_claims={
-                "email": "user@example.com",
-                "display_name": "Test User",
-            },
-        )
-
-        payload = jwt_service.verify_access_token(token)  # noqa: F821
-        assert payload["email"] == "user@example.com"
-        assert payload["display_name"] == "Test User"
-
-    def test_decode_unsafe(self, jwt_service):
-        """测试不安全解码（不验证签名）"""
-
-        # 不验证签名地解码
-        payload = jwt_service.decode_token_unsafe(token)  # noqa: F821
-        assert payload is not None
-        assert payload["sub"] == "user-123"
-
-        # 无效 Token 应该返回 None
-        invalid_payload = jwt_service.decode_token_unsafe("invalid.token")
-        assert invalid_payload is None

tests/unit/test_list_members_use_case.py

@@ -1,241 +0,0 @@
-"""
-获取成员列表 Use Case 测试
-"""
-
-from datetime import datetime, timezone
-from unittest.mock import Mock
-
-import pytest
-
-from packages.application.workspace.list_members_use_case import (
-    ListMembersRequest,
-    ListMembersUseCase,
-)
-from packages.domain.entities import (
-    User,
-    Workspace,
-    WorkspaceMember,
-    WorkspaceMemberRole,
-)
-
-
-class TestListMembersUseCase:
-    """获取成员列表测试"""
-
-    @pytest.fixture
-    def mock_workspace_repo(self):
-        repo = Mock()
-        repo.find_by_id = Mock(return_value=None)
-        return repo
-
-    @pytest.fixture
-    def mock_member_repo(self):
-        repo = Mock()
-        repo.find_by_workspace_and_user = Mock(return_value=None)
-        repo.find_by_workspace = Mock(return_value=[])
-        return repo
-
-    @pytest.fixture
-    def mock_user_repo(self):
-        repo = Mock()
-        repo.find_by_id = Mock(return_value=None)
-        return repo
-
-    @pytest.fixture
-    def use_case(self, mock_workspace_repo, mock_member_repo, mock_user_repo):
-        return ListMembersUseCase(
-            workspace_repository=mock_workspace_repo,
-            workspace_member_repository=mock_member_repo,
-            user_repository=mock_user_repo,
-        )
-
-    @pytest.fixture
-    def test_workspace(self):
-        return Workspace(
-            id="workspace-123",
-            name="Test Workspace",
-            owner_user_id="owner-id",
-        )
-
-    def test_list_members_success(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        mock_user_repo,
-        test_workspace,
-    ):
-        """测试获取成员列表成功"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-
-        # 请求者是 Admin
-        requester_member = WorkspaceMember(
-            id="member-1",
-            user_id="admin-id",
-            role=WorkspaceMemberRole.ADMIN,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = requester_member
-
-        # 3 个成员
-        member1 = WorkspaceMember(
-            id="member-1",
-            user_id="owner-id",
-            role=WorkspaceMemberRole.OWNER,
-            invited_by=None,
-        )
-
-        member2 = WorkspaceMember(
-            id="member-2",
-            user_id="admin-id",
-            role=WorkspaceMemberRole.ADMIN,
-            invited_by="owner-id",
-        )
-
-        member3 = WorkspaceMember(
-            id="member-3",
-            user_id="user-id",
-            role=WorkspaceMemberRole.MEMBER,
-            invited_by="admin-id",
-        )
-
-        mock_member_repo.find_by_workspace.return_value = [member1, member2, member3]
-
-        # 用户信息
-        user1 = User(
-            id="owner-id",
-            email="owner@test.com",
-            username="owner",
-            display_name="Owner User",
-        )
-
-        user2 = User(
-            id="admin-id",
-            email="admin@test.com",
-            username="admin",
-            display_name="Admin User",
-        )
-
-        user3 = User(
-            id="user-id",
-            email="user@test.com",
-            username="user",
-            display_name="Regular User",
-        )
-
-        mock_user_repo.find_by_id.side_effect = [user1, user2, user3]
-
-        request = ListMembersRequest(
-            requester_user_id="admin-id",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert error is None
-        assert response is not None
-        assert len(response.members) == 3
-
-        # 验证第一个成员（Owner）
-        m1 = response.members[0]
-        assert m1.user_id == "owner-id"
-        assert m1.username == "owner"
-        assert m1.email == "owner@test.com"
-        assert m1.role == "owner"
-        assert m1.invited_by is None
-
-        # 验证第二个成员（Admin）
-        m2 = response.members[1]
-        assert m2.user_id == "admin-id"
-        assert m2.role == "admin"
-        assert m2.invited_by == "owner-id"
-
-        # 验证第三个成员（Member）
-        m3 = response.members[2]
-        assert m3.user_id == "user-id"
-        assert m3.role == "member"
-        assert m3.invited_by == "admin-id"
-
-    def test_list_members_workspace_not_found(
-        self,
-        use_case,
-        mock_workspace_repo,
-    ):
-        """测试工作空间不存在"""
-        mock_workspace_repo.find_by_id.return_value = None
-
-        request = ListMembersRequest(
-            requester_user_id="user-id",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "Workspace not found"
-
-    def test_list_members_requester_not_member(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试请求者不是成员"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-        mock_member_repo.find_by_workspace_and_user.return_value = None
-
-        request = ListMembersRequest(
-            requester_user_id="outsider-id",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "You are not a member of this workspace"
-
-    def test_list_members_empty_workspace(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试空工作空间（理论上不应该发生）"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-
-        requester_member = WorkspaceMember(
-            id="member-1",
-            user_id="user-id",
-            role=WorkspaceMemberRole.MEMBER,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = requester_member
-        mock_member_repo.find_by_workspace.return_value = []
-
-        request = ListMembersRequest(
-            requester_user_id="user-id",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert error is None
-        assert response is not None
-        assert len(response.members) == 0
-
-        """测试缺少工作空间 ID"""
-        request = ListMembersRequest(
-            requester_user_id="user-id",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "Workspace ID is required"
-
-    def test_list_members_missing_requester_id(self, use_case):
-        """测试缺少请求者 ID"""
-        request = ListMembersRequest(
-            requester_user_id="",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "Requester user ID is required"

tests/unit/test_login_use_case.py

@@ -1,199 +0,0 @@
-"""
-用户登录 Use Case 测试
-"""
-
-from datetime import datetime, timezone
-from unittest.mock import Mock
-
-import pytest
-
-from packages.application.auth import (
-    LoginRequest,
-    LoginUseCase,
-    LogoutRequest,
-    LogoutUseCase,
-)
-from packages.application.auth.password_hasher import password_hasher
-from packages.domain.entities import User
-
-
-class TestLoginUseCase:
-    """登录用例测试"""
-
-    @pytest.fixture
-    def mock_user_repo(self):
-        repo = Mock()
-        repo.find_by_email = Mock(return_value=None)
-        repo.save = Mock()
-        return repo
-
-    @pytest.fixture
-    def use_case(self, mock_user_repo):
-        session_store = Mock()
-        session_store.save_session.return_value = True
-        return LoginUseCase(user_repository=mock_user_repo, session_store=session_store)
-
-    @pytest.fixture
-    def test_user(self):
-        """创建测试用户"""
-        password_hash = password_hasher.hash_password("SecurePass123")
-        return User(
-            id="user-123",
-            email="test@example.com",
-            username="testuser",
-            display_name="Test User",
-            password_hash=password_hash,
-            email_verified=True,
-        )
-
-    def test_login_success(self, use_case, mock_user_repo, test_user):
-        """测试登录成功"""
-        mock_user_repo.find_by_email.return_value = test_user
-
-        request = LoginRequest(
-            email="test@example.com",
-            password="SecurePass123",
-            device_info="Chrome/Windows",
-            ip_address="192.168.1.1",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert error is None
-        assert response is not None
-        assert response.user_id == "user-123"
-        assert response.email == "test@example.com"
-        assert response.username == "testuser"
-        assert response.access_token != ""
-        assert response.refresh_token != ""
-        assert response.expires_in > 0
-
-        # 验证保存了 session
-        use_case.session_store.save_session.assert_called_once()
-
-        # 验证更新了最后登录信息
-        mock_user_repo.save.assert_called_once()
-
-    def test_login_invalid_email(self, use_case, mock_user_repo):
-        """测试邮箱不存在"""
-        mock_user_repo.find_by_email.return_value = None
-
-        request = LoginRequest(
-            email="nonexistent@example.com",
-            password="SecurePass123",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "Invalid email or password"
-
-    def test_login_wrong_password(self, use_case, mock_user_repo, test_user):
-        """测试密码错误"""
-        mock_user_repo.find_by_email.return_value = test_user
-
-        request = LoginRequest(
-            email="test@example.com",
-            password="WrongPassword123",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "Invalid email or password"
-
-    def test_login_missing_email(self, use_case):
-        """测试缺少邮箱"""
-        request = LoginRequest(
-            email="",
-            password="SecurePass123",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "Email is required"
-
-    def test_login_missing_password(self, use_case, mock_user_repo, test_user):
-        """测试缺少密码"""
-        mock_user_repo.find_by_email.return_value = test_user
-
-        request = LoginRequest(
-            email="test@example.com",
-            password="",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "Password is required"
-
-
-class TestLogoutUseCase:
-    """登出用例测试"""
-
-    @pytest.fixture
-    def use_case(self):
-        session_store = Mock()
-        return LogoutUseCase(session_store=session_store)
-
-    def test_logout_current_device(self, use_case):
-        """测试登出当前设备"""
-        use_case.session_store.delete_session.return_value = True
-
-        request = LogoutRequest(
-            user_id="user-123",
-            session_id="session-abc",
-            logout_all_devices=False,
-        )
-
-        success, error = use_case.execute(request)
-
-        assert success is True
-        assert error is None
-
-        use_case.session_store.delete_session.assert_called_once_with("session-abc")
-
-    def test_logout_all_devices(self, use_case):
-        """测试登出所有设备"""
-        use_case.session_store.delete_all_user_sessions.return_value = 3
-
-        request = LogoutRequest(
-            user_id="user-123",
-            logout_all_devices=True,
-        )
-
-        success, error = use_case.execute(request)
-
-        assert success is True
-        assert error is None
-
-        use_case.session_store.delete_all_user_sessions.assert_called_once_with("user-123")
-
-    def test_logout_session_not_found(self, use_case):
-        """测试 session 不存在"""
-        use_case.session_store.delete_session.return_value = False
-
-        request = LogoutRequest(
-            user_id="user-123",
-            session_id="nonexistent",
-            logout_all_devices=False,
-        )
-
-        success, error = use_case.execute(request)
-
-        assert success is False
-        assert error == "Session not found"
-
-    def test_logout_missing_session_id(self, use_case):
-        """测试缺少 session_id"""
-        request = LogoutRequest(
-            user_id="user-123",
-            session_id=None,
-            logout_all_devices=False,
-        )
-
-        success, error = use_case.execute(request)
-
-        assert success is False
-        assert error == "Session ID is required"

tests/unit/test_permissions.py

@@ -1,193 +0,0 @@
-"""
-权限验证辅助函数测试
-"""
-
-from unittest.mock import Mock
-
-import pytest
-
-from packages.domain.entities import WorkspaceMember, WorkspaceMemberRole
-from packages.domain.permissions import Permission, PermissionChecker, has_permission
-
-
-class TestPermissionChecker:
-    """权限检查器测试"""
-
-    @pytest.fixture
-    def mock_member_repo(self):
-        repo = Mock()
-        repo.find_by_workspace_and_user = Mock(return_value=None)
-        return repo
-
-    @pytest.fixture
-    def checker(self, mock_member_repo):
-        return PermissionChecker(workspace_member_repository=mock_member_repo)
-
-    def test_check_workspace_access_has_access(self, checker, mock_member_repo):
-        """测试有访问权限"""
-        member = WorkspaceMember(
-            id="member-1",
-            user_id="user-123",
-            role=WorkspaceMemberRole.MEMBER,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = member
-
-        has_access, role = checker.check_workspace_access("workspace-123", "user-123")
-
-        assert has_access is True
-        assert role == "member"
-
-    def test_check_workspace_access_no_access(self, checker, mock_member_repo):
-        """测试无访问权限"""
-        mock_member_repo.find_by_workspace_and_user.return_value = None
-
-        has_access, role = checker.check_workspace_access("workspace-123", "user-123")
-
-        assert has_access is False
-        assert role is None
-
-    def test_check_is_owner_true(self, checker, mock_member_repo):
-        """测试是 Owner"""
-        member = WorkspaceMember(
-            id="member-1",
-            user_id="user-123",
-            role=WorkspaceMemberRole.OWNER,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = member
-
-        is_owner = checker.check_is_owner("workspace-123", "user-123")
-
-        assert is_owner is True
-
-    def test_check_is_owner_false(self, checker, mock_member_repo):
-        """测试不是 Owner"""
-        member = WorkspaceMember(
-            id="member-1",
-            user_id="user-123",
-            role=WorkspaceMemberRole.ADMIN,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = member
-
-        is_owner = checker.check_is_owner("workspace-123", "user-123")
-
-        assert is_owner is False
-
-    def test_check_is_admin_or_owner_admin(self, checker, mock_member_repo):
-        """测试是 Admin"""
-        member = WorkspaceMember(
-            id="member-1",
-            user_id="user-123",
-            role=WorkspaceMemberRole.ADMIN,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = member
-
-        is_admin_or_owner = checker.check_is_admin_or_owner("workspace-123", "user-123")
-
-        assert is_admin_or_owner is True
-
-    def test_check_is_admin_or_owner_owner(self, checker, mock_member_repo):
-        """测试是 Owner"""
-        member = WorkspaceMember(
-            id="member-1",
-            user_id="user-123",
-            role=WorkspaceMemberRole.OWNER,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = member
-
-        is_admin_or_owner = checker.check_is_admin_or_owner("workspace-123", "user-123")
-
-        assert is_admin_or_owner is True
-
-    def test_check_is_admin_or_owner_member(self, checker, mock_member_repo):
-        """测试是普通成员"""
-        member = WorkspaceMember(
-            id="member-1",
-            user_id="user-123",
-            role=WorkspaceMemberRole.MEMBER,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = member
-
-        is_admin_or_owner = checker.check_is_admin_or_owner("workspace-123", "user-123")
-
-        assert is_admin_or_owner is False
-
-    def test_check_can_create_project_member(self, checker, mock_member_repo):
-        """测试 Member 可以创建项目"""
-        member = WorkspaceMember(
-            id="member-1",
-            user_id="user-123",
-            role=WorkspaceMemberRole.MEMBER,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = member
-
-        can_create = checker.check_can_create_project("workspace-123", "user-123")
-
-        assert can_create is True
-
-    def test_check_can_create_project_viewer(self, checker, mock_member_repo):
-        """测试 Viewer 不能创建项目"""
-        member = WorkspaceMember(
-            id="member-1",
-            user_id="user-123",
-            role=WorkspaceMemberRole.VIEWER,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = member
-
-        can_create = checker.check_can_create_project("workspace-123", "user-123")
-
-        assert can_create is False
-
-    def test_check_can_delete_project_member(self, checker, mock_member_repo):
-        """测试 Member 不能删除项目"""
-        member = WorkspaceMember(
-            id="member-1",
-            user_id="user-123",
-            role=WorkspaceMemberRole.MEMBER,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = member
-
-        can_delete = checker.check_can_delete_project("workspace-123", "user-123")
-
-        assert can_delete is False
-
-    def test_check_can_delete_project_admin(self, checker, mock_member_repo):
-        """测试 Admin 可以删除项目"""
-        member = WorkspaceMember(
-            id="member-1",
-            user_id="user-123",
-            role=WorkspaceMemberRole.ADMIN,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = member
-
-        can_delete = checker.check_can_delete_project("workspace-123", "user-123")
-
-        assert can_delete is True
-
-
-class TestPermissionFunctions:
-    """权限函数测试"""
-
-    def test_has_permission_owner(self):
-        """测试 Owner 权限"""
-        assert has_permission(WorkspaceMemberRole.OWNER, Permission.WORKSPACE_DELETE) is True
-        assert has_permission(WorkspaceMemberRole.OWNER, Permission.MEMBER_REMOVE) is True
-        assert has_permission(WorkspaceMemberRole.OWNER, Permission.PROJECT_CREATE) is True
-
-    def test_has_permission_admin(self):
-        """测试 Admin 权限"""
-        assert has_permission(WorkspaceMemberRole.ADMIN, Permission.WORKSPACE_EDIT) is True
-        assert has_permission(WorkspaceMemberRole.ADMIN, Permission.MEMBER_REMOVE) is True
-        assert has_permission(WorkspaceMemberRole.ADMIN, Permission.WORKSPACE_DELETE) is False
-
-    def test_has_permission_member(self):
-        """测试 Member 权限"""
-        assert has_permission(WorkspaceMemberRole.MEMBER, Permission.PROJECT_CREATE) is True
-        assert has_permission(WorkspaceMemberRole.MEMBER, Permission.PROJECT_DELETE) is False
-        assert has_permission(WorkspaceMemberRole.MEMBER, Permission.MEMBER_INVITE) is False
-
-    def test_has_permission_viewer(self):
-        """测试 Viewer 权限"""
-        assert has_permission(WorkspaceMemberRole.VIEWER, Permission.WORKSPACE_VIEW) is True
-        assert has_permission(WorkspaceMemberRole.VIEWER, Permission.PROJECT_VIEW) is True
-        assert has_permission(WorkspaceMemberRole.VIEWER, Permission.PROJECT_CREATE) is False
-        assert has_permission(WorkspaceMemberRole.VIEWER, Permission.ASSET_UPLOAD) is False

tests/unit/test_quota.py

@@ -1,285 +0,0 @@
-"""
-配额检查服务测试
-"""
-
-from unittest.mock import Mock
-
-import pytest
-
-from packages.domain.entities import Workspace
-from packages.domain.quota import QuotaChecker, QuotaWarningLevel, get_warning_level
-
-
-class TestQuotaChecker:
-    """配额检查器测试"""
-
-    @pytest.fixture
-    def mock_workspace_repo(self):
-        repo = Mock()
-        repo.find_by_id = Mock(return_value=None)
-        repo.save = Mock()
-        return repo
-
-    @pytest.fixture
-    def mock_project_repo(self):
-        repo = Mock()
-        repo.count_by_workspace = Mock(return_value=0)
-        return repo
-
-    @pytest.fixture
-    def checker(self, mock_workspace_repo, mock_project_repo):
-        return QuotaChecker(
-            workspace_repository=mock_workspace_repo,
-            project_repository=mock_project_repo,
-        )
-
-    def test_check_can_create_project_within_limit(
-        self,
-        checker,
-        mock_workspace_repo,
-        mock_project_repo,
-    ):
-        """测试在配额内可以创建项目"""
-        workspace = Workspace(
-            id="workspace-123",
-            name="Test",
-            owner_user_id="owner-id",
-            subscription_plan="free",
-            max_projects=3,
-        )
-        mock_workspace_repo.find_by_id.return_value = workspace
-        mock_project_repo.count_by_workspace.return_value = 2
-
-        can_create, error = checker.check_can_create_project("workspace-123")
-
-        assert can_create is True
-        assert error is None
-
-    def test_check_can_create_project_at_limit(
-        self,
-        checker,
-        mock_workspace_repo,
-        mock_project_repo,
-    ):
-        """测试达到配额限制"""
-        workspace = Workspace(
-            id="workspace-123",
-            name="Test",
-            owner_user_id="owner-id",
-            subscription_plan="free",
-            max_projects=3,
-        )
-        mock_workspace_repo.find_by_id.return_value = workspace
-        mock_project_repo.count_by_workspace.return_value = 3
-
-        can_create, error = checker.check_can_create_project("workspace-123")
-
-        assert can_create is False
-        assert "Project limit reached" in error
-
-    def test_check_can_create_project_unlimited(
-        self,
-        checker,
-        mock_workspace_repo,
-        mock_project_repo,
-    ):
-        """测试无限制计划"""
-        workspace = Workspace(
-            id="workspace-123",
-            name="Test",
-            owner_user_id="owner-id",
-            subscription_plan="pro",
-            max_projects=999999,
-        )
-        mock_workspace_repo.find_by_id.return_value = workspace
-        mock_project_repo.count_by_workspace.return_value = 1000
-
-        can_create, error = checker.check_can_create_project("workspace-123")
-
-        assert can_create is True
-        assert error is None
-
-    def test_check_storage_available_within_limit(
-        self,
-        checker,
-        mock_workspace_repo,
-    ):
-        """测试存储在配额内"""
-        workspace = Workspace(
-            id="workspace-123",
-            name="Test",
-            owner_user_id="owner-id",
-            subscription_plan="free",
-            max_storage_gb=10,
-            used_storage_gb=5.0,
-        )
-        mock_workspace_repo.find_by_id.return_value = workspace
-
-        can_store, error = checker.check_storage_available("workspace-123", 3.0)
-
-        assert can_store is True
-        assert error is None
-
-    def test_check_storage_available_exceeded(
-        self,
-        checker,
-        mock_workspace_repo,
-    ):
-        """测试存储超出配额"""
-        workspace = Workspace(
-            id="workspace-123",
-            name="Test",
-            owner_user_id="owner-id",
-            subscription_plan="free",
-            max_storage_gb=10,
-            used_storage_gb=8.0,
-        )
-        mock_workspace_repo.find_by_id.return_value = workspace
-
-        can_store, error = checker.check_storage_available("workspace-123", 3.0)
-
-        assert can_store is False
-        assert "Storage limit exceeded" in error
-
-    def test_get_quota_status(
-        self,
-        checker,
-        mock_workspace_repo,
-        mock_project_repo,
-    ):
-        """测试获取配额状态"""
-        workspace = Workspace(
-            id="workspace-123",
-            name="Test",
-            owner_user_id="owner-id",
-            subscription_plan="free",
-            max_projects=3,
-            max_storage_gb=10,
-            used_storage_gb=7.5,
-        )
-        mock_workspace_repo.find_by_id.return_value = workspace
-        mock_project_repo.count_by_workspace.return_value = 2
-
-        status = checker.get_quota_status("workspace-123")
-
-        assert status is not None
-        assert status["subscription_plan"] == "free"
-        assert status["projects"]["used"] == 2
-        assert status["projects"]["limit"] == 3
-        assert status["projects"]["usage_percent"] == pytest.approx(66.67, rel=0.1)
-        assert status["storage"]["used_gb"] == 7.5
-        assert status["storage"]["limit_gb"] == 10
-        assert status["storage"]["remaining_gb"] == 2.5
-        assert status["storage"]["usage_percent"] == 75.0
-
-    def test_get_quota_status_unlimited(
-        self,
-        checker,
-        mock_workspace_repo,
-        mock_project_repo,
-    ):
-        """测试无限制配额状态"""
-        workspace = Workspace(
-            id="workspace-123",
-            name="Test",
-            owner_user_id="owner-id",
-            subscription_plan="pro",
-            max_projects=999999,
-            max_storage_gb=100,
-            used_storage_gb=50.0,
-        )
-        mock_workspace_repo.find_by_id.return_value = workspace
-        mock_project_repo.count_by_workspace.return_value = 1000
-
-        status = checker.get_quota_status("workspace-123")
-
-        assert status["projects"]["unlimited"] is True
-        assert status["projects"]["usage_percent"] == 0
-
-    def test_update_storage_usage_increase(
-        self,
-        checker,
-        mock_workspace_repo,
-    ):
-        """测试增加存储使用量"""
-        workspace = Workspace(
-            id="workspace-123",
-            name="Test",
-            owner_user_id="owner-id",
-            max_storage_gb=10,
-            used_storage_gb=5.0,
-        )
-        mock_workspace_repo.find_by_id.return_value = workspace
-
-        success, error = checker.update_storage_usage("workspace-123", 2.5)
-
-        assert success is True
-        assert error is None
-        assert workspace.used_storage_gb == 7.5
-        mock_workspace_repo.save.assert_called_once()
-
-    def test_update_storage_usage_decrease(
-        self,
-        checker,
-        mock_workspace_repo,
-    ):
-        """测试减少存储使用量"""
-        workspace = Workspace(
-            id="workspace-123",
-            name="Test",
-            owner_user_id="owner-id",
-            max_storage_gb=10,
-            used_storage_gb=5.0,
-        )
-        mock_workspace_repo.find_by_id.return_value = workspace
-
-        success, error = checker.update_storage_usage("workspace-123", -2.0)
-
-        assert success is True
-        assert workspace.used_storage_gb == 3.0
-
-    def test_update_storage_usage_prevent_negative(
-        self,
-        checker,
-        mock_workspace_repo,
-    ):
-        """测试防止负数存储"""
-        workspace = Workspace(
-            id="workspace-123",
-            name="Test",
-            owner_user_id="owner-id",
-            max_storage_gb=10,
-            used_storage_gb=2.0,
-        )
-        mock_workspace_repo.find_by_id.return_value = workspace
-
-        success, error = checker.update_storage_usage("workspace-123", -5.0)
-
-        assert success is True
-        assert workspace.used_storage_gb == 0.0
-
-
-class TestWarningLevel:
-    """警告级别测试"""
-
-    def test_get_warning_level_normal(self):
-        """测试正常级别"""
-        assert get_warning_level(50.0) == QuotaWarningLevel.NORMAL
-        assert get_warning_level(79.9) == QuotaWarningLevel.NORMAL
-
-    def test_get_warning_level_warning(self):
-        """测试警告级别"""
-        assert get_warning_level(80.0) == QuotaWarningLevel.WARNING
-        assert get_warning_level(85.0) == QuotaWarningLevel.WARNING
-        assert get_warning_level(89.9) == QuotaWarningLevel.WARNING
-
-    def test_get_warning_level_critical(self):
-        """测试严重级别"""
-        assert get_warning_level(90.0) == QuotaWarningLevel.CRITICAL
-        assert get_warning_level(95.0) == QuotaWarningLevel.CRITICAL
-        assert get_warning_level(99.9) == QuotaWarningLevel.CRITICAL
-
-    def test_get_warning_level_exceeded(self):
-        """测试超出级别"""
-        assert get_warning_level(100.0) == QuotaWarningLevel.EXCEEDED
-        assert get_warning_level(105.0) == QuotaWarningLevel.EXCEEDED

tests/unit/test_release_scripts.py

@@ -1,333 +0,0 @@
-from pathlib import Path
-
-
-def test_restore_postgres_plan_is_non_destructive():
-    script = Path("scripts/restore_postgres_plan.sh").read_text(encoding="utf-8")
-    executable_prefix = script.split("cat <<EOF", 1)[0]
-
-    assert "pg_restore" in script
-    assert "This script intentionally does not execute restore automatically." in script
-    assert "pg_restore" not in executable_prefix
-
-
-def test_gitea_production_deploy_uses_production_ports():
-    workflow = Path(".gitea/workflows/deploy.yml").read_text(encoding="utf-8")
-    production_section = workflow.split("deploy-production:", 1)[1]
-
-    assert "GITHUB_TOKEN: ${{ github.token }}" in workflow
-    assert "WEB_PORT=3001" not in production_section
-    assert "http://127.0.0.1:8001/health" in production_section
-    assert "RELEASE_VERSION='${GITHUB_REF_NAME}' sh -s" in production_section
-    assert "http://127.0.0.1:8000/health" not in production_section
-
-
-def test_deploy_production_uses_production_infra_and_project():
-    script = Path("infra/docker/deploy-production.sh").read_text(encoding="utf-8")
-
-    assert 'HOST_PREFIX="${HOST_PREFIX-/host}"' in script
-    assert "apps/web/dist/index.html" in script
-    assert "Production deploy must not build frontend assets on the server" in script
-    assert "WEB_DOCKERFILE=infra/docker/web-artifact.Dockerfile" in script
-    assert "WEB_NGINX_CONF=infra/docker/nginx-production.conf" in script
-    assert "Skipping production API/worker/web image builds" in script
-    assert "RELEASE_VERSION" in script
-    assert "RUNTIME_IMAGE_TAR" in script
-    assert 'docker load -i "$RUNTIME_IMAGE_TAR"' in script
-    assert 'export APP_VERSION="$RELEASE_VERSION"' in script
-    assert 'export WORKER_CONCURRENCY="${WORKER_CONCURRENCY:-1}"' in script
-    assert 'export WORKER_MAX_TASKS_PER_CHILD="${WORKER_MAX_TASKS_PER_CHILD:-100}"' in script
-    assert 'docker image inspect "${API_IMAGE:-xiaoxia-saas-api:dev}"' in script
-    assert 'docker image inspect "${WORKER_IMAGE:-xiaoxia-saas-worker:dev}"' in script
-    assert "ALLOW_PRODUCTION_BUILDS=true" in script
-    assert "xiaoxia-postgres-production" in script
-    assert "xiaoxia-redis-production" in script
-    assert "xiaoxia-postgres\n" not in script
-    assert "xiaoxia-redis\n" not in script
-    assert "COMPOSE_PROJECT_NAME=xiaoxia-production-app" in script
-    assert '--env-file "$ENV_FILE"' in script
-    assert "python /app/scripts/validate_release_env.py --from-environ --strict-external" in script
-    assert "alembic upgrade head" in script
-    assert 'docker compose --env-file "$ENV_FILE" up -d api worker' in script
-    assert 'docker compose --env-file "$ENV_FILE" up -d --force-recreate web' in script
-    assert "nginx resolves the current API container IP" in script
-    assert "/etc/cron.d/xiaoxia-production-resource-check" in script
-    assert "EXPECTED_VERSION=$RELEASE_VERSION" in script
-
-
-def test_production_workflow_preserves_previous_web_assets():
-    workflow = Path(".gitea/workflows/deploy.yml").read_text(encoding="utf-8")
-    production_section = workflow.split("deploy-production:", 1)[1]
-
-    assert 'old_assets_dir="/tmp/xiaoxia-previous-web-assets-${RELEASE_VERSION}"' in production_section
-    assert 'docker cp xiaoxia-web-production:/usr/share/nginx/html/assets/. "$old_assets_dir"/' in production_section
-    assert 'cp -a /var/lib/xiaoxia-saas-production/repo/apps/web/dist/assets/. "$old_assets_dir"/' in production_section
-    assert 'if [ ! -e "/var/lib/xiaoxia-saas-production/repo/apps/web/dist/assets/$name" ]; then' in production_section
-
-
-def test_production_deploy_recreates_web_after_api_for_nginx_dns():
-    script = Path("infra/docker/deploy-production.sh").read_text(encoding="utf-8")
-
-    api_up = script.index('docker compose --env-file "$ENV_FILE" up -d api worker')
-    web_up = script.index('docker compose --env-file "$ENV_FILE" up -d --force-recreate web')
-
-    assert api_up < web_up
-    assert "up -d api worker web" not in script
-
-
-def test_production_nginx_static_upstream_requires_web_recreate():
-    config = Path("infra/docker/nginx-production.conf").read_text(encoding="utf-8")
-    script = Path("infra/docker/deploy-production.sh").read_text(encoding="utf-8")
-
-    assert "proxy_pass http://xiaoxia-api-production:8000/api/;" in config
-    assert "client_max_body_size 800m;" in config
-    assert "proxy_read_timeout 300s;" in config
-    assert "proxy_request_buffering off;" in config
-    assert "location = /index.html" in config
-    assert 'Cache-Control "no-store, no-cache, must-revalidate" always' in config
-    assert "--force-recreate web" in script
-
-
-def test_production_infra_uses_separate_containers_and_ports():
-    compose = Path("infra/docker/infra-production.yml").read_text(encoding="utf-8")
-
-    assert "name: xiaoxia-production" in compose
-    assert "container_name: xiaoxia-postgres-production" in compose
-    assert "container_name: xiaoxia-redis-production" in compose
-    assert "${POSTGRES_PORT:-5433}:5432" in compose
-    assert "${REDIS_PORT:-6380}:6379" in compose
-    assert "postgres_data:" not in compose
-    assert "redis_data:" not in compose
-
-
-def test_init_production_env_is_non_deploying_and_separated():
-    script = Path("scripts/init_production_env.sh").read_text(encoding="utf-8")
-
-    assert "APP_ENV=production" in script
-    assert "ENVIRONMENT=production" in script
-    assert "xiaoxia-postgres-production" in script
-    assert "xiaoxia-redis-production" in script
-    assert "GENERATED_FILES_HOST_DIR=$GENERATED_DIR" in script
-    assert "ENABLE_EMAIL_DELIVERY=false" in script
-    assert "docker compose" not in script
-    assert "docker run" not in script
-
-
-def test_staging_deploy_supports_host_prefix_and_web_build():
-    script = Path("infra/docker/deploy-staging.sh").read_text(encoding="utf-8")
-
-    assert 'HOST_PREFIX="${HOST_PREFIX-/host}"' in script
-    assert 'ROOT_DIR="$HOST_PREFIX/var/lib/xiaoxia-saas-staging/repo"' in script
-    assert 'ENV_FILE="$HOST_PREFIX/var/lib/xiaoxia-saas-staging/.env"' in script
-    assert '"$HOST_PREFIX/var/lib/xiaoxia-saas-staging/generated"' in script
-    assert "docker compose build --pull=false web" in script
-
-
-def test_deploy_scripts_build_web_image_explicitly():
-    staging_script = Path("infra/docker/deploy-staging.sh").read_text(encoding="utf-8")
-    production_script = Path("infra/docker/deploy-production.sh").read_text(encoding="utf-8")
-    compose = Path("infra/docker/compose.yml").read_text(encoding="utf-8")
-
-    assert "docker compose build --pull=false web" in staging_script
-    assert 'docker compose --env-file "$ENV_FILE" build --pull=false web' in production_script
-    assert "dockerfile: ${WEB_DOCKERFILE:-infra/docker/web.Dockerfile}" in compose
-    assert "NGINX_CONF: ${WEB_NGINX_CONF:-infra/docker/nginx.conf}" in compose
-    assert "APP_VERSION: ${APP_VERSION:-0.1.0}" in compose
-    assert "WORKER_CONCURRENCY: ${WORKER_CONCURRENCY:-1}" in compose
-
-
-def test_production_deploy_prunes_old_unused_docker_artifacts():
-    script = Path("infra/docker/deploy-production.sh").read_text(encoding="utf-8")
-
-    assert "PRUNE_UNUSED_DOCKER_AFTER_DEPLOY" in script
-    assert 'docker image prune -af --filter "until=168h"' in script
-    assert 'docker builder prune -af --filter "until=168h"' in script
-
-
-def test_worker_runtime_is_constrained_by_environment():
-    dockerfile = Path("infra/docker/worker.Dockerfile").read_text(encoding="utf-8")
-
-    assert "--concurrency=${WORKER_CONCURRENCY:-1}" in dockerfile
-    assert "--max-tasks-per-child=${WORKER_MAX_TASKS_PER_CHILD:-100}" in dockerfile
-
-
-def test_production_nginx_proxies_to_production_api_container():
-    config = Path("infra/docker/nginx-production.conf").read_text(encoding="utf-8")
-
-    assert "proxy_pass http://xiaoxia-api-production:8000/api/;" in config
-    assert "proxy_pass http://api:8000/api/;" not in config
-
-
-def test_web_artifact_dockerfile_does_not_build_frontend_on_server():
-    dockerfile = Path("infra/docker/web-artifact.Dockerfile").read_text(encoding="utf-8")
-
-    assert "COPY apps/web/dist ./" in dockerfile
-    assert "ARG NGINX_CONF=infra/docker/nginx.conf" in dockerfile
-    assert "COPY ${NGINX_CONF} /etc/nginx/conf.d/default.conf" in dockerfile
-    assert "npm" not in dockerfile
-    assert "node" not in dockerfile.lower()
-
-
-def test_release_artifact_script_builds_web_before_packaging():
-    script = Path("scripts/package_release_artifact.sh").read_text(encoding="utf-8")
-
-    assert "npm ci" in script
-    assert "npm run build" in script
-    assert "--exclude=apps/web/node_modules" in script
-
-
-def test_web_dockerfile_uses_reachable_base_image_mirror():
-    dockerfile = Path("infra/docker/web.Dockerfile").read_text(encoding="utf-8")
-
-    assert "FROM docker.m.daocloud.io/library/node:20 AS builder" in dockerfile
-    assert "npm ci" in dockerfile
-    assert "COPY apps/web/package.json apps/web/package-lock.json" in dockerfile
-    assert "FROM docker.m.daocloud.io/library/nginx:alpine AS runner" in dockerfile
-    assert "FROM node:20" not in dockerfile
-    assert "FROM nginx:alpine" not in dockerfile
-
-
-def test_admin_routes_use_coming_soon_until_backend_exists():
-    router = Path("apps/web/src/router/index.tsx").read_text(encoding="utf-8")
-    sidebar = Path("apps/web/src/components/layout/Sidebar.tsx").read_text(encoding="utf-8")
-
-    assert "AdminComingSoon" in router
-    assert "@/pages/admin/Dashboard" not in router
-    assert "@/pages/admin/UserManagement" not in router
-    assert "@/pages/admin/Analytics" not in router
-    assert "@/pages/admin/SystemMonitor" not in router
-    assert "@/pages/admin/LogViewer" not in router
-    assert "Admin（暂未开放）" in sidebar
-    assert "disabled: true" in sidebar
-
-
-def test_subscription_api_is_not_wired_to_ui_until_backend_exists():
-    source_files = list(Path("apps/web/src").rglob("*.ts")) + list(Path("apps/web/src").rglob("*.tsx"))
-    importers = []
-    for path in source_files:
-        if path.as_posix().endswith("apps/web/src/api/subscription.ts"):
-            continue
-        text = path.read_text(encoding="utf-8")
-        if "@/api/subscription" in text or "../api/subscription" in text:
-            importers.append(path.as_posix())
-
-    assert importers == []
-
-
-def test_gitea_production_deploy_requires_runtime_builder_job():
-    workflow = Path(".gitea/workflows/deploy.yml").read_text(encoding="utf-8")
-    build_section = workflow.split("build-production-runtime-images:", 1)[1].split("deploy-production:", 1)[0]
-    production_section = workflow.split("deploy-production:", 1)[1]
-
-    assert "runs-on: runtime-builder" in build_section
-    assert 'scripts/build_release_images.sh "${GITHUB_REF_NAME}"' in build_section
-    assert "GITHUB_TOKEN: ${{ github.token }}" in build_section
-    assert "docker.m.daocloud.io/library/node:20" in build_section
-    assert "npm ci && npm run build" in build_section
-    assert "dist/release-artifacts/xiaoxia-release-${GITHUB_REF_NAME}.tar.gz" in build_section
-    assert "--exclude=./dist" in build_section
-    assert "--exclude=dist " not in build_section
-    assert "PRODUCTION_SSH_HOST" in build_section
-    assert "PRODUCTION_SSH_USER" in build_section
-    assert "PRODUCTION_SSH_KEY" in build_section
-    assert "/root/.ssh/xiaoxia_runtime_builder" in build_section
-    assert "runtime-images-${GITHUB_REF_NAME}.tar" in build_section
-    assert "release-${GITHUB_REF_NAME}.tar.gz" in build_section
-    assert "needs: build-production-runtime-images" in production_section
-    assert "runs-on: runtime-builder" in production_section
-    assert "Deploy production over SSH" in production_section
-    assert 'release_tar="/var/lib/xiaoxia-saas-production/release-${RELEASE_VERSION}.tar.gz"' in production_section
-    assert "runtime-images-${RELEASE_VERSION}.tar" in production_section
-    assert "apps/web/dist/index.html" in production_section
-    assert (
-        'HOST_PREFIX= WEB_IMAGE="xiaoxia-saas-web:${RELEASE_VERSION}" WEB_IMAGE_TAR="/var/lib/xiaoxia-saas-production/web-${RELEASE_VERSION}.tar" sh /var/lib/xiaoxia-saas-production/repo/infra/docker/deploy-production.sh'
-        in production_section
-    )
-
-
-def test_build_host_runbook_requires_off_production_runtime_builds():
-    runbook = Path("docs/BUILD-HOST-RUNBOOK.md").read_text(encoding="utf-8")
-
-    assert "Production host currently runs Gitea" in runbook
-    assert "scripts/build_release_images.sh v0.1.5" in runbook
-    assert "runtime-images-v0.1.5.tar" in runbook
-    assert "The deploy must fail if the runtime image tar is missing" in runbook
-    assert "python scripts/smoke_public_upload_flow.py" in runbook
-
-
-def test_runtime_builder_runner_runbook_matches_workflow():
-    runbook = Path("docs/GITEA-RUNTIME-BUILDER-RUNNER.md").read_text(encoding="utf-8")
-
-    assert "runs-on: runtime-builder" in runbook
-    assert "PRODUCTION_SSH_HOST" in runbook
-    assert "PRODUCTION_SSH_USER" in runbook
-    assert "PRODUCTION_SSH_KEY" in runbook
-    assert "/root/.ssh/xiaoxia_runtime_builder" in runbook
-    assert "Do not install this runner on the current production host" in runbook
-    assert "production deploy waits for the image job" in runbook
-    assert "release-<tag>.tar.gz" in runbook
-    assert "prebuilt Web dist" in runbook
-
-
-def test_deployment_docs_forbid_production_runtime_builds():
-    docs = Path("docs/DEPLOYMENT.md").read_text(encoding="utf-8")
-
-    assert "runtime-images-<tag>.tar" in docs
-    assert "API/Worker" in docs
-    assert "不是构建机" in docs
-    assert "infra/docker/deploy-production.sh" in docs
-
-
-def test_runtime_image_release_scripts_keep_builds_off_production():
-    build_script = Path("scripts/build_release_images.sh").read_text(encoding="utf-8")
-    deploy_script = Path("scripts/deploy_release_images_production.sh").read_text(encoding="utf-8")
-
-    assert "docker build --pull=false -f infra/docker/api.Dockerfile" in build_script
-    assert "docker build --pull=false -f infra/docker/worker.Dockerfile" in build_script
-    assert "docker save" in build_script
-    assert "Refusing to build runtime images on a host that is running production services." in build_script
-    assert "ALLOW_SHARED_PRODUCTION_BUILD_HOST=true" in build_script
-    assert "docker load -i" in deploy_script
-    assert 'API_IMAGE="xiaoxia-saas-api:$VERSION"' in deploy_script
-    assert 'WORKER_IMAGE="xiaoxia-saas-worker:$VERSION"' in deploy_script
-    assert "ALLOW_PRODUCTION_BUILDS=false" in deploy_script
-    assert 'docker compose --env-file "$ENV_FILE" build --pull=false web' in deploy_script
-    assert 'docker compose --env-file "$ENV_FILE" build --pull=false api' not in deploy_script
-    assert 'docker compose --env-file "$ENV_FILE" build --pull=false worker' not in deploy_script
-
-
-def test_production_release_checklist_matches_automatic_release_contract():
-    checklist = Path("docs/PRODUCTION-RELEASE-CHECKLIST.md").read_text(encoding="utf-8")
-
-    assert "runtime-builder" in checklist
-    assert "release-<tag>.tar.gz" in checklist
-    assert "apps/web/dist/index.html" in checklist
-    assert "HOST_PREFIX=" in checklist
-    assert "curl -fsS http://127.0.0.1:8001/health" in checklist
-    assert "xiaoxia-postgres-production" in checklist
-    assert "v0.1.6" in checklist
-    assert "v0.1.9" in checklist
-    assert "生产机补 build" in checklist
-
-
-def test_release_automation_retrospective_records_failed_probe_tags():
-    retrospective = Path("docs/RELEASE-AUTOMATION-RETROSPECTIVE-2026-06-22.md").read_text(encoding="utf-8")
-
-    assert "v0.1.9" in retrospective
-    assert "first verified end-to-end automatic production release" in retrospective
-    assert "v0.1.6" in retrospective
-    assert "v0.1.7" in retrospective
-    assert "v0.1.8" in retrospective
-    assert "Do not use `v0.1.6`, `v0.1.7`, or `v0.1.8` as rollback targets" in retrospective
-    assert "HOST_PREFIX=" in retrospective
-    assert "--exclude=./dist" in retrospective
-    assert "public_upload_flow=ok" in retrospective
-
-
-def test_backup_postgres_writes_manifest_and_version():
-    script = Path("scripts/backup_postgres.sh").read_text(encoding="utf-8")
-
-    assert "pg_dump" in script
-    assert "alembic_version.txt" in script
-    assert "manifest.txt" in script
-    assert "test -s" not in script
-    assert 'if [ ! -s "$DUMP_PATH" ]' in script

tests/unit/test_remove_member_use_case.py

@@ -1,357 +0,0 @@
-"""
-移除成员 Use Case 测试
-"""
-
-from unittest.mock import Mock
-
-import pytest
-
-from packages.application.workspace.remove_member_use_case import (
-    LeaveWorkspaceRequest,
-    LeaveWorkspaceUseCase,
-    RemoveMemberRequest,
-    RemoveMemberUseCase,
-)
-from packages.domain.entities import Workspace, WorkspaceMember, WorkspaceMemberRole
-
-
-class TestRemoveMemberUseCase:
-    """移除成员测试"""
-
-    @pytest.fixture
-    def mock_workspace_repo(self):
-        repo = Mock()
-        repo.find_by_id = Mock(return_value=None)
-        return repo
-
-    @pytest.fixture
-    def mock_member_repo(self):
-        repo = Mock()
-        repo.find_by_workspace_and_user = Mock(return_value=None)
-        repo.delete = Mock(return_value=True)
-        return repo
-
-    @pytest.fixture
-    def use_case(self, mock_workspace_repo, mock_member_repo):
-        return RemoveMemberUseCase(
-            workspace_repository=mock_workspace_repo,
-            workspace_member_repository=mock_member_repo,
-        )
-
-    @pytest.fixture
-    def test_workspace(self):
-        return Workspace(
-            id="workspace-123",
-            name="Test Workspace",
-            owner_user_id="owner-id",
-        )
-
-    def test_remove_member_success_by_owner(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试 Owner 移除成员成功"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-
-        owner_member = WorkspaceMember(
-            id="member-1",
-            user_id="owner-id",
-            role=WorkspaceMemberRole.OWNER,
-        )
-
-        target_member = WorkspaceMember(
-            id="member-2",
-            user_id="target-id",
-            role=WorkspaceMemberRole.MEMBER,
-        )
-
-        mock_member_repo.find_by_workspace_and_user.side_effect = [
-            owner_member,
-            target_member,
-        ]
-
-        request = RemoveMemberRequest(
-            requester_user_id="owner-id",
-            target_user_id="target-id",
-        )
-
-        success, error = use_case.execute(request)
-
-        assert success is True
-        assert error is None
-        mock_member_repo.delete.assert_called_once_with("member-2")
-
-    def test_remove_member_success_by_admin(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试 Admin 移除普通成员成功"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-
-        admin_member = WorkspaceMember(
-            id="member-1",
-            user_id="admin-id",
-            role=WorkspaceMemberRole.ADMIN,
-        )
-
-        target_member = WorkspaceMember(
-            id="member-2",
-            user_id="target-id",
-            role=WorkspaceMemberRole.VIEWER,
-        )
-
-        mock_member_repo.find_by_workspace_and_user.side_effect = [
-            admin_member,
-            target_member,
-        ]
-
-        request = RemoveMemberRequest(
-            requester_user_id="admin-id",
-            target_user_id="target-id",
-        )
-
-        success, error = use_case.execute(request)
-
-        assert success is True
-        assert error is None
-
-    def test_remove_member_cannot_remove_owner(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试不能移除 Owner"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-
-        admin_member = WorkspaceMember(
-            id="member-1",
-            user_id="admin-id",
-            role=WorkspaceMemberRole.ADMIN,
-        )
-
-        owner_member = WorkspaceMember(
-            id="member-2",
-            user_id="owner-id",
-            role=WorkspaceMemberRole.OWNER,
-        )
-
-        mock_member_repo.find_by_workspace_and_user.side_effect = [
-            admin_member,
-            owner_member,
-        ]
-
-        request = RemoveMemberRequest(
-            requester_user_id="admin-id",
-            target_user_id="owner-id",
-        )
-
-        success, error = use_case.execute(request)
-
-        assert success is False
-        assert error == "Cannot remove the workspace owner"
-
-    def test_remove_member_admin_cannot_remove_admin(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试 Admin 不能移除另一个 Admin"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-
-        admin_member1 = WorkspaceMember(
-            id="member-1",
-            user_id="admin-id-1",
-            role=WorkspaceMemberRole.ADMIN,
-        )
-
-        admin_member2 = WorkspaceMember(
-            id="member-2",
-            user_id="admin-id-2",
-            role=WorkspaceMemberRole.ADMIN,
-        )
-
-        mock_member_repo.find_by_workspace_and_user.side_effect = [
-            admin_member1,
-            admin_member2,
-        ]
-
-        request = RemoveMemberRequest(
-            requester_user_id="admin-id-1",
-            target_user_id="admin-id-2",
-        )
-
-        success, error = use_case.execute(request)
-
-        assert success is False
-        assert error == "Admins cannot remove other admins"
-
-    def test_remove_member_cannot_remove_self(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试不能移除自己"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-
-        admin_member = WorkspaceMember(
-            id="member-1",
-            user_id="admin-id",
-            role=WorkspaceMemberRole.ADMIN,
-        )
-
-        mock_member_repo.find_by_workspace_and_user.return_value = admin_member
-
-        request = RemoveMemberRequest(
-            requester_user_id="admin-id",
-            target_user_id="admin-id",
-        )
-
-        success, error = use_case.execute(request)
-
-        assert success is False
-        assert "Cannot remove yourself" in error
-
-    def test_remove_member_no_permission(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试普通成员没有权限移除"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-
-        regular_member = WorkspaceMember(
-            id="member-1",
-            user_id="user-id",
-            role=WorkspaceMemberRole.MEMBER,
-        )
-
-        mock_member_repo.find_by_workspace_and_user.return_value = regular_member
-
-        request = RemoveMemberRequest(
-            requester_user_id="user-id",
-            target_user_id="target-id",
-        )
-
-        success, error = use_case.execute(request)
-
-        assert success is False
-        assert error == "Only owners and admins can remove members"
-
-
-class TestLeaveWorkspaceUseCase:
-    """离开 Workspace 测试"""
-
-    @pytest.fixture
-    def mock_workspace_repo(self):
-        repo = Mock()
-        repo.find_by_id = Mock(return_value=None)
-        return repo
-
-    @pytest.fixture
-    def mock_member_repo(self):
-        repo = Mock()
-        repo.find_by_workspace_and_user = Mock(return_value=None)
-        repo.delete = Mock(return_value=True)
-        return repo
-
-    @pytest.fixture
-    def use_case(self, mock_workspace_repo, mock_member_repo):
-        return LeaveWorkspaceUseCase(
-            workspace_repository=mock_workspace_repo,
-            workspace_member_repository=mock_member_repo,
-        )
-
-    @pytest.fixture
-    def test_workspace(self):
-        return Workspace(
-            id="workspace-123",
-            name="Test Workspace",
-            owner_user_id="owner-id",
-        )
-
-    def test_leave_workspace_success(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试离开工作空间成功"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-
-        member = WorkspaceMember(
-            id="member-1",
-            user_id="user-id",
-            role=WorkspaceMemberRole.MEMBER,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = member
-
-        request = LeaveWorkspaceRequest(
-            user_id="user-id",
-        )
-
-        success, error = use_case.execute(request)
-
-        assert success is True
-        assert error is None
-        mock_member_repo.delete.assert_called_once_with("member-1")
-
-    def test_leave_workspace_owner_cannot_leave(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试 Owner 不能离开"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-
-        owner_member = WorkspaceMember(
-            id="member-1",
-            user_id="owner-id",
-            role=WorkspaceMemberRole.OWNER,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = owner_member
-
-        request = LeaveWorkspaceRequest(
-            user_id="owner-id",
-        )
-
-        success, error = use_case.execute(request)
-
-        assert success is False
-        assert "Owner cannot leave workspace" in error
-
-    def test_leave_workspace_not_member(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试不是成员"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-        mock_member_repo.find_by_workspace_and_user.return_value = None
-
-        request = LeaveWorkspaceRequest(
-            user_id="user-id",
-        )
-
-        success, error = use_case.execute(request)
-
-        assert success is False
-        assert error == "You are not a member of this workspace"

tests/unit/test_smoke_external_services.py

@@ -1,107 +0,0 @@
-import types
-
-import scripts.smoke_external_services as smoke
-
-
-def test_external_smoke_skips_disabled_services(monkeypatch, capsys):
-    for key in [
-        "ENABLE_REDIS_SESSIONS",
-        "ENABLE_EMAIL_DELIVERY",
-        "OSS_ENDPOINT",
-        "OSS_ACCESS_KEY_ID",
-        "OSS_ACCESS_KEY_SECRET",
-        "OSS_BUCKET_NAME",
-    ]:
-        monkeypatch.delenv(key, raising=False)
-
-    smoke.smoke_redis(strict=False)
-    smoke.smoke_smtp(strict=False, send_email_to=None)
-    smoke.smoke_oss(strict=False)
-
-    output = capsys.readouterr().out
-    assert "SKIP redis sessions" in output
-    assert "SKIP smtp" in output
-    assert "SKIP oss" in output
-
-
-def test_external_smoke_requires_oss_in_strict_mode(monkeypatch):
-    monkeypatch.delenv("OSS_ENDPOINT", raising=False)
-
-    try:
-        smoke.smoke_oss(strict=True)
-    except RuntimeError as error:
-        assert "missing required env: OSS_ENDPOINT" in str(error)
-    else:
-        raise AssertionError("strict OSS smoke should fail without credentials")
-
-
-def test_external_smoke_redis_round_trip(monkeypatch, capsys):
-    class FakeRedis:
-        def __init__(self):
-            self.values = {}
-
-        def ping(self):
-            return True
-
-        def setex(self, key, seconds, value):
-            self.values[key] = value
-
-        def get(self, key):
-            return self.values[key]
-
-        def delete(self, key):
-            self.values.pop(key, None)
-
-    fake_client = FakeRedis()
-    monkeypatch.setenv("ENABLE_REDIS_SESSIONS", "true")
-    monkeypatch.setenv("REDIS_URL", "redis://redis:6379/0")
-    monkeypatch.setattr(smoke.redis.Redis, "from_url", lambda *args, **kwargs: fake_client)
-
-    smoke.smoke_redis(strict=False)
-
-    assert "OK redis sessions" in capsys.readouterr().out
-
-
-def test_external_smoke_can_skip_smtp_in_strict_mode(monkeypatch, capsys):
-    monkeypatch.setattr(smoke, "smoke_redis", lambda strict: print("OK redis sessions"))
-    monkeypatch.setattr(smoke, "smoke_oss", lambda strict: print("OK oss upload/download/delete"))
-
-    assert smoke.main(["--strict", "--skip-smtp"]) == 0
-
-    output = capsys.readouterr().out
-    assert "SKIP smtp: explicitly skipped" in output
-    assert "OK oss upload/download/delete" in output
-
-
-def test_external_smoke_smtp_connect_login(monkeypatch, capsys):
-    actions = []
-
-    class FakeSMTP:
-        def __init__(self, host, port, timeout):
-            actions.append((host, port, timeout))
-
-        def __enter__(self):
-            return self
-
-        def __exit__(self, exc_type, exc, tb):
-            return False
-
-        def starttls(self):
-            actions.append("tls")
-
-        def login(self, user, password):
-            actions.append(("login", user, password))
-
-    monkeypatch.setenv("ENABLE_EMAIL_DELIVERY", "true")
-    monkeypatch.setenv("SMTP_HOST", "smtp.local")
-    monkeypatch.setenv("SMTP_PORT", "587")
-    monkeypatch.setenv("SMTP_USER", "mailer")
-    monkeypatch.setenv("SMTP_PASSWORD", "secret")
-    monkeypatch.setenv("SMTP_FROM_EMAIL", "noreply@example.test")
-    monkeypatch.setenv("SMTP_USE_TLS", "true")
-    monkeypatch.setattr(smoke.smtplib, "SMTP", FakeSMTP)
-
-    smoke.smoke_smtp(strict=False, send_email_to=None)
-
-    assert actions == [("smtp.local", 587, 10), "tls", ("login", "mailer", "secret")]
-    assert "OK smtp connect/login" in capsys.readouterr().out

tests/unit/test_subscription_use_case.py

@@ -1,338 +0,0 @@
-"""
-Subscription 管理 Use Case 测试
-"""
-
-from unittest.mock import Mock
-
-import pytest
-
-from packages.application.workspace.subscription_use_case import (
-    CancelSubscriptionRequest,
-    CancelSubscriptionUseCase,
-    UpgradeSubscriptionRequest,
-    UpgradeSubscriptionUseCase,
-)
-from packages.domain.entities import Workspace, WorkspaceMember, WorkspaceMemberRole
-
-
-class TestUpgradeSubscriptionUseCase:
-    """升级订阅测试"""
-
-    @pytest.fixture
-    def mock_workspace_repo(self):
-        repo = Mock()
-        repo.find_by_id = Mock(return_value=None)
-        repo.save = Mock()
-        return repo
-
-    @pytest.fixture
-    def mock_member_repo(self):
-        repo = Mock()
-        repo.find_by_workspace_and_user = Mock(return_value=None)
-        return repo
-
-    @pytest.fixture
-    def use_case(self, mock_workspace_repo, mock_member_repo):
-        return UpgradeSubscriptionUseCase(
-            workspace_repository=mock_workspace_repo,
-            workspace_member_repository=mock_member_repo,
-        )
-
-    def test_upgrade_from_free_to_pro(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-    ):
-        """测试从 Free 升级到 Pro"""
-        workspace = Workspace(
-            id="workspace-123",
-            name="Test Workspace",
-            owner_user_id="owner-id",
-            subscription_plan="free",
-        )
-        mock_workspace_repo.find_by_id.return_value = workspace
-
-        owner_member = WorkspaceMember(
-            id="member-1",
-            user_id="owner-id",
-            role=WorkspaceMemberRole.OWNER,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = owner_member
-
-        request = UpgradeSubscriptionRequest(
-            requester_user_id="owner-id",
-            new_plan="pro",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert error is None
-        assert response is not None
-        assert response.old_plan == "free"
-        assert response.new_plan == "pro"
-        assert response.max_projects == 999999
-        assert response.max_storage_gb == 100
-
-        # 验证更新了 workspace
-        assert workspace.subscription_plan == "pro"
-        assert workspace.max_projects == 999999
-        assert workspace.subscription_expires_at is not None
-
-    def test_upgrade_from_pro_to_enterprise(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-    ):
-        """测试从 Pro 升级到 Enterprise"""
-        workspace = Workspace(
-            id="workspace-123",
-            name="Test Workspace",
-            owner_user_id="owner-id",
-            subscription_plan="pro",
-            max_projects=999999,
-            max_storage_gb=100,
-        )
-        mock_workspace_repo.find_by_id.return_value = workspace
-
-        owner_member = WorkspaceMember(
-            id="member-1",
-            user_id="owner-id",
-            role=WorkspaceMemberRole.OWNER,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = owner_member
-
-        request = UpgradeSubscriptionRequest(
-            requester_user_id="owner-id",
-            new_plan="enterprise",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert error is None
-        assert response is not None
-        assert response.old_plan == "pro"
-        assert response.new_plan == "enterprise"
-        assert response.max_storage_gb == 1000
-
-    def test_upgrade_cannot_downgrade(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-    ):
-        """测试不能降级"""
-        workspace = Workspace(
-            id="workspace-123",
-            name="Test Workspace",
-            owner_user_id="owner-id",
-            subscription_plan="pro",
-        )
-        mock_workspace_repo.find_by_id.return_value = workspace
-
-        owner_member = WorkspaceMember(
-            id="member-1",
-            user_id="owner-id",
-            role=WorkspaceMemberRole.OWNER,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = owner_member
-
-        request = UpgradeSubscriptionRequest(
-            requester_user_id="owner-id",
-            new_plan="free",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert "Cannot downgrade" in error
-
-    def test_upgrade_already_on_plan(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-    ):
-        """测试已经是该计划"""
-        workspace = Workspace(
-            id="workspace-123",
-            name="Test Workspace",
-            owner_user_id="owner-id",
-            subscription_plan="pro",
-        )
-        mock_workspace_repo.find_by_id.return_value = workspace
-
-        owner_member = WorkspaceMember(
-            id="member-1",
-            user_id="owner-id",
-            role=WorkspaceMemberRole.OWNER,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = owner_member
-
-        request = UpgradeSubscriptionRequest(
-            requester_user_id="owner-id",
-            new_plan="pro",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert "already on pro plan" in error
-
-    def test_upgrade_only_owner_can_upgrade(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-    ):
-        """测试只有 Owner 可以升级"""
-        workspace = Workspace(
-            id="workspace-123",
-            name="Test Workspace",
-            owner_user_id="owner-id",
-            subscription_plan="free",
-        )
-        mock_workspace_repo.find_by_id.return_value = workspace
-
-        admin_member = WorkspaceMember(
-            id="member-1",
-            user_id="admin-id",
-            role=WorkspaceMemberRole.ADMIN,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = admin_member
-
-        request = UpgradeSubscriptionRequest(
-            requester_user_id="admin-id",
-            new_plan="pro",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert "Only workspace owner" in error
-
-
-class TestCancelSubscriptionUseCase:
-    """取消订阅测试"""
-
-    @pytest.fixture
-    def mock_workspace_repo(self):
-        repo = Mock()
-        repo.find_by_id = Mock(return_value=None)
-        repo.save = Mock()
-        return repo
-
-    @pytest.fixture
-    def mock_member_repo(self):
-        repo = Mock()
-        repo.find_by_workspace_and_user = Mock(return_value=None)
-        return repo
-
-    @pytest.fixture
-    def use_case(self, mock_workspace_repo, mock_member_repo):
-        return CancelSubscriptionUseCase(
-            workspace_repository=mock_workspace_repo,
-            workspace_member_repository=mock_member_repo,
-        )
-
-    def test_cancel_subscription_success(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-    ):
-        """测试取消订阅成功"""
-        workspace = Workspace(
-            id="workspace-123",
-            name="Test Workspace",
-            owner_user_id="owner-id",
-            subscription_plan="pro",
-            max_projects=999999,
-            max_storage_gb=100,
-        )
-        mock_workspace_repo.find_by_id.return_value = workspace
-
-        owner_member = WorkspaceMember(
-            id="member-1",
-            user_id="owner-id",
-            role=WorkspaceMemberRole.OWNER,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = owner_member
-
-        request = CancelSubscriptionRequest(
-            requester_user_id="owner-id",
-        )
-
-        success, error = use_case.execute(request)
-
-        assert success is True
-        assert error is None
-
-        # 验证降级到 free
-        assert workspace.subscription_plan == "free"
-        assert workspace.max_projects == 3
-        assert workspace.max_storage_gb == 10
-        assert workspace.subscription_expires_at is None
-
-    def test_cancel_already_free(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-    ):
-        """测试已经是免费计划"""
-        workspace = Workspace(
-            id="workspace-123",
-            name="Test Workspace",
-            owner_user_id="owner-id",
-            subscription_plan="free",
-        )
-        mock_workspace_repo.find_by_id.return_value = workspace
-
-        owner_member = WorkspaceMember(
-            id="member-1",
-            user_id="owner-id",
-            role=WorkspaceMemberRole.OWNER,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = owner_member
-
-        request = CancelSubscriptionRequest(
-            requester_user_id="owner-id",
-        )
-
-        success, error = use_case.execute(request)
-
-        assert success is False
-        assert "already on free plan" in error
-
-    def test_cancel_only_owner_can_cancel(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-    ):
-        """测试只有 Owner 可以取消"""
-        workspace = Workspace(
-            id="workspace-123",
-            name="Test Workspace",
-            owner_user_id="owner-id",
-            subscription_plan="pro",
-        )
-        mock_workspace_repo.find_by_id.return_value = workspace
-
-        admin_member = WorkspaceMember(
-            id="member-1",
-            user_id="admin-id",
-            role=WorkspaceMemberRole.ADMIN,
-        )
-        mock_member_repo.find_by_workspace_and_user.return_value = admin_member
-
-        request = CancelSubscriptionRequest(
-            requester_user_id="admin-id",
-        )
-
-        success, error = use_case.execute(request)
-
-        assert success is False
-        assert "Only workspace owner" in error

tests/unit/test_update_member_role_use_case.py

@@ -1,313 +0,0 @@
-"""
-修改成员角色 Use Case 测试
-"""
-
-from unittest.mock import Mock
-
-import pytest
-
-from packages.application.workspace.update_member_role_use_case import (
-    UpdateMemberRoleRequest,
-    UpdateMemberRoleUseCase,
-)
-from packages.domain.entities import Workspace, WorkspaceMember, WorkspaceMemberRole
-
-
-class TestUpdateMemberRoleUseCase:
-    """修改成员角色测试"""
-
-    @pytest.fixture
-    def mock_workspace_repo(self):
-        repo = Mock()
-        repo.find_by_id = Mock(return_value=None)
-        return repo
-
-    @pytest.fixture
-    def mock_member_repo(self):
-        repo = Mock()
-        repo.find_by_workspace_and_user = Mock(return_value=None)
-        repo.save = Mock()
-        return repo
-
-    @pytest.fixture
-    def use_case(self, mock_workspace_repo, mock_member_repo):
-        return UpdateMemberRoleUseCase(
-            workspace_repository=mock_workspace_repo,
-            workspace_member_repository=mock_member_repo,
-        )
-
-    @pytest.fixture
-    def test_workspace(self):
-        return Workspace(
-            id="workspace-123",
-            name="Test Workspace",
-            owner_user_id="owner-id",
-        )
-
-    def test_update_role_success_by_owner(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试 Owner 修改成员角色成功"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-
-        owner_member = WorkspaceMember(
-            id="member-1",
-            user_id="owner-id",
-            role=WorkspaceMemberRole.OWNER,
-        )
-
-        target_member = WorkspaceMember(
-            id="member-2",
-            user_id="target-id",
-            role=WorkspaceMemberRole.MEMBER,
-        )
-
-        mock_member_repo.find_by_workspace_and_user.side_effect = [
-            owner_member,
-            target_member,
-        ]
-
-        request = UpdateMemberRoleRequest(
-            requester_user_id="owner-id",
-            target_user_id="target-id",
-            new_role="admin",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert error is None
-        assert response is not None
-        assert response.user_id == "target-id"
-        assert response.old_role == "member"
-        assert response.new_role == "admin"
-
-        # 验证更新了角色
-        assert target_member.role == "admin"
-        mock_member_repo.save.assert_called_once()
-
-    def test_update_role_success_by_admin(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试 Admin 修改普通成员角色成功"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-
-        admin_member = WorkspaceMember(
-            id="member-1",
-            user_id="admin-id",
-            role=WorkspaceMemberRole.ADMIN,
-        )
-
-        target_member = WorkspaceMember(
-            id="member-2",
-            user_id="target-id",
-            role=WorkspaceMemberRole.VIEWER,
-        )
-
-        mock_member_repo.find_by_workspace_and_user.side_effect = [
-            admin_member,
-            target_member,
-        ]
-
-        request = UpdateMemberRoleRequest(
-            requester_user_id="admin-id",
-            target_user_id="target-id",
-            new_role="member",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert error is None
-        assert response is not None
-
-    def test_update_role_cannot_change_owner(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试不能修改 Owner 角色"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-
-        admin_member = WorkspaceMember(
-            id="member-1",
-            user_id="admin-id",
-            role=WorkspaceMemberRole.ADMIN,
-        )
-
-        owner_member = WorkspaceMember(
-            id="member-2",
-            user_id="owner-id",
-            role=WorkspaceMemberRole.OWNER,
-        )
-
-        mock_member_repo.find_by_workspace_and_user.side_effect = [
-            admin_member,
-            owner_member,
-        ]
-
-        request = UpdateMemberRoleRequest(
-            requester_user_id="admin-id",
-            target_user_id="owner-id",
-            new_role="member",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "Cannot change the owner's role"
-
-    def test_update_role_admin_cannot_change_admin(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试 Admin 不能修改另一个 Admin 角色"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-
-        admin_member1 = WorkspaceMember(
-            id="member-1",
-            user_id="admin-id-1",
-            role=WorkspaceMemberRole.ADMIN,
-        )
-
-        admin_member2 = WorkspaceMember(
-            id="member-2",
-            user_id="admin-id-2",
-            role=WorkspaceMemberRole.ADMIN,
-        )
-
-        mock_member_repo.find_by_workspace_and_user.side_effect = [
-            admin_member1,
-            admin_member2,
-        ]
-
-        request = UpdateMemberRoleRequest(
-            requester_user_id="admin-id-1",
-            target_user_id="admin-id-2",
-            new_role="member",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "Admins cannot change other admins' roles"
-
-    def test_update_role_cannot_change_self(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试不能修改自己的角色"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-
-        admin_member = WorkspaceMember(
-            id="member-1",
-            user_id="admin-id",
-            role=WorkspaceMemberRole.ADMIN,
-        )
-
-        mock_member_repo.find_by_workspace_and_user.return_value = admin_member
-
-        request = UpdateMemberRoleRequest(
-            requester_user_id="admin-id",
-            target_user_id="admin-id",
-            new_role="member",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "Cannot change your own role"
-
-    def test_update_role_invalid_role(self, use_case):
-        """测试无效角色（不能改为 owner）"""
-        request = UpdateMemberRoleRequest(
-            requester_user_id="admin-id",
-            target_user_id="target-id",
-            new_role="owner",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert "Invalid role" in error
-
-    def test_update_role_already_has_role(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试角色相同"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-
-        owner_member = WorkspaceMember(
-            id="member-1",
-            user_id="owner-id",
-            role=WorkspaceMemberRole.OWNER,
-        )
-
-        target_member = WorkspaceMember(
-            id="member-2",
-            user_id="target-id",
-            role=WorkspaceMemberRole.ADMIN,
-        )
-
-        mock_member_repo.find_by_workspace_and_user.side_effect = [
-            owner_member,
-            target_member,
-        ]
-
-        request = UpdateMemberRoleRequest(
-            requester_user_id="owner-id",
-            target_user_id="target-id",
-            new_role="admin",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert "already has the admin role" in error
-
-    def test_update_role_no_permission(
-        self,
-        use_case,
-        mock_workspace_repo,
-        mock_member_repo,
-        test_workspace,
-    ):
-        """测试普通成员没有权限"""
-        mock_workspace_repo.find_by_id.return_value = test_workspace
-
-        regular_member = WorkspaceMember(
-            id="member-1",
-            user_id="user-id",
-            role=WorkspaceMemberRole.MEMBER,
-        )
-
-        mock_member_repo.find_by_workspace_and_user.return_value = regular_member
-
-        request = UpdateMemberRoleRequest(
-            requester_user_id="user-id",
-            target_user_id="target-id",
-            new_role="admin",
-        )
-
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert error == "Only owners and admins can change member roles"

tests/unit/test_validate_release_env.py

@@ -1,87 +0,0 @@
-from pathlib import Path
-
-from scripts.validate_release_env import parse_env_file, validate
-
-
-def test_validate_release_env_accepts_strict_production(tmp_path: Path):
-    env_file = tmp_path / ".env.production"
-    env_file.write_text(
-        "\n".join(
-            [
-                "APP_ENV=production",
-                "DATABASE_URL=postgresql+psycopg://user:pass@db:5432/app",
-                "JWT_SECRET_KEY=abcdefghijklmnopqrstuvwxyz123456",
-                "REDIS_URL=redis://redis:6379/0",
-                "ENABLE_EMAIL_DELIVERY=true",
-                "ENABLE_REDIS_SESSIONS=true",
-                "SMTP_HOST=smtp.example.internal",
-                "SMTP_PORT=587",
-                "SMTP_USER=mailer",
-                "SMTP_PASSWORD=strong-password",
-                "SMTP_FROM_EMAIL=noreply@xiaoxia.local",
-                "SMTP_FROM_NAME=Xiaoxia",
-                "OSS_ENDPOINT=oss-cn-hangzhou.aliyuncs.com",
-                "OSS_ACCESS_KEY_ID=ak-real",
-                "OSS_ACCESS_KEY_SECRET=sk-real-secret",
-                "OSS_BUCKET_NAME=xiaoxia-prod",
-                "GENERATED_FILES_HOST_DIR=/var/lib/xiaoxia-saas-production/generated",
-                "DEBUG=false",
-                "AUTO_CREATE_SCHEMA=false",
-            ]
-        ),
-        encoding="utf-8",
-    )
-
-    assert validate(parse_env_file(env_file), strict_external=True) == []
-
-
-def test_validate_release_env_allows_disabled_smtp_with_strict_external(tmp_path: Path):
-    env_file = tmp_path / ".env.production"
-    env_file.write_text(
-        "\n".join(
-            [
-                "APP_ENV=production",
-                "DATABASE_URL=postgresql+psycopg://user:pass@db:5432/app",
-                "JWT_SECRET_KEY=abcdefghijklmnopqrstuvwxyz123456",
-                "REDIS_URL=redis://redis:6379/0",
-                "ENABLE_EMAIL_DELIVERY=false",
-                "ENABLE_REDIS_SESSIONS=true",
-                "OSS_ENDPOINT=oss-cn-hangzhou.aliyuncs.com",
-                "OSS_ACCESS_KEY_ID=ak-real",
-                "OSS_ACCESS_KEY_SECRET=sk-real-secret",
-                "OSS_BUCKET_NAME=xiaoxia-prod",
-                "GENERATED_FILES_HOST_DIR=/var/lib/xiaoxia-saas-production/generated",
-                "DEBUG=false",
-                "AUTO_CREATE_SCHEMA=false",
-            ]
-        ),
-        encoding="utf-8",
-    )
-
-    assert validate(parse_env_file(env_file), strict_external=True) == []
-
-
-def test_validate_release_env_rejects_placeholders_and_staging_paths(tmp_path: Path):
-    env_file = tmp_path / ".env.production"
-    env_file.write_text(
-        "\n".join(
-            [
-                "APP_ENV=production",
-                "DATABASE_URL=postgresql+psycopg://user:pass@db:5432/app",
-                "JWT_SECRET_KEY=your-super-secret-key-change-this-in-production-min-32-chars",
-                "REDIS_URL=redis://redis:6379/0",
-                "GENERATED_FILES_HOST_DIR=/var/lib/xiaoxia-saas-staging/generated",
-                "DEBUG=true",
-                "AUTO_CREATE_SCHEMA=true",
-            ]
-        ),
-        encoding="utf-8",
-    )
-
-    errors = validate(parse_env_file(env_file), strict_external=True)
-
-    assert "placeholder value remains in JWT_SECRET_KEY" in errors
-    assert "GENERATED_FILES_HOST_DIR must not point at staging in production" in errors
-    assert "DEBUG must be false in production" in errors
-    assert "AUTO_CREATE_SCHEMA must not be enabled outside development" in errors
-    assert "missing external-service env: OSS_ACCESS_KEY_ID" in errors

tests/unit/test_worker_ingest_status.py

@@ -1,9 +0,0 @@
-from pathlib import Path
-
-
-def test_ingest_marks_created_assets_ready_with_size_metadata():
-    source = Path("apps/worker/worker_app/tasks/ingest.py").read_text(encoding="utf-8")
-
-    assert "status=AssetStatus.READY" in source
-    assert 'file_size=int(metadata["size_bytes"])' in source
-    assert 'duration=float(metadata["duration"])' in source