Compare commits

...

2 Commits

Author SHA1 Message Date
-LAN-
5d1ed222e6 test(api): fix external controller auth fixture import 2026-04-08 22:06:03 +08:00
-LAN-
91f21fe5b4 fix(api): scope external API use checks to the tenant 2026-04-08 19:44:13 +08:00
5 changed files with 51 additions and 15 deletions

View File

@@ -227,10 +227,11 @@ class ExternalApiUseCheckApi(Resource):
@login_required
@account_initialization_required
def get(self, external_knowledge_api_id):
_, current_tenant_id = current_account_with_tenant()
external_knowledge_api_id = str(external_knowledge_api_id)
external_knowledge_api_is_using, count = ExternalDatasetService.external_knowledge_api_use_check(
external_knowledge_api_id
external_knowledge_api_id, current_tenant_id
)
return {"is_using": external_knowledge_api_is_using, "count": count}, 200

View File

@@ -148,18 +148,23 @@ class ExternalDatasetService:
db.session.commit()
@staticmethod
def external_knowledge_api_use_check(external_knowledge_api_id: str) -> tuple[bool, int]:
def external_knowledge_api_use_check(external_knowledge_api_id: str, tenant_id: str) -> tuple[bool, int]:
"""
Return usage for an external knowledge API within a single tenant.
The caller already scopes access by tenant, so this query must do the
same; otherwise the endpoint becomes a cross-tenant UUID oracle.
"""
count = (
db.session.scalar(
select(func.count(ExternalKnowledgeBindings.id)).where(
ExternalKnowledgeBindings.external_knowledge_api_id == external_knowledge_api_id
ExternalKnowledgeBindings.external_knowledge_api_id == external_knowledge_api_id,
ExternalKnowledgeBindings.tenant_id == tenant_id,
)
)
or 0
)
if count > 0:
return True, count
return False, 0
return count > 0, count
@staticmethod
def get_external_knowledge_binding_with_dataset_id(tenant_id: str, dataset_id: str) -> ExternalKnowledgeBindings:

View File

@@ -1,3 +1,4 @@
from importlib import import_module
from unittest.mock import MagicMock, PropertyMock, patch
import pytest
@@ -11,6 +12,7 @@ from controllers.console.datasets.external import (
BedrockRetrievalApi,
ExternalApiTemplateApi,
ExternalApiTemplateListApi,
ExternalApiUseCheckApi,
ExternalDatasetCreateApi,
ExternalKnowledgeHitTestingApi,
)
@@ -19,6 +21,8 @@ from services.external_knowledge_service import ExternalDatasetService
from services.hit_testing_service import HitTestingService
from services.knowledge_service import ExternalDatasetTestService
external_controller = import_module("controllers.console.datasets.external")
def unwrap(func):
while hasattr(func, "__wrapped__"):
@@ -44,10 +48,11 @@ def current_user():
@pytest.fixture(autouse=True)
def mock_auth(mocker, current_user):
mocker.patch(
"controllers.console.datasets.external.current_account_with_tenant",
return_value=(current_user, "tenant-1"),
def mock_auth(monkeypatch, current_user):
monkeypatch.setattr(
external_controller,
"current_account_with_tenant",
lambda: (current_user, "tenant-1"),
)
@@ -136,6 +141,26 @@ class TestExternalApiTemplateApi:
method(api, "api-id")
class TestExternalApiUseCheckApi:
def test_get_scopes_usage_check_to_current_tenant(self, app):
api = ExternalApiUseCheckApi()
method = unwrap(api.get)
with (
app.test_request_context("/"),
patch.object(
ExternalDatasetService,
"external_knowledge_api_use_check",
return_value=(True, 2),
) as mock_use_check,
):
response, status = method(api, "api-id")
assert status == 200
assert response == {"is_using": True, "count": 2}
mock_use_check.assert_called_once_with("api-id", "tenant-1")
class TestExternalDatasetCreateApi:
def test_create_success(self, app):
api = ExternalDatasetCreateApi()

View File

@@ -396,10 +396,11 @@ class TestExternalDatasetServiceUsageAndBindings:
mock_db_session.scalar.return_value = 3
in_use, count = ExternalDatasetService.external_knowledge_api_use_check("api-1")
in_use, count = ExternalDatasetService.external_knowledge_api_use_check("api-1", "tenant-1")
assert in_use is True
assert count == 3
assert "tenant_id" in str(mock_db_session.scalar.call_args.args[0])
def test_external_knowledge_api_use_check_not_in_use(self, mock_db_session: MagicMock):
"""
@@ -408,7 +409,7 @@ class TestExternalDatasetServiceUsageAndBindings:
mock_db_session.scalar.return_value = 0
in_use, count = ExternalDatasetService.external_knowledge_api_use_check("api-1")
in_use, count = ExternalDatasetService.external_knowledge_api_use_check("api-1", "tenant-1")
assert in_use is False
assert count == 0

View File

@@ -974,26 +974,29 @@ class TestExternalDatasetServiceAPIUseCheck:
"""Test API use check when API has one binding."""
# Arrange
api_id = "api-123"
tenant_id = "tenant-123"
mock_db.session.scalar.return_value = 1
# Act
in_use, count = ExternalDatasetService.external_knowledge_api_use_check(api_id)
in_use, count = ExternalDatasetService.external_knowledge_api_use_check(api_id, tenant_id)
# Assert
assert in_use is True
assert count == 1
assert "tenant_id" in str(mock_db.session.scalar.call_args.args[0])
@patch("services.external_knowledge_service.db")
def test_external_knowledge_api_use_check_in_use_multiple(self, mock_db, factory):
"""Test API use check with multiple bindings."""
# Arrange
api_id = "api-123"
tenant_id = "tenant-123"
mock_db.session.scalar.return_value = 10
# Act
in_use, count = ExternalDatasetService.external_knowledge_api_use_check(api_id)
in_use, count = ExternalDatasetService.external_knowledge_api_use_check(api_id, tenant_id)
# Assert
assert in_use is True
@@ -1004,11 +1007,12 @@ class TestExternalDatasetServiceAPIUseCheck:
"""Test API use check when API is not in use."""
# Arrange
api_id = "api-123"
tenant_id = "tenant-123"
mock_db.session.scalar.return_value = 0
# Act
in_use, count = ExternalDatasetService.external_knowledge_api_use_check(api_id)
in_use, count = ExternalDatasetService.external_knowledge_api_use_check(api_id, tenant_id)
# Assert
assert in_use is False