test(api): replace mock doubles in knowledge-fs service tests with typed fakes

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

api/tests/unit_tests/knowledge_fs_core_fakes.py

@@ -0,0 +1,102 @@
+"""Small typed test fakes shared by the knowledge-fs core/service unit tests.
+
+They replace generic mock doubles with plain objects that record calls in lists.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass, field
+
+from services.knowledge_fs.observability import (
+    KnowledgeFSBatchStatusMetric,
+    KnowledgeFSCapabilityIssuanceMetric,
+    KnowledgeFSControlSpaceStateMetric,
+    KnowledgeFSLifecycleTaskMetric,
+)
+
+
+@dataclass(frozen=True)
+class RecordedCall:
+    args: tuple[object, ...]
+    kwargs: dict[str, object]
+
+
+@dataclass
+class RecordingCallable:
+    """A callable that records each invocation and returns or raises a configured result."""
+
+    return_value: object = None
+    side_effect: BaseException | Callable[..., object] | None = None
+    name: str = ""
+    calls: list[RecordedCall] = field(default_factory=list)
+
+    def __call__(self, *args: object, **kwargs: object) -> object:
+        self.calls.append(RecordedCall(args=args, kwargs=kwargs))
+        if isinstance(self.side_effect, BaseException):
+            raise self.side_effect
+        if self.side_effect is not None:
+            return self.side_effect(*args, **kwargs)
+        return self.return_value
+
+    @property
+    def call_count(self) -> int:
+        return len(self.calls)
+
+    @property
+    def last_call(self) -> RecordedCall:
+        assert self.calls, f"{self.name or 'callable'} was not called"
+        return self.calls[-1]
+
+    def assert_called_once_with(self, *args: object, **kwargs: object) -> None:
+        assert self.calls == [RecordedCall(args=args, kwargs=kwargs)], self.calls
+
+    def assert_not_called(self) -> None:
+        assert self.calls == [], self.calls
+
+
+@dataclass(frozen=True)
+class Marker:
+    """An opaque, identity-comparable stand-in for a collaborator nobody inspects."""
+
+    label: str
+
+
+@dataclass(frozen=True)
+class FakeCapabilityClaims:
+    """The subset of signed Capability v2 claims that brokers read back after issuance."""
+
+    exp: int
+    iat: int = 0
+
+
+@dataclass(frozen=True)
+class FakeIssuedCapability:
+    token: str
+    claims: FakeCapabilityClaims
+
+
+@dataclass
+class RecordingOperationalMetrics:
+    """KnowledgeFS operational metrics port that keeps every emitted event in order."""
+
+    batch_statuses: list[KnowledgeFSBatchStatusMetric] = field(default_factory=list)
+    capability_issuances: list[KnowledgeFSCapabilityIssuanceMetric] = field(default_factory=list)
+    control_space_states: list[KnowledgeFSControlSpaceStateMetric] = field(default_factory=list)
+    lifecycle_tasks: list[KnowledgeFSLifecycleTaskMetric] = field(default_factory=list)
+    gauge_readers: list[Callable[[], Mapping[str, int]]] = field(default_factory=list)
+
+    def record_batch_status(self, event: KnowledgeFSBatchStatusMetric) -> None:
+        self.batch_statuses.append(event)
+
+    def record_capability_issuance(self, event: KnowledgeFSCapabilityIssuanceMetric) -> None:
+        self.capability_issuances.append(event)
+
+    def record_control_space_state(self, event: KnowledgeFSControlSpaceStateMetric) -> None:
+        self.control_space_states.append(event)
+
+    def record_lifecycle_task(self, event: KnowledgeFSLifecycleTaskMetric) -> None:
+        self.lifecycle_tasks.append(event)
+
+    def register_control_space_state_gauge(self, read_counts: Callable[[], Mapping[str, int]]) -> None:
+        self.gauge_readers.append(read_counts)

api/tests/unit_tests/services/enterprise/test_rbac_service.py

@@ -907,7 +907,7 @@ class TestKnowledgeFSRoleMutations:
     @pytest.fixture
     def events(self, monkeypatch: pytest.MonkeyPatch, transport: RBACTransport) -> list[str]:
         events: list[str] = []
-        send = transport.send
+        forward_to_transport = transport.send
 
         def record_remote(
             method: str,
@@ -920,7 +920,7 @@ class TestKnowledgeFSRoleMutations:
             timeout: float,
         ) -> JsonValue:
             events.append("remote")
-            return send(
+            return forward_to_transport(
                 method,
                 endpoint,
                 tenant_id=tenant_id,

api/tests/unit_tests/services/test_knowledge_fs_batch_capability.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-from types import SimpleNamespace
-
 import pytest
 from sqlalchemy.orm import Session, sessionmaker
 
@@ -16,15 +14,16 @@ from services.knowledge_fs.batch_capability import (
 )
 from services.knowledge_fs.product_remote import KnowledgeFSOperationUnavailableError
 from services.knowledge_fs_capability import CapabilityIssueRequest
+from tests.unit_tests.knowledge_fs_core_fakes import FakeCapabilityClaims, FakeIssuedCapability
 
 
 class FakeIssuer:
     def __init__(self) -> None:
         self.requests: list[CapabilityIssueRequest] = []
 
-    def issue(self, request: CapabilityIssueRequest) -> SimpleNamespace:
+    def issue(self, request: CapabilityIssueRequest) -> FakeIssuedCapability:
         self.requests.append(request)
-        return SimpleNamespace(token="batch-token", claims=SimpleNamespace(exp=2_000_000_000))
+        return FakeIssuedCapability(token="batch-token", claims=FakeCapabilityClaims(exp=2_000_000_000))
 
 
 class FakeCutoverGate:

api/tests/unit_tests/services/test_knowledge_fs_capability_broker.py

@@ -2,8 +2,8 @@ from __future__ import annotations
 
 import uuid
 from collections.abc import Callable
-from types import SimpleNamespace
-from typing import Literal, cast, override
+from dataclasses import dataclass
+from typing import Literal, override
 
 import pytest
 import sqlalchemy as sa
@@ -76,28 +76,38 @@ class FakeProduct:
         )
 
 
+@dataclass(frozen=True)
+class _IssuedClaims:
+    iat: int
+    exp: int
+
+
+@dataclass(frozen=True)
+class _IssuedCapability:
+    token: str
+    claims: _IssuedClaims
+
+
+def _signed_capability() -> _IssuedCapability:
+    return _IssuedCapability(token="signed-capability", claims=_IssuedClaims(iat=1_999_999_940, exp=2_000_000_000))
+
+
 class FakeIssuer:
     def __init__(self) -> None:
         self.requests: list[CapabilityIssueRequest] = []
 
-    def issue(self, request: CapabilityIssueRequest) -> SimpleNamespace:
+    def issue(self, request: CapabilityIssueRequest) -> _IssuedCapability:
         self.requests.append(request)
-        return SimpleNamespace(
-            token="signed-capability",
-            claims=SimpleNamespace(iat=1_999_999_940, exp=2_000_000_000),
-        )
+        return _signed_capability()
 
 
 class AuditFailingOnceIssuer(FakeIssuer):
     @override
-    def issue(self, request: CapabilityIssueRequest) -> SimpleNamespace:
+    def issue(self, request: CapabilityIssueRequest) -> _IssuedCapability:
         self.requests.append(request)
         if len(self.requests) == 1:
             raise RuntimeError("issuance audit persistence failed")
-        return SimpleNamespace(
-            token="signed-capability",
-            claims=SimpleNamespace(iat=1_999_999_940, exp=2_000_000_000),
-        )
+        return _signed_capability()
 
 
 class NarrowingIssuer(FakeIssuer):
@@ -113,7 +123,7 @@ class NarrowingIssuer(FakeIssuer):
         self.saw_committed_reservation = False
 
     @override
-    def issue(self, request: CapabilityIssueRequest) -> SimpleNamespace:
+    def issue(self, request: CapabilityIssueRequest) -> _IssuedCapability:
         with self._maker() as session:
             reservation = session.scalar(
                 sa.select(KnowledgeFSCapabilityIssuanceReservation).where(
@@ -900,13 +910,37 @@ def test_uncutover_workspace_rejects_service_and_app_capabilities_before_issuanc
     if principal_kind == "service":
         with pytest.raises(KnowledgeFSOperationUnavailableError, match="not cut over"):
             broker.issue_service(
-                profile=cast(KnowledgeFSServiceApiProfile, SimpleNamespace(tenant_id="tenant-1")),
+                profile=KnowledgeFSServiceApiProfile(
+                    tenant_id="tenant-1",
+                    control_space_id="control-space-1",
+                    api_token_id="api-token-1",
+                    principal_id="api-token-1",
+                    knowledge_space_id="space-1",
+                    knowledge_space_revision=0,
+                    membership_epoch=1,
+                    space_acl_epoch=2,
+                    external_access_epoch=3,
+                    content_policy_revision=4,
+                ),
                 operation_id="createQuery",
             )
     else:
         with pytest.raises(KnowledgeFSOperationUnavailableError, match="not cut over"):
             broker.issue_app(
-                profile=cast(KnowledgeFSAppPrincipalProfile, SimpleNamespace(tenant_id="tenant-1")),
+                profile=KnowledgeFSAppPrincipalProfile(
+                    tenant_id="tenant-1",
+                    control_space_id="control-space-1",
+                    app_id="app-1",
+                    join_id="join-1",
+                    caller_kind=KnowledgeFSAppSpaceJoinType.AGENT,
+                    action="queries.create",
+                    knowledge_space_id="space-1",
+                    knowledge_space_revision=0,
+                    membership_epoch=1,
+                    space_acl_epoch=2,
+                    external_access_epoch=3,
+                    content_policy_revision=4,
+                ),
                 operation_id="createQuery",
             )
 

api/tests/unit_tests/services/test_knowledge_fs_control_space_lifecycle.py

@@ -1,37 +1,79 @@
 from __future__ import annotations
 
+from dataclasses import dataclass, field
 from datetime import datetime, timedelta
-from types import SimpleNamespace
-from unittest.mock import MagicMock
 
-from models.knowledge_fs import KnowledgeFSControlSpaceState
+from models.knowledge_fs import KnowledgeFSControlSpace, KnowledgeFSControlSpaceState
+from repositories.knowledge_fs_control_space_repository import KnowledgeFSControlSpaceCASUpdate
 from services.knowledge_fs.control_space_lifecycle import KnowledgeFSControlSpaceLifecycleService
+from services.knowledge_fs.observability import KnowledgeFSControlSpaceStateMetric
+
+
+@dataclass
+class _SequencedRepository:
+    reads: list[KnowledgeFSControlSpace]
+    cas_updates: list[KnowledgeFSControlSpaceCASUpdate] = field(default_factory=list)
+    get_calls: list[tuple[str, str]] = field(default_factory=list)
+
+    def get(self, *, tenant_id: str, control_space_id: str) -> KnowledgeFSControlSpace | None:
+        self.get_calls.append((tenant_id, control_space_id))
+        return self.reads.pop(0)
+
+    def compare_and_set_lifecycle(self, update: KnowledgeFSControlSpaceCASUpdate) -> bool:
+        self.cas_updates.append(update)
+        return True
+
+
+@dataclass
+class _RecordingMetrics:
+    control_space_states: list[KnowledgeFSControlSpaceStateMetric] = field(default_factory=list)
+
+    def record_control_space_state(self, event: KnowledgeFSControlSpaceStateMetric) -> None:
+        self.control_space_states.append(event)
+
+
+def _control_space(
+    *,
+    knowledge_space_id: str | None,
+    resource_version: int,
+    state: KnowledgeFSControlSpaceState,
+    updated_at: datetime,
+) -> KnowledgeFSControlSpace:
+    space = KnowledgeFSControlSpace(
+        tenant_id="tenant-secret",
+        owner_account_id="account-1",
+        provisioning_key="provision-1",
+    )
+    space.id = "control-secret"
+    space.deletion_irreversible_at = None
+    space.knowledge_space_id = knowledge_space_id
+    space.knowledge_space_revision = 0
+    space.resource_version = resource_version
+    space.state = state
+    space.updated_at = updated_at
+    return space
 
 
 def test_successful_transition_records_previous_state_dwell_without_resource_labels() -> None:
     entered_at = datetime(2026, 7, 21, 10, 0, 0)
-    before = SimpleNamespace(
-        deletion_irreversible_at=None,
+    before = _control_space(
         knowledge_space_id=None,
-        knowledge_space_revision=0,
         resource_version=3,
         state=KnowledgeFSControlSpaceState.PROVISIONING,
         updated_at=entered_at,
     )
-    after = SimpleNamespace(
-        **{
-            **vars(before),
-            "knowledge_space_id": "space-1",
-            "resource_version": 4,
-            "state": KnowledgeFSControlSpaceState.ACTIVE,
-            "updated_at": entered_at + timedelta(seconds=12),
-        }
+    after = _control_space(
+        knowledge_space_id="space-1",
+        resource_version=4,
+        state=KnowledgeFSControlSpaceState.ACTIVE,
+        updated_at=entered_at + timedelta(seconds=12),
+    )
+    repository = _SequencedRepository(reads=[before, after])
+    metrics = _RecordingMetrics()
+    service = KnowledgeFSControlSpaceLifecycleService(
+        repository,  # type: ignore[arg-type]
+        metrics=metrics,  # type: ignore[arg-type]
     )
-    repository = MagicMock()
-    repository.get.side_effect = (before, after)
-    repository.compare_and_set_lifecycle.return_value = True
-    metrics = MagicMock()
-    service = KnowledgeFSControlSpaceLifecycleService(repository, metrics=metrics)
 
     transitioned = service.transition(
         tenant_id="tenant-secret",
@@ -44,12 +86,13 @@ def test_successful_transition_records_previous_state_dwell_without_resource_lab
     )
 
     assert transitioned is after
-    assert metrics.record_control_space_state.call_args.args[0] == (
+    assert len(repository.cas_updates) == 1
+    assert metrics.control_space_states[-1] == (
         12.0,
         "provisioning",
         "active",
     )
-    serialized = str(metrics.record_control_space_state.call_args_list)
+    serialized = str(metrics.control_space_states)
     assert "tenant-secret" not in serialized
     assert "control-secret" not in serialized
     assert "space-1" not in serialized

api/tests/unit_tests/services/test_knowledge_fs_cutover_runtime_gate.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 from datetime import UTC, datetime
-from unittest.mock import MagicMock
 
 import pytest
 from sqlalchemy.orm import Session, sessionmaker
@@ -13,6 +12,7 @@ from models.knowledge_fs_cutover import (
 )
 from services.knowledge_fs.cutover_runtime_gate import SQLKnowledgeFSWorkspaceRuntimeGate
 from services.knowledge_fs.product_remote import KnowledgeFSOperationUnavailableError
+from tests.unit_tests.knowledge_fs_core_fakes import RecordingCallable
 
 _WATERMARK: KnowledgeFSCutoverRevisionWatermark = {
     "membership_epoch": 1,
@@ -33,10 +33,14 @@ def test_runtime_gate_fails_closed_without_a_cutover_ledger(sqlite_session: Sess
         gate.require_capability_v2(tenant_id="tenant-1")
 
 
+class _Initializer:
+    def __init__(self, ensure_initialized: RecordingCallable) -> None:
+        self.ensure_initialized = ensure_initialized
+
+
 @pytest.mark.parametrize("sqlite_session", [(KnowledgeFSWorkspaceCutoverLedger,)], indirect=True)
 def test_runtime_gate_initializes_a_missing_greenfield_workspace_once(sqlite_session: Session) -> None:
     session_maker = sessionmaker(bind=sqlite_session.get_bind(), expire_on_commit=False)
-    initializer = MagicMock()
 
     def initialize(*, tenant_id: str) -> None:
         with session_maker.begin() as session:
@@ -55,13 +59,14 @@ def test_runtime_gate_initializes_a_missing_greenfield_workspace_once(sqlite_ses
                 )
             )
 
-    initializer.ensure_initialized.side_effect = initialize
+    ensure_initialized = RecordingCallable(side_effect=initialize)
+    initializer = _Initializer(ensure_initialized)
     gate = SQLKnowledgeFSWorkspaceRuntimeGate(session_maker, initializer=initializer)
 
     gate.require_product_routes(tenant_id="tenant-1")
     gate.require_capability_v2(tenant_id="tenant-1")
 
-    initializer.ensure_initialized.assert_called_once_with(tenant_id="tenant-1")
+    ensure_initialized.assert_called_once_with(tenant_id="tenant-1")
 
 
 @pytest.mark.parametrize("sqlite_session", [(KnowledgeFSWorkspaceCutoverLedger,)], indirect=True)

api/tests/unit_tests/services/test_knowledge_fs_download_service.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 from collections.abc import Generator
 from pathlib import Path
 from typing import override
-from unittest.mock import MagicMock
 from zipfile import ZipFile
 
 import pytest
@@ -45,6 +44,16 @@ class FakeObjectStorage(KnowledgeFSObjectStorageService):
         return None if body is None else (chunk for chunk in (body[:2], body[2:]))
 
 
+class FailingObjectStorage(FakeObjectStorage):
+    def __init__(self, error: Exception) -> None:
+        super().__init__({})
+        self.error = error
+
+    @override
+    def head_object(self, *, key: str) -> KnowledgeFSObjectMetadata | None:
+        raise self.error
+
+
 def descriptor(
     *, document_id: str, filename: str, object_key: str, size_bytes: int
 ) -> KnowledgeFSDocumentDownloadDescriptor:
@@ -92,9 +101,7 @@ def test_load_stream_translates_object_storage_errors(
     storage_error: Exception,
     expected_error: type[Exception],
 ) -> None:
-    object_storage = MagicMock(spec=KnowledgeFSObjectStorageService)
-    object_storage.head_object.side_effect = storage_error
-    service = KnowledgeFSDownloadService(object_storage=object_storage)
+    service = KnowledgeFSDownloadService(object_storage=FailingObjectStorage(storage_error))
 
     with pytest.raises(expected_error):
         service.load_stream(descriptor(document_id="document-1", filename="a.txt", object_key="object-1", size_bytes=4))

api/tests/unit_tests/services/test_knowledge_fs_greenfield_initializer.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from unittest.mock import MagicMock
+from dataclasses import dataclass, field
 from uuid import uuid4
 
 import pytest
@@ -13,6 +13,14 @@ from services.knowledge_fs.cutover import KnowledgeFSCutoverGateBlockedError
 from services.knowledge_fs.greenfield_initializer import KnowledgeFSWorkspaceGreenfieldInitializer
 
 
+@dataclass
+class _RecordingCutover:
+    greenfield_initializations: list[tuple[str, str]] = field(default_factory=list)
+
+    def initialize_greenfield(self, *, tenant_id: str, owner_account_id: str) -> None:
+        self.greenfield_initializations.append((tenant_id, owner_account_id))
+
+
 @pytest.fixture
 def greenfield_session_maker(sqlite_engine: Engine) -> sessionmaker[Session]:
     tables = [
@@ -39,15 +47,15 @@ def test_greenfield_initializer_uses_the_unique_workspace_owner(
                 role=TenantAccountRole.OWNER,
             )
         )
-    cutover = MagicMock()
-    initializer = KnowledgeFSWorkspaceGreenfieldInitializer(greenfield_session_maker, cutover=cutover)
+    cutover = _RecordingCutover()
+    initializer = KnowledgeFSWorkspaceGreenfieldInitializer(
+        greenfield_session_maker,
+        cutover=cutover,  # type: ignore[arg-type]
+    )
 
     initializer.ensure_initialized(tenant_id=tenant_id)
 
-    cutover.initialize_greenfield.assert_called_once_with(
-        tenant_id=tenant_id,
-        owner_account_id=owner_account_id,
-    )
+    assert cutover.greenfield_initializations == [(tenant_id, owner_account_id)]
 
 
 def test_greenfield_initializer_rejects_a_workspace_without_one_owner(
@@ -58,7 +66,10 @@ def test_greenfield_initializer_rejects_a_workspace_without_one_owner(
         session.add(tenant)
         session.flush()
         tenant_id = tenant.id
-    initializer = KnowledgeFSWorkspaceGreenfieldInitializer(greenfield_session_maker, cutover=MagicMock())
+    initializer = KnowledgeFSWorkspaceGreenfieldInitializer(
+        greenfield_session_maker,
+        cutover=_RecordingCutover(),  # type: ignore[arg-type]
+    )
 
     with pytest.raises(KnowledgeFSCutoverGateBlockedError, match="exactly one Workspace owner"):
         initializer.ensure_initialized(tenant_id=tenant_id)
@@ -72,10 +83,13 @@ def test_greenfield_initializer_rejects_an_archived_workspace(
         session.add(tenant)
         session.flush()
         tenant_id = tenant.id
-    cutover = MagicMock()
-    initializer = KnowledgeFSWorkspaceGreenfieldInitializer(greenfield_session_maker, cutover=cutover)
+    cutover = _RecordingCutover()
+    initializer = KnowledgeFSWorkspaceGreenfieldInitializer(
+        greenfield_session_maker,
+        cutover=cutover,  # type: ignore[arg-type]
+    )
 
     with pytest.raises(KnowledgeFSCutoverGateBlockedError, match="not active"):
         initializer.ensure_initialized(tenant_id=tenant_id)
 
-    cutover.initialize_greenfield.assert_not_called()
+    assert cutover.greenfield_initializations == []

api/tests/unit_tests/services/test_knowledge_fs_initial_source_preview_job.py

@@ -1,9 +1,8 @@
 import json
-from types import SimpleNamespace
-from typing import cast
-from unittest.mock import MagicMock, patch
+from unittest.mock import patch
 
 import pytest
+from sqlalchemy.orm import sessionmaker
 
 from models.account import Account
 from services.knowledge_fs.initial_source_preview_job import (
@@ -31,9 +30,15 @@ def _payload() -> KnowledgeFSInitialWebsiteSourcePreviewPayload:
     )
 
 
+def _account() -> Account:
+    account = Account(name="Account One", email="account-1@example.com")
+    account.id = "account-1"
+    return account
+
+
 def test_start_validates_binding_and_enqueues_preview_without_running_it_inline() -> None:
-    service = KnowledgeFSInitialSourcePreviewJobService(MagicMock())
-    account = cast(Account, SimpleNamespace(id="account-1"))
+    service = KnowledgeFSInitialSourcePreviewJobService(sessionmaker())
+    account = _account()
 
     with (
         patch(
@@ -76,8 +81,8 @@ def test_start_validates_binding_and_enqueues_preview_without_running_it_inline(
 
 
 def test_start_rejects_a_second_active_preview_for_the_same_account() -> None:
-    service = KnowledgeFSInitialSourcePreviewJobService(MagicMock())
-    account = cast(Account, SimpleNamespace(id="account-1"))
+    service = KnowledgeFSInitialSourcePreviewJobService(sessionmaker())
+    account = _account()
 
     with (
         patch(
@@ -93,8 +98,8 @@ def test_start_rejects_a_second_active_preview_for_the_same_account() -> None:
 
 
 def test_start_releases_the_active_preview_slot_when_enqueue_fails() -> None:
-    service = KnowledgeFSInitialSourcePreviewJobService(MagicMock())
-    account = cast(Account, SimpleNamespace(id="account-1"))
+    service = KnowledgeFSInitialSourcePreviewJobService(sessionmaker())
+    account = _account()
 
     with (
         patch(

api/tests/unit_tests/services/test_knowledge_fs_lifecycle_saga.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 from collections import deque
 from datetime import datetime, timedelta
 from typing import override
-from unittest.mock import MagicMock, call
 
 import pytest
 from sqlalchemy.orm import Session, sessionmaker
@@ -31,6 +30,7 @@ from services.knowledge_fs.lifecycle_port import (
 )
 from services.knowledge_fs.lifecycle_saga import KnowledgeFSLifecycleSagaRunner
 from services.knowledge_fs.observability import KnowledgeFSLifecycleTaskMetric
+from tests.unit_tests.knowledge_fs_core_fakes import RecordingOperationalMetrics
 from tests.unit_tests.services.knowledge_fs_fakes import UnexpectedLifecycleRemote
 
 
@@ -129,7 +129,7 @@ def test_integrated_provision_saga_registers_remote_space_and_acks(sqlite_sessio
         command = session.get(KnowledgeFSLifecycleOutbox, intent.outbox.id)
         assert command is not None
         command.created_at = now - timedelta(seconds=12)
-    metrics = MagicMock()
+    metrics = RecordingOperationalMetrics()
     runner = KnowledgeFSLifecycleSagaRunner(session_maker, remote, metrics=metrics)
 
     result = runner.dispatch_one(
@@ -153,12 +153,12 @@ def test_integrated_provision_saga_registers_remote_space_and_acks(sqlite_sessio
     assert remote.provision_requests[0].model_intent["pluginId"] == "langgenius/openai"
     assert remote.provision_requests[0].profile_intent is not None
     assert remote.provision_requests[0].profile_intent["defaultMode"] == "fast"
-    assert metrics.record_lifecycle_task.call_args_list == [
-        call(KnowledgeFSLifecycleTaskMetric(None, "provision", "running")),
-        call(KnowledgeFSLifecycleTaskMetric(12.0, "provision", "succeeded")),
+    assert metrics.lifecycle_tasks == [
+        KnowledgeFSLifecycleTaskMetric(None, "provision", "running"),
+        KnowledgeFSLifecycleTaskMetric(12.0, "provision", "succeeded"),
     ]
-    assert "tenant-1" not in str(metrics.record_lifecycle_task.call_args_list)
-    assert intent.control_space.id not in str(metrics.record_lifecycle_task.call_args_list)
+    assert "tenant-1" not in str(metrics.lifecycle_tasks)
+    assert intent.control_space.id not in str(metrics.lifecycle_tasks)
 
 
 @pytest.mark.parametrize(
@@ -232,7 +232,7 @@ def test_deletion_saga_crosses_irreversible_point_and_remains_retryable_when_pro
             KnowledgeFSDeletionProgress(KnowledgeFSDeletionPhase.COMPLETED, 4, irreversible_at),
         )
     )
-    metrics = MagicMock()
+    metrics = RecordingOperationalMetrics()
     runner = KnowledgeFSLifecycleSagaRunner(session_maker, remote, metrics=metrics)
 
     first = runner.dispatch_one(
@@ -260,7 +260,7 @@ def test_deletion_saga_crosses_irreversible_point_and_remains_retryable_when_pro
     assert command is not None
     assert command.status is KnowledgeFSLifecycleOutboxStatus.SUCCEEDED
     assert len(remote.deletion_requests) == 2
-    assert [metric.status for ((metric,), _) in metrics.record_lifecycle_task.call_args_list] == [
+    assert [metric.status for metric in metrics.lifecycle_tasks] == [
         "running",
         "retry",
         "running",

api/tests/unit_tests/services/test_knowledge_fs_observability.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
-from unittest.mock import MagicMock
+from collections.abc import Callable, Iterable, Mapping
+from dataclasses import dataclass, field
+
+from opentelemetry.metrics import Observation
 
 from services.knowledge_fs.observability import (
     KnowledgeFSBatchStatusMetric,
@@ -9,14 +12,59 @@ from services.knowledge_fs.observability import (
     KnowledgeFSLifecycleTaskMetric,
     OpenTelemetryKnowledgeFSOperationalMetrics,
 )
+from tests.unit_tests.knowledge_fs_core_fakes import RecordedCall
+
+
+@dataclass
+class _FakeInstrument:
+    name: str
+    log: list[RecordedCall]
+    calls: list[RecordedCall] = field(default_factory=list)
+
+    def add(self, amount: int, *, attributes: dict[str, str]) -> None:
+        self._record(amount, attributes=attributes)
+
+    def record(self, amount: float, *, attributes: dict[str, str]) -> None:
+        self._record(amount, attributes=attributes)
+
+    def _record(self, *args: object, **kwargs: object) -> None:
+        call = RecordedCall(args=args, kwargs=kwargs)
+        self.calls.append(call)
+        self.log.append(call)
+
+
+@dataclass
+class _FakeMeter:
+    log: list[RecordedCall] = field(default_factory=list)
+    counters: dict[str, _FakeInstrument] = field(default_factory=dict)
+    histograms: dict[str, _FakeInstrument] = field(default_factory=dict)
+    gauge_calls: list[RecordedCall] = field(default_factory=list)
+
+    def create_counter(self, name: str, *, description: str, unit: str) -> _FakeInstrument:
+        self.log.append(RecordedCall(args=(name,), kwargs={"description": description, "unit": unit}))
+        return self.counters.setdefault(name, _FakeInstrument(name=name, log=self.log))
+
+    def create_histogram(self, name: str, *, description: str, unit: str) -> _FakeInstrument:
+        self.log.append(RecordedCall(args=(name,), kwargs={"description": description, "unit": unit}))
+        return self.histograms.setdefault(name, _FakeInstrument(name=name, log=self.log))
+
+    def create_observable_gauge(
+        self,
+        name: str,
+        *,
+        callbacks: tuple[Callable[[object], Iterable[Observation]], ...],
+        description: str,
+        unit: str,
+    ) -> object:
+        call = RecordedCall(args=(name,), kwargs={"callbacks": callbacks, "description": description, "unit": unit})
+        self.gauge_calls.append(call)
+        self.log.append(call)
+        return object()
 
 
 def test_operational_metrics_use_only_bounded_labels_and_numeric_measurements() -> None:
-    meter = MagicMock()
-    counters: dict[str, MagicMock] = {}
-    histograms: dict[str, MagicMock] = {}
-    meter.create_counter.side_effect = lambda name, **_: counters.setdefault(name, MagicMock())
-    meter.create_histogram.side_effect = lambda name, **_: histograms.setdefault(name, MagicMock())
+    meter = _FakeMeter()
+    counters = meter.counters
     metrics = OpenTelemetryKnowledgeFSOperationalMetrics(meter=meter)
 
     metrics.record_capability_issuance(
@@ -51,8 +99,8 @@ def test_operational_metrics_use_only_bounded_labels_and_numeric_measurements()
         )
     )
     capability_counter = counters["dify.knowledge_fs.capability_issuance"]
-    assert capability_counter.add.call_args.args == (1,)
-    assert capability_counter.add.call_args.kwargs == {
+    assert capability_counter.calls[-1].args == (1,)
+    assert capability_counter.calls[-1].kwargs == {
         "attributes": {
             "caller_kind": "interactive",
             "operation_id": "getKnowledgeSpace",
@@ -60,7 +108,7 @@ def test_operational_metrics_use_only_bounded_labels_and_numeric_measurements()
             "reason": "success",
         }
     }
-    all_calls = str(meter.mock_calls)
+    all_calls = str(meter.log)
     assert "tenant_id" not in all_calls
     assert "control_space_id" not in all_calls
     assert "jti" not in all_calls
@@ -68,17 +116,23 @@ def test_operational_metrics_use_only_bounded_labels_and_numeric_measurements()
 
 
 def test_control_space_current_state_gauge_uses_bounded_states_and_is_best_effort() -> None:
-    meter = MagicMock()
+    meter = _FakeMeter()
     metrics = OpenTelemetryKnowledgeFSOperationalMetrics(meter=meter)
-    read_counts = MagicMock(return_value={"deleting": 2, "error": 1, "provisioning": 3})
+    counts_error: list[Exception] = []
+
+    def read_counts() -> Mapping[str, int]:
+        if counts_error:
+            raise counts_error[0]
+        return {"deleting": 2, "error": 1, "provisioning": 3}
 
     metrics.register_control_space_state_gauge(read_counts)
 
-    gauge_call = meter.create_observable_gauge.call_args
+    assert len(meter.gauge_calls) == 1
+    gauge_call = meter.gauge_calls[0]
     assert gauge_call.args == ("dify.knowledge_fs.control_spaces",)
     assert gauge_call.kwargs["unit"] == "{space}"
     callback = gauge_call.kwargs["callbacks"][0]
-    observations = tuple(callback(MagicMock()))
+    observations = tuple(callback(object()))
     assert [(observation.value, observation.attributes) for observation in observations] == [
         (3, {"aggregation_scope": "global_database_snapshot", "state": "provisioning"}),
         (2, {"aggregation_scope": "global_database_snapshot", "state": "deleting"}),
@@ -87,7 +141,7 @@ def test_control_space_current_state_gauge_uses_bounded_states_and_is_best_effor
 
     assert "aggregate replica series with max, never sum" in gauge_call.kwargs["description"]
 
-    read_counts.side_effect = RuntimeError("database unavailable")
-    assert tuple(callback(MagicMock())) == ()
+    counts_error.append(RuntimeError("database unavailable"))
+    assert tuple(callback(object())) == ()
     metrics.register_control_space_state_gauge(read_counts)
-    assert meter.create_observable_gauge.call_count == 1
+    assert len(meter.gauge_calls) == 1

api/tests/unit_tests/services/test_knowledge_fs_product_authorization.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
-from unittest.mock import MagicMock, patch
+from unittest.mock import patch
+
+from sqlalchemy.orm import Session
 
 from core.rbac.entities import RBACPermission
 from services.enterprise import rbac_service
@@ -29,7 +31,7 @@ def test_enterprise_knowledge_fs_permissions_use_one_batch_request() -> None:
             "tenant-1",
             "account-1",
             ["control-1", "control-2"],
-            session=MagicMock(),
+            session=Session(),
         )
 
     inner_call.assert_called_once_with(
@@ -67,7 +69,7 @@ def test_knowledge_fs_capabilities_reuse_the_legacy_dataset_permission_points()
 def test_rbac_disabled_fallback_grants_every_dataset_point_knowledge_fs_consults() -> None:
     with config_overrides_context(RBAC_ENABLED=False):
         permissions = rbac_service.RBACService.KnowledgeFSPermissions.batch_get(
-            "tenant-1", "account-1", ["control-1"], session=MagicMock()
+            "tenant-1", "account-1", ["control-1"], session=Session()
         )
 
     assert permissions["control-1"] == list(KNOWLEDGE_FS_RBAC_PERMISSION_KEYS)
@@ -98,7 +100,7 @@ def test_rbac_transport_failure_filters_every_control_space() -> None:
     port = DifyKnowledgeFSProductRBACPort()
     with patch.object(rbac_service.RBACService.KnowledgeFSPermissions, "batch_get", side_effect=RuntimeError("down")):
         allowed = port.filter_authorized_control_space_ids(
-            session=MagicMock(),
+            session=Session(),
             tenant_id="tenant-1",
             account_id="account-1",
             control_space_ids=["control-1"],
@@ -119,7 +121,7 @@ def test_rbac_permission_map_ignores_unknown_keys_and_preserves_one_resource_bat
         },
     ) as batch_get:
         permissions = port.permission_keys_by_control_space(
-            session=MagicMock(),
+            session=Session(),
             tenant_id="tenant-1",
             account_id="account-1",
             control_space_ids=["control-1", "control-2"],

api/tests/unit_tests/services/test_knowledge_fs_product_service.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 from collections.abc import Sequence
 from datetime import UTC, datetime
 from typing import Never, override
-from unittest.mock import MagicMock
 
 import pytest
 from sqlalchemy import event
@@ -36,6 +35,7 @@ from services.knowledge_fs.product_remote import (
     UnavailableKnowledgeFSProductRemote,
 )
 from services.knowledge_fs.product_service import KnowledgeFSProductService
+from tests.unit_tests.knowledge_fs_core_fakes import RecordingOperationalMetrics
 
 
 class FakeRBAC:
@@ -268,7 +268,7 @@ def test_list_filters_locally_then_fetches_one_explicit_authorized_batch(sqlite_
     remote = FakeRemote({"space-owner": summary, "space-hidden": summary})
     batch_capabilities = FakeBatchCapabilities()
     rbac = FakeRBAC()
-    metrics = MagicMock()
+    metrics = RecordingOperationalMetrics()
     clock = iter((10.0, 10.25))
     service = KnowledgeFSProductService(
         sessionmaker(bind=sqlite_session.get_bind(), expire_on_commit=False),
@@ -326,8 +326,8 @@ def test_list_filters_locally_then_fetches_one_explicit_authorized_batch(sqlite_
         KnowledgeFSProductPermission.READ,
         KnowledgeFSProductPermission.QUERY,
     }
-    assert metrics.record_batch_status.call_args.args[0] == (0.25, 2, "degraded", 3, 1)
-    assert "tenant-1" not in str(metrics.record_batch_status.call_args_list)
+    assert metrics.batch_statuses[-1] == (0.25, 2, "degraded", 3, 1)
+    assert "tenant-1" not in str(metrics.batch_statuses)
 
 
 @pytest.mark.parametrize(
@@ -768,7 +768,7 @@ def test_list_fails_closed_without_remote_io_when_batch_capability_cannot_be_iss
     )
     remote = FakeRemote({"space-1": summary})
     batch_capabilities = FakeBatchCapabilities(fail=True)
-    metrics = MagicMock()
+    metrics = RecordingOperationalMetrics()
     clock = iter((20.0, 20.5))
     service = KnowledgeFSProductService(
         sessionmaker(bind=sqlite_session.get_bind(), expire_on_commit=False),
@@ -785,7 +785,7 @@ def test_list_fails_closed_without_remote_io_when_batch_capability_cannot_be_iss
     assert response.data[0].technical_status == "unavailable"
     assert len(batch_capabilities.calls) == 1
     assert remote.batch_calls == []
-    assert metrics.record_batch_status.call_args.args[0] == (0.5, 1, "failed", 1, 0)
+    assert metrics.batch_statuses[-1] == (0.5, 1, "failed", 1, 0)
 
 
 @pytest.mark.parametrize(

api/tests/unit_tests/services/test_knowledge_fs_query_images.py

@@ -2,12 +2,15 @@ from __future__ import annotations
 
 from collections.abc import Callable, Iterator
 from contextlib import nullcontext
-from types import SimpleNamespace
+from datetime import datetime
 
 import pytest
 
+from extensions.storage.storage_type import StorageType
 from graphon.file import File, FileTransferMethod, FileType
 from models.enums import CreatorUserRole
+from models.model import UploadFile
+from models.tools import ToolFile
 from services.knowledge_fs import query_images
 
 
@@ -25,25 +28,28 @@ def _upload(
     mime_type: str = "image/png",
     size: int = 12,
     tenant_id: str = "tenant-1",
-) -> SimpleNamespace:
-    return SimpleNamespace(
-        created_by=account_id,
-        created_by_role=CreatorUserRole.ACCOUNT,
-        id=upload_file_id,
-        key=f"uploads/{upload_file_id}",
-        mime_type=mime_type,
-        size=size,
+) -> UploadFile:
+    upload = UploadFile(
         tenant_id=tenant_id,
+        storage_type=StorageType.LOCAL,
+        key=f"uploads/{upload_file_id}",
+        name="image.png",
+        size=size,
+        extension="png",
+        mime_type=mime_type,
+        created_by_role=CreatorUserRole.ACCOUNT,
+        created_by=account_id,
+        created_at=datetime(2026, 7, 21, 12, 0),
         used=False,
-        used_at=None,
-        used_by=None,
     )
+    upload.id = upload_file_id
+    return upload
 
 
-def _install_files(monkeypatch: pytest.MonkeyPatch, files: list[SimpleNamespace]) -> _Session:
+def _install_files(monkeypatch: pytest.MonkeyPatch, files: list[UploadFile]) -> _Session:
     session = _Session()
 
-    def get_upload_files_by_ids(_tenant_id: str, _ids: list[str], *, session: _Session) -> dict[str, SimpleNamespace]:
+    def get_upload_files_by_ids(_tenant_id: str, _ids: list[str], *, session: _Session) -> dict[str, UploadFile]:
         assert session is not None
         return {item.id: item for item in files}
 
@@ -59,7 +65,8 @@ def _install_files(monkeypatch: pytest.MonkeyPatch, files: list[SimpleNamespace]
 def test_validate_query_images_checks_actor_bounds_and_marks_usage(monkeypatch: pytest.MonkeyPatch) -> None:
     upload = _upload()
     session = _install_files(monkeypatch, [upload])
-    monkeypatch.setattr(query_images, "naive_utc_now", lambda: "now")
+    used_at = datetime(2026, 7, 21, 12, 30)
+    monkeypatch.setattr(query_images, "naive_utc_now", lambda: used_at)
 
     result = query_images.validate_query_image_references(
         tenant_id="tenant-1",
@@ -77,7 +84,7 @@ def test_validate_query_images_checks_actor_bounds_and_marks_usage(monkeypatch:
     ]
     assert upload.used is True
     assert upload.used_by == "account-1"
-    assert upload.used_at == "now"
+    assert upload.used_at == used_at
     assert session.committed is True
 
 
@@ -92,7 +99,7 @@ def test_validate_query_images_checks_actor_bounds_and_marks_usage(monkeypatch:
 )
 def test_validate_query_images_rejects_hidden_or_invalid_files(
     monkeypatch: pytest.MonkeyPatch,
-    upload: SimpleNamespace,
+    upload: UploadFile,
     code: str,
 ) -> None:
     _install_files(monkeypatch, [upload])
@@ -221,7 +228,7 @@ def test_workflow_grant_is_file_tenant_subject_scoped_and_loads_without_actor_ow
     _install_files(monkeypatch, [upload])
 
     class _Controller:
-        def get_upload_file(self, **_kwargs: object) -> SimpleNamespace:
+        def get_upload_file(self, **_kwargs: object) -> UploadFile:
             return upload
 
     monkeypatch.setattr(query_images, "DatabaseFileAccessController", _Controller)
@@ -263,21 +270,23 @@ def test_workflow_grant_loads_tool_files(
     config_overrides: Callable[..., None],
 ) -> None:
     body = b"\x89PNG\r\n\x1a\nrest"
-    tool_file = SimpleNamespace(
+    tool_file = ToolFile(
+        user_id="account-1",
+        tenant_id="tenant-1",
+        conversation_id=None,
         file_key="tool-files/image.png",
-        id="00000000-0000-4000-8000-000000000002",
         mimetype="image/png",
         size=len(body),
-        tenant_id="tenant-1",
     )
+    tool_file.id = "00000000-0000-4000-8000-000000000002"
 
     class _ToolFileSession(_Session):
-        def get(self, model: object, file_id: str) -> SimpleNamespace | None:
+        def get(self, model: object, file_id: str) -> ToolFile | None:
             assert model is query_images.ToolFile
             return tool_file if file_id == tool_file.id else None
 
     class _Controller:
-        def get_tool_file(self, **_kwargs: object) -> SimpleNamespace:
+        def get_tool_file(self, **_kwargs: object) -> ToolFile:
             return tool_file
 
     monkeypatch.setattr(query_images.session_factory, "create_session", lambda: nullcontext(_ToolFileSession()))
@@ -348,7 +357,7 @@ def test_workflow_grant_rejects_cross_tenant_files_and_expiry(monkeypatch: pytes
     upload = _upload(tenant_id="other-tenant")
 
     class _Controller:
-        def get_upload_file(self, **_kwargs: object) -> SimpleNamespace:
+        def get_upload_file(self, **_kwargs: object) -> UploadFile:
             return upload
 
     monkeypatch.setattr(query_images, "DatabaseFileAccessController", _Controller)

api/tests/unit_tests/services/test_knowledge_fs_remote_registry.py

@@ -1,23 +1,23 @@
 from __future__ import annotations
 
-from unittest.mock import MagicMock
-
 import pytest
+from sqlalchemy.orm import sessionmaker
 
 from core.db.session_factory import session_factory
 from repositories import sqlalchemy_knowledge_fs_capability_issuance_auditor
 from services import knowledge_fs_capability
 from services.knowledge_fs import lifecycle_remote_http, remote_registry
 from tests.unit_tests.config_override import apply_config_overrides
+from tests.unit_tests.knowledge_fs_core_fakes import Marker
 
 
 def test_configured_lifecycle_remote_assembles_capability_v2_http_client(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    maker = MagicMock(name="session-maker")
-    auditor = MagicMock(name="capability-auditor")
-    issuer = MagicMock(name="capability-issuer")
-    client = MagicMock(name="lifecycle-http-client")
+    maker = sessionmaker()
+    auditor = Marker("capability-auditor")
+    issuer = Marker("capability-issuer")
+    client = Marker("lifecycle-http-client")
     captured: dict[str, object] = {}
 
     apply_config_overrides(
@@ -71,7 +71,7 @@ def test_lifecycle_remote_assembly_fails_closed_without_endpoint_or_capability_i
 def test_explicit_lifecycle_remote_factory_remains_available_for_isolated_process_tests(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    remote = MagicMock(name="remote")
+    remote = Marker("remote")
     monkeypatch.setattr(remote_registry, "_remote_factory", None)
 
     remote_registry.configure_knowledge_fs_lifecycle_remote_factory(lambda: remote)

api/tests/unit_tests/services/test_knowledge_fs_runtime.py

@@ -1,28 +1,27 @@
 from __future__ import annotations
 
-from types import SimpleNamespace
-from unittest.mock import MagicMock
-
 import pytest
+from sqlalchemy.orm import sessionmaker
 
 from services.knowledge_fs import runtime
 from services.knowledge_fs.product_remote import KnowledgeFSOperationUnavailableError
 from tests.unit_tests.config_override import apply_config_overrides
+from tests.unit_tests.knowledge_fs_core_fakes import Marker, RecordingCallable
 
 
 def test_runtime_fails_closed_when_the_remote_endpoint_is_not_configured(monkeypatch: pytest.MonkeyPatch) -> None:
     apply_config_overrides(monkeypatch, KNOWLEDGE_FS_BASE_URL=None)
 
     with pytest.raises(KnowledgeFSOperationUnavailableError, match="not configured"):
-        runtime.create_knowledge_fs_runtime(MagicMock())
+        runtime.create_knowledge_fs_runtime(sessionmaker())
 
 
 def test_process_runtime_is_built_once_for_the_application_session_factory(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    session_maker = MagicMock(name="session_maker")
-    built_runtime = SimpleNamespace(component="runtime")
-    factory = MagicMock(return_value=built_runtime)
+    session_maker = sessionmaker()
+    built_runtime = Marker("runtime")
+    factory = RecordingCallable(return_value=built_runtime)
     monkeypatch.setattr(runtime, "create_knowledge_fs_runtime", factory)
 
     first = runtime.get_knowledge_fs_runtime(session_maker)
@@ -66,14 +65,13 @@ def test_runtime_wires_one_shared_authorization_and_remote_graph(monkeypatch: py
         "create_configured_knowledge_fs_capability_issuer",
         "get_knowledge_fs_lifecycle_remote",
     )
-    factories: dict[str, MagicMock] = {}
+    factories: dict[str, RecordingCallable] = {}
     for name in factory_names:
-        factory = MagicMock(name=name)
-        factory.return_value = SimpleNamespace(component=name)
+        factory = RecordingCallable(return_value=Marker(name), name=name)
         factories[name] = factory
         monkeypatch.setattr(runtime, name, factory)
 
-    session_maker = MagicMock(name="session_maker")
+    session_maker = sessionmaker()
     result = runtime.create_knowledge_fs_runtime(session_maker)
 
     assert result.application is factories["KnowledgeFSProductApplicationService"].return_value
@@ -95,7 +93,7 @@ def test_runtime_wires_one_shared_authorization_and_remote_graph(monkeypatch: py
         audit=factories["SQLAlchemyKnowledgeFSCapabilityIssuanceAuditor"].return_value
     )
     assert (
-        factories["KnowledgeFSProductService"].call_args.kwargs["cutover_gate"]
+        factories["KnowledgeFSProductService"].last_call.kwargs["cutover_gate"]
         is factories["SQLKnowledgeFSWorkspaceRuntimeGate"].return_value
     )
     factories["KnowledgeFSWorkspaceCutoverService"].assert_called_once_with(
@@ -111,7 +109,7 @@ def test_runtime_wires_one_shared_authorization_and_remote_graph(monkeypatch: py
         initializer=factories["KnowledgeFSWorkspaceGreenfieldInitializer"].return_value,
     )
     assert (
-        factories["KnowledgeFSProductApplicationService"].call_args.kwargs["rbac"]
+        factories["KnowledgeFSProductApplicationService"].last_call.kwargs["rbac"]
         is factories["DifyKnowledgeFSProductRBACPort"].return_value
     )
-    assert factories["KnowledgeFSAppExecutionCapabilityService"].call_args.kwargs["admission"] is result.app_admission
+    assert factories["KnowledgeFSAppExecutionCapabilityService"].last_call.kwargs["admission"] is result.app_admission

api/tests/unit_tests/services/test_knowledge_fs_service_query_images.py

@@ -1,16 +1,19 @@
 from __future__ import annotations
 
 from contextlib import nullcontext
+from datetime import datetime
 from io import BytesIO
-from types import SimpleNamespace
 
 import pytest
 from werkzeug.datastructures import FileStorage
 
 from core.knowledge_fs.errors import KnowledgeFSProductRequestRejectedError, KnowledgeFSQueryImageError
+from extensions.storage.storage_type import StorageType
 from models.enums import ApiTokenType, CreatorUserRole
+from models.model import UploadFile
 from services.knowledge_fs import query_images, service_query_image_upload
 from services.knowledge_fs.service_api_authorization import KnowledgeFSServiceApiProfile
+from tests.unit_tests.config_override import apply_config_overrides
 
 TOKEN_ID = "00000000-0000-4000-8000-000000000001"
 OTHER_TOKEN_ID = "00000000-0000-4000-8000-000000000002"
@@ -52,7 +55,7 @@ def _install(monkeypatch: pytest.MonkeyPatch) -> tuple[_Session, dict[str, bytes
     monkeypatch.setattr(service_query_image_upload.storage, "save", lambda key, body: objects.__setitem__(key, body))
     monkeypatch.setattr(service_query_image_upload.storage, "delete", lambda key: objects.pop(key, None))
     monkeypatch.setattr(query_images.storage, "load", lambda key, **_kwargs: iter([objects[key]]))
-    monkeypatch.setattr(service_query_image_upload, "dify_config", SimpleNamespace(STORAGE_TYPE="local"))
+    apply_config_overrides(monkeypatch, STORAGE_TYPE="local")
     return session, objects
 
 
@@ -60,19 +63,22 @@ def _file(body: bytes = PNG, *, mime_type: str = "image/png", filename: str = "i
     return FileStorage(stream=BytesIO(body), filename=filename, content_type=mime_type)
 
 
-def _record(*, role: CreatorUserRole = CreatorUserRole.DATASET_API_KEY, owner: str = TOKEN_ID):
-    return SimpleNamespace(
-        id=FILE_ID,
+def _record(*, role: CreatorUserRole = CreatorUserRole.DATASET_API_KEY, owner: str = TOKEN_ID) -> UploadFile:
+    record = UploadFile(
         tenant_id=TENANT_ID,
+        storage_type=StorageType.LOCAL,
+        key="image-key",
+        name="image.png",
+        size=len(PNG),
+        extension="png",
+        mime_type="image/png",
         created_by_role=role,
         created_by=owner,
-        mime_type="image/png",
-        size=len(PNG),
-        key="image-key",
+        created_at=datetime(2026, 7, 21, 12, 0),
         used=False,
-        used_by=None,
-        used_at=None,
     )
+    record.id = FILE_ID
+    return record
 
 
 def test_upload_admission_and_inner_load_use_same_active_key_owner(monkeypatch: pytest.MonkeyPatch) -> None:

api/tests/unit_tests/services/test_knowledge_fs_space_tag_service.py

@@ -1,4 +1,4 @@
-from unittest.mock import MagicMock
+from dataclasses import dataclass, field
 
 import pytest
 import sqlalchemy as sa
@@ -38,8 +38,29 @@ def _tag(tag_id: str, *, tenant_id: str = "tenant-1", tag_type: TagType = TagTyp
     return tag
 
 
-def _service(factory: sessionmaker[Session]) -> tuple[KnowledgeFSSpaceTagService, MagicMock]:
-    product = MagicMock()
+@dataclass
+class _RecordingAuthorization:
+    permissions: list[KnowledgeFSProductPermission] = field(default_factory=list)
+
+    def authorize_control_space_in_session(
+        self,
+        *,
+        session: Session,
+        tenant_id: str,
+        account_id: str,
+        control_space_id: str,
+        permission: KnowledgeFSProductPermission,
+        require_active: bool = False,
+    ) -> object:
+        assert session.is_active
+        assert (tenant_id, account_id, control_space_id) == ("tenant-1", "account-1", "control-1")
+        assert require_active is False
+        self.permissions.append(permission)
+        return object()
+
+
+def _service(factory: sessionmaker[Session]) -> tuple[KnowledgeFSSpaceTagService, _RecordingAuthorization]:
+    product = _RecordingAuthorization()
     return KnowledgeFSSpaceTagService(factory, product=product), product
 
 
@@ -68,8 +89,7 @@ def test_replace_tags_is_idempotent_and_lists_final_set(sqlite_session_factory:
     ] == ["tag-1"]
     with sqlite_session_factory() as session:
         assert session.scalar(sa.select(sa.func.count(KnowledgeFSSpaceTagBinding.id))) == 1
-    first_authorization = product.authorize_control_space_in_session.call_args_list[0]
-    assert first_authorization.kwargs["permission"] is KnowledgeFSProductPermission.EDIT
+    assert product.permissions[0] is KnowledgeFSProductPermission.EDIT
 
 
 @pytest.mark.parametrize(

api/tests/unit_tests/services/test_knowledge_fs_text_store.py

@@ -1,7 +1,7 @@
 import hashlib
 from contextlib import nullcontext
-from types import SimpleNamespace
-from unittest.mock import MagicMock, patch
+from dataclasses import dataclass
+from unittest.mock import patch
 
 import pytest
 from pydantic import ValidationError
@@ -24,11 +24,20 @@ from services.knowledge_fs.text_store_native import (
 from services.knowledge_fs.vector_store import VectorStoreUnavailableError
 from services.knowledge_fs.vector_store_elasticsearch import ElasticsearchVectorStore
 from services.knowledge_fs.vector_store_weaviate import WeaviateVectorStore
+from tests.unit_tests.knowledge_fs_core_fakes import Marker
 
 A, B, C, GENERATION = (f"00000000-0000-4000-8000-{n:012d}" for n in (1, 2, 3, 4))
 SCOPE = {"tenant_id": A, "knowledge_space_id": B, "index_version": "fts-v1"}
 
 
+@dataclass(frozen=True)
+class _StaticTextClient:
+    result: dict
+
+    def execute(self, _payload: TextPayload) -> dict:
+        return self.result
+
+
 def point(identifier=A, text="refund policy 退 款"):
     return {
         "id": identifier,
@@ -139,7 +148,7 @@ def test_checksum_and_extra_configuration_rejected():
 )
 def test_invalid_backend_readback_rejected(result):
     with pytest.raises(VectorStoreUnavailableError):
-        execute_text_request(SimpleNamespace(execute=lambda _: result), request("get", ids=[A]))
+        execute_text_request(_StaticTextClient(result), request("get", ids=[A]))
 
 
 @pytest.mark.parametrize(
@@ -148,12 +157,12 @@ def test_invalid_backend_readback_rejected(result):
 def test_unauthorized_duplicate_and_nonfinite_matches_rejected(matches):
     with pytest.raises(VectorStoreUnavailableError):
         execute_text_request(
-            SimpleNamespace(execute=lambda _: {"points": [], "matches": matches}),
+            _StaticTextClient({"points": [], "matches": matches}),
             request("search", ids=[A], query="refund"),
         )
     with pytest.raises(VectorStoreUnavailableError):
         execute_text_request(
-            SimpleNamespace(execute=lambda _: {"points": [], "matches": matches * 2}),
+            _StaticTextClient({"points": [], "matches": matches * 2}),
             request("search", ids=[A], query="refund", limit=1),
         )
 
@@ -167,7 +176,7 @@ def test_unauthorized_duplicate_and_nonfinite_matches_rejected(matches):
 )
 def test_selection_reuses_vector_configuration(config_overrides, backend, wrapper, expected):
     config_overrides(VECTOR_STORE=backend)
-    native = wrapper(MagicMock())
+    native = wrapper(Marker("native-client"))
     with patch(
         "services.knowledge_fs.vector_store.configured_vector_client", return_value=nullcontext(native)
     ) as factory:

api/tests/unit_tests/services/test_knowledge_fs_text_store_tidb.py

@@ -1,87 +1,140 @@
-from types import SimpleNamespace
-from unittest.mock import MagicMock, patch
+from collections.abc import Callable, Sequence
+from dataclasses import dataclass, field
+from unittest.mock import patch
 
 import pymysql
 import pytest
 
+from models.dataset import TidbAuthBinding
+from models.enums import TidbAuthBindingStatus
 from services.knowledge_fs.text_store import TextIndexPendingError, configured_text_client, execute_text_request
 from services.knowledge_fs.text_store_tidb import TidbTextStore, _sql_endpoint
 from services.knowledge_fs.vector_store import VectorStoreUnavailableError
 from tests.unit_tests.services.test_knowledge_fs_text_store import GENERATION, A, point, request
 
 
+@dataclass
+class _FakeCursor:
+    """pymysql cursor double: queued fetchone rows win over the steady fetchone value."""
+
+    fetchone_queue: list[object] = field(default_factory=list)
+    fetchone_value: object = (1,)
+    fetchall_value: list[tuple[object, ...]] = field(default_factory=list)
+    execute_error: BaseException | Callable[..., None] | None = None
+    executed: list[tuple[object, ...]] = field(default_factory=list)
+    executemany_calls: list[tuple[str, Sequence[object]]] = field(default_factory=list)
+
+    def __enter__(self) -> "_FakeCursor":
+        return self
+
+    def __exit__(self, *_exc_info: object) -> None:
+        return None
+
+    def execute(self, sql: str, *args: object) -> None:
+        self.executed.append((sql, *args))
+        if isinstance(self.execute_error, BaseException):
+            raise self.execute_error
+        if self.execute_error is not None:
+            self.execute_error(sql, *args)
+
+    def executemany(self, sql: str, rows: Sequence[object]) -> None:
+        self.executemany_calls.append((sql, rows))
+
+    def fetchone(self) -> object:
+        if self.fetchone_queue:
+            return self.fetchone_queue.pop(0)
+        return self.fetchone_value
+
+    def fetchall(self) -> list[tuple[object, ...]]:
+        return self.fetchall_value
+
+
+@dataclass
+class _FakeConnection:
+    cursor_double: _FakeCursor = field(default_factory=_FakeCursor)
+    close_count: int = 0
+
+    def cursor(self) -> _FakeCursor:
+        return self.cursor_double
+
+    def close(self) -> None:
+        self.close_count += 1
+
+
 @pytest.fixture
-def store():
-    connection = MagicMock()
-    cursor = connection.cursor.return_value.__enter__.return_value
-    cursor.fetchone.return_value = (1,)
-    cursor.fetchall.return_value = []
-    return TidbTextStore(connection)
+def connection() -> _FakeConnection:
+    return _FakeConnection()
 
 
-def test_native_tidb_ddl_write_readback_query_and_delete(store):
-    cursor = store.connection.cursor.return_value.__enter__.return_value
-    cursor.fetchone.side_effect = [None, None, (1,)]
+@pytest.fixture
+def cursor(connection: _FakeConnection) -> _FakeCursor:
+    return connection.cursor_double
+
+
+@pytest.fixture
+def store(connection: _FakeConnection):
+    return TidbTextStore(connection)  # type: ignore[arg-type]
+
+
+def test_native_tidb_ddl_write_readback_query_and_delete(store, connection, cursor):
+    cursor.fetchone_queue = [None, None, (1,)]
     execute_text_request(store, request("upsert", points=[point()]))
-    sqls = [call.args[0] for call in cursor.execute.call_args_list]
+    sqls = [call[0] for call in cursor.executed]
     assert any("FULLTEXT INDEX" in sql and "MULTILINGUAL" in sql for sql in sqls)
-    assert "ON DUPLICATE KEY UPDATE id=id" in cursor.executemany.call_args.args[0]
-    cursor.fetchone.side_effect = None
-    cursor.fetchone.return_value = (1,)
-    cursor.fetchall.return_value = [(A, GENERATION, point()["content_hash"], point()["text"])]
+    assert "ON DUPLICATE KEY UPDATE id=id" in cursor.executemany_calls[-1][0]
+    cursor.fetchone_queue = []
+    cursor.fetchone_value = (1,)
+    cursor.fetchall_value = [(A, GENERATION, point()["content_hash"], point()["text"])]
     assert execute_text_request(store, request("get", ids=[A]))["points"] == [point()]
-    cursor.fetchall.return_value = [(A, 1.25)]
+    cursor.fetchall_value = [(A, 1.25)]
     assert execute_text_request(store, request("search", ids=[A], query="退 款"))["matches"] == [
         {"id": A, "score": 1.25}
     ]
-    sql, params = cursor.execute.call_args.args
+    sql, params = cursor.executed[-1]
     assert "IGNORE INDEX (PRIMARY)" in sql
     assert "READ_FROM_STORAGE(TIFLASH[" in sql
     assert "CONCAT(' ',text,' ') LIKE %s ESCAPE '!'" in sql
     assert "id IN (%s)" in sql
     assert params == ("退 款", "退 款", A, "% 款 %", "% 退 %", 10)
-    cursor.fetchall.return_value = []
+    cursor.fetchall_value = []
     execute_text_request(store, request("delete", ids=[A]))
-    cursor.fetchall.return_value = [(A, GENERATION, point()["content_hash"], point()["text"])]
+    cursor.fetchall_value = [(A, GENERATION, point()["content_hash"], point()["text"])]
     with pytest.raises(VectorStoreUnavailableError, match="not visible"):
         execute_text_request(store, request("delete", ids=[A]))
     store.close()
-    store.connection.close.assert_called_once()
+    assert connection.close_count == 1
 
 
 @pytest.mark.parametrize("operation", ["get", "delete", "search"])
-def test_missing_index_distinct_from_provider_failure(store, operation):
-    cursor = store.connection.cursor.return_value.__enter__.return_value
-    cursor.fetchone.return_value = None
+def test_missing_index_distinct_from_provider_failure(store, cursor, operation):
+    cursor.fetchone_value = None
     payload = request(operation, ids=[A], **({"query": "refund"} if operation == "search" else {}))
     if operation == "search":
         with pytest.raises(VectorStoreUnavailableError, match="missing"):
             execute_text_request(store, payload)
     else:
         assert execute_text_request(store, payload) == {"points": [], "matches": []}
-    cursor.execute.side_effect = OSError("unavailable")
+    cursor.execute_error = OSError("unavailable")
     with pytest.raises(OSError):
         execute_text_request(store, payload)
 
 
 @pytest.mark.parametrize("replica", [None, (0,)])
-def test_unready_columnar_index_requires_retry(store, replica):
-    cursor = store.connection.cursor.return_value.__enter__.return_value
-    cursor.fetchone.side_effect = [(1,), replica]
+def test_unready_columnar_index_requires_retry(store, cursor, replica):
+    cursor.fetchone_queue = [(1,), replica]
     with pytest.raises(TextIndexPendingError):
         execute_text_request(store, request("search", ids=[A], query="refund"))
 
 
 @pytest.mark.parametrize("code", [1061, 8200])
-def test_concurrent_ddl_only_accepts_duplicate_index(store, code):
-    cursor = store.connection.cursor.return_value.__enter__.return_value
-    cursor.fetchone.side_effect = [(1,), None, (1,)]
+def test_concurrent_ddl_only_accepts_duplicate_index(store, cursor, code):
+    cursor.fetchone_queue = [(1,), None, (1,)]
 
     def execute(sql, *_args):
         if "ALTER TABLE" in sql:
             raise pymysql.err.OperationalError(code, "duplicate or unsupported")
 
-    cursor.execute.side_effect = execute
+    cursor.execute_error = execute
     if code == 1061:
         execute_text_request(store, request("upsert", points=[point()]))
     else:
@@ -94,7 +147,15 @@ def test_uses_existing_dify_tenant_binding_and_tls_metadata_endpoint(config_over
     config_overrides(TIDB_API_URL="https://cloud.example.test")
     config_overrides(TIDB_PUBLIC_KEY="test-public")
     config_overrides(TIDB_PRIVATE_KEY="test-private")
-    binding = SimpleNamespace(cluster_id="cluster-test", account="test-account", password="test-password")
+    binding = TidbAuthBinding(
+        tenant_id=A,
+        cluster_id="cluster-test",
+        cluster_name="cluster",
+        active=True,
+        status=TidbAuthBindingStatus.ACTIVE,
+        account="test-account",
+        password="test-password",
+    )
     _sql_endpoint.cache_clear()
     with (
         patch("services.tidb_binding_service.resolve_tidb_auth_binding", return_value=binding) as resolver,

api/tests/unit_tests/services/test_tidb_binding_service.py

@@ -2,7 +2,7 @@ import threading
 import uuid
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import contextmanager
-from types import SimpleNamespace
+from dataclasses import dataclass
 from unittest.mock import MagicMock, patch
 
 import httpx
@@ -10,17 +10,53 @@ import pytest
 from dify_vdb_tidb_on_qdrant.tidb_service import TidbService
 from redis.exceptions import LockError
 from sqlalchemy import select
+from sqlalchemy.engine import Engine
 from sqlalchemy.orm import Session
 
+from core.rag.embedding.embedding_base import Embeddings
 from models.account import Tenant
-from models.dataset import TidbAuthBinding
+from models.dataset import Dataset, TidbAuthBinding
 from models.enums import TidbAuthBindingStatus
 from services import tidb_binding_service as service
+from tests.unit_tests.knowledge_fs_core_fakes import RecordingCallable
+
+
+@dataclass(frozen=True)
+class _Db:
+    engine: Engine
+
+
+@dataclass(frozen=True)
+class _Redis:
+    lock: RecordingCallable
+
+
+@dataclass(frozen=True)
+class _Environment:
+    engine: Engine
+    tenants: list[Tenant]
+    redis: _Redis
+    create: MagicMock
+    get: MagicMock
+
+
+class _UnusedEmbeddings(Embeddings):
+    def embed_documents(self, texts: list[str]) -> list[list[float]]:
+        raise NotImplementedError
+
+    def embed_multimodal_documents(self, multimodel_documents: list[dict[str, object]]) -> list[list[float]]:
+        raise NotImplementedError
+
+    def embed_query(self, text: str) -> list[float]:
+        raise NotImplementedError
+
+    def embed_multimodal_query(self, multimodel_document: dict[str, object]) -> list[float]:
+        raise NotImplementedError
 
 
 @pytest.fixture
 def environment(sqlite_engine, monkeypatch, config_overrides):
-    monkeypatch.setattr(service, "db", SimpleNamespace(engine=sqlite_engine))
+    monkeypatch.setattr(service, "db", _Db(engine=sqlite_engine))
     lock = threading.Lock()
 
     @contextmanager
@@ -28,7 +64,7 @@ def environment(sqlite_engine, monkeypatch, config_overrides):
         with lock:
             yield
 
-    redis = SimpleNamespace(lock=MagicMock(side_effect=acquire))
+    redis = _Redis(lock=RecordingCallable(side_effect=acquire))
     monkeypatch.setattr(service, "redis_client", redis)
     config_overrides(
         TIDB_PROJECT_ID="project",
@@ -54,7 +90,7 @@ def environment(sqlite_engine, monkeypatch, config_overrides):
             },
         ) as get,
     ):
-        yield SimpleNamespace(engine=sqlite_engine, tenants=tenants, redis=redis, create=create, get=get)
+        yield _Environment(engine=sqlite_engine, tenants=tenants, redis=redis, create=create, get=get)
 
 
 def add_binding(env, *, tenant=None, active=False, status=TidbAuthBindingStatus.ACTIVE, **changes):
@@ -353,7 +389,7 @@ def test_legacy_dataset_factory_uses_shared_binding_and_resumes_pending(environm
     from dify_vdb_tidb_on_qdrant import tidb_on_qdrant_vector as legacy
 
     binding = add_binding(environment, tenant=environment.tenants[0].id, active=True)
-    dataset = SimpleNamespace(id=str(uuid.uuid4()), tenant_id=binding.tenant_id, index_struct_dict=None)
+    dataset = Dataset(id=str(uuid.uuid4()), tenant_id=binding.tenant_id, index_struct=None)
     with (
         patch.object(
             legacy, "resolve_tidb_auth_binding", side_effect=[service.TidbBindingPendingError(), binding]
@@ -361,7 +397,7 @@ def test_legacy_dataset_factory_uses_shared_binding_and_resumes_pending(environm
         patch.object(legacy.time, "sleep") as sleep,
         patch.object(legacy, "TidbOnQdrantVector") as vector,
     ):
-        legacy.TidbOnQdrantVectorFactory().init_vector(dataset, [], MagicMock(), session=None)
+        legacy.TidbOnQdrantVectorFactory().init_vector(dataset, [], _UnusedEmbeddings(), session=None)
     assert resolve.call_count == 2
     resolve.assert_called_with(binding.tenant_id, allow_create=True)
     sleep.assert_called_once_with(5)
@@ -372,12 +408,12 @@ def test_legacy_dataset_factory_uses_shared_binding_and_resumes_pending(environm
 def test_legacy_dataset_factory_bounds_readiness_wait(environment):
     from dify_vdb_tidb_on_qdrant import tidb_on_qdrant_vector as legacy
 
-    dataset = SimpleNamespace(id=str(uuid.uuid4()), tenant_id=environment.tenants[0].id)
+    dataset = Dataset(id=str(uuid.uuid4()), tenant_id=environment.tenants[0].id)
     with (
         patch.object(legacy, "resolve_tidb_auth_binding", side_effect=service.TidbBindingPendingError()),
         patch.object(legacy.time, "monotonic", side_effect=[0, 901]),
         patch.object(legacy.time, "sleep") as sleep,
     ):
         with pytest.raises(service.TidbBindingPendingError):
-            legacy.TidbOnQdrantVectorFactory().init_vector(dataset, [], MagicMock(), session=None)
+            legacy.TidbOnQdrantVectorFactory().init_vector(dataset, [], _UnusedEmbeddings(), session=None)
     sleep.assert_not_called()