diff --git a/api/tests/test_containers_integration_tests/controllers/console/test_apikey.py b/api/tests/test_containers_integration_tests/controllers/console/test_apikey.py new file mode 100644 index 0000000000..7df63aae1a --- /dev/null +++ b/api/tests/test_containers_integration_tests/controllers/console/test_apikey.py @@ -0,0 +1,153 @@ +"""Integration tests for console API key endpoints using testcontainers.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest +from flask.testing import FlaskClient +from sqlalchemy import delete +from sqlalchemy.orm import Session + +from models.enums import ApiTokenType +from models.model import ApiToken, App, AppMode +from tests.test_containers_integration_tests.controllers.console.helpers import ( + authenticate_console_client, + create_console_account_and_tenant, + create_console_app, +) + + +@pytest.fixture +def setup_app( + db_session_with_containers: Session, + test_client_with_containers: FlaskClient, +) -> tuple[FlaskClient, dict[str, str], App]: + """Create an authenticated client with an app for API key tests.""" + account, tenant = create_console_account_and_tenant(db_session_with_containers) + app = create_console_app(db_session_with_containers, tenant.id, account.id, AppMode.CHAT) + headers = authenticate_console_client(test_client_with_containers, account) + return test_client_with_containers, headers, app + + +@pytest.fixture(autouse=True) +def cleanup_api_tokens(db_session_with_containers: Session): + """Remove API tokens created during each test.""" + yield + db_session_with_containers.execute(delete(ApiToken)) + db_session_with_containers.commit() + + +class TestAppApiKeyListResource: + """Tests for GET/POST /apps//api-keys.""" + + def test_get_empty_keys(self, setup_app: tuple[FlaskClient, dict[str, str], App]) -> None: + client, headers, app = setup_app + resp = client.get(f"/console/api/apps/{app.id}/api-keys", headers=headers) + assert resp.status_code == 200 + assert resp.json is not None + assert resp.json["data"] == [] + + def test_create_api_key(self, setup_app: tuple[FlaskClient, dict[str, str], App]) -> None: + client, headers, app = setup_app + resp = client.post(f"/console/api/apps/{app.id}/api-keys", headers=headers) + assert resp.status_code == 201 + data = resp.json + assert data is not None + assert data["token"].startswith("app-") + assert data["id"] is not None + + def test_get_keys_after_create(self, setup_app: tuple[FlaskClient, dict[str, str], App]) -> None: + client, headers, app = setup_app + client.post(f"/console/api/apps/{app.id}/api-keys", headers=headers) + client.post(f"/console/api/apps/{app.id}/api-keys", headers=headers) + + resp = client.get(f"/console/api/apps/{app.id}/api-keys", headers=headers) + assert resp.status_code == 200 + assert resp.json is not None + assert len(resp.json["data"]) == 2 + + def test_create_key_max_limit( + self, + setup_app: tuple[FlaskClient, dict[str, str], App], + db_session_with_containers: Session, + ) -> None: + client, headers, app = setup_app + # Create 10 keys (the max) + for _ in range(10): + client.post(f"/console/api/apps/{app.id}/api-keys", headers=headers) + + # 11th should fail + resp = client.post(f"/console/api/apps/{app.id}/api-keys", headers=headers) + assert resp.status_code == 400 + + def test_get_keys_for_nonexistent_app( + self, + setup_app: tuple[FlaskClient, dict[str, str], App], + ) -> None: + client, headers, _ = setup_app + resp = client.get( + "/console/api/apps/00000000-0000-0000-0000-000000000000/api-keys", + headers=headers, + ) + assert resp.status_code == 404 + + +class TestAppApiKeyResource: + """Tests for DELETE /apps//api-keys/.""" + + def test_delete_key_success(self, setup_app: tuple[FlaskClient, dict[str, str], App]) -> None: + client, headers, app = setup_app + create_resp = client.post(f"/console/api/apps/{app.id}/api-keys", headers=headers) + assert create_resp.json is not None + key_id = create_resp.json["id"] + + resp = client.delete(f"/console/api/apps/{app.id}/api-keys/{key_id}", headers=headers) + assert resp.status_code == 204 + + def test_delete_nonexistent_key(self, setup_app: tuple[FlaskClient, dict[str, str], App]) -> None: + client, headers, app = setup_app + resp = client.delete( + f"/console/api/apps/{app.id}/api-keys/00000000-0000-0000-0000-000000000000", + headers=headers, + ) + assert resp.status_code == 404 + + def test_delete_key_nonexistent_app( + self, + setup_app: tuple[FlaskClient, dict[str, str], App], + ) -> None: + client, headers, _ = setup_app + resp = client.delete( + "/console/api/apps/00000000-0000-0000-0000-000000000000/api-keys/00000000-0000-0000-0000-000000000000", + headers=headers, + ) + assert resp.status_code == 404 + + def test_delete_forbidden_for_non_admin( + self, + flask_app_with_containers, + ) -> None: + """A non-admin member cannot delete API keys via the controller permission check.""" + from werkzeug.exceptions import Forbidden + + from controllers.console.apikey import BaseApiKeyResource + + resource = BaseApiKeyResource() + resource.resource_type = ApiTokenType.APP + resource.resource_model = MagicMock() + resource.resource_id_field = "app_id" + + non_admin = MagicMock() + non_admin.is_admin_or_owner = False + + with ( + flask_app_with_containers.test_request_context("/"), + patch( + "controllers.console.apikey.current_account_with_tenant", + return_value=(non_admin, "tenant-id"), + ), + patch("controllers.console.apikey._get_resource"), + ): + with pytest.raises(Forbidden): + BaseApiKeyResource.delete(resource, "rid", "kid") diff --git a/api/tests/unit_tests/controllers/console/test_apikey.py b/api/tests/unit_tests/controllers/console/test_apikey.py deleted file mode 100644 index 2dff9c4037..0000000000 --- a/api/tests/unit_tests/controllers/console/test_apikey.py +++ /dev/null @@ -1,139 +0,0 @@ -from unittest.mock import MagicMock, patch - -import pytest -from werkzeug.exceptions import Forbidden - -from controllers.console.apikey import ( - BaseApiKeyListResource, - BaseApiKeyResource, - _get_resource, -) -from models.enums import ApiTokenType - - -@pytest.fixture -def tenant_context_admin(): - with patch("controllers.console.apikey.current_account_with_tenant") as mock: - user = MagicMock() - user.is_admin_or_owner = True - mock.return_value = (user, "tenant-123") - yield mock - - -@pytest.fixture -def tenant_context_non_admin(): - with patch("controllers.console.apikey.current_account_with_tenant") as mock: - user = MagicMock() - user.is_admin_or_owner = False - mock.return_value = (user, "tenant-123") - yield mock - - -@pytest.fixture -def db_mock(): - with patch("controllers.console.apikey.db") as mock_db: - mock_db.session = MagicMock() - yield mock_db - - -@pytest.fixture(autouse=True) -def bypass_permissions(): - with patch( - "controllers.console.apikey.edit_permission_required", - lambda f: f, - ): - yield - - -class DummyApiKeyListResource(BaseApiKeyListResource): - resource_type = ApiTokenType.APP - resource_model = MagicMock() - resource_id_field = "app_id" - token_prefix = "app-" - - -class DummyApiKeyResource(BaseApiKeyResource): - resource_type = ApiTokenType.APP - resource_model = MagicMock() - resource_id_field = "app_id" - - -class TestGetResource: - def test_get_resource_success(self): - fake_resource = MagicMock() - - with ( - patch("controllers.console.apikey.select") as mock_select, - patch("controllers.console.apikey.Session") as mock_session, - patch("controllers.console.apikey.db") as mock_db, - ): - mock_db.engine = MagicMock() - mock_select.return_value.filter_by.return_value = MagicMock() - - session = mock_session.return_value.__enter__.return_value - session.execute.return_value.scalar_one_or_none.return_value = fake_resource - - result = _get_resource("rid", "tid", MagicMock) - assert result == fake_resource - - def test_get_resource_not_found(self): - with ( - patch("controllers.console.apikey.select") as mock_select, - patch("controllers.console.apikey.Session") as mock_session, - patch("controllers.console.apikey.db") as mock_db, - patch("controllers.console.apikey.flask_restx.abort") as abort, - ): - mock_db.engine = MagicMock() - mock_select.return_value.filter_by.return_value = MagicMock() - - session = mock_session.return_value.__enter__.return_value - session.execute.return_value.scalar_one_or_none.return_value = None - - _get_resource("rid", "tid", MagicMock) - - abort.assert_called_once() - - -class TestBaseApiKeyListResource: - def test_get_apikeys_success(self, tenant_context_admin, db_mock): - resource = DummyApiKeyListResource() - - with patch("controllers.console.apikey._get_resource"): - db_mock.session.scalars.return_value.all.return_value = [MagicMock(), MagicMock()] - - result = DummyApiKeyListResource.get.__wrapped__(resource, "resource-id") - assert "items" in result - - -class TestBaseApiKeyResource: - def test_delete_forbidden(self, tenant_context_non_admin, db_mock): - resource = DummyApiKeyResource() - - with patch("controllers.console.apikey._get_resource"): - with pytest.raises(Forbidden): - DummyApiKeyResource.delete(resource, "rid", "kid") - - def test_delete_key_not_found(self, tenant_context_admin, db_mock): - resource = DummyApiKeyResource() - db_mock.session.scalar.return_value = None - - with patch("controllers.console.apikey._get_resource"): - with pytest.raises(Exception) as exc_info: - DummyApiKeyResource.delete(resource, "rid", "kid") - - # flask_restx.abort raises HTTPException with message in data attribute - assert exc_info.value.data["message"] == "API key not found" - - def test_delete_success(self, tenant_context_admin, db_mock): - resource = DummyApiKeyResource() - db_mock.session.scalar.return_value = MagicMock() - - with ( - patch("controllers.console.apikey._get_resource"), - patch("controllers.console.apikey.ApiTokenCache.delete"), - ): - result, status = DummyApiKeyResource.delete(resource, "rid", "kid") - - assert status == 204 - assert result == {"result": "success"} - db_mock.session.commit.assert_called_once()