opensource/dify
3
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-10-16 15:35:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/deploy/konwledge' into deploy/konwledge

Browse Source
This commit is contained in:
Stephen Zhou
2026-10-15 20:02:49 +08:00
parent 4722f30ec8 1079a07d60
commit bb0d3d46f2
7 changed files with 769 additions and 492 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/tests/unit_tests/controllers/console/datasets/test_datasets.py
+34 -24
View File
@@ -1,4 +1,5 @@
import datetime
from dataclasses import dataclass
from inspect import unwrap
from types import SimpleNamespace
from unittest.mock import MagicMock, create_autospec, patch
@@ -42,7 +43,7 @@ from controllers.console.datasets.error import (
IndexingEstimateError,
)
from machinery.context import RequestContext
from models.knowledge_fs import KnowledgeFSUpgradeJobStatus, KnowledgeFSUpgradeStage
from models.knowledge_fs import KnowledgeFSUpgradeJob, KnowledgeFSUpgradeJobStatus, KnowledgeFSUpgradeStage
from services.data_source.entities.notion_import import NotionPageType
from services.dataset_knowledge_fs_upgrade_service import (
KnowledgeFSUpgradeConflictError,
@@ -344,32 +345,41 @@ def test_new_source_estimate_authorizes_before_execution(dataset_id, scene, loca
estimates.estimate_new_sources.assert_not_called()
def _upgrade_job(dataset_id: str = str(DATASET_ID), **overrides):
values = {
"id": "upgrade-job-1",
"old_dataset_id": dataset_id,
"new_control_space_id": None,
"status": KnowledgeFSUpgradeJobStatus.QUEUED,
"stage": KnowledgeFSUpgradeStage.VALIDATING,
"snapshot_at": datetime.datetime(2026, 8, 17, tzinfo=datetime.UTC),
"total_documents": 1,
"completed_documents": 0,
"total_sources": 0,
"completed_sources": 0,
"last_error_code": None,
"last_error_message": None,
"completed_at": None,
}
values.update(overrides)
return SimpleNamespace(**values)
def _upgrade_job(job_id: str = "upgrade-job-1") -> KnowledgeFSUpgradeJob:
job = KnowledgeFSUpgradeJob(
tenant_id="tenant-1",
old_dataset_id=str(DATASET_ID),
requested_by_account_id="account-1",
owner_account_id="account-1",
idempotency_key="upgrade-request-1",
snapshot_at=datetime.datetime(2026, 8, 17, tzinfo=datetime.UTC),
config_snapshot={},
permission_snapshot={},
app_binding_snapshot=[],
tag_ids_snapshot=[],
status=KnowledgeFSUpgradeJobStatus.QUEUED,
stage=KnowledgeFSUpgradeStage.VALIDATING,
total_documents=1,
)
job.id = job_id
return job
@dataclass(frozen=True)
class _UpgradeKnowledgeServices:
upgrades: KnowledgeUpgradeApplicationService
@dataclass(frozen=True)
class _UpgradeApplicationServices:
knowledge: _UpgradeKnowledgeServices
@pytest.fixture
def upgrades(monkeypatch):
service = create_autospec(KnowledgeUpgradeApplicationService, instance=True, spec_set=True)
monkeypatch.setattr(
controller, "application_services", lambda: SimpleNamespace(knowledge=SimpleNamespace(upgrades=service))
)
services = _UpgradeApplicationServices(knowledge=_UpgradeKnowledgeServices(upgrades=service))
monkeypatch.setattr(controller, "application_services", lambda: services)
return service
@@ -384,7 +394,7 @@ def test_upgrade_writes_require_dataset_edit_permission() -> None:
def test_upgrade_jobs_serializes_recoverable_jobs(app, upgrades):
upgrades.list_recoverable_jobs.return_value = [_upgrade_job(id="job-1")]
upgrades.list_recoverable_jobs.return_value = [_upgrade_job(job_id="job-1")]
with app.test_request_context("/datasets/knowledge-fs-upgrade-jobs"):
response = unwrap(DatasetKnowledgeFSUpgradeJobsApi.get)(DatasetKnowledgeFSUpgradeJobsApi(), CONTEXT)
upgrades.list_recoverable_jobs.assert_called_once_with(CONTEXT)
@@ -453,7 +463,7 @@ def test_upgrade_job_status_maps_missing_job_to_not_found(app, upgrades):
def test_upgrade_retry_returns_queued(app, upgrades):
upgrades.retry.return_value = _upgrade_job(id="job-1")
upgrades.retry.return_value = _upgrade_job(job_id="job-1")
with app.test_request_context(f"/datasets/{DATASET_ID}/knowledge-fs-upgrades/job-1"):
response, status = unwrap(DatasetKnowledgeFSUpgradeJobApi.post)(
DatasetKnowledgeFSUpgradeJobApi(), CONTEXT, DATASET_ID, "job-1"
api/tests/unit_tests/controllers/inner_api/plugin/test_plugin.py
+29 -14
View File
@@ -36,7 +36,10 @@ from controllers.inner_api.plugin.plugin import (
PluginModelCatalogApi,
PluginUploadFileRequestApi,
)
from core.datasource.entities.datasource_entities import DatasourceProviderType
from core.plugin.entities.request import RequestInvokeDatasource, RequestInvokeTextEmbedding, RequestListModels
from core.workflow.file_reference import build_file_reference
from graphon.model_runtime.entities.model_entities import ModelType
from models import Account, Tenant
from tests.unit_tests.config_override import apply_config_overrides
from tests.unit_tests.model_factories import make_tenant
@@ -52,6 +55,10 @@ def _user() -> Account:
return user
def _text_embedding_payload() -> RequestInvokeTextEmbedding:
return RequestInvokeTextEmbedding(provider="openai", model="text-embedding-3-small", texts=["hello"])
def _extract_raw_post(cls):
"""Extract the raw post() method from a plugin endpoint class.
@@ -121,9 +128,9 @@ class TestPluginInvokeTextEmbeddingApi:
result = raw_post(
api_instance,
user_model=MagicMock(id="user-id"),
tenant_model=MagicMock(id="tenant-id"),
payload=MagicMock(),
user_model=_user(),
tenant_model=_tenant(),
payload=_text_embedding_payload(),
)
assert result == {
@@ -141,9 +148,9 @@ class TestPluginInvokeTextEmbeddingApi:
result = raw_post(
api_instance,
user_model=MagicMock(id="user-id"),
tenant_model=MagicMock(id="tenant-id"),
payload=MagicMock(),
user_model=_user(),
tenant_model=_tenant(),
payload=_text_embedding_payload(),
)
assert result == {
@@ -184,9 +191,9 @@ class TestPluginModelCatalogApi:
@patch("controllers.inner_api.plugin.plugin.PluginModelBackwardsInvocation")
def test_post_lists_models_in_the_resolved_tenant_scope(self, mock_invocation, api_instance, app: Flask):
mock_invocation.list_models.return_value = {"items": [], "next_offset": None}
mock_tenant = MagicMock(id="tenant-id")
mock_user = MagicMock(id="user-id")
mock_payload = MagicMock()
mock_tenant = _tenant()
mock_user = _user()
mock_payload = RequestListModels(model_type=ModelType.LLM)
raw_post = _extract_raw_post(PluginModelCatalogApi)
result = raw_post(
@@ -218,11 +225,19 @@ class TestPluginInvokeDatasourceApi:
api_instance,
app: Flask,
):
mock_tenant = MagicMock(id="tenant-id")
mock_user = MagicMock(id="user-id")
mock_payload = MagicMock()
invocation = MagicMock()
stream = MagicMock()
mock_tenant = _tenant()
mock_user = _user()
mock_payload = RequestInvokeDatasource(
tenant_id="tenant-id",
user_id="user-id",
provider="langgenius/notion_datasource/notion_datasource",
datasource="notion_datasource",
datasource_type=DatasourceProviderType.ONLINE_DOCUMENT,
credential_id="credential-id",
operation="get_online_document_pages",
)
invocation = object()
stream = object()
mock_invocation.invoke.return_value = invocation
mock_invocation.convert_to_event_stream.return_value = stream
mock_length_prefixed_response.return_value = "response"
api/tests/unit_tests/controllers/knowledge_fs_controller_fakes.py
+103
View File
@@ -0,0 +1,103 @@
"""Typed recording fakes shared by the KnowledgeFS controller tests.
Controllers reach KnowledgeFS through a runtime object whose components expose
plain methods. These fakes replace those components without mock objects:
``Recorder`` stores every call in a list and returns or raises what the test
configured, and ``FakeComponent`` exposes only the operations a test declares,
so an unexpected delegation fails with ``AttributeError``.
"""
from __future__ import annotations
from collections.abc import Callable, Iterable, Iterator
from dataclasses import dataclass, field
@dataclass(frozen=True)
class RecordedCall:
args: tuple[object, ...] = ()
kwargs: dict[str, object] = field(default_factory=dict)
def invocation(*args: object, **kwargs: object) -> RecordedCall:
return RecordedCall(args=args, kwargs=kwargs)
class Recorder:
"""Callable test double that records calls and returns, raises or delegates.
``raises`` is raised on every call, ``results`` is consumed one value per
call (exceptions in it are raised), and ``effect`` computes the result from
the call arguments. Otherwise ``return_value`` is returned.
"""
def __init__(
self,
return_value: object = None,
*,
raises: BaseException | None = None,
results: Iterable[object] | None = None,
effect: Callable[..., object] | None = None,
) -> None:
self.return_value = return_value
self.raises = raises
self.effect = effect
self._results: Iterator[object] | None = iter(results) if results is not None else None
self.calls: list[RecordedCall] = []
def __call__(self, *args: object, **kwargs: object) -> object:
self.calls.append(RecordedCall(args=args, kwargs=kwargs))
if self.raises is not None:
raise self.raises
if self._results is not None:
result = next(self._results)
if isinstance(result, BaseException):
raise result
return result
if self.effect is not None:
return self.effect(*args, **kwargs)
return self.return_value
@property
def last_call(self) -> RecordedCall:
assert self.calls, "expected at least one call"
return self.calls[-1]
@property
def only_call(self) -> RecordedCall:
assert len(self.calls) == 1, f"expected exactly one call, got {len(self.calls)}"
return self.calls[0]
class FakeComponent:
"""Runtime component exposing exactly the named operations given to it."""
def __init__(self, **operations: Recorder) -> None:
self._operations = operations
def __getattr__(self, name: str) -> Recorder:
operations = self.__dict__.get("_operations", {})
if name not in operations:
raise AttributeError(name)
return operations[name]
@dataclass
class FakeRuntime:
"""The subset of ``KnowledgeFSRuntime`` that controllers resolve per request."""
facade: FakeComponent = field(default_factory=FakeComponent)
application: FakeComponent = field(default_factory=FakeComponent)
control_plane: FakeComponent = field(default_factory=FakeComponent)
app_bindings: FakeComponent = field(default_factory=FakeComponent)
broker: FakeComponent = field(default_factory=FakeComponent)
service_api_authorization: FakeComponent = field(default_factory=FakeComponent)
def raising(error: BaseException) -> Callable[..., object]:
"""Return a plain callable that raises ``error`` when invoked."""
def operation(*_args: object, **_kwargs: object) -> object:
raise error
return operation
api/tests/unit_tests/controllers/openapi/test_knowledge_fs.py
+29 -23
View File
@@ -3,7 +3,6 @@ from __future__ import annotations
import inspect
import uuid
from typing import Protocol, runtime_checkable
from unittest.mock import Mock
import pytest
from flask import Flask
@@ -15,6 +14,13 @@ from controllers.openapi import bp as openapi_bp
from controllers.openapi.auth.context import Context
from controllers.openapi.auth.subjects import AccountSubject
from libs.oauth_bearer import AuthContext, TokenType
from tests.unit_tests.controllers.knowledge_fs_controller_fakes import (
FakeComponent,
FakeRuntime,
Recorder,
invocation,
raising,
)
@runtime_checkable
@@ -68,14 +74,17 @@ def test_list_controller_adapts_public_pagination_to_the_product_operation(monke
from controllers.openapi.knowledge_fs import KnowledgeFsEntryListApi
from services.knowledge_fs.product_dto import KnowledgeFSListResponse
facade = Mock()
facade.list_knowledge_fs.return_value = KnowledgeFSListResponse.model_validate(
{
"items": [],
"nextCursor": "opaque-next-page",
"path": "/knowledge",
"truncated": True,
}
facade = FakeComponent(
list_knowledge_fs=Recorder(
KnowledgeFSListResponse.model_validate(
{
"items": [],
"nextCursor": "opaque-next-page",
"path": "/knowledge",
"truncated": True,
}
)
)
)
monkeypatch.setattr(module, "_knowledge_fs_facade", lambda: facade)
account_id = uuid.uuid4()
@@ -103,7 +112,7 @@ def test_list_controller_adapts_public_pagination_to_the_product_operation(monke
expires_at=None,
)
),
session=Mock(spec=Session),
session=Session(),
view_args={},
),
)
@@ -118,8 +127,7 @@ def test_list_controller_adapts_public_pagination_to_the_product_operation(monke
"preview": None,
"truncated": True,
}
facade.list_knowledge_fs.assert_called_once()
call = facade.list_knowledge_fs.call_args.kwargs
call = facade.list_knowledge_fs.only_call.kwargs
assert call["tenant_id"] == "workspace-1"
assert call["account_id"] == str(account_id)
assert call["control_space_id"] == "space-1"
@@ -215,7 +223,6 @@ def test_each_custom_interface_delegates_with_an_internal_product_query(
from controllers.openapi import knowledge_fs as module
from services.knowledge_fs import product_dto
facade = Mock()
product_responses = {
"tree_knowledge_fs": product_dto.KnowledgeFSTreeResponse.model_validate(
{
@@ -247,8 +254,8 @@ def test_each_custom_interface_delegates_with_an_internal_product_query(
}
),
}
delegate = Mock(return_value=product_responses[facade_method])
facade.configure_mock(**{facade_method: delegate})
delegate = Recorder(product_responses[facade_method])
facade = FakeComponent(**{facade_method: delegate})
monkeypatch.setattr(module, "_knowledge_fs_facade", lambda: facade)
account_id = uuid.uuid4()
api_class, handler = {
@@ -291,15 +298,14 @@ def test_each_custom_interface_delegates_with_an_internal_product_query(
expires_at=None,
)
),
session=Mock(spec=Session),
session=Session(),
view_args={},
),
**({"body": public_model} if method_name == "post" else {"query": public_model}),
)
assert isinstance(result, BaseModel)
delegate.assert_called_once()
call = delegate.call_args.kwargs
call = delegate.only_call.kwargs
assert call["tenant_id"] == "workspace-1"
assert call["account_id"] == str(account_id)
assert call["control_space_id"] == "space-1"
@@ -322,13 +328,13 @@ def test_knowledge_fs_facade_uses_the_configured_runtime(monkeypatch: pytest.Mon
from controllers.openapi import knowledge_fs as module
session_maker = object()
facade = object()
get_runtime = Mock(return_value=Mock(facade=facade))
facade = FakeComponent()
get_runtime = Recorder(FakeRuntime(facade=facade))
monkeypatch.setattr(module.session_factory, "get_session_maker", lambda: session_maker)
monkeypatch.setattr(module, "get_knowledge_fs_runtime", get_runtime)
assert module._knowledge_fs_facade() is facade
get_runtime.assert_called_once_with(session_maker)
assert get_runtime.calls == [invocation(session_maker)]
def test_knowledge_fs_error_adapter_uses_stable_domain_errors() -> None:
@@ -365,10 +371,10 @@ def test_knowledge_fs_error_adapter_uses_stable_domain_errors() -> None:
)
for domain_error, openapi_error in mappings:
fail = module._knowledge_fs_errors(Mock(side_effect=domain_error))
guarded = module._knowledge_fs_errors(raising(domain_error))
with pytest.raises(openapi_error):
fail()
guarded()
@pytest.mark.parametrize("operation", ["ls", "tree", "grep", "find", "cat", "stat", "diff"])
api/tests/unit_tests/controllers/test_knowledge_fs_product_controllers.py
+273 -210
View File
@@ -4,12 +4,12 @@ import ast
import inspect
from collections.abc import Generator
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import UTC, datetime
from io import BytesIO
from pathlib import Path
from types import FunctionType, SimpleNamespace
from types import FunctionType
from typing import Never
from unittest.mock import MagicMock
import pytest
from flask import Flask
@@ -27,6 +27,10 @@ from controllers.console.knowledge_fs.error import (
from controllers.console.wraps import RBACPermission
from controllers.service_api import service_api_ns
from controllers.service_api.knowledge_fs import resources as service_resources
from models import Account
from models.enums import ApiTokenType
from models.model import ApiToken
from services.knowledge_fs.capability_broker import KnowledgeFSIssuedProductCapability
from services.knowledge_fs.control_space_commands import KnowledgeFSControlSpaceIntentConflictError
from services.knowledge_fs.download_service import KnowledgeFSDownloadUnavailableError
from services.knowledge_fs.object_storage import KnowledgeFSObjectMetadata
@@ -55,10 +59,64 @@ from services.knowledge_fs.service_api_authorization import (
KnowledgeFSServiceApiScopeError,
)
from tests.unit_tests.config_override import apply_config_overrides
from tests.unit_tests.controllers.knowledge_fs_controller_fakes import FakeComponent, FakeRuntime, Recorder, invocation
_API_ROOT = Path(__file__).resolve().parents[3]
@dataclass(frozen=True)
class _FacadeRuntime[T]:
facade: T
@dataclass(frozen=True)
class _BrokerRuntime[T]:
broker: T
def _account(account_id: str = "account-1") -> Account:
account = Account(name="Test User", email="user@example.com")
account.id = account_id
return account
def _dataset_api_token(tenant_id: str | None = "tenant-1") -> ApiToken:
api_token = ApiToken()
api_token.id = "token-1"
api_token.tenant_id = tenant_id
api_token.type = ApiTokenType.DATASET
api_token.token = "dataset-test-key"
return api_token
def _issued_capability(
*, token: str, operation_id: str, knowledge_space_id: str = "space-1"
) -> KnowledgeFSIssuedProductCapability:
return KnowledgeFSIssuedProductCapability(
token=token,
expires_at=datetime(2030, 1, 1, tzinfo=UTC),
operation_id=operation_id,
knowledge_space_id=knowledge_space_id,
knowledge_space_revision=1,
trace_id="trace-1",
)
def _service_profile(control_space_id: str) -> KnowledgeFSServiceApiProfile:
return KnowledgeFSServiceApiProfile(
tenant_id="tenant-1",
control_space_id=control_space_id,
api_token_id="token-1",
principal_id="token-1",
knowledge_space_id="space-1",
knowledge_space_revision=1,
membership_epoch=0,
space_acl_epoch=0,
external_access_epoch=0,
content_policy_revision=0,
)
def _rbac_wrapper(view: FunctionType) -> FunctionType:
current = inspect.unwrap(
view,
@@ -194,12 +252,19 @@ def test_console_and_service_api_routes_are_registered() -> None:
def test_initial_website_preview_job_routes_delegate_to_namespace_capabilities(
monkeypatch: pytest.MonkeyPatch,
) -> None:
account = SimpleNamespace(id="account-1")
preview_service = SimpleNamespace(require_visible_credential=MagicMock())
facade = SimpleNamespace(
create_namespace_source_preview=MagicMock(return_value=SimpleNamespace(job_id="job-1")),
get_namespace_source_preview=MagicMock(
return_value=KnowledgeFSNamespacePreviewJobResponse(
account = _account()
preview_service = FakeComponent(require_visible_credential=Recorder())
facade = FakeComponent(
create_namespace_source_preview=Recorder(
KnowledgeFSNamespacePreviewJobResponse(
job_id="job-1",
status="queued",
configuration_fingerprint="f" * 64,
expires_at="2030-01-01T00:00:00Z",
)
),
get_namespace_source_preview=Recorder(
KnowledgeFSNamespacePreviewJobResponse(
job_id="job-1",
status="completed",
configuration_fingerprint="f" * 64,
@@ -213,8 +278,8 @@ def test_initial_website_preview_job_routes_delegate_to_namespace_capabilities(
],
)
),
cancel_namespace_source_preview=MagicMock(
return_value=KnowledgeFSNamespacePreviewJobResponse(
cancel_namespace_source_preview=Recorder(
KnowledgeFSNamespacePreviewJobResponse(
job_id="job-1",
status="consumed",
configuration_fingerprint="f" * 64,
@@ -224,7 +289,7 @@ def test_initial_website_preview_job_routes_delegate_to_namespace_capabilities(
)
monkeypatch.setattr(console_resources, "current_account_with_tenant", lambda: (account, "tenant-1"))
monkeypatch.setattr(console_resources, "KnowledgeFSInitialSourcePreviewService", lambda _: preview_service)
monkeypatch.setattr(console_resources, "_console_services", lambda: SimpleNamespace(facade=facade))
monkeypatch.setattr(console_resources, "_console_services", lambda: FakeRuntime(facade=facade))
app = Flask(__name__)
with app.test_request_context(
@@ -243,8 +308,8 @@ def test_initial_website_preview_job_routes_delegate_to_namespace_capabilities(
)
assert status == 202
assert response == {"job_id": "job-1", "status": "pending"}
preview_service.require_visible_credential.assert_called_once()
facade.create_namespace_source_preview.assert_called_once()
assert len(preview_service.require_visible_credential.calls) == 1
assert len(facade.create_namespace_source_preview.calls) == 1
with app.test_request_context():
response = inspect.unwrap(console_resources.KnowledgeFSInitialSourcePreviewJobApi.get)(
@@ -301,12 +366,12 @@ def test_document_download_routes_deny_callers_without_dataset_download_permissi
method_name: FunctionType,
path_args: dict[str, str],
) -> None:
permission_gate = MagicMock(side_effect=Forbidden())
permission_gate = Recorder(raises=Forbidden())
apply_config_overrides(monkeypatch, RBAC_ENABLED=True)
monkeypatch.setattr(
common_wraps,
"current_account_with_tenant",
lambda: (SimpleNamespace(id="account-1"), "tenant-1"),
lambda: (_account(), "tenant-1"),
)
monkeypatch.setattr("controllers.common.rbac.checks.RBACService.CheckAccess.check", permission_gate)
# The download routes carry the knowledge rate limit too; it needs a logged-in account.
@@ -321,13 +386,15 @@ def test_document_download_routes_deny_callers_without_dataset_download_permissi
with pytest.raises(Forbidden):
permission_wrapper(api_class(), **path_args)
permission_gate.assert_called_once_with(
"tenant-1",
"account-1",
scene=RBACPermission.DATASET_DOCUMENT_DOWNLOAD,
resource_type=None,
resource_id=None,
)
assert permission_gate.calls == [
invocation(
"tenant-1",
"account-1",
scene=RBACPermission.DATASET_DOCUMENT_DOWNLOAD,
resource_type=None,
resource_id=None,
)
]
def test_single_document_download_allows_dataset_download_permission_and_keeps_tenant_scope(
@@ -341,20 +408,20 @@ def test_single_document_download_allows_dataset_download_permission_and_keeps_t
sha256="a" * 64,
size_bytes=4,
)
facade = SimpleNamespace(prepare_logical_document_download=MagicMock(return_value=descriptor))
download_service = SimpleNamespace(load_stream=MagicMock(return_value=iter((b"body",))))
permission_gate = MagicMock()
facade = FakeComponent(prepare_logical_document_download=Recorder(descriptor))
download_service = FakeComponent(load_stream=Recorder(iter((b"body",))))
permission_gate = Recorder(True)
apply_config_overrides(monkeypatch, RBAC_ENABLED=True)
monkeypatch.setattr(
common_wraps,
"current_account_with_tenant",
lambda: (SimpleNamespace(id="account-1"), "tenant-1"),
lambda: (_account(), "tenant-1"),
)
monkeypatch.setattr("controllers.common.rbac.checks.RBACService.CheckAccess.check", permission_gate)
# The download routes carry the knowledge rate limit too; it needs a logged-in account.
monkeypatch.setattr(console_wraps, "check_knowledge_rate_limit", lambda: None)
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_console_services", lambda: SimpleNamespace(facade=facade))
monkeypatch.setattr(console_resources, "_console_services", lambda: FakeRuntime(facade=facade))
monkeypatch.setattr(console_resources, "KnowledgeFSDownloadService", lambda: download_service)
permission_wrapper = _rbac_wrapper(console_resources.KnowledgeFSSpaceLogicalDocumentDownloadApi.get)
app = Flask(__name__)
@@ -368,42 +435,44 @@ def test_single_document_download_allows_dataset_download_permission_and_keeps_t
assert response.get_data() == b"body"
assert response.headers["Content-Disposition"] == "attachment; filename*=UTF-8''guide.md"
permission_gate.assert_called_once_with(
"tenant-1",
"account-1",
scene=RBACPermission.DATASET_DOCUMENT_DOWNLOAD,
resource_type=None,
resource_id=None,
)
facade.prepare_logical_document_download.assert_called_once_with(
tenant_id="tenant-1",
account_id="account-1",
control_space_id="control-1",
document_id="document-1",
)
assert permission_gate.calls == [
invocation(
"tenant-1",
"account-1",
scene=RBACPermission.DATASET_DOCUMENT_DOWNLOAD,
resource_type=None,
resource_id=None,
)
]
assert facade.prepare_logical_document_download.calls == [
invocation(
tenant_id="tenant-1",
account_id="account-1",
control_space_id="control-1",
document_id="document-1",
)
]
def test_single_document_download_preserves_resource_not_found_after_permission_allows(
monkeypatch: pytest.MonkeyPatch,
) -> None:
facade = SimpleNamespace(
prepare_logical_document_download=MagicMock(
side_effect=KnowledgeFSProductResourceNotFoundError("document not found")
)
facade = FakeComponent(
prepare_logical_document_download=Recorder(raises=KnowledgeFSProductResourceNotFoundError("document not found"))
)
permission_gate = MagicMock()
download_service_factory = MagicMock()
permission_gate = Recorder(True)
download_service_factory = Recorder()
apply_config_overrides(monkeypatch, RBAC_ENABLED=True)
monkeypatch.setattr(
common_wraps,
"current_account_with_tenant",
lambda: (SimpleNamespace(id="account-1"), "tenant-1"),
lambda: (_account(), "tenant-1"),
)
monkeypatch.setattr("controllers.common.rbac.checks.RBACService.CheckAccess.check", permission_gate)
# The download routes carry the knowledge rate limit too; it needs a logged-in account.
monkeypatch.setattr(console_wraps, "check_knowledge_rate_limit", lambda: None)
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_console_services", lambda: SimpleNamespace(facade=facade))
monkeypatch.setattr(console_resources, "_console_services", lambda: FakeRuntime(facade=facade))
monkeypatch.setattr(console_resources, "KnowledgeFSDownloadService", download_service_factory)
permission_wrapper = _rbac_wrapper(console_resources.KnowledgeFSSpaceLogicalDocumentDownloadApi.get)
app = Flask(__name__)
@@ -415,14 +484,16 @@ def test_single_document_download_preserves_resource_not_found_after_permission_
document_id="document-1",
)
permission_gate.assert_called_once()
facade.prepare_logical_document_download.assert_called_once_with(
tenant_id="tenant-1",
account_id="account-1",
control_space_id="foreign-control",
document_id="document-1",
)
download_service_factory.assert_not_called()
assert len(permission_gate.calls) == 1
assert facade.prepare_logical_document_download.calls == [
invocation(
tenant_id="tenant-1",
account_id="account-1",
control_space_id="foreign-control",
document_id="document-1",
)
]
assert download_service_factory.calls == []
def test_single_document_download_maps_storage_unavailable_after_permission_allows(
@@ -436,22 +507,22 @@ def test_single_document_download_maps_storage_unavailable_after_permission_allo
sha256="a" * 64,
size_bytes=4,
)
facade = SimpleNamespace(prepare_logical_document_download=MagicMock(return_value=descriptor))
download_service = SimpleNamespace(
load_stream=MagicMock(side_effect=KnowledgeFSDownloadUnavailableError("storage unavailable"))
facade = FakeComponent(prepare_logical_document_download=Recorder(descriptor))
download_service = FakeComponent(
load_stream=Recorder(raises=KnowledgeFSDownloadUnavailableError("storage unavailable"))
)
permission_gate = MagicMock()
permission_gate = Recorder(True)
apply_config_overrides(monkeypatch, RBAC_ENABLED=True)
monkeypatch.setattr(
common_wraps,
"current_account_with_tenant",
lambda: (SimpleNamespace(id="account-1"), "tenant-1"),
lambda: (_account(), "tenant-1"),
)
monkeypatch.setattr("controllers.common.rbac.checks.RBACService.CheckAccess.check", permission_gate)
# The download routes carry the knowledge rate limit too; it needs a logged-in account.
monkeypatch.setattr(console_wraps, "check_knowledge_rate_limit", lambda: None)
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_console_services", lambda: SimpleNamespace(facade=facade))
monkeypatch.setattr(console_resources, "_console_services", lambda: FakeRuntime(facade=facade))
monkeypatch.setattr(console_resources, "KnowledgeFSDownloadService", lambda: download_service)
permission_wrapper = _rbac_wrapper(console_resources.KnowledgeFSSpaceLogicalDocumentDownloadApi.get)
app = Flask(__name__)
@@ -463,8 +534,8 @@ def test_single_document_download_maps_storage_unavailable_after_permission_allo
document_id="document-1",
)
permission_gate.assert_called_once()
download_service.load_stream.assert_called_once_with(descriptor)
assert len(permission_gate.calls) == 1
assert download_service.load_stream.calls == [invocation(descriptor)]
def test_batch_document_download_maps_storage_unavailable_after_permission_allows(
@@ -478,22 +549,26 @@ def test_batch_document_download_maps_storage_unavailable_after_permission_allow
sha256="a" * 64,
size_bytes=4,
)
facade = SimpleNamespace(prepare_logical_document_download=MagicMock(return_value=descriptor))
zip_context = MagicMock()
zip_context.__enter__.side_effect = KnowledgeFSDownloadUnavailableError("storage unavailable")
download_service = SimpleNamespace(build_zip_tempfile=MagicMock(return_value=zip_context))
permission_gate = MagicMock()
facade = FakeComponent(prepare_logical_document_download=Recorder(descriptor))
@contextmanager
def unavailable_zip() -> Generator[Never]:
raise KnowledgeFSDownloadUnavailableError("storage unavailable")
yield
download_service = FakeComponent(build_zip_tempfile=Recorder(unavailable_zip()))
permission_gate = Recorder(True)
apply_config_overrides(monkeypatch, RBAC_ENABLED=True)
monkeypatch.setattr(
common_wraps,
"current_account_with_tenant",
lambda: (SimpleNamespace(id="account-1"), "tenant-1"),
lambda: (_account(), "tenant-1"),
)
monkeypatch.setattr("controllers.common.rbac.checks.RBACService.CheckAccess.check", permission_gate)
# The download routes carry the knowledge rate limit too; it needs a logged-in account.
monkeypatch.setattr(console_wraps, "check_knowledge_rate_limit", lambda: None)
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_console_services", lambda: SimpleNamespace(facade=facade))
monkeypatch.setattr(console_resources, "_console_services", lambda: FakeRuntime(facade=facade))
monkeypatch.setattr(console_resources, "KnowledgeFSDownloadService", lambda: download_service)
permission_wrapper = _rbac_wrapper(console_resources.KnowledgeFSSpaceLogicalDocumentsDownloadApi.post)
app = Flask(__name__)
@@ -507,8 +582,8 @@ def test_batch_document_download_maps_storage_unavailable_after_permission_allow
control_space_id="control-1",
)
permission_gate.assert_called_once()
download_service.build_zip_tempfile.assert_called_once_with([descriptor])
assert len(permission_gate.calls) == 1
assert download_service.build_zip_tempfile.calls == [invocation([descriptor])]
def test_knowledge_fs_request_and_response_schemas_are_registered() -> None:
@@ -631,9 +706,9 @@ def test_document_multimodal_manifest_console_bff_exposes_only_authorized_asset_
f"tenant-1/spaces/018f0d60-7a49-7cc2-9c1b-5b36f18f2c43/documents/{document_id}/assets/figure-1-full.png"
)
)
facade = SimpleNamespace(get_document_multimodal_manifest=MagicMock(return_value=manifest))
facade = FakeComponent(get_document_multimodal_manifest=Recorder(manifest))
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_console_services", lambda: SimpleNamespace(facade=facade))
monkeypatch.setattr(console_resources, "_console_services", lambda: FakeRuntime(facade=facade))
apply_config_overrides(monkeypatch, CONSOLE_API_URL="")
app = Flask(__name__)
@@ -653,12 +728,14 @@ def test_document_multimodal_manifest_console_bff_exposes_only_authorized_asset_
assert item["parse_element_id"] == "parse-element-1"
assert "objectKey" not in str(response)
assert "sha256" not in str(response)
facade.get_document_multimodal_manifest.assert_called_once_with(
tenant_id="tenant-1",
account_id="account-1",
control_space_id="control-1",
document_id=document_id,
)
assert facade.get_document_multimodal_manifest.calls == [
invocation(
tenant_id="tenant-1",
account_id="account-1",
control_space_id="control-1",
document_id=document_id,
)
]
def test_document_multimodal_asset_console_bff_reauthorizes_and_streams_scoped_images(
@@ -669,10 +746,10 @@ def test_document_multimodal_asset_console_bff_reauthorizes_and_streams_scoped_i
f"tenant-1/spaces/018f0d60-7a49-7cc2-9c1b-5b36f18f2c43/documents/{document_id}/assets/figure-1-full.png"
)
manifest = _document_multimodal_manifest(object_key=object_key)
facade = SimpleNamespace(get_document_multimodal_manifest=MagicMock(return_value=manifest))
object_storage = SimpleNamespace(
head_object=MagicMock(
return_value=KnowledgeFSObjectMetadata(
facade = FakeComponent(get_document_multimodal_manifest=Recorder(manifest))
object_storage = FakeComponent(
head_object=Recorder(
KnowledgeFSObjectMetadata(
checksum_sha256_base64="checksum",
content_type="image/png",
key=object_key,
@@ -681,10 +758,10 @@ def test_document_multimodal_asset_console_bff_reauthorizes_and_streams_scoped_i
)
)
)
download_service = SimpleNamespace(load_stream=MagicMock(return_value=iter((b"im", b"age"))))
download_service_factory = MagicMock(return_value=download_service)
download_service = FakeComponent(load_stream=Recorder(iter((b"im", b"age"))))
download_service_factory = Recorder(download_service)
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_console_services", lambda: SimpleNamespace(facade=facade))
monkeypatch.setattr(console_resources, "_console_services", lambda: FakeRuntime(facade=facade))
monkeypatch.setattr(console_resources, "KnowledgeFSObjectStorageService", lambda: object_storage)
monkeypatch.setattr(console_resources, "KnowledgeFSDownloadService", download_service_factory)
app = Flask(__name__)
@@ -703,9 +780,9 @@ def test_document_multimodal_asset_console_bff_reauthorizes_and_streams_scoped_i
assert response.headers["Content-Disposition"] == "inline"
assert response.headers["Content-Security-Policy"] == "default-src 'none'; sandbox"
assert response.headers["X-Content-Type-Options"] == "nosniff"
object_storage.head_object.assert_called_once_with(key=object_key)
download_service_factory.assert_called_once_with(object_storage=object_storage)
download_service.load_stream.assert_called_once()
assert object_storage.head_object.calls == [invocation(key=object_key)]
assert download_service_factory.calls == [invocation(object_storage=object_storage)]
assert len(download_service.load_stream.calls) == 1
def test_document_multimodal_asset_console_bff_rejects_cross_document_object_keys(
@@ -718,10 +795,10 @@ def test_document_multimodal_asset_console_bff_rejects_cross_document_object_key
"documents/018f0d60-7a49-7cc2-9c1b-5b36f18f2cff/assets/foreign.png"
)
)
facade = SimpleNamespace(get_document_multimodal_manifest=MagicMock(return_value=manifest))
object_storage_factory = MagicMock()
facade = FakeComponent(get_document_multimodal_manifest=Recorder(manifest))
object_storage_factory = Recorder()
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_console_services", lambda: SimpleNamespace(facade=facade))
monkeypatch.setattr(console_resources, "_console_services", lambda: FakeRuntime(facade=facade))
monkeypatch.setattr(console_resources, "KnowledgeFSObjectStorageService", object_storage_factory)
app = Flask(__name__)
get = inspect.unwrap(console_resources.KnowledgeFSSpaceDocumentMultimodalAssetApi.get)
@@ -734,7 +811,7 @@ def test_document_multimodal_asset_console_bff_rejects_cross_document_object_key
item_id="figure:1",
)
object_storage_factory.assert_not_called()
assert object_storage_factory.calls == []
def test_source_workflow_import_contract_is_discriminated_idempotent_and_accepted() -> None:
@@ -785,7 +862,7 @@ def test_small_file_console_bff_reads_only_through_facade_and_returns_no_capabil
)
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_console_services", lambda: SimpleNamespace(facade=Facade()))
monkeypatch.setattr(console_resources, "_console_services", lambda: _FacadeRuntime(facade=Facade()))
app = Flask(__name__)
with app.test_request_context(
@@ -851,7 +928,7 @@ def test_document_upload_console_bff_reads_only_through_facade_and_returns_accep
)
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_console_services", lambda: SimpleNamespace(facade=Facade()))
monkeypatch.setattr(console_resources, "_console_services", lambda: _FacadeRuntime(facade=Facade()))
app = Flask(__name__)
with app.test_request_context(
@@ -876,11 +953,14 @@ def test_document_upload_console_bff_reads_only_through_facade_and_returns_accep
def test_staged_document_claim_passes_only_upload_id_to_the_admission_service(
monkeypatch: pytest.MonkeyPatch,
) -> None:
staged_uploads = MagicMock()
staged_uploads.claim.return_value = KnowledgeFSDocumentStagedUploadAcceptedResponse(
upload_id="staged-1",
document_asset_id="asset-1",
compilation_job_id="job-1",
staged_uploads = FakeComponent(
claim=Recorder(
KnowledgeFSDocumentStagedUploadAcceptedResponse(
upload_id="staged-1",
document_asset_id="asset-1",
compilation_job_id="job-1",
)
)
)
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_staged_uploads", lambda: staged_uploads)
@@ -897,8 +977,8 @@ def test_staged_document_claim_passes_only_upload_id_to_the_admission_service(
"document_asset_id": "asset-1",
"compilation_job_id": "job-1",
}
staged_uploads.claim.assert_called_once()
assert staged_uploads.claim.call_args.kwargs["payload"].upload_id == "staged-1"
assert len(staged_uploads.claim.calls) == 1
assert staged_uploads.claim.last_call.kwargs["payload"].upload_id == "staged-1"
@pytest.mark.parametrize("file_size_limit_mb", [15, 50])
@@ -922,8 +1002,7 @@ def test_workspace_staging_upload_persists_before_space_creation(
admission_events.append("release")
admission = RecordingAdmission()
staged_uploads = MagicMock()
staged_uploads.stage.return_value = KnowledgeFSStagedUploadResponse(
staged = KnowledgeFSStagedUploadResponse(
id="staged-1",
file_name="notes.md",
content_type="text/markdown",
@@ -935,10 +1014,10 @@ def test_workspace_staging_upload_persists_before_space_creation(
def stage(**_: object) -> KnowledgeFSStagedUploadResponse:
assert admission.active
admission_events.append("stage")
return staged_uploads.stage.return_value
return staged
staged_uploads.stage.side_effect = stage
account = SimpleNamespace(id="account-1")
staged_uploads = FakeComponent(stage=Recorder(effect=stage))
account = _account()
monkeypatch.setattr(console_resources, "current_account_with_tenant", lambda: (account, "tenant-1"))
monkeypatch.setattr(console_resources, "_staged_uploads", lambda: staged_uploads)
monkeypatch.setattr(console_resources, "DEFAULT_KNOWLEDGE_FS_BUFFERED_UPLOAD_ADMISSION", admission)
@@ -960,14 +1039,16 @@ def test_workspace_staging_upload_persists_before_space_creation(
assert status == 201
assert response["id"] == "staged-1"
assert admission_events == [f"admit:{file_size_limit_mb * 1024 * 1024}", "stage", "release"]
staged_uploads.stage.assert_called_once_with(
tenant_id="tenant-1",
account=account,
file_name="notes.md",
content_type="text/markdown",
body=b"# Notes",
file_size_limit_mb=file_size_limit_mb,
)
assert staged_uploads.stage.calls == [
invocation(
tenant_id="tenant-1",
account=account,
file_name="notes.md",
content_type="text/markdown",
body=b"# Notes",
file_size_limit_mb=file_size_limit_mb,
)
]
def test_workspace_staging_zero_byte_plan_limit_returns_request_too_large_before_admission(
@@ -978,8 +1059,8 @@ def test_workspace_staging_zero_byte_plan_limit_returns_request_too_large_before
_ = reserved_bytes
raise AssertionError("a disabled upload limit must fail before admission")
staged_uploads = MagicMock()
account = SimpleNamespace(id="account-1")
staged_uploads = FakeComponent(stage=Recorder())
account = _account()
monkeypatch.setattr(console_resources, "current_account_with_tenant", lambda: (account, "tenant-1"))
monkeypatch.setattr(console_resources, "_staged_uploads", lambda: staged_uploads)
monkeypatch.setattr(console_resources, "DEFAULT_KNOWLEDGE_FS_BUFFERED_UPLOAD_ADMISSION", FailingAdmission())
@@ -1000,7 +1081,7 @@ def test_workspace_staging_zero_byte_plan_limit_returns_request_too_large_before
post(console_resources.KnowledgeFSStagedUploadsApi())
assert rejected.value.status_code == 413
staged_uploads.stage.assert_not_called()
assert staged_uploads.stage.calls == []
def test_logical_document_delete_accepts_initial_row_version(
@@ -1031,7 +1112,7 @@ def test_logical_document_delete_accepts_initial_row_version(
)
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_console_services", lambda: SimpleNamespace(facade=Facade()))
monkeypatch.setattr(console_resources, "_console_services", lambda: _FacadeRuntime(facade=Facade()))
app = Flask(__name__)
with app.test_request_context(
@@ -1261,7 +1342,7 @@ def test_jwks_http_resource_returns_only_public_keys(monkeypatch: pytest.MonkeyP
{"alg": "RS256", "e": "AQAB", "kid": "previous", "kty": "RSA", "n": "modulus-2", "use": "sig"},
]
}
issuer = SimpleNamespace(public_jwks=lambda: public_jwks)
issuer = FakeComponent(public_jwks=Recorder(public_jwks))
apply_config_overrides(
monkeypatch,
KNOWLEDGE_FS_CAPABILITY_V2_ENABLED=True,
@@ -1293,14 +1374,9 @@ def test_service_profile_authorizes_the_dataset_key_for_the_route_space(monkeypa
external_access_epoch=0,
content_policy_revision=0,
)
authorization = MagicMock()
authorization.authorize.return_value = profile
runtime = SimpleNamespace(service_api_authorization=authorization)
monkeypatch.setattr(
service_resources,
"validate_and_get_api_token",
MagicMock(return_value=SimpleNamespace(id="token-1", tenant_id="tenant-1")),
)
authorization = FakeComponent(authorize=Recorder(profile))
runtime = FakeRuntime(service_api_authorization=authorization)
monkeypatch.setattr(service_resources, "validate_and_get_api_token", Recorder(_dataset_api_token()))
app = Flask(__name__)
with app.test_request_context(headers={"Authorization": "Bearer dataset-test-key"}):
result = service_resources._profile(
@@ -1310,22 +1386,24 @@ def test_service_profile_authorizes_the_dataset_key_for_the_route_space(monkeypa
)
assert result is profile
authorization.authorize.assert_called_once_with(
api_token_id="token-1",
tenant_id="tenant-1",
control_space_id="control-2",
)
assert authorization.authorize.calls == [
invocation(
api_token_id="token-1",
tenant_id="tenant-1",
control_space_id="control-2",
)
]
def test_service_profile_applies_the_knowledge_rate_limit_before_authorizing(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Every KnowledgeFS service route shares the legacy dataset service API rate limit."""
authorization = MagicMock()
runtime = SimpleNamespace(service_api_authorization=authorization)
api_token = SimpleNamespace(id="token-1", tenant_id="tenant-1")
monkeypatch.setattr(service_resources, "validate_and_get_api_token", MagicMock(return_value=api_token))
rate_limit = MagicMock(side_effect=Forbidden("rate limited"))
authorization = FakeComponent(authorize=Recorder())
runtime = FakeRuntime(service_api_authorization=authorization)
api_token = _dataset_api_token()
monkeypatch.setattr(service_resources, "validate_and_get_api_token", Recorder(api_token))
rate_limit = Recorder(raises=Forbidden("rate limited"))
monkeypatch.setattr(service_resources, "check_knowledge_rate_limit", rate_limit)
with pytest.raises(Forbidden, match="rate limited"):
@@ -1335,18 +1413,14 @@ def test_service_profile_applies_the_knowledge_rate_limit_before_authorizing(
control_space_id="control-2",
)
rate_limit.assert_called_once_with(api_token)
authorization.authorize.assert_not_called()
assert rate_limit.calls == [invocation(api_token)]
assert authorization.authorize.calls == []
def test_service_profile_rejects_a_dataset_key_without_a_workspace(monkeypatch: pytest.MonkeyPatch) -> None:
authorization = MagicMock()
runtime = SimpleNamespace(service_api_authorization=authorization)
monkeypatch.setattr(
service_resources,
"validate_and_get_api_token",
MagicMock(return_value=SimpleNamespace(id="token-1", tenant_id=None)),
)
authorization = FakeComponent(authorize=Recorder())
runtime = FakeRuntime(service_api_authorization=authorization)
monkeypatch.setattr(service_resources, "validate_and_get_api_token", Recorder(_dataset_api_token(tenant_id=None)))
with pytest.raises(KnowledgeFSServiceApiAuthorizationError, match="not workspace-scoped"):
service_resources._profile(
@@ -1355,7 +1429,7 @@ def test_service_profile_rejects_a_dataset_key_without_a_workspace(monkeypatch:
control_space_id="control-2",
)
authorization.authorize.assert_not_called()
assert authorization.authorize.calls == []
def test_product_modules_do_not_import_dify_dataset_or_document_services() -> None:
@@ -1415,20 +1489,16 @@ def test_query_stream_capability_issues_exact_space_grant_without_token_in_url(
calls: list[dict[str, object]] = []
class Broker:
def issue_interactive(self, **kwargs: object) -> SimpleNamespace:
def issue_interactive(self, **kwargs: object) -> KnowledgeFSIssuedProductCapability:
calls.append(kwargs)
return SimpleNamespace(
token="query-capability",
trace_id="trace-1",
expires_at=datetime(2030, 1, 1, tzinfo=UTC),
)
return _issued_capability(token="query-capability", operation_id="createQuery")
apply_config_overrides(monkeypatch, CONSOLE_API_URL="https://dify.test")
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(
console_resources,
"_console_services",
lambda: SimpleNamespace(broker=Broker()),
lambda: _BrokerRuntime(broker=Broker()),
)
app = Flask(__name__)
@@ -1458,21 +1528,16 @@ def test_query_admission_binds_validated_mode_to_resolved_kfs_space(monkeypatch:
calls: list[dict[str, object]] = []
class Broker:
def issue_interactive(self, **kwargs: object) -> SimpleNamespace:
def issue_interactive(self, **kwargs: object) -> KnowledgeFSIssuedProductCapability:
calls.append(kwargs)
return SimpleNamespace(
token="query-capability",
trace_id="trace-1",
expires_at=datetime(2030, 1, 1, tzinfo=UTC),
knowledge_space_id="space-1",
)
return _issued_capability(token="query-capability", operation_id="createQuery")
apply_config_overrides(monkeypatch, CONSOLE_API_URL="https://dify.test")
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(
console_resources,
"_console_services",
lambda: SimpleNamespace(broker=Broker()),
lambda: _BrokerRuntime(broker=Broker()),
)
app = Flask(__name__)
@@ -1503,10 +1568,14 @@ def test_query_admission_binds_validated_mode_to_resolved_kfs_space(monkeypatch:
def test_console_query_stream_proxy_forwards_only_the_admitted_capability(
monkeypatch: pytest.MonkeyPatch,
) -> None:
close = MagicMock()
facade = SimpleNamespace(
stream_query=MagicMock(
return_value=KnowledgeFSRemoteSSEResponse(
closed: list[str] = []
def close() -> None:
closed.append("closed")
facade = FakeComponent(
stream_query=Recorder(
KnowledgeFSRemoteSSEResponse(
status_code=200,
headers=(
("content-type", "text/event-stream"),
@@ -1517,7 +1586,7 @@ def test_console_query_stream_proxy_forwards_only_the_admitted_capability(
)
)
)
monkeypatch.setattr(console_resources, "_console_services", lambda: SimpleNamespace(facade=facade))
monkeypatch.setattr(console_resources, "_console_services", lambda: FakeRuntime(facade=facade))
app = Flask(__name__)
with app.test_request_context(
@@ -1541,8 +1610,8 @@ def test_console_query_stream_proxy_forwards_only_the_admitted_capability(
assert response.headers["Content-Type"] == "text/event-stream"
assert response.headers["X-Accel-Buffering"] == "no"
assert b"".join(response.response) == b'event: answer\ndata: {"answer":"ok"}\n\n'
close.assert_called_once_with()
call = facade.stream_query.call_args.kwargs
assert closed == ["closed"]
call = facade.stream_query.last_call.kwargs
assert call["capability_token"] == "capability-token"
assert call["trace_id"] == "trace-1"
assert call["payload"].knowledge_space_id == "space-1"
@@ -1552,10 +1621,14 @@ def test_console_query_stream_proxy_forwards_only_the_admitted_capability(
def test_service_query_stream_proxy_forwards_only_the_admitted_capability(
monkeypatch: pytest.MonkeyPatch,
) -> None:
close = MagicMock()
facade = SimpleNamespace(
stream_query=MagicMock(
return_value=KnowledgeFSRemoteSSEResponse(
closed: list[str] = []
def close() -> None:
closed.append("closed")
facade = FakeComponent(
stream_query=Recorder(
KnowledgeFSRemoteSSEResponse(
status_code=200,
headers=(
("content-type", "text/event-stream"),
@@ -1566,7 +1639,7 @@ def test_service_query_stream_proxy_forwards_only_the_admitted_capability(
)
)
)
monkeypatch.setattr(service_resources, "_runtime", lambda: SimpleNamespace(facade=facade))
monkeypatch.setattr(service_resources, "_runtime", lambda: FakeRuntime(facade=facade))
app = Flask(__name__)
with app.test_request_context(
@@ -1590,8 +1663,8 @@ def test_service_query_stream_proxy_forwards_only_the_admitted_capability(
assert response.headers["Content-Type"] == "text/event-stream"
assert response.headers["X-Accel-Buffering"] == "no"
assert b"".join(response.response) == b'event: answer\ndata: {"answer":"ok"}\n\n'
close.assert_called_once_with()
call = facade.stream_query.call_args.kwargs
assert closed == ["closed"]
call = facade.stream_query.last_call.kwargs
assert call["capability_token"] == "capability-token"
assert call["trace_id"] == "trace-1"
assert call["payload"].knowledge_space_id == "space-1"
@@ -1624,15 +1697,11 @@ def test_task_stream_capability_uses_broker(
calls: list[dict[str, object]] = []
class Broker:
def issue_interactive(self, **kwargs: object) -> SimpleNamespace:
def issue_interactive(self, **kwargs: object) -> KnowledgeFSIssuedProductCapability:
calls.append(kwargs)
return SimpleNamespace(
token="direct-capability",
expires_at=datetime(2030, 1, 1, tzinfo=UTC),
knowledge_space_id="space-1",
)
return _issued_capability(token="direct-capability", operation_id="streamResearchTask")
runtime = SimpleNamespace(broker=Broker())
runtime = _BrokerRuntime(broker=Broker())
apply_config_overrides(monkeypatch, CONSOLE_API_URL="https://dify.test")
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_console_services", lambda: runtime)
@@ -1656,29 +1725,21 @@ def test_task_stream_capability_uses_broker(
def test_service_query_admission_uses_broker(monkeypatch: pytest.MonkeyPatch) -> None:
calls: list[dict[str, object]] = []
profile = SimpleNamespace(tenant_id="tenant-1", control_space_id="control-1")
authorization = MagicMock()
authorization.authorize.return_value = profile
profile = _service_profile("control-1")
authorization = FakeComponent(authorize=Recorder(profile))
class Broker:
def issue_service(self, **kwargs: object) -> SimpleNamespace:
def issue_service(self, **kwargs: object) -> KnowledgeFSIssuedProductCapability:
calls.append(kwargs)
return SimpleNamespace(
token="service-capability",
trace_id="trace-1",
expires_at=datetime(2030, 1, 1, tzinfo=UTC),
knowledge_space_id="space-1",
)
return _issued_capability(token="service-capability", operation_id="createQuery")
runtime = SimpleNamespace(
service_api_authorization=authorization,
broker=Broker(),
)
monkeypatch.setattr(
service_resources,
"validate_and_get_api_token",
MagicMock(return_value=SimpleNamespace(id="token-1", tenant_id="tenant-1")),
)
@dataclass(frozen=True)
class ServiceRuntime:
service_api_authorization: FakeComponent
broker: Broker
runtime = ServiceRuntime(service_api_authorization=authorization, broker=Broker())
monkeypatch.setattr(service_resources, "validate_and_get_api_token", Recorder(_dataset_api_token()))
apply_config_overrides(monkeypatch, SERVICE_API_URL="https://api.dify.test")
monkeypatch.setattr(service_resources, "_runtime", lambda: runtime)
app = Flask(__name__)
@@ -1694,11 +1755,13 @@ def test_service_query_admission_uses_broker(monkeypatch: pytest.MonkeyPatch) ->
assert response["trace_id"] == "trace-1"
assert response["request"]["knowledgeSpaceId"] == "space-1"
assert response["url"] == "https://api.dify.test/v1/knowledge-fs/query-stream"
authorization.authorize.assert_called_once_with(
api_token_id="token-1",
tenant_id="tenant-1",
control_space_id="control-1",
)
assert authorization.authorize.calls == [
invocation(
api_token_id="token-1",
tenant_id="tenant-1",
control_space_id="control-1",
)
]
assert calls == [{"profile": profile, "operation_id": "createQuery"}]
api/tests/unit_tests/controllers/test_knowledge_fs_resource_delegation.py
+263 -193
View File
@@ -1,13 +1,12 @@
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import UTC, datetime
from hashlib import sha256
from http import HTTPStatus
from inspect import unwrap
from types import SimpleNamespace
from typing import Literal
from unittest.mock import MagicMock
from typing import Literal, cast
import pytest
from flask import Flask
@@ -43,19 +42,30 @@ from controllers.service_api.knowledge_fs.error import (
KnowledgeFSServiceUpstreamUnavailableHTTPError,
)
from libs.exception import BaseHTTPException
from models import Account
from models.knowledge_fs import KnowledgeFSAppSpaceJoinType
from services.knowledge_fs.capability_broker import KnowledgeFSIssuedProductCapability
from services.knowledge_fs.product_dto import (
KnowledgeFSOverviewBaseStatsResponse,
KnowledgeFSOverviewQueryOutcomesResponse,
KnowledgeFSPublicFailureResponse,
KnowledgeFSQueryCreatePayload,
KnowledgeFSSourceResponse,
KnowledgeFSSourceSyncPolicyResponse,
KnowledgeFSSourceUpdatePayload,
KnowledgeFSSourceWorkflowResponse,
KnowledgeFSStreamCapabilityPayload,
)
from services.knowledge_fs.product_remote import KnowledgeFSOperationUnavailableError
from services.knowledge_fs.runtime import KnowledgeFSRuntime
from tests.unit_tests.config_override import apply_config_overrides
from tests.unit_tests.controllers.knowledge_fs_controller_fakes import (
FakeComponent,
FakeRuntime,
Recorder,
invocation,
raising,
)
_RAW_RESULT = object()
@@ -79,6 +89,52 @@ def source_response(**overrides: object) -> KnowledgeFSSourceResponse:
)
def sync_policy_response(revision: int) -> KnowledgeFSSourceSyncPolicyResponse:
return KnowledgeFSSourceSyncPolicyResponse.model_validate(
{
"created_at": datetime(2026, 1, 1, tzinfo=UTC),
"enabled": True,
"expected_source_version": 1,
"id": "policy-1",
"knowledge_space_id": "space-1",
"mode": "interval",
"revision": revision,
"source_id": "source-1",
"updated_at": datetime(2026, 1, 1, tzinfo=UTC),
}
)
def issued_capability() -> KnowledgeFSIssuedProductCapability:
return KnowledgeFSIssuedProductCapability(
token="capability-token",
expires_at=datetime(2026, 7, 21, tzinfo=UTC),
operation_id="createQuery",
knowledge_space_id="knowledge-space-1",
knowledge_space_revision=1,
trace_id="trace-1",
)
def _account(account_id: str) -> Account:
account = Account(name="Test User", email="user@example.com")
account.id = account_id
return account
@dataclass(frozen=True)
class _SettingsUpdateResult:
"""The only field the settings route reads before dumping the facade result."""
migration: None = None
@dataclass(frozen=True)
class _MembersPayload:
members: tuple[str, ...]
query_images: list[str] = field(default_factory=list)
def _invoke[R](resource_type: type[Resource], handler: Callable[..., R], *args: object) -> R:
return unwrap(handler)(resource_type(), *args)
@@ -674,23 +730,18 @@ def test_console_resources_delegate_one_tenant_scoped_product_operation(
payload = (
KnowledgeFSSourceUpdatePayload(name="Renamed")
if class_name.__name__ == "KnowledgeFSSpaceSourceApi" and method_name.__name__ == "patch"
else SimpleNamespace(members=("member-1",), query_images=[])
else _MembersPayload(members=("member-1",))
)
runtime = SimpleNamespace(
application=MagicMock(),
control_plane=MagicMock(),
app_bindings=MagicMock(),
facade=MagicMock(),
delegate = Recorder(_RAW_RESULT)
components = {name: FakeComponent() for name in ("application", "control_plane", "app_bindings", "facade")}
components[component_name] = FakeComponent(**{delegate_name: delegate})
runtime = FakeRuntime(
application=components["application"],
control_plane=components["control_plane"],
app_bindings=components["app_bindings"],
facade=components["facade"],
)
component = {
"application": runtime.application,
"control_plane": runtime.control_plane,
"app_bindings": runtime.app_bindings,
"facade": runtime.facade,
}[component_name]
delegate = MagicMock(return_value=_RAW_RESULT)
component.configure_mock(**{delegate_name: delegate})
dump_response = MagicMock(side_effect=lambda schema, raw: (schema.__name__, raw))
dump_response = Recorder(effect=lambda schema, raw: (schema.__name__, raw))
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_console_services", lambda: runtime)
monkeypatch.setattr(console_resources, "_payload", lambda _: payload)
@@ -702,8 +753,8 @@ def test_console_resources_delegate_one_tenant_scoped_product_operation(
with app.test_request_context("/", method="POST"):
result = _invoke(class_name, method_name, *route_args)
delegate.assert_called_once()
call_fields = delegate.call_args.kwargs
assert len(delegate.calls) == 1
call_fields = delegate.last_call.kwargs
assert call_fields["tenant_id"] == "tenant-1"
assert call_fields.get("account_id", call_fields.get("actor_account_id")) == "account-1"
for field_name, expected in expected_fields.items():
@@ -717,16 +768,15 @@ def test_console_resources_delegate_one_tenant_scoped_product_operation(
assert call_fields["members"] == ("member-1",)
if "idempotency_key" in call_fields:
assert call_fields["idempotency_key"] == "idempotency-1"
if dump_response.called:
assert dump_response.call_args.args[1] is _RAW_RESULT
if dump_response.calls:
assert dump_response.last_call.args[1] is _RAW_RESULT
else:
assert result == ("", HTTPStatus.NO_CONTENT)
def test_console_document_reference_delegates_exact_asset_identity(monkeypatch: pytest.MonkeyPatch) -> None:
facade = MagicMock()
facade.resolve_document_reference.return_value = _RAW_RESULT
runtime = SimpleNamespace(facade=facade)
facade = FakeComponent(resolve_document_reference=Recorder(_RAW_RESULT))
runtime = FakeRuntime(facade=facade)
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_console_services", lambda: runtime)
monkeypatch.setattr(console_resources, "dump_response", lambda _schema, raw: raw)
@@ -740,19 +790,20 @@ def test_console_document_reference_delegates_exact_asset_identity(monkeypatch:
)
assert result is _RAW_RESULT
facade.resolve_document_reference.assert_called_once_with(
tenant_id="tenant-1",
account_id="account-1",
control_space_id="space-1",
document_asset_id="asset-1",
document_asset_version=7,
)
assert facade.resolve_document_reference.calls == [
invocation(
tenant_id="tenant-1",
account_id="account-1",
control_space_id="space-1",
document_asset_id="asset-1",
document_asset_version=7,
)
]
def test_console_space_list_preserves_repeated_creator_filters(monkeypatch: pytest.MonkeyPatch) -> None:
application = MagicMock()
application.list_spaces.return_value = _RAW_RESULT
runtime = SimpleNamespace(application=application)
application = FakeComponent(list_spaces=Recorder(_RAW_RESULT))
runtime = FakeRuntime(application=application)
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_console_services", lambda: runtime)
monkeypatch.setattr(console_resources, "dump_response", lambda _schema, raw: raw)
@@ -763,24 +814,25 @@ def test_console_space_list_preserves_repeated_creator_filters(monkeypatch: pyte
):
result = _invoke(console_resources.KnowledgeFSSpacesApi, console_resources.KnowledgeFSSpacesApi.get)
application.list_spaces.assert_called_once_with(
tenant_id="tenant-1",
account_id="account-1",
page=2,
limit=10,
creator_ids=["creator-1", "creator-2"],
tag_ids=["tag-1", "tag-2"],
query="Support",
)
assert application.list_spaces.calls == [
invocation(
tenant_id="tenant-1",
account_id="account-1",
page=2,
limit=10,
creator_ids=["creator-1", "creator-2"],
tag_ids=["tag-1", "tag-2"],
query="Support",
)
]
assert result is _RAW_RESULT
def test_console_source_update_does_not_start_sync_before_resource_selection(
monkeypatch: pytest.MonkeyPatch,
) -> None:
facade = MagicMock()
facade.update_source.return_value = _RAW_RESULT
runtime = SimpleNamespace(facade=facade)
facade = FakeComponent(update_source=Recorder(_RAW_RESULT))
runtime = FakeRuntime(facade=facade)
payload = KnowledgeFSSourceUpdatePayload.model_validate(
{
"expectedVersion": 3,
@@ -802,7 +854,7 @@ def test_console_source_update_does_not_start_sync_before_resource_selection(
"source-1",
)
forwarded = facade.update_source.call_args.kwargs["payload"]
forwarded = facade.update_source.last_call.kwargs["payload"]
assert forwarded is not payload
assert forwarded.status == "disabled"
assert forwarded.sync_after_update is None
@@ -844,8 +896,6 @@ def test_console_source_update_commits_complete_edit_selection(
selection: dict[str, object],
expected_import_kind: str,
) -> None:
facade = MagicMock()
facade.update_source.return_value = SimpleNamespace(version=4)
selection_kind = selection["kind"]
assert isinstance(selection_kind, str)
provider_kind = {"online_document": "online-document", "online_drive": "online-drive"}.get(selection_kind)
@@ -854,10 +904,28 @@ def test_console_source_update_commits_complete_edit_selection(
type="web" if selection_kind == "website_crawl" else "connector",
uri="https://old.example.com",
)
updated = SimpleNamespace(sync_workflow=None)
facade.get_source.side_effect = [original, updated]
workflow = SimpleNamespace(id="workflow-1")
runtime = SimpleNamespace(facade=facade)
updated = source_response(sync_workflow=None)
facade = FakeComponent(
get_source=Recorder(results=[original, updated]),
update_source=Recorder(source_response(version=4)),
)
workflow = KnowledgeFSSourceWorkflowResponse.model_validate(
{
"checkpoint": "queued",
"created_at": datetime(2026, 1, 1, tzinfo=UTC),
"execution_attempts": 0,
"id": "workflow-1",
"knowledge_space_id": "space-1",
"kind": "source-import",
"max_execution_attempts": 3,
"progress_completed": 0,
"progress_failed": 0,
"progress_skipped": 0,
"state": "queued",
"updated_at": datetime(2026, 1, 1, tzinfo=UTC),
}
)
runtime = FakeRuntime(facade=facade)
payload = KnowledgeFSSourceUpdatePayload.model_validate(
{
"expectedVersion": 3,
@@ -866,7 +934,7 @@ def test_console_source_update_commits_complete_edit_selection(
"syncPolicy": {"enabled": True, "mode": "interval"},
}
)
commit = MagicMock(return_value=workflow)
commit = Recorder(workflow)
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_console_services", lambda: runtime)
monkeypatch.setattr(console_resources, "_payload", lambda _: payload)
@@ -882,12 +950,12 @@ def test_console_source_update_commits_complete_edit_selection(
"source-1",
)
facade.update_source.assert_not_called()
source_payload = commit.call_args.kwargs["source_update"]
assert facade.update_source.calls == []
source_payload = commit.last_call.kwargs["source_update"]
assert source_payload.status == "disabled"
import_payload = commit.call_args.kwargs["payload"]
import_payload = commit.last_call.kwargs["payload"]
assert import_payload.kind == expected_import_kind
assert commit.call_args.kwargs["idempotency_key"] == "source-edit:source-1:3"
assert commit.last_call.kwargs["idempotency_key"] == "source-edit:source-1:3"
assert updated.sync_workflow is workflow
assert result is updated
@@ -895,7 +963,6 @@ def test_console_source_update_commits_complete_edit_selection(
def test_console_source_update_does_not_import_an_unchanged_complete_selection(
monkeypatch: pytest.MonkeyPatch,
) -> None:
facade = MagicMock()
source = source_response(
metadata={"crawled": {"https://example.com/a": {}}},
sync_policy=None,
@@ -903,17 +970,20 @@ def test_console_source_update_does_not_import_an_unchanged_complete_selection(
uri="https://example.com",
version=3,
)
facade.get_source.return_value = source
facade.get_source_sync_policy.return_value = SimpleNamespace(revision=2)
facade.update_source.return_value = source
runtime = SimpleNamespace(facade=facade)
facade = FakeComponent(
get_source=Recorder(source),
get_source_sync_policy=Recorder(sync_policy_response(revision=2)),
update_source=Recorder(source),
update_source_sync_policy=Recorder(),
)
runtime = FakeRuntime(facade=facade)
payload = KnowledgeFSSourceUpdatePayload.model_validate(
{
"selection": {"kind": "website_crawl", "sourceUrls": ["https://example.com/a"]},
"syncPolicy": {"enabled": True, "mode": "interval"},
}
)
commit = MagicMock()
commit = Recorder()
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_console_services", lambda: runtime)
monkeypatch.setattr(console_resources, "_payload", lambda _: payload)
@@ -929,10 +999,10 @@ def test_console_source_update_does_not_import_an_unchanged_complete_selection(
"source-1",
)
facade.update_source.assert_called_once()
commit.assert_not_called()
facade.update_source_sync_policy.assert_not_called()
command = facade.update_source.call_args.kwargs["payload"]
assert len(facade.update_source.calls) == 1
assert commit.calls == []
assert facade.update_source_sync_policy.calls == []
command = facade.update_source.last_call.kwargs["payload"]
assert command.expected_version == 3
assert command.expected_policy_revision == 2
assert command.sync_policy.mode == "interval"
@@ -1062,18 +1132,20 @@ def test_source_edit_reimports_an_unchanged_selection_when_source_is_in_error()
def test_console_source_update_applies_policy_without_creating_import_work(
monkeypatch: pytest.MonkeyPatch,
) -> None:
facade = MagicMock()
source = source_response(version=5, sync_policy=None)
facade.get_source.return_value = source
facade.get_source_sync_policy.return_value = SimpleNamespace(revision=2)
policy = SimpleNamespace(revision=3)
policy = sync_policy_response(revision=3)
source.sync_policy = policy
facade.update_source.return_value = source
runtime = SimpleNamespace(facade=facade)
facade = FakeComponent(
get_source=Recorder(source),
get_source_sync_policy=Recorder(sync_policy_response(revision=2)),
update_source=Recorder(source),
update_source_sync_policy=Recorder(),
)
runtime = FakeRuntime(facade=facade)
payload = KnowledgeFSSourceUpdatePayload.model_validate(
{"syncPolicy": {"customIntervalSeconds": 7200, "enabled": True, "mode": "custom"}}
)
commit = MagicMock()
commit = Recorder()
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_console_services", lambda: runtime)
monkeypatch.setattr(console_resources, "_payload", lambda _: payload)
@@ -1089,10 +1161,10 @@ def test_console_source_update_applies_policy_without_creating_import_work(
"source-1",
)
facade.update_source.assert_called_once()
commit.assert_not_called()
facade.update_source_sync_policy.assert_not_called()
policy_payload = facade.update_source.call_args.kwargs["payload"]
assert len(facade.update_source.calls) == 1
assert commit.calls == []
assert facade.update_source_sync_policy.calls == []
policy_payload = facade.update_source.last_call.kwargs["payload"]
assert policy_payload.expected_policy_revision == 2
assert policy_payload.expected_version == 5
assert policy_payload.sync_policy.custom_interval_seconds == 7200
@@ -1101,9 +1173,8 @@ def test_console_source_update_applies_policy_without_creating_import_work(
def test_console_metadata_delete_forwards_row_version_cas(monkeypatch: pytest.MonkeyPatch) -> None:
facade = MagicMock()
facade.delete_metadata_field.return_value = _RAW_RESULT
runtime = SimpleNamespace(facade=facade)
facade = FakeComponent(delete_metadata_field=Recorder(_RAW_RESULT))
runtime = FakeRuntime(facade=facade)
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_console_services", lambda: runtime)
monkeypatch.setattr(console_resources, "dump_response", lambda _schema, raw: raw)
@@ -1117,13 +1188,15 @@ def test_console_metadata_delete_forwards_row_version_cas(monkeypatch: pytest.Mo
"field-1",
)
facade.delete_metadata_field.assert_called_once_with(
tenant_id="tenant-1",
account_id="account-1",
control_space_id="space-1",
field_id="field-1",
expected_row_version=7,
)
assert facade.delete_metadata_field.calls == [
invocation(
tenant_id="tenant-1",
account_id="account-1",
control_space_id="space-1",
field_id="field-1",
expected_row_version=7,
)
]
assert result is _RAW_RESULT
@@ -1188,14 +1261,14 @@ def test_console_overview_stats_composes_kfs_metrics_with_dify_app_bindings(
"window": "7d",
}
)
runtime = SimpleNamespace(
facade=SimpleNamespace(
get_overview_query_outcomes=MagicMock(return_value=outcomes),
get_overview_stats=MagicMock(return_value=stats),
runtime = FakeRuntime(
facade=FakeComponent(
get_overview_query_outcomes=Recorder(outcomes),
get_overview_stats=Recorder(stats),
),
app_bindings=SimpleNamespace(count_active=MagicMock(return_value=7)),
app_bindings=FakeComponent(count_active=Recorder(7)),
)
dump_response = MagicMock(side_effect=lambda _schema, raw: raw)
dump_response = Recorder(effect=lambda _schema, raw: raw)
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(console_resources, "_console_services", lambda: runtime)
monkeypatch.setattr(console_resources, "dump_response", dump_response)
@@ -1208,22 +1281,28 @@ def test_console_overview_stats_composes_kfs_metrics_with_dify_app_bindings(
"control-space-1",
)
runtime.facade.get_overview_stats.assert_called_once_with(
tenant_id="tenant-1",
account_id="account-1",
control_space_id="control-space-1",
)
runtime.facade.get_overview_query_outcomes.assert_called_once_with(
tenant_id="tenant-1",
account_id="account-1",
control_space_id="control-space-1",
window="7d",
)
runtime.app_bindings.count_active.assert_called_once_with(
tenant_id="tenant-1",
actor_account_id="account-1",
control_space_id="control-space-1",
)
assert runtime.facade.get_overview_stats.calls == [
invocation(
tenant_id="tenant-1",
account_id="account-1",
control_space_id="control-space-1",
)
]
assert runtime.facade.get_overview_query_outcomes.calls == [
invocation(
tenant_id="tenant-1",
account_id="account-1",
control_space_id="control-space-1",
window="7d",
)
]
assert runtime.app_bindings.count_active.calls == [
invocation(
tenant_id="tenant-1",
actor_account_id="account-1",
control_space_id="control-space-1",
)
]
assert result.documents == 13
assert result.linked_apps == 7
assert result.queries.value == 120
@@ -1233,7 +1312,7 @@ def test_console_overview_stats_composes_kfs_metrics_with_dify_app_bindings(
assert result.answer_rate.change_percentage_points == pytest.approx(-5)
assert result.freshness_seconds == 3600
assert result.generated_at == outcomes.generated_at
assert dump_response.call_args.args[1] is result
assert dump_response.last_call.args[1] is result
_SERVICE_DELEGATION_CASES = (
@@ -1476,8 +1555,8 @@ def test_service_resources_bind_route_identifiers_to_one_declared_operation(
expected_fields: dict[str, object],
) -> None:
payload = object()
execute = MagicMock(return_value=_RAW_RESULT)
dump_response = MagicMock(side_effect=lambda schema, raw: (schema.__name__, raw))
execute = Recorder(_RAW_RESULT)
dump_response = Recorder(effect=lambda schema, raw: (schema.__name__, raw))
monkeypatch.setattr(service_resources, "_execute_service_operation", execute)
monkeypatch.setattr(service_resources, "_payload", lambda _: payload)
monkeypatch.setattr(service_resources, "_idempotency_key", lambda: "idempotency-1")
@@ -1500,8 +1579,8 @@ def test_service_resources_bind_route_identifiers_to_one_declared_operation(
with app.test_request_context("/", method="POST", headers={"Idempotency-Key": "idempotency-1"}):
_invoke(class_name, method_name, *route_args)
execute.assert_called_once()
call_fields = execute.call_args.kwargs
assert len(execute.calls) == 1
call_fields = execute.last_call.kwargs
assert call_fields["control_space_id"] == "space-1"
assert call_fields["operation_id"] == operation_id
for field_name, expected in expected_fields.items():
@@ -1512,7 +1591,7 @@ def test_service_resources_bind_route_identifiers_to_one_declared_operation(
assert call_fields["headers"] == (("Idempotency-Key", "idempotency-1"),)
if "query" in call_fields:
assert call_fields["query"] == (("normalized", "true"),)
assert dump_response.call_args.args[1] is _RAW_RESULT
assert dump_response.last_call.args[1] is _RAW_RESULT
@pytest.mark.parametrize(
@@ -1563,13 +1642,13 @@ def test_service_credential_routes_validate_profile_before_facade_delegation(
operation_id: str,
cursor: str | None,
) -> None:
facade = SimpleNamespace(
execute_service=MagicMock(return_value=_RAW_RESULT),
update_service_settings=MagicMock(return_value=SimpleNamespace(migration=None)),
facade = FakeComponent(
execute_service=Recorder(_RAW_RESULT),
update_service_settings=Recorder(_SettingsUpdateResult()),
)
runtime = SimpleNamespace(facade=facade)
runtime = FakeRuntime(facade=facade)
profile = object()
validate_profile = MagicMock(return_value=profile)
validate_profile = Recorder(profile)
monkeypatch.setattr(service_resources, "_runtime", lambda: runtime)
monkeypatch.setattr(service_resources, "_profile", validate_profile)
monkeypatch.setattr(service_resources, "_payload", lambda _: object())
@@ -1580,16 +1659,16 @@ def test_service_credential_routes_validate_profile_before_facade_delegation(
with app.test_request_context("/", method="POST", query_string=query_string):
_invoke(class_name, method_name, "space-1")
validate_profile.assert_called_once_with(runtime, operation_id=operation_id, control_space_id="space-1")
assert validate_profile.calls == [invocation(runtime, operation_id=operation_id, control_space_id="space-1")]
if operation_id == "updateSettings":
assert facade.update_service_settings.call_args.kwargs["profile"] is profile
facade.execute_service.assert_not_called()
assert facade.update_service_settings.last_call.kwargs["profile"] is profile
assert facade.execute_service.calls == []
return
assert facade.execute_service.call_args.kwargs["profile"] is profile
assert facade.execute_service.call_args.kwargs["operation_id"] == operation_id
if "query" in facade.execute_service.call_args.kwargs:
assert facade.execute_service.last_call.kwargs["profile"] is profile
assert facade.execute_service.last_call.kwargs["operation_id"] == operation_id
if "query" in facade.execute_service.last_call.kwargs:
expected_query = (("limit", "50"),) + ((("cursor", cursor),) if cursor else ())
assert facade.execute_service.call_args.kwargs["query"] == expected_query
assert facade.execute_service.last_call.kwargs["query"] == expected_query
@pytest.mark.parametrize(
@@ -1604,13 +1683,7 @@ def test_deprecated_buffered_routes_fail_closed(class_name: type[console_resourc
def test_console_stream_capabilities_bind_the_authorized_resource(monkeypatch: pytest.MonkeyPatch) -> None:
issued = SimpleNamespace(
token="capability-token",
trace_id="trace-1",
expires_at=datetime(2026, 7, 21, tzinfo=UTC),
knowledge_space_id="knowledge-space-1",
)
broker = SimpleNamespace(issue_interactive=MagicMock(return_value=issued))
broker = FakeComponent(issue_interactive=Recorder(issued_capability()))
payloads = iter(
[
KnowledgeFSQueryCreatePayload(query="question", mode="deep"),
@@ -1622,7 +1695,7 @@ def test_console_stream_capabilities_bind_the_authorized_resource(monkeypatch: p
monkeypatch.setattr(
console_resources,
"_console_services",
lambda: SimpleNamespace(broker=broker),
lambda: FakeRuntime(broker=broker),
)
monkeypatch.setattr(console_resources, "_payload", lambda _: next(payloads))
monkeypatch.setattr(console_resources, "dump_response", lambda _, response: response)
@@ -1652,26 +1725,20 @@ def test_console_stream_capabilities_bind_the_authorized_resource(monkeypatch: p
"?knowledgeSpaceId=knowledge-space-1"
)
assert legacy_query.url == "https://dify.example/console/api/knowledge-fs/query-stream"
assert broker.issue_interactive.call_args_list[0].kwargs["operation_id"] == "createQuery"
assert broker.issue_interactive.call_args_list[1].kwargs == {
assert broker.issue_interactive.calls[0].kwargs["operation_id"] == "createQuery"
assert broker.issue_interactive.calls[1].kwargs == {
"tenant_id": "tenant-1",
"account_id": "account-1",
"control_space_id": "space-2",
"operation_id": "streamResearchTask",
"resource_id": "task/1",
}
assert broker.issue_interactive.call_args_list[2].kwargs["operation_id"] == "createQuery"
assert broker.issue_interactive.calls[2].kwargs["operation_id"] == "createQuery"
def test_service_query_admission_binds_profile_space_and_payload(monkeypatch: pytest.MonkeyPatch) -> None:
issued = SimpleNamespace(
token="capability-token",
trace_id="trace-1",
expires_at=datetime(2026, 7, 21, tzinfo=UTC),
knowledge_space_id="knowledge-space-1",
)
broker = SimpleNamespace(issue_service=MagicMock(return_value=issued))
runtime = SimpleNamespace(broker=broker)
broker = FakeComponent(issue_service=Recorder(issued_capability()))
runtime = FakeRuntime(broker=broker)
profile = object()
apply_config_overrides(monkeypatch, SERVICE_API_URL="https://api.dify.example")
monkeypatch.setattr(service_resources, "_runtime", lambda: runtime)
@@ -1687,7 +1754,7 @@ def test_service_query_admission_binds_profile_space_and_payload(monkeypatch: py
"space-1",
)
broker.issue_service.assert_called_once_with(profile=profile, operation_id="createQuery")
assert broker.issue_service.calls == [invocation(profile=profile, operation_id="createQuery")]
assert response.request.knowledge_space_id == "knowledge-space-1"
assert response.request.query == "question"
assert response.url == "https://api.dify.example/v1/knowledge-fs/query-stream"
@@ -1696,21 +1763,21 @@ def test_service_query_admission_binds_profile_space_and_payload(monkeypatch: py
def test_console_resource_helpers_validate_feature_payload_headers_and_query_pairs(
monkeypatch: pytest.MonkeyPatch,
) -> None:
runtime = object()
runtime = FakeRuntime()
session_maker = object()
get_runtime = MagicMock(return_value=runtime)
get_runtime = Recorder(runtime)
apply_config_overrides(monkeypatch, KNOWLEDGE_FS_ENABLED=True)
monkeypatch.setattr(console_resources.session_factory, "get_session_maker", lambda: session_maker)
monkeypatch.setattr(console_resources, "get_knowledge_fs_runtime", get_runtime)
monkeypatch.setattr(
console_resources,
"current_account_with_tenant",
lambda: (SimpleNamespace(id="account-1"), "tenant-1"),
lambda: (_account("account-1"), "tenant-1"),
)
app = Flask(__name__)
assert console_resources._console_services() is runtime
get_runtime.assert_called_once_with(session_maker)
assert get_runtime.calls == [invocation(session_maker)]
assert console_resources._actor() == ("account-1", "tenant-1")
query = KnowledgeFSQueryCreatePayload(query="question", mode="fast", active_document_ids=["doc-1"])
assert console_resources._query_pairs(query) == (
@@ -1736,16 +1803,16 @@ def test_console_resource_helpers_validate_feature_payload_headers_and_query_pai
def test_service_resource_helpers_validate_feature_bearer_headers_and_boolean_queries(
monkeypatch: pytest.MonkeyPatch,
) -> None:
runtime = object()
runtime = FakeRuntime()
session_maker = object()
get_runtime = MagicMock(return_value=runtime)
get_runtime = Recorder(runtime)
apply_config_overrides(monkeypatch, KNOWLEDGE_FS_ENABLED=True)
monkeypatch.setattr(service_resources.session_factory, "get_session_maker", lambda: session_maker)
monkeypatch.setattr(service_resources, "get_knowledge_fs_runtime", get_runtime)
app = Flask(__name__)
assert service_resources._runtime() is runtime
get_runtime.assert_called_once_with(session_maker)
assert get_runtime.calls == [invocation(session_maker)]
class Query(BaseModel):
enabled: bool = True
@@ -1786,12 +1853,12 @@ def test_console_request_rejections_preserve_http_status_contracts() -> None:
504: KnowledgeFSTimeoutHTTPError,
}
for status, http_error in expected.items():
reject = console_resources._knowledge_fs_errors(
MagicMock(side_effect=KnowledgeFSProductRequestRejectedError(status_code=status))
guarded = console_resources._knowledge_fs_errors(
raising(KnowledgeFSProductRequestRejectedError(status_code=status))
)
with pytest.raises(http_error) as raised:
reject()
guarded()
assert raised.value.code == status
assert raised.value.data == {
@@ -1836,13 +1903,13 @@ def test_console_unavailable_and_timeout_rejections_preserve_safe_failure_metada
failure = KnowledgeFSPublicFailureResponse.model_validate(
{**expected_failure, "message": "Private upstream diagnostic that must not reach the client"}
)
reject = console_resources._knowledge_fs_errors(
MagicMock(side_effect=KnowledgeFSProductRequestRejectedError(status_code=status, failure=failure))
guarded = console_resources._knowledge_fs_errors(
raising(KnowledgeFSProductRequestRejectedError(status_code=status, failure=failure))
)
http_error = KnowledgeFSOperationUnavailableHTTPError if status == 503 else KnowledgeFSTimeoutHTTPError
with pytest.raises(http_error) as raised:
reject()
guarded()
assert raised.value.code == status
assert raised.value.data == {
@@ -1866,12 +1933,12 @@ def test_console_request_rejection_preserves_safe_failure_metadata() -> None:
"traceId": "trace-model",
}
)
reject = console_resources._knowledge_fs_errors(
MagicMock(side_effect=KnowledgeFSProductRequestRejectedError(status_code=422, failure=failure))
guarded = console_resources._knowledge_fs_errors(
raising(KnowledgeFSProductRequestRejectedError(status_code=422, failure=failure))
)
with pytest.raises(KnowledgeFSRequestRejectedHTTPError) as raised:
reject()
guarded()
assert raised.value.data == {
"code": "knowledge_fs_request_rejected",
@@ -1940,10 +2007,10 @@ def test_console_error_adapter_maps_every_domain_boundary_to_the_stable_http_con
)
for domain_error, http_error in mappings:
fail = console_resources._knowledge_fs_errors(MagicMock(side_effect=domain_error))
guarded = console_resources._knowledge_fs_errors(raising(domain_error))
with pytest.raises(http_error):
fail()
guarded()
assert KnowledgeFSUpstreamUnavailableHTTPError.code == HTTPStatus.SERVICE_UNAVAILABLE
@@ -1973,10 +2040,10 @@ def test_service_error_adapter_maps_every_domain_boundary_to_the_stable_http_con
)
for domain_error, http_error in mappings:
fail = service_resources._service_api_errors(MagicMock(side_effect=domain_error))
guarded = service_resources._service_api_errors(raising(domain_error))
with pytest.raises(http_error):
fail()
guarded()
rejected_mappings = (
(400, KnowledgeFSServiceInvalidRequestHTTPError),
@@ -1987,11 +2054,11 @@ def test_service_error_adapter_maps_every_domain_boundary_to_the_stable_http_con
(429, KnowledgeFSServiceRateLimitHTTPError),
)
for status, http_error in rejected_mappings:
fail = service_resources._service_api_errors(
MagicMock(side_effect=KnowledgeFSProductRequestRejectedError(status_code=status))
guarded = service_resources._service_api_errors(
raising(KnowledgeFSProductRequestRejectedError(status_code=status))
)
with pytest.raises(http_error):
fail()
guarded()
def test_console_small_file_reader_rejects_malformed_and_empty_multipart_bodies() -> None:
@@ -2027,12 +2094,12 @@ def test_console_app_binding_route_rejects_unknown_caller_kind_before_revoke(
) -> None:
from services.knowledge_fs.app_binding_management import KnowledgeFSAppBindingManagementError
revoke = MagicMock()
revoke = Recorder()
monkeypatch.setattr(console_resources, "_actor", lambda: ("account-1", "tenant-1"))
monkeypatch.setattr(
console_resources,
"_console_services",
lambda: SimpleNamespace(app_bindings=SimpleNamespace(revoke=revoke)),
lambda: FakeRuntime(app_bindings=FakeComponent(revoke=revoke)),
)
with pytest.raises(KnowledgeFSAppBindingManagementError, match="caller kind"):
@@ -2044,15 +2111,16 @@ def test_console_app_binding_route_rejects_unknown_caller_kind_before_revoke(
"app-1",
)
revoke.assert_not_called()
assert revoke.calls == []
def test_service_profile_hides_unknown_operations_before_dataset_key_authorization(
monkeypatch: pytest.MonkeyPatch,
) -> None:
authorization = SimpleNamespace(authorize=MagicMock())
runtime = MagicMock(spec=KnowledgeFSRuntime, service_api_authorization=authorization)
monkeypatch.setattr(service_resources, "product_operation_action", MagicMock(side_effect=KeyError("unknown")))
authorization = FakeComponent(authorize=Recorder())
# The profile check must fail before it touches any runtime component but authorization.
runtime = cast(KnowledgeFSRuntime, FakeRuntime(service_api_authorization=authorization))
monkeypatch.setattr(service_resources, "product_operation_action", raising(KeyError("unknown")))
with pytest.raises(KnowledgeFSOperationUnavailableError, match="unknownOperation"):
service_resources._profile(
@@ -2061,7 +2129,7 @@ def test_service_profile_hides_unknown_operations_before_dataset_key_authorizati
control_space_id="space-1",
)
authorization.authorize.assert_not_called()
assert authorization.authorize.calls == []
def test_service_operation_helper_forwards_the_complete_validated_request(
@@ -2069,10 +2137,10 @@ def test_service_operation_helper_forwards_the_complete_validated_request(
) -> None:
profile = object()
payload = KnowledgeFSQueryCreatePayload(query="question")
facade = SimpleNamespace(execute_service=MagicMock(return_value={"id": "result-1"}))
runtime = SimpleNamespace(facade=facade)
facade = FakeComponent(execute_service=Recorder({"id": "result-1"}))
runtime = FakeRuntime(facade=facade)
monkeypatch.setattr(service_resources, "_runtime", lambda: runtime)
monkeypatch.setattr(service_resources, "_profile", MagicMock(return_value=profile))
monkeypatch.setattr(service_resources, "_profile", Recorder(profile))
result = service_resources._execute_service_operation(
control_space_id="space-1",
@@ -2086,13 +2154,15 @@ def test_service_operation_helper_forwards_the_complete_validated_request(
)
assert result == {"id": "result-1"}
facade.execute_service.assert_called_once_with(
profile=profile,
operation_id="createResearchTask",
payload=payload,
query=(("cursor", "next"),),
bind_space_in_body=True,
resource_id="task-1",
path_parameters=(("taskId", "task-1"),),
headers=(("Idempotency-Key", "request-key"),),
)
assert facade.execute_service.calls == [
invocation(
profile=profile,
operation_id="createResearchTask",
payload=payload,
query=(("cursor", "next"),),
bind_space_in_body=True,
resource_id="task-1",
path_parameters=(("taskId", "task-1"),),
headers=(("Idempotency-Key", "request-key"),),
)
]
api/tests/unit_tests/controllers/test_knowledge_fs_service_contracts.py
+38 -28
View File
@@ -1,8 +1,5 @@
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import MagicMock
import pytest
from flask import Flask
from pydantic import ValidationError
@@ -24,7 +21,9 @@ from services.knowledge_fs.product_dto import (
KnowledgeFSServiceSourceCreatePayload,
KnowledgeFSServiceSourceUpdatePayload,
)
from services.knowledge_fs.service_api_authorization import KnowledgeFSServiceApiProfile
from tests.unit_tests.config_override import apply_config_overrides
from tests.unit_tests.controllers.knowledge_fs_controller_fakes import FakeComponent, FakeRuntime, Recorder, invocation
_NOW = "2026-09-17T00:00:00Z"
_FAILURE = {
@@ -68,10 +67,21 @@ def _workflow() -> dict[str, object]:
@pytest.fixture
def runtime(monkeypatch: pytest.MonkeyPatch):
facade = SimpleNamespace(execute_service=MagicMock())
value = SimpleNamespace(facade=facade)
value = FakeRuntime(facade=FakeComponent(execute_service=Recorder()))
profile = KnowledgeFSServiceApiProfile(
tenant_id="tenant-1",
control_space_id="control-1",
api_token_id="token-1",
principal_id="principal-1",
knowledge_space_id="space-1",
knowledge_space_revision=1,
membership_epoch=1,
space_acl_epoch=1,
external_access_epoch=1,
content_policy_revision=1,
)
monkeypatch.setattr(resources, "_runtime", lambda: value)
monkeypatch.setattr(resources, "_profile", lambda *_args, **_kwargs: SimpleNamespace(tenant_id="tenant-1"))
monkeypatch.setattr(resources, "_profile", lambda *_args, **_kwargs: profile)
return value
@@ -183,7 +193,7 @@ def test_logical_read_exposes_current_cas_and_pagination(runtime) -> None:
result = resources.KnowledgeFSServiceLogicalDocumentsApi().get("control-1")
assert result["data"][0]["row_version"] == 7
assert result["data"][0]["id"] == "doc-1"
assert runtime.facade.execute_service.call_args.kwargs["query"] == (("limit", "17"), ("cursor", "prev"))
assert runtime.facade.execute_service.last_call.kwargs["query"] == (("limit", "17"), ("cursor", "prev"))
def test_deletion_links_and_partial_batch_results_stay_on_public_origin(
@@ -238,7 +248,7 @@ def test_legacy_source_import_adapts_coordinates_and_admits_durable_workflow(run
result, status = resource.post("control-1", "source-1")
assert status == 202
assert result["id"] == "run-1"
kwargs = runtime.facade.execute_service.call_args.kwargs
kwargs = runtime.facade.execute_service.last_call.kwargs
assert kwargs["operation_id"] == "importSourceWorkflow"
assert kwargs["headers"] == (("Idempotency-Key", "import-once"),)
assert kwargs["payload"].model_dump(mode="json", by_alias=True)["items"][0]["providerItemId"] == coordinate
@@ -249,12 +259,12 @@ def test_malformed_upstream_is_502_and_invalid_input_is_400_without_input_echo(r
with Flask(__name__).test_request_context("/"):
with pytest.raises(KnowledgeFSServiceUpstreamUnavailableHTTPError):
resources.KnowledgeFSServiceLogicalDocumentApi().get("control-1", "doc-1")
runtime.facade.execute_service.reset_mock()
runtime.facade.execute_service.calls.clear()
with Flask(__name__).test_request_context("/?limit=private-invalid-value"):
with pytest.raises(KnowledgeFSServiceInvalidRequestHTTPError) as caught:
resources.KnowledgeFSServiceLogicalDocumentsApi().get("control-1")
assert "private-invalid-value" not in str(caught.value.data)
runtime.facade.execute_service.assert_not_called()
assert runtime.facade.execute_service.calls == []
def test_resume_binds_original_task_and_processing_history_preserves_superseded(runtime) -> None:
@@ -276,7 +286,7 @@ def test_resume_binds_original_task_and_processing_history_preserves_superseded(
with Flask(__name__).test_request_context("/"):
response = resources.KnowledgeFSServiceDocumentProcessingTasksApi().get("control-1", "doc-1")
assert response["data"][0]["state"] == "superseded"
assert runtime.facade.execute_service.call_args.kwargs["resource_id"] == "doc-1"
assert runtime.facade.execute_service.last_call.kwargs["resource_id"] == "doc-1"
runtime.facade.execute_service.return_value = {
"id": "task-1",
"knowledgeSpaceId": "space-1",
@@ -290,8 +300,8 @@ def test_resume_binds_original_task_and_processing_history_preserves_superseded(
with Flask(__name__).test_request_context("/", method="POST"):
response = resources.KnowledgeFSServiceResearchTaskResumeApi().post("control-1", "task-1")
assert response["id"] == "task-1"
assert runtime.facade.execute_service.call_args.kwargs["operation_id"] == "resumeResearchTask"
assert runtime.facade.execute_service.call_args.kwargs["resource_id"] == "task-1"
assert runtime.facade.execute_service.last_call.kwargs["operation_id"] == "resumeResearchTask"
assert runtime.facade.execute_service.last_call.kwargs["resource_id"] == "task-1"
@pytest.mark.parametrize("status", [503, 504])
@@ -315,21 +325,21 @@ def test_service_preserves_temporary_failure_status_and_safe_violations(status:
def test_service_image_upload_uses_authenticated_profile_and_typed_response(monkeypatch: pytest.MonkeyPatch) -> None:
import io
upload = MagicMock(return_value={"upload_file_id": "image-1", "byte_size": 8, "mime_type": "image/png"})
upload = Recorder({"upload_file_id": "image-1", "byte_size": 8, "mime_type": "image/png"})
monkeypatch.setattr(resources, "upload_service_query_image", upload)
with Flask(__name__).test_request_context("/", method="POST", data={"file": (io.BytesIO(b"fakepng"), "image.png")}):
response, status = resources.KnowledgeFSServiceQueryImagesApi().post("control-1")
assert status == 201
assert response == {"upload_file_id": "image-1", "byte_size": 8, "mime_type": "image/png"}
assert upload.call_args.kwargs["profile"].tenant_id == "tenant-1"
assert upload.call_args.kwargs["file"].filename == "image.png"
assert upload.last_call.kwargs["profile"].tenant_id == "tenant-1"
assert upload.last_call.kwargs["file"].filename == "image.png"
def test_image_admission_checks_ownership_before_issuing_a_capability(runtime, monkeypatch: pytest.MonkeyPatch) -> None:
from core.knowledge_fs.errors import KnowledgeFSProductRequestRejectedError
runtime.broker = MagicMock()
validator = MagicMock(side_effect=KnowledgeFSProductRequestRejectedError(status_code=422))
runtime.broker = FakeComponent(issue_service=Recorder())
validator = Recorder(raises=KnowledgeFSProductRequestRejectedError(status_code=422))
monkeypatch.setattr(resources, "validate_service_query_image_references", validator)
with Flask(__name__).test_request_context(
"/",
@@ -339,8 +349,8 @@ def test_image_admission_checks_ownership_before_issuing_a_capability(runtime, m
with pytest.raises(Exception) as error:
resources.KnowledgeFSServiceQueryAdmissionApi().post("control-1")
assert error.value.code == 422
runtime.broker.issue_service.assert_not_called()
validator.assert_called_once()
assert runtime.broker.issue_service.calls == []
assert len(validator.calls) == 1
def test_atomic_crawl_edit_rejects_inline_document_content() -> None:
@@ -411,15 +421,15 @@ def test_recovery_reads_keep_the_authorized_space_and_child_resource_binding(
path_parameters,
) -> None:
profile = object()
authorize = MagicMock(return_value=profile)
authorize = Recorder(profile)
monkeypatch.setattr(resources, "_profile", authorize)
monkeypatch.setattr(resources, "_dump_response", lambda _schema, raw: raw)
upstream = {"id": resource_id}
runtime.facade.execute_service.return_value = upstream
with Flask(__name__).test_request_context("/", method="GET"):
assert controller().get(*arguments) is upstream
authorize.assert_called_once_with(runtime, operation_id=operation, control_space_id="control-1")
call = runtime.facade.execute_service.call_args.kwargs
assert authorize.calls == [invocation(runtime, operation_id=operation, control_space_id="control-1")]
call = runtime.facade.execute_service.last_call.kwargs
assert call["profile"] is profile
assert call["operation_id"] == operation
assert call["resource_id"] == resource_id
@@ -432,7 +442,7 @@ def test_source_workflow_history_keeps_source_filter_and_bounded_pagination(
monkeypatch.setattr(resources, "_dump_response", lambda _schema, raw: raw)
with Flask(__name__).test_request_context("/?source_id=source-1&cursor=next-page&limit=25"):
resources.KnowledgeFSServiceSourceWorkflowsApi().get("control-1")
call = runtime.facade.execute_service.call_args.kwargs
call = runtime.facade.execute_service.last_call.kwargs
assert call["operation_id"] == "listSourceWorkflows"
assert dict(call["query"]) == {"sourceId": "source-1", "cursor": "next-page", "limit": "25"}
@@ -448,7 +458,7 @@ def test_preview_selection_is_a_durable_idempotent_import_admission(runtime) ->
result, status = resources.KnowledgeFSServiceCrawlPreviewSelectionApi().post("control-1", "run-1")
assert status == 202
assert result["id"] == "run-1"
call = runtime.facade.execute_service.call_args.kwargs
call = runtime.facade.execute_service.last_call.kwargs
assert call["operation_id"] == "selectCrawlPreviewPages"
assert call["resource_id"] == "run-1"
assert call["headers"] == (("Idempotency-Key", "selection-request-1"),)
@@ -600,7 +610,7 @@ def test_durable_source_contracts_bind_the_original_profile_and_resource(
requires_key,
) -> None:
profile = object()
authorize = MagicMock(return_value=profile)
authorize = Recorder(profile)
monkeypatch.setattr(resources, "_profile", authorize)
monkeypatch.setattr(resources, "_dump_response", lambda _schema, raw: raw)
with Flask(__name__).test_request_context(
@@ -610,8 +620,8 @@ def test_durable_source_contracts_bind_the_original_profile_and_resource(
headers={"Idempotency-Key": "source-command-1"},
):
method(controller(), *arguments)
authorize.assert_called_once_with(runtime, operation_id=operation, control_space_id="control-1")
call = runtime.facade.execute_service.call_args.kwargs
assert authorize.calls == [invocation(runtime, operation_id=operation, control_space_id="control-1")]
call = runtime.facade.execute_service.last_call.kwargs
assert call["profile"] is profile
assert call["resource_id"] == resource_id
assert call["path_parameters"] == path_parameters