From ac093ae55bb6a2d26a25dcdce25392454fd7a8e0 Mon Sep 17 00:00:00 2001 From: Stephen Zhou <38493346+hyoban@users.noreply.github.com> Date: Thu, 15 Oct 2026 19:47:24 +0800 Subject: [PATCH] test(api): replace mocks in knowledge fs control space management tests Co-Authored-By: Claude Opus 5.5 --- ...t_knowledge_fs_control_space_management.py | 105 ++++++++++++------ 1 file changed, 74 insertions(+), 31 deletions(-) diff --git a/api/tests/unit_tests/services/test_knowledge_fs_control_space_management.py b/api/tests/unit_tests/services/test_knowledge_fs_control_space_management.py index 1070d4a072a..ef65fc1cf7c 100644 --- a/api/tests/unit_tests/services/test_knowledge_fs_control_space_management.py +++ b/api/tests/unit_tests/services/test_knowledge_fs_control_space_management.py @@ -1,8 +1,7 @@ from __future__ import annotations +from dataclasses import dataclass, field from datetime import datetime -from types import SimpleNamespace -from unittest.mock import MagicMock import pytest from sqlalchemy import select @@ -23,6 +22,54 @@ from services.knowledge_fs.control_space_management import ( ) +def _control_space(control_space_id: str) -> KnowledgeFSControlSpace: + control_space = KnowledgeFSControlSpace( + tenant_id="tenant-1", + owner_account_id="account-1", + provisioning_key=f"provision-{control_space_id}", + knowledge_space_id="space-1", + state=KnowledgeFSControlSpaceState.ACTIVE, + ) + control_space.id = control_space_id + return control_space + + +@dataclass +class _QueuedRegister: + """Replays queued registration outcomes and records each registration it receives.""" + + outcomes: list[tuple[KnowledgeFSControlSpace, bool]] + calls: list[KnowledgeFSControlSpaceRegistration] = field(default_factory=list) + + def __call__(self, registration: KnowledgeFSControlSpaceRegistration) -> tuple[KnowledgeFSControlSpace, bool]: + self.calls.append(registration) + return self.outcomes.pop(0) + + +@dataclass +class _RecordingRepositoryFactory: + sessions: list[Session] = field(default_factory=list) + + def __call__(self, session: Session) -> object: + self.sessions.append(session) + return self + + +@dataclass +class _RecordingLifecycle: + result: KnowledgeFSControlSpace + repositories: list[object] = field(default_factory=list) + transitions: list[dict[str, object]] = field(default_factory=list) + + def __call__(self, repository: object) -> _RecordingLifecycle: + self.repositories.append(repository) + return self + + def transition(self, **kwargs: object) -> KnowledgeFSControlSpace: + self.transitions.append(kwargs) + return self.result + + @pytest.mark.parametrize( "sqlite_session", [ @@ -110,39 +157,31 @@ def test_registration_rejects_reusing_either_remote_or_local_identity(sqlite_ses def test_backfill_is_dry_run_by_default_and_accounts_for_replays(monkeypatch: pytest.MonkeyPatch) -> None: - service = KnowledgeFSControlSpaceManagementService(MagicMock()) + service = KnowledgeFSControlSpaceManagementService(sessionmaker()) registrations = ( KnowledgeFSControlSpaceRegistration("tenant-1", "account-1", "provision-1", "space-1", 1), KnowledgeFSControlSpaceRegistration("tenant-1", "account-1", "provision-2", "space-2", 1), ) - register = MagicMock( - side_effect=( - (SimpleNamespace(id="control-1"), False), - (SimpleNamespace(id="control-2"), True), - ) + register = _QueuedRegister( + outcomes=[ + (_control_space("control-1"), False), + (_control_space("control-2"), True), + ] ) monkeypatch.setattr(service, "register", register) assert service.backfill(registrations, apply=False) == (2, 0, 0) - register.assert_not_called() + assert register.calls == [] assert service.backfill(registrations, apply=True) == (2, 1, 1) - assert register.call_count == 2 + assert len(register.calls) == 2 def test_repair_registration_delegates_an_active_cas_transition(monkeypatch: pytest.MonkeyPatch) -> None: - session = MagicMock() - transaction = MagicMock() - transaction.__enter__.return_value = session - session_maker = MagicMock() - session_maker.begin.return_value = transaction - repository = MagicMock() - repository_factory = MagicMock(return_value=repository) - lifecycle = MagicMock() - lifecycle.transition.return_value = SimpleNamespace(id="control-1") - lifecycle_factory = MagicMock(return_value=lifecycle) + repository_factory = _RecordingRepositoryFactory() + lifecycle = _RecordingLifecycle(result=_control_space("control-1")) monkeypatch.setattr(management_module, "SQLAlchemyKnowledgeFSControlSpaceRepository", repository_factory) - monkeypatch.setattr(management_module, "KnowledgeFSControlSpaceLifecycleService", lifecycle_factory) - service = KnowledgeFSControlSpaceManagementService(session_maker) + monkeypatch.setattr(management_module, "KnowledgeFSControlSpaceLifecycleService", lifecycle) + service = KnowledgeFSControlSpaceManagementService(sessionmaker()) result = service.repair_registration( tenant_id="tenant-1", @@ -153,12 +192,16 @@ def test_repair_registration_delegates_an_active_cas_transition(monkeypatch: pyt ) assert result.id == "control-1" - lifecycle.transition.assert_called_once_with( - tenant_id="tenant-1", - control_space_id="control-1", - expected_resource_version=4, - new_state=KnowledgeFSControlSpaceState.ACTIVE, - lifecycle_operation_id="management-repair", - knowledge_space_id="space-1", - knowledge_space_revision=9, - ) + assert len(repository_factory.sessions) == 1 + assert lifecycle.repositories == [repository_factory] + assert lifecycle.transitions == [ + { + "tenant_id": "tenant-1", + "control_space_id": "control-1", + "expected_resource_version": 4, + "new_state": KnowledgeFSControlSpaceState.ACTIVE, + "lifecycle_operation_id": "management-repair", + "knowledge_space_id": "space-1", + "knowledge_space_revision": 9, + } + ]