From 7f80f2db87c0856fd2b3607ee6e0d77ccc3f6e39 Mon Sep 17 00:00:00 2001 From: Stephen Zhou <38493346+hyoban@users.noreply.github.com> Date: Thu, 15 Oct 2026 20:25:25 +0800 Subject: [PATCH] test(api): type KnowledgeFS task tests for strict pyrefly Make the shared facade fake return payloads checked against their real DTO type, record failed-retrieval captures as typed calls, narrow nested JSON metadata before indexing, and annotate the investigation task tests. Co-Authored-By: Claude Opus 5.5 --- .../tasks/knowledge_fs_task_fakes.py | 17 ++++- ..._knowledge_fs_agent_investigation_tasks.py | 20 +++-- ...est_knowledge_fs_failed_retrieval_tasks.py | 44 ++++++++--- .../test_knowledge_fs_initial_source_tasks.py | 74 +++++++++++-------- .../test_knowledge_fs_source_import_tasks.py | 41 ++++++---- 5 files changed, 129 insertions(+), 67 deletions(-) diff --git a/api/tests/unit_tests/tasks/knowledge_fs_task_fakes.py b/api/tests/unit_tests/tasks/knowledge_fs_task_fakes.py index 23eb2d168b8..632be4179d4 100644 --- a/api/tests/unit_tests/tasks/knowledge_fs_task_fakes.py +++ b/api/tests/unit_tests/tasks/knowledge_fs_task_fakes.py @@ -4,6 +4,7 @@ from __future__ import annotations from collections.abc import Callable, Mapping from dataclasses import dataclass, field +from typing import TypeVar, cast from celery.exceptions import Retry @@ -25,6 +26,8 @@ from services.knowledge_fs.product_dto import ( KnowledgeFSSourceWorkflowResponse, ) +_PayloadT = TypeVar("_PayloadT") + @dataclass class RecordingKnowledgeFSMetrics: @@ -85,9 +88,11 @@ class RecordingKnowledgeFSFacade: def call_count(self, method: str) -> int: return len(self.calls.get(method, [])) - def payload(self, method: str) -> object: - """Return the ``payload`` keyword of the most recent call to ``method``.""" - return self.calls[method][-1]["payload"] + def payload(self, method: str, payload_type: type[_PayloadT]) -> _PayloadT: + """Return the ``payload`` keyword of the most recent call to ``method``, checked against ``payload_type``.""" + payload = self.calls[method][-1]["payload"] + assert isinstance(payload, payload_type), f"{method} payload is {type(payload).__name__}" + return payload def _respond(self, method: str, kwargs: dict[str, object]) -> object: self.calls.setdefault(method, []).append(kwargs) @@ -139,6 +144,12 @@ class RecordingKnowledgeFSFacade: return self._respond("update_source_sync_policy", kwargs) +def json_object(value: object) -> dict[str, object]: + """Narrow recorded JSON metadata, or an object nested in it, to a string-keyed dict.""" + assert isinstance(value, dict), f"expected a JSON object, got {type(value).__name__}" + return cast(dict[str, object], value) + + @dataclass(frozen=True) class FacadeRuntime: """The ``KnowledgeFSRuntime`` member the source tasks read.""" diff --git a/api/tests/unit_tests/tasks/test_knowledge_fs_agent_investigation_tasks.py b/api/tests/unit_tests/tasks/test_knowledge_fs_agent_investigation_tasks.py index cc051fc8853..711b91220c5 100644 --- a/api/tests/unit_tests/tasks/test_knowledge_fs_agent_investigation_tasks.py +++ b/api/tests/unit_tests/tasks/test_knowledge_fs_agent_investigation_tasks.py @@ -1,4 +1,5 @@ from dataclasses import dataclass, field +from typing import Literal from uuid import uuid4 import pytest @@ -25,19 +26,20 @@ class _Gateway: raise self.error -def report(): +def report() -> tuple[dict[str, object], str]: space = str(uuid4()) + attempts: list[dict[str, object]] = [] return { "investigation_id": str(uuid4()), "execution_context": {"tenant_id": "tenant", "agent_mode": "agent_app", "invoke_from": "published"}, "bindings": [{"id": "docs", "name": "Docs", "control_space_id": space}], "query": "refund", "status": "completed", - "attempts": [], + "attempts": attempts, }, space -def test_dataset_worker_reconstructs_report_for_each_delivery(monkeypatch): +def test_dataset_worker_reconstructs_report_for_each_delivery(monkeypatch: pytest.MonkeyPatch) -> None: gateway = _Gateway() monkeypatch.setattr(module, "AgentKnowledgeGateway", lambda: gateway) payload, space = report() @@ -51,7 +53,7 @@ def test_dataset_worker_reconstructs_report_for_each_delivery(monkeypatch): assert str(first[0].investigation_id) == payload["investigation_id"] -def test_revoked_authority_stops_publication_and_transient_failures_retry(monkeypatch): +def test_revoked_authority_stops_publication_and_transient_failures_retry(monkeypatch: pytest.MonkeyPatch) -> None: gateway = _Gateway(error=KnowledgeFsError("KNOWLEDGE_SCOPE_CHANGED", "revoked", 403)) monkeypatch.setattr(module, "AgentKnowledgeGateway", lambda: gateway) payload, space = report() @@ -67,7 +69,9 @@ def test_revoked_authority_stops_publication_and_transient_failures_retry(monkey @pytest.mark.parametrize("status_code", [400, 403, 409, 413, 422]) -def test_permanent_remote_rejections_do_not_retry(monkeypatch, status_code): +def test_permanent_remote_rejections_do_not_retry( + monkeypatch: pytest.MonkeyPatch, status_code: Literal[400, 403, 409, 413, 422] +) -> None: gateway = _Gateway(error=KnowledgeFSProductRequestRejectedError(status_code=status_code)) monkeypatch.setattr(module, "AgentKnowledgeGateway", lambda: gateway) task = module.capture_agent_knowledge_investigation_task @@ -82,7 +86,9 @@ def test_permanent_remote_rejections_do_not_retry(monkeypatch, status_code): @pytest.mark.parametrize(("retries", "countdown"), [(0, 30), (1, 60), (2, 120)]) -def test_rate_limited_report_retries_with_backoff(monkeypatch, retries, countdown): +def test_rate_limited_report_retries_with_backoff( + monkeypatch: pytest.MonkeyPatch, retries: int, countdown: int +) -> None: error = KnowledgeFSProductRequestRejectedError(status_code=429) gateway = _Gateway(error=error) monkeypatch.setattr(module, "AgentKnowledgeGateway", lambda: gateway) @@ -98,7 +104,7 @@ def test_rate_limited_report_retries_with_backoff(monkeypatch, retries, countdow assert retry.calls == [{"exc": error, "countdown": countdown}] -def test_malformed_queued_report_is_rejected_before_gateway_publication(monkeypatch): +def test_malformed_queued_report_is_rejected_before_gateway_publication(monkeypatch: pytest.MonkeyPatch) -> None: gateway = _Gateway() monkeypatch.setattr(module, "AgentKnowledgeGateway", lambda: gateway) task = module.capture_agent_knowledge_investigation_task diff --git a/api/tests/unit_tests/tasks/test_knowledge_fs_failed_retrieval_tasks.py b/api/tests/unit_tests/tasks/test_knowledge_fs_failed_retrieval_tasks.py index 72d167d09bb..50829bbfdce 100644 --- a/api/tests/unit_tests/tasks/test_knowledge_fs_failed_retrieval_tasks.py +++ b/api/tests/unit_tests/tasks/test_knowledge_fs_failed_retrieval_tasks.py @@ -1,14 +1,19 @@ from __future__ import annotations from dataclasses import dataclass, field -from typing import Literal, TypedDict +from typing import Literal import pytest from celery.app.task import Context from celery.exceptions import Retry +from typing_extensions import TypedDict +from core.knowledge_fs.resource import KnowledgeResourceRef from services.knowledge_fs.app_admission_service import KnowledgeFSAppAdmissionError -from services.knowledge_fs.product_dto import KnowledgeFSWorkflowFailedRetrievalCaptureResponse +from services.knowledge_fs.product_dto import ( + KnowledgeFSWorkflowFailedRetrievalCapturePayload, + KnowledgeFSWorkflowFailedRetrievalCaptureResponse, +) from services.knowledge_fs.product_remote import KnowledgeFSProductRequestRejectedError from tasks import knowledge_fs_failed_retrieval_tasks as task_module from tests.unit_tests.tasks.knowledge_fs_task_fakes import RecordingRetry @@ -17,7 +22,7 @@ from tests.unit_tests.tasks.task_options import task_options EVENT_ID = "019fac9f-bfb0-75ee-9af5-252ebafbac1e" -class FailedRetrievalTaskArguments(TypedDict): +class FailedRetrievalTaskArguments(TypedDict, closed=True): event_id: str tenant_id: str app_id: str @@ -39,13 +44,28 @@ def _task_kwargs() -> FailedRetrievalTaskArguments: } +@dataclass(frozen=True) +class _CaptureCall: + tenant_id: str + app_id: str + resource: KnowledgeResourceRef + payload: KnowledgeFSWorkflowFailedRetrievalCapturePayload + + @dataclass class _AppCapabilities: outcome: KnowledgeFSWorkflowFailedRetrievalCaptureResponse | Exception - calls: list[dict[str, object]] = field(default_factory=list) + calls: list[_CaptureCall] = field(default_factory=list) - def capture_workflow_failed_retrieval(self, **kwargs: object) -> KnowledgeFSWorkflowFailedRetrievalCaptureResponse: - self.calls.append(kwargs) + def capture_workflow_failed_retrieval( + self, + *, + tenant_id: str, + app_id: str, + resource: KnowledgeResourceRef, + payload: KnowledgeFSWorkflowFailedRetrievalCapturePayload, + ) -> KnowledgeFSWorkflowFailedRetrievalCaptureResponse: + self.calls.append(_CaptureCall(tenant_id=tenant_id, app_id=app_id, resource=resource, payload=payload)) if isinstance(self.outcome, Exception): raise self.outcome return self.outcome @@ -91,12 +111,12 @@ def test_failed_retrieval_task_runs_on_dataset_queue_and_reauthorizes_app( assert task_options(task_module.capture_workflow_failed_retrieval_task)["queue"] == "dataset" assert len(capability.calls) == 1 call = capability.calls[0] - assert call["tenant_id"] == "tenant-1" - assert call["app_id"] == "app-1" - assert call["resource"].control_space_id == "control-1" - assert str(call["payload"].event_id) == EVENT_ID - assert call["payload"].query == "missing answer" - assert call["payload"].retrieval_trace_id == "retrieval-trace-1" + assert call.tenant_id == "tenant-1" + assert call.app_id == "app-1" + assert call.resource.control_space_id == "control-1" + assert str(call.payload.event_id) == EVENT_ID + assert call.payload.query == "missing answer" + assert call.payload.retrieval_trace_id == "retrieval-trace-1" def test_enqueue_is_idempotency_aware_and_never_propagates_broker_failure(monkeypatch: pytest.MonkeyPatch) -> None: diff --git a/api/tests/unit_tests/tasks/test_knowledge_fs_initial_source_tasks.py b/api/tests/unit_tests/tasks/test_knowledge_fs_initial_source_tasks.py index 036f5f84766..b25836a8e2b 100644 --- a/api/tests/unit_tests/tasks/test_knowledge_fs_initial_source_tasks.py +++ b/api/tests/unit_tests/tasks/test_knowledge_fs_initial_source_tasks.py @@ -10,9 +10,15 @@ from models.account import Account from models.knowledge_fs import KnowledgeFSControlSpace, KnowledgeFSControlSpaceState from models.oauth import DatasourceProvider from services.knowledge_fs.product_dto import ( + KnowledgeFSCrawlImportPayload, KnowledgeFSInitialOnlineDocumentSourcePayload, KnowledgeFSInitialOnlineDriveSourcePayload, KnowledgeFSInitialWebsiteSourcePayload, + KnowledgeFSSourceConnectionCreatePayload, + KnowledgeFSSourceCreatePayload, + KnowledgeFSSourceSyncPolicyPayload, + KnowledgeFSSourceUpdatePayload, + KnowledgeFSSourceWorkflowImportPayload, knowledge_fs_initial_preview_configuration_fingerprint, ) from services.knowledge_fs.product_remote import ( @@ -31,6 +37,7 @@ from tasks.knowledge_fs_initial_source_tasks import ( from tests.unit_tests.tasks.knowledge_fs_task_fakes import ( FacadeRuntime, RecordingKnowledgeFSFacade, + json_object, source_connection_list_response, source_connection_response, source_list_response, @@ -286,7 +293,7 @@ def test_initial_website_source_import_recrawls_exact_selection_and_configures_d assert _start(facade, _payload()) == "workflow-1" - source_payload = facade.payload("create_source") + source_payload = facade.payload("create_source", KnowledgeFSSourceCreatePayload) assert source_payload.status == "disabled" assert source_payload.connection_id == "connection-1" assert source_payload.metadata["datasourceParameterMode"] == "exact" @@ -299,16 +306,17 @@ def test_initial_website_source_import_recrawls_exact_selection_and_configures_d "limit": 25, "url": "https://docs.dify.ai", } - import_payload = facade.payload("import_selected_source_crawl") + import_payload = facade.payload("import_selected_source_crawl", KnowledgeFSCrawlImportPayload) assert import_payload.source_urls == [ "https://docs.dify.ai/a", "https://docs.dify.ai/b", ] - source_update_payload = facade.payload("update_source") + source_update_payload = facade.payload("update_source", KnowledgeFSSourceUpdatePayload) + source_update_metadata = json_object(source_update_payload.metadata) assert source_update_payload.expected_version == 3 - assert source_update_payload.metadata == {"preview": False} + assert source_update_metadata == {"preview": False} assert source_update_payload.status == "active" - sync_payload = facade.payload("update_source_sync_policy") + sync_payload = facade.payload("update_source_sync_policy", KnowledgeFSSourceSyncPolicyPayload) assert sync_payload.mode == "interval" assert sync_payload.enabled is True assert sync_payload.expected_revision == 0 @@ -334,7 +342,7 @@ def test_initial_website_source_import_preserves_distinct_urls_with_the_same_can assert _start(facade, payload) == "workflow-1" - import_payload = facade.payload("import_selected_source_crawl") + import_payload = facade.payload("import_selected_source_crawl", KnowledgeFSCrawlImportPayload) assert import_payload.source_urls == [ "https://docs.dify.ai/a/", "https://DOCS.dify.ai:443/a#section", @@ -346,7 +354,7 @@ def test_initial_website_source_import_preserves_legacy_crawl_option_projection( assert _start(facade, _legacy_payload()) == "workflow-1" - source_payload = facade.payload("create_source") + source_payload = facade.payload("create_source", KnowledgeFSSourceCreatePayload) assert "datasourceParameterMode" not in source_payload.metadata assert "parameters" not in source_payload.metadata assert source_payload.metadata["crawlOptions"] == { @@ -436,7 +444,7 @@ def test_initial_connector_source_import_uses_exact_binding_and_selection( ) assert result == "connector-workflow-1" - connection_payload = facade.payload("create_source_connection") + connection_payload = facade.payload("create_source_connection", KnowledgeFSSourceConnectionCreatePayload) assert connection_payload.provider_id == provider_id assert connection_payload.configuration == { "credentialId": credential_id, @@ -445,17 +453,17 @@ def test_initial_connector_source_import_uses_exact_binding_and_selection( "provider": payload.provider, "providerKind": "online-document" if workflow_kind == "online-document-import" else "online-drive", } - source_payload = facade.payload("create_source") + source_payload = facade.payload("create_source", KnowledgeFSSourceCreatePayload) assert source_payload.connection_id == "connector-1" assert source_payload.name == expected_source_name assert source_payload.type == "connector" assert source_payload.uri == "connector://connector-1" assert source_payload.metadata["providerId"] == provider_id assert source_payload.metadata["providerName"] == expected_provider_name - import_payload = facade.payload("import_source_workflow").root + import_payload = facade.payload("import_source_workflow", KnowledgeFSSourceWorkflowImportPayload).root assert import_payload.kind == workflow_kind assert import_payload.items == payload.selection - sync_payload = facade.payload("update_source_sync_policy") + sync_payload = facade.payload("update_source_sync_policy", KnowledgeFSSourceSyncPolicyPayload) assert sync_payload.mode == ("manual" if workflow_kind == "online-drive-import" else "interval") @@ -488,12 +496,13 @@ def test_initial_website_source_import_reuses_source_across_pages_and_preserves_ assert _start(facade, _payload()) == "failed-workflow" assert facade.call_count("create_source") == 0 assert facade.call_count("list_source_connections") == 0 - source_update_payload = facade.payload("update_source") + source_update_payload = facade.payload("update_source", KnowledgeFSSourceUpdatePayload) + source_update_metadata = json_object(source_update_payload.metadata) assert source_update_payload.status == "disabled" - assert "parameters" not in source_update_payload.metadata - assert source_update_payload.metadata["preview"] is False - assert source_update_payload.metadata["initialImport"]["state"] == "failed" - assert source_update_payload.metadata["initialImport"]["syncPolicy"] == { + assert "parameters" not in source_update_metadata + assert source_update_metadata["preview"] is False + assert json_object(source_update_metadata["initialImport"])["state"] == "failed" + assert json_object(source_update_metadata["initialImport"])["syncPolicy"] == { "customIntervalSeconds": None, "enabled": True, "mode": "interval", @@ -522,12 +531,13 @@ def test_initial_website_source_import_exposes_preview_consume_rejection() -> No with pytest.raises(KnowledgeFSProductRequestRejectedError): _start(facade, payload) - source_update = facade.payload("update_source") + source_update = facade.payload("update_source", KnowledgeFSSourceUpdatePayload) + source_update_metadata = json_object(source_update.metadata) assert source_update.status == "disabled" - assert source_update.metadata["preview"] is False - assert source_update.metadata["initialImport"]["state"] == "failed" - assert source_update.metadata["initialImport"]["errorCode"] == "SOURCE_WORKFLOW_FAILED" - assert source_update.metadata["initialImport"]["syncPolicy"] == { + assert source_update_metadata["preview"] is False + assert json_object(source_update_metadata["initialImport"])["state"] == "failed" + assert json_object(source_update_metadata["initialImport"])["errorCode"] == "SOURCE_WORKFLOW_FAILED" + assert json_object(source_update_metadata["initialImport"])["syncPolicy"] == { "customIntervalSeconds": None, "enabled": True, "mode": "interval", @@ -567,7 +577,7 @@ def test_initial_website_source_import_configures_remaining_sync_modes( facade.script("get_source_sync_policy", sync_policy_response(revision=7)) assert _start(facade, _payload(sync_policy)) == "workflow-1" - sync_payload = facade.payload("update_source_sync_policy") + sync_payload = facade.payload("update_source_sync_policy", KnowledgeFSSourceSyncPolicyPayload) assert sync_payload.mode == expected_mode assert sync_payload.enabled is expected_enabled assert sync_payload.custom_interval_seconds == expected_custom_interval_seconds @@ -592,7 +602,7 @@ def test_initial_website_source_import_does_not_recommit_active_source_after_ret assert _start(facade, _payload()) == "workflow-1" assert facade.call_count("update_source") == 0 - sync_payload = facade.payload("update_source_sync_policy") + sync_payload = facade.payload("update_source_sync_policy", KnowledgeFSSourceSyncPolicyPayload) assert sync_payload.expected_source_version == 4 @@ -688,7 +698,7 @@ def test_initial_website_source_import_creates_missing_firecrawl_connection() -> assert _start(facade, _payload()) == "workflow-1" - connection_payload = facade.payload("create_source_connection") + connection_payload = facade.payload("create_source_connection", KnowledgeFSSourceConnectionCreatePayload) assert connection_payload.auth_kind == "endpoint" assert connection_payload.configuration == { "credentialId": "firecrawl-credential-1", @@ -699,7 +709,7 @@ def test_initial_website_source_import_creates_missing_firecrawl_connection() -> } assert connection_payload.credentials == {} assert connection_payload.provider_id == "plugin-daemon-website" - source_payload = facade.payload("create_source") + source_payload = facade.payload("create_source", KnowledgeFSSourceCreatePayload) assert source_payload.connection_id == "connection-created" @@ -754,7 +764,8 @@ def test_initial_website_source_import_polls_existing_workflow_without_recreatin assert facade.call_count("list_source_connections") == 0 assert facade.call_count("create_source") == 0 assert facade.call_count("import_selected_source_crawl") == 0 - source_update_payload = facade.payload("update_source") + source_update_payload = facade.payload("update_source", KnowledgeFSSourceUpdatePayload) + source_update_metadata = json_object(source_update_payload.metadata) assert source_update_payload.status == "active" @@ -788,10 +799,11 @@ def test_initial_website_source_import_exposes_failed_source_without_activating_ assert _start(facade, _payload(), workflow_id="workflow-1") == "workflow-1" - source_update_payload = facade.payload("update_source") + source_update_payload = facade.payload("update_source", KnowledgeFSSourceUpdatePayload) + source_update_metadata = json_object(source_update_payload.metadata) assert source_update_payload.status == "disabled" - assert source_update_payload.metadata["preview"] is False - initial_import = source_update_payload.metadata["initialImport"] + assert source_update_metadata["preview"] is False + initial_import = json_object(source_update_metadata["initialImport"]) assert initial_import["errorCode"] == "SOURCE_DOCUMENT_MATERIALIZATION_FAILED" assert initial_import["errorMessage"] == "Source document materialization failed" assert initial_import["state"] == "failed" @@ -803,7 +815,9 @@ def test_initial_website_source_import_exposes_failed_source_without_activating_ "enabled": True, "mode": "interval", } - assert len(initial_import["configurationFingerprint"]) == 64 + fingerprint = initial_import["configurationFingerprint"] + assert isinstance(fingerprint, str) + assert len(fingerprint) == 64 assert facade.call_count("update_source_sync_policy") == 0 diff --git a/api/tests/unit_tests/tasks/test_knowledge_fs_source_import_tasks.py b/api/tests/unit_tests/tasks/test_knowledge_fs_source_import_tasks.py index 561a7db5739..2a108586672 100644 --- a/api/tests/unit_tests/tasks/test_knowledge_fs_source_import_tasks.py +++ b/api/tests/unit_tests/tasks/test_knowledge_fs_source_import_tasks.py @@ -4,7 +4,12 @@ from unittest.mock import patch import pytest from celery.exceptions import Retry -from services.knowledge_fs.product_dto import KnowledgeFSSourceResponse, KnowledgeFSSourceWorkflowResponse +from services.knowledge_fs.product_dto import ( + KnowledgeFSSourceResponse, + KnowledgeFSSourceSyncPolicyPayload, + KnowledgeFSSourceUpdatePayload, + KnowledgeFSSourceWorkflowResponse, +) from services.knowledge_fs.product_remote import ( KnowledgeFSProductRequestRejectedError, KnowledgeFSProductResourceNotFoundError, @@ -18,6 +23,7 @@ from tests.unit_tests.tasks.knowledge_fs_task_fakes import ( FacadeRuntime, RecordingKnowledgeFSFacade, RecordingRetry, + json_object, source_response, source_workflow_response, ) @@ -102,16 +108,17 @@ def test_finalize_source_import_activates_source_then_applies_policy_for_new_ver assert _run(facade) == "import-1" - policy = facade.payload("update_source_sync_policy") + policy = facade.payload("update_source_sync_policy", KnowledgeFSSourceSyncPolicyPayload) assert policy.enabled is False assert policy.mode == "manual" assert policy.expected_revision == 0 assert policy.expected_source_version == 5 - update = facade.payload("update_source") + update = facade.payload("update_source", KnowledgeFSSourceUpdatePayload) + update_metadata = json_object(update.metadata) assert update.status == "active" - assert "parameters" not in update.metadata - assert update.metadata["pendingImport"] is None - assert update.metadata["lastImport"]["state"] == "completed" + assert "parameters" not in update_metadata + assert update_metadata["pendingImport"] is None + assert json_object(update_metadata["lastImport"])["state"] == "completed" def test_finalize_source_import_retries_policy_after_source_activation() -> None: @@ -137,7 +144,7 @@ def test_finalize_source_import_retries_policy_after_source_activation() -> None assert _run(facade) == "import-1" assert facade.call_count("update_source") == 0 - policy = facade.payload("update_source_sync_policy") + policy = facade.payload("update_source_sync_policy", KnowledgeFSSourceSyncPolicyPayload) assert policy.expected_source_version == 5 @@ -184,11 +191,13 @@ def test_finalize_source_import_recovers_a_stale_failed_marker_after_workflow_co assert _run(facade) == "import-1" - update = facade.payload("update_source") + update = facade.payload("update_source", KnowledgeFSSourceUpdatePayload) + + update_metadata = json_object(update.metadata) assert update.status == "active" - assert update.metadata["lastImport"]["state"] == "completed" - assert update.metadata["pendingImport"] is None - policy = facade.payload("update_source_sync_policy") + assert json_object(update_metadata["lastImport"])["state"] == "completed" + assert update_metadata["pendingImport"] is None + policy = facade.payload("update_source_sync_policy", KnowledgeFSSourceSyncPolicyPayload) assert policy.expected_source_version == 6 @@ -206,10 +215,12 @@ def test_finalize_source_import_persists_failure_on_visible_source() -> None: assert _run(facade) == "import-1" - update = facade.payload("update_source") + update = facade.payload("update_source", KnowledgeFSSourceUpdatePayload) + + update_metadata = json_object(update.metadata) assert update.status == "error" - assert "parameters" not in update.metadata - assert update.metadata["lastImport"] == { + assert "parameters" not in update_metadata + assert update_metadata["lastImport"] == { "errorCode": "PROVIDER_FAILED", "errorMessage": "provider failed", "kind": "crawl-preview-selection", @@ -218,7 +229,7 @@ def test_finalize_source_import_persists_failure_on_visible_source() -> None: "syncPolicy": {"enabled": False, "mode": "manual"}, "workflowId": "import-1", } - assert update.metadata["pendingImport"] is None + assert update_metadata["pendingImport"] is None assert facade.call_count("update_source_sync_policy") == 0