Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 45 additions & 21 deletions tests/fixtures/client.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
import hashlib
import importlib
import sys
from contextlib import nullcontext
from unittest.mock import MagicMock, patch

import pytest
from fastapi.testclient import TestClient
Expand All @@ -9,6 +11,20 @@
from keep.api.models.db.tenant import TenantApiKey


def _mock_oidc_discovery():
"""Return a context manager that stubs OIDC discovery HTTP calls.

Prevents real network requests to Auth0 during app startup when
AUTH_TYPE=AUTH0 is set in tests.
"""
mock_resp = MagicMock()
mock_resp.json.return_value = {
"jwks_uri": "https://test-domain.auth0.com/.well-known/jwks.json"
}
mock_resp.raise_for_status = MagicMock()
return patch("requests.get", return_value=mock_resp)


@pytest.fixture
def test_app(monkeypatch, request, db_session):
# Store original setup_logging function
Expand All @@ -21,11 +37,13 @@ def test_app(monkeypatch, request, db_session):

try:
monkeypatch.setenv("KEEP_USE_LIMITER", "false")
is_auth0 = False
# Check if request.param is a dict or a string
if isinstance(request.param, dict):
# Set environment variables based on the provided dictionary
for key, value in request.param.items():
monkeypatch.setenv(key, str(value))
is_auth0 = request.param.get("AUTH_TYPE") == "AUTH0"
else:
# Old behavior for string parameters
auth_type = request.param
Expand All @@ -35,27 +53,33 @@ def test_app(monkeypatch, request, db_session):
if auth_type == "MULTI_TENANT":
monkeypatch.setenv("AUTH0_DOMAIN", "https://auth0domain.com")

# Clear and reload modules to ensure environment changes are reflected
for module in list(sys.modules):
if module.startswith("keep.api.routes"):
del sys.modules[module]

# Bug in db patching
elif module.startswith("keep.providers.providers_service"):
importlib.reload(sys.modules[module])

if "keep.api.api" in sys.modules:
importlib.reload(sys.modules["keep.api.api"])

if "keep.api.config" in sys.modules:
importlib.reload(sys.modules["keep.api.config"])

# Import and return the app instance
from keep.api.api import get_app
from keep.api.config import provision_resources

provision_resources()
app = get_app()
# When AUTH_TYPE=AUTH0, the authverifier module makes a real HTTP call
# at import time to fetch the OIDC discovery document. The mock must
# wrap the module reload as well, because deleted route modules get
# re-imported during reload and cascade into re-importing the verifier.
ctx = _mock_oidc_discovery() if is_auth0 else nullcontext()
with ctx:
# Clear and reload modules to ensure environment changes are reflected
for module in list(sys.modules):
if module.startswith("keep.api.routes"):
del sys.modules[module]

# Bug in db patching
elif module.startswith("keep.providers.providers_service"):
importlib.reload(sys.modules[module])

if "keep.api.api" in sys.modules:
importlib.reload(sys.modules["keep.api.api"])

if "keep.api.config" in sys.modules:
importlib.reload(sys.modules["keep.api.config"])

# Import and return the app instance
from keep.api.api import get_app
from keep.api.config import provision_resources

provision_resources()
app = get_app()
return app
finally:
# Restore the original setup_logging function
Expand Down
Loading