test: migrate apikey controller tests to testcontainers (#34286)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
Desel72
2026-03-31 16:06:42 +03:00
committed by GitHub
parent 90f94be2b3
commit b818cc0766
2 changed files with 153 additions and 139 deletions

View File

@@ -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/<resource_id>/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/<resource_id>/api-keys/<api_key_id>."""
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")

View File

@@ -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()