diff --git a/tests/unit/test_accept_invitation_use_case.py b/tests/unit/test_accept_invitation_use_case.py deleted file mode 100644 index d3e97bf09..000000000 --- a/tests/unit/test_accept_invitation_use_case.py +++ /dev/null @@ -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" diff --git a/tests/unit/test_alembic_preflight.py b/tests/unit/test_alembic_preflight.py deleted file mode 100644 index f7cec0004..000000000 --- a/tests/unit/test_alembic_preflight.py +++ /dev/null @@ -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" diff --git a/tests/unit/test_architecture_boundaries.py b/tests/unit/test_architecture_boundaries.py deleted file mode 100644 index 4828e5f42..000000000 --- a/tests/unit/test_architecture_boundaries.py +++ /dev/null @@ -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 == [] diff --git a/tests/unit/test_asset_library_repository.py b/tests/unit/test_asset_library_repository.py deleted file mode 100644 index 3297a48f5..000000000 --- a/tests/unit/test_asset_library_repository.py +++ /dev/null @@ -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 diff --git a/tests/unit/test_auth.py b/tests/unit/test_auth.py deleted file mode 100644 index 95ba7c698..000000000 --- a/tests/unit/test_auth.py +++ /dev/null @@ -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 diff --git a/tests/unit/test_auth_middleware.py b/tests/unit/test_auth_middleware.py deleted file mode 100644 index b4ac5d812..000000000 --- a/tests/unit/test_auth_middleware.py +++ /dev/null @@ -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()) diff --git a/tests/unit/test_create_workspace_use_case.py b/tests/unit/test_create_workspace_use_case.py deleted file mode 100644 index f72bcf50c..000000000 --- a/tests/unit/test_create_workspace_use_case.py +++ /dev/null @@ -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 diff --git a/tests/unit/test_invite_member_use_case.py b/tests/unit/test_invite_member_use_case.py deleted file mode 100644 index 8f5ca3373..000000000 --- a/tests/unit/test_invite_member_use_case.py +++ /dev/null @@ -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 diff --git a/tests/unit/test_jwt_service.py b/tests/unit/test_jwt_service.py deleted file mode 100644 index 41d414d6d..000000000 --- a/tests/unit/test_jwt_service.py +++ /dev/null @@ -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 diff --git a/tests/unit/test_list_members_use_case.py b/tests/unit/test_list_members_use_case.py deleted file mode 100644 index 13ddfe736..000000000 --- a/tests/unit/test_list_members_use_case.py +++ /dev/null @@ -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" diff --git a/tests/unit/test_login_use_case.py b/tests/unit/test_login_use_case.py deleted file mode 100644 index 8129b5021..000000000 --- a/tests/unit/test_login_use_case.py +++ /dev/null @@ -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" diff --git a/tests/unit/test_permissions.py b/tests/unit/test_permissions.py deleted file mode 100644 index b3df90f29..000000000 --- a/tests/unit/test_permissions.py +++ /dev/null @@ -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 diff --git a/tests/unit/test_quota.py b/tests/unit/test_quota.py deleted file mode 100644 index e0ab90586..000000000 --- a/tests/unit/test_quota.py +++ /dev/null @@ -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 diff --git a/tests/unit/test_release_scripts.py b/tests/unit/test_release_scripts.py deleted file mode 100644 index 142b03682..000000000 --- a/tests/unit/test_release_scripts.py +++ /dev/null @@ -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 <.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-.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-.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 diff --git a/tests/unit/test_remove_member_use_case.py b/tests/unit/test_remove_member_use_case.py deleted file mode 100644 index 613c7667a..000000000 --- a/tests/unit/test_remove_member_use_case.py +++ /dev/null @@ -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" diff --git a/tests/unit/test_smoke_external_services.py b/tests/unit/test_smoke_external_services.py deleted file mode 100644 index 4c1de4012..000000000 --- a/tests/unit/test_smoke_external_services.py +++ /dev/null @@ -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 diff --git a/tests/unit/test_subscription_use_case.py b/tests/unit/test_subscription_use_case.py deleted file mode 100644 index d5c914cd9..000000000 --- a/tests/unit/test_subscription_use_case.py +++ /dev/null @@ -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 diff --git a/tests/unit/test_update_member_role_use_case.py b/tests/unit/test_update_member_role_use_case.py deleted file mode 100644 index da255c05d..000000000 --- a/tests/unit/test_update_member_role_use_case.py +++ /dev/null @@ -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" diff --git a/tests/unit/test_validate_release_env.py b/tests/unit/test_validate_release_env.py deleted file mode 100644 index e259270e0..000000000 --- a/tests/unit/test_validate_release_env.py +++ /dev/null @@ -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 diff --git a/tests/unit/test_worker_ingest_status.py b/tests/unit/test_worker_ingest_status.py deleted file mode 100644 index 2b8a82af4..000000000 --- a/tests/unit/test_worker_ingest_status.py +++ /dev/null @@ -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