From 06c9ec2d73cd8363d31e1ce9ec01c80ecaba09d7 Mon Sep 17 00:00:00 2001 From: FFXN Date: Wed, 26 Aug 2026 16:05:27 +0800 Subject: [PATCH] test(knowledge_fs): add tests for service profile and issuance authorization errors --- .../test_knowledge_fs_product_controllers.py | 24 ++++- .../test_knowledge_fs_capability_broker.py | 96 +++++++++++++++++++ 2 files changed, 119 insertions(+), 1 deletion(-) diff --git a/api/tests/unit_tests/controllers/test_knowledge_fs_product_controllers.py b/api/tests/unit_tests/controllers/test_knowledge_fs_product_controllers.py index e5c8c8ecfad..6bab8270851 100644 --- a/api/tests/unit_tests/controllers/test_knowledge_fs_product_controllers.py +++ b/api/tests/unit_tests/controllers/test_knowledge_fs_product_controllers.py @@ -40,7 +40,10 @@ from services.knowledge_fs.product_remote import ( KnowledgeFSRemoteMultipartFile, KnowledgeFSRemoteSSEResponse, ) -from services.knowledge_fs.service_api_authorization import KnowledgeFSServiceApiProfile +from services.knowledge_fs.service_api_authorization import ( + KnowledgeFSServiceApiAuthorizationError, + KnowledgeFSServiceApiProfile, +) _API_ROOT = Path(__file__).resolve().parents[3] @@ -1085,6 +1088,25 @@ def test_service_profile_authorizes_the_dataset_key_for_the_route_space(monkeypa ) +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)), + ) + + with pytest.raises(KnowledgeFSServiceApiAuthorizationError, match="not workspace-scoped"): + service_resources._profile( + runtime, # type: ignore[arg-type] + operation_id="listDocuments", + control_space_id="control-2", + ) + + authorization.authorize.assert_not_called() + + def test_product_modules_do_not_import_dify_dataset_or_document_services() -> None: paths = [ *(path for path in _API_ROOT.glob("services/knowledge_fs/*.py") if not path.name.startswith("upgrade_")), diff --git a/api/tests/unit_tests/services/test_knowledge_fs_capability_broker.py b/api/tests/unit_tests/services/test_knowledge_fs_capability_broker.py index e2d079a9dca..220b6d1b7c4 100644 --- a/api/tests/unit_tests/services/test_knowledge_fs_capability_broker.py +++ b/api/tests/unit_tests/services/test_knowledge_fs_capability_broker.py @@ -584,6 +584,102 @@ def test_stale_external_profile_is_revalidated_before_reservation( assert sqlite_session.scalar(sa.select(sa.func.count(KnowledgeFSCapabilityIssuanceReservation.id))) == 0 +@pytest.mark.parametrize("sqlite_session", [_ISSUANCE_FENCE_MODELS], indirect=True) +def test_service_issuance_rejects_a_space_disabled_after_authorization(sqlite_session: Session) -> None: + space, service_profile, _ = _seed_external_principals(sqlite_session) + policy = sqlite_session.scalar( + sa.select(KnowledgeFSExternalAccessPolicy).where( + KnowledgeFSExternalAccessPolicy.control_space_id == service_profile.control_space_id + ) + ) + assert policy is not None + policy.service_api_enabled = False + sqlite_session.commit() + issuer = FakeIssuer() + broker = KnowledgeFSCapabilityBroker( + sessionmaker(bind=sqlite_session.get_bind(), expire_on_commit=False), + cutover_gate=FakeCutoverGate(), + product=FakeProduct(space), # type: ignore[arg-type] + issuer=issuer, # type: ignore[arg-type] + ) + + with pytest.raises(KnowledgeFSOperationUnavailableError, match="no longer authorized"): + broker.issue_service( + profile=service_profile, + operation_id="createQuery", + trace_id="trace-disabled-service-api", + ) + + assert issuer.requests == [] + assert sqlite_session.scalar(sa.select(sa.func.count(KnowledgeFSCapabilityIssuanceReservation.id))) == 0 + + +@pytest.mark.parametrize( + ("operation_id", "issuer", "trace_id", "message"), + [ + ("unknownOperation", FakeIssuer(), "trace-invalid-operation", "operation is unavailable"), + ("createQuery", None, "trace-disabled-capability", "Capability v2 is disabled"), + ("createQuery", FakeIssuer(), " ", "trace id is required"), + ], +) +@pytest.mark.parametrize("sqlite_session", [_ISSUANCE_FENCE_MODELS], indirect=True) +def test_interactive_issuance_rejects_invalid_preflight_inputs( + sqlite_session: Session, + operation_id: str, + issuer: FakeIssuer | None, + trace_id: str, + message: str, +) -> None: + space, _, _ = _seed_external_principals(sqlite_session) + broker = KnowledgeFSCapabilityBroker( + sessionmaker(bind=sqlite_session.get_bind(), expire_on_commit=False), + cutover_gate=FakeCutoverGate(), + product=FakeProduct(space), # type: ignore[arg-type] + issuer=issuer, # type: ignore[arg-type] + ) + + with pytest.raises(KnowledgeFSOperationUnavailableError, match=message): + broker.issue_interactive( + tenant_id="tenant-1", + account_id="account-1", + control_space_id=space.id, + operation_id=operation_id, + trace_id=trace_id, + ) + + +@pytest.mark.parametrize("sqlite_session", [_ISSUANCE_FENCE_MODELS], indirect=True) +def test_workflow_app_issuance_uses_the_workflow_caller_kind(sqlite_session: Session) -> None: + space, _, app_profile = _seed_external_principals(sqlite_session) + join = sqlite_session.get(AppKnowledgeFSSpaceJoin, app_profile.join_id) + assert join is not None + join.join_type = KnowledgeFSAppSpaceJoinType.WORKFLOW + policy = sqlite_session.scalar( + sa.select(KnowledgeFSExternalAccessPolicy).where( + KnowledgeFSExternalAccessPolicy.control_space_id == app_profile.control_space_id + ) + ) + assert policy is not None + policy.workflow_enabled = True + sqlite_session.commit() + app_profile = app_profile._replace(caller_kind=KnowledgeFSAppSpaceJoinType.WORKFLOW) + issuer = FakeIssuer() + broker = KnowledgeFSCapabilityBroker( + sessionmaker(bind=sqlite_session.get_bind(), expire_on_commit=False), + cutover_gate=FakeCutoverGate(), + product=FakeProduct(space), # type: ignore[arg-type] + issuer=issuer, # type: ignore[arg-type] + ) + + broker.issue_app( + profile=app_profile, + operation_id="createQuery", + trace_id="trace-workflow-app", + ) + + assert issuer.requests[0].caller_kind == "workflow" + + @pytest.mark.parametrize("principal_kind", ["service", "app"]) @pytest.mark.parametrize("sqlite_session", [_ISSUANCE_FENCE_MODELS], indirect=True) def test_external_issuance_uses_fresh_authorization_revision(