Testing : API Key Fixture

backend/app/crud/__init__.py

@@ -30,6 +30,7 @@
     get_api_keys_by_project,
     get_api_key_by_project_user,
     delete_api_key,
+    get_api_key_by_user_id,
 )
 
 from .credentials import (

backend/app/crud/api_key.py

@@ -162,3 +162,21 @@ def get_api_keys_by_project(session: Session, project_id: int) -> list[APIKeyPub
         result.append(APIKeyPublic.model_validate(key_dict))
 
     return result
+
+
+def get_api_key_by_user_id(session: Session, user_id: int) -> APIKeyPublic | None:
+    """
+    Retrieves the API key associated with a user by their user_id.
+    """
+    api_key = (
+        session.query(APIKey)
+        .filter(APIKey.user_id == user_id, APIKey.is_deleted == False)
+        .first()
+    )
+
+    if not api_key:
+        return None
+
+    key_dict = api_key.model_dump()
+    key_dict["key"] = decrypt_api_key(api_key.key)
+    return APIKeyPublic.model_validate(key_dict)

backend/app/seed_data/seed_data.json

@@ -21,14 +21,14 @@
   ],
   "users": [
     {
-      "email": "superuser@example.com",
+      "email": "{{SUPERUSER_EMAIL}}",
       "password": "changethis",
       "full_name": "SUPERUSER",
       "is_active": true,
       "is_superuser": true
     },
     {
-      "email": "org_admin@example.com",
+      "email": "{{ADMIN_EMAIL}}",
       "password": "admin123",
       "full_name": "ADMIN",
       "is_active": true,
@@ -38,15 +38,15 @@
   "apikeys": [
     {
       "organization_name": "Project Tech4dev",
-      "user_email": "superuser@example.com",
+      "user_email": "{{SUPERUSER_EMAIL}}",
       "project_name": "Glific",
       "api_key": "ApiKey No3x47A5qoIGhm0kVKjQ77dhCqEdWRIQZlEPzzzh7i8",
       "is_deleted": false,
       "deleted_at": null
     },
     {
       "organization_name": "Project Tech4dev",
-      "user_email": "org_admin@example.com",
+      "user_email": "{{ADMIN_EMAIL}}",
       "project_name": "Dalgo",
       "api_key": "ApiKey Px8y47B6roJHin1lWLkR88eiDrFdXSJRZmFQazzai8j9",
       "is_deleted": false,

backend/app/seed_data/seed_data.py

@@ -8,6 +8,7 @@
 from sqlmodel import Session, delete, select
 
 from app.core.db import engine
+from app.core import settings
 from app.core.security import encrypt_api_key, get_password_hash
 from app.models import APIKey, Organization, Project, User, Credential, Assistant
 
@@ -308,6 +309,12 @@ def seed_database(session: Session) -> None:
                 f"Created organization: {organization.name} (ID: {organization.id})"
             )
 
+        for user_data in seed_data["users"]:
+            if user_data["email"] == "{{SUPERUSER_EMAIL}}":
+                user_data["email"] = settings.FIRST_SUPERUSER
+            elif user_data["email"] == "{{ADMIN_EMAIL}}":
+                user_data["email"] = settings.EMAIL_TEST_USER
+
         # Create users
         users = []
         for user_data in seed_data["users"]:
@@ -322,6 +329,12 @@ def seed_database(session: Session) -> None:
             projects.append(project)
             logging.info(f"Created project: {project.name} (ID: {project.id})")
 
+        for api_key_data in seed_data["apikeys"]:
+            if api_key_data["user_email"] == "{{SUPERUSER_EMAIL}}":
+                api_key_data["user_email"] = settings.FIRST_SUPERUSER
+            elif api_key_data["user_email"] == "{{ADMIN_EMAIL}}":
+                api_key_data["user_email"] = settings.EMAIL_TEST_USER
+
         # Create API keys
         api_keys = []
         for api_key_data in seed_data["apikeys"]:

backend/app/tests/api/routes/collections/test_collection_info.py

@@ -10,8 +10,6 @@
 
 client = TestClient(app)
 
-original_api_key = "ApiKey No3x47A5qoIGhm0kVKjQ77dhCqEdWRIQZlEPzzzh7i8"
-
 
 def create_collection(
     db,
@@ -38,8 +36,10 @@ def create_collection(
     return collection
 
 
-def test_collection_info_processing(db: Session, client: TestClient):
-    headers = {"X-API-KEY": original_api_key}
+def test_collection_info_processing(
+    db: Session, client: TestClient, normal_user_api_key_headers
+):
+    headers = normal_user_api_key_headers
     user = get_user_from_api_key(db, headers)
     collection = create_collection(db, user, status=CollectionStatus.processing)
 
@@ -57,8 +57,10 @@ def test_collection_info_processing(db: Session, client: TestClient):
     assert data["llm_service_name"] is None
 
 
-def test_collection_info_successful(db: Session, client: TestClient):
-    headers = {"X-API-KEY": original_api_key}
+def test_collection_info_successful(
+    db: Session, client: TestClient, normal_user_api_key_headers
+):
+    headers = normal_user_api_key_headers
     user = get_user_from_api_key(db, headers)
     collection = create_collection(
         db, user, status=CollectionStatus.successful, with_llm=True
@@ -78,8 +80,10 @@ def test_collection_info_successful(db: Session, client: TestClient):
     assert data["llm_service_name"] == "gpt-4o"
 
 
-def test_collection_info_failed(db: Session, client: TestClient):
-    headers = {"X-API-KEY": original_api_key}
+def test_collection_info_failed(
+    db: Session, client: TestClient, normal_user_api_key_headers
+):
+    headers = normal_user_api_key_headers
     user = get_user_from_api_key(db, headers)
     collection = create_collection(db, user, status=CollectionStatus.failed)
 

backend/app/tests/api/routes/collections/test_create_collections.py

@@ -47,9 +47,7 @@ class TestCollectionRouteCreate:
 
     @openai_responses.mock()
     def test_create_collection_success(
-        self,
-        client: TestClient,
-        db: Session,
+        self, client: TestClient, db: Session, normal_user_api_key_headers
     ):
         store = DocumentStore(db)
         documents = store.fill(self._n_documents)
@@ -62,14 +60,10 @@ def test_create_collection_success(
             "instructions": "Test collection assistant.",
             "temperature": 0.1,
         }
-        original_api_key = "ApiKey No3x47A5qoIGhm0kVKjQ77dhCqEdWRIQZlEPzzzh7i8"
-
-        headers = {"X-API-KEY": original_api_key}
+        headers = normal_user_api_key_headers
 
         response = client.post(
-            f"{settings.API_V1_STR}/collections/create",
-            json=body,
-            headers=headers,
+            f"{settings.API_V1_STR}/collections/create", json=body, headers=headers
         )
 
         assert response.status_code == 200

backend/app/tests/api/routes/documents/test_route_document_upload.py

@@ -25,7 +25,7 @@ def put(self, route: Route, scratch: Path):
         with scratch.open("rb") as fp:
             return self.client.post(
                 str(route),
-                headers=self.superuser_token_headers,
+                headers=self.normal_user_api_key_headers,
                 files={
                     "src": (str(scratch), fp, mtype),
                 },
@@ -45,8 +45,8 @@ def route():
 
 
 @pytest.fixture
-def uploader(client: TestClient, superuser_token_headers: dict[str, str]):
-    return WebUploader(client, superuser_token_headers)
+def uploader(client: TestClient, normal_user_api_key_headers: dict[str, str]):
+    return WebUploader(client, normal_user_api_key_headers)
 
 
 @pytest.fixture(scope="class")

backend/app/tests/api/routes/test_assistants.py

@@ -4,16 +4,11 @@
 from app.tests.utils.openai import mock_openai_assistant
 
 
-@pytest.fixture
-def normal_user_api_key_header():
-    return {"X-API-KEY": "ApiKey Px8y47B6roJHin1lWLkR88eiDrFdXSJRZmFQazzai8j9"}
-
-
 @patch("app.api.routes.assistants.fetch_assistant_from_openai")
 def test_ingest_assistant_success(
     mock_fetch_assistant,
     client: TestClient,
-    normal_user_api_key_header: str,
+    normal_user_api_key_headers: dict[str, str],
 ):
     """Test successful assistant ingestion from OpenAI."""
     mock_assistant = mock_openai_assistant()
@@ -22,7 +17,7 @@ def test_ingest_assistant_success(
 
     response = client.post(
         f"/api/v1/assistant/{mock_assistant.id}/ingest",
-        headers=normal_user_api_key_header,
+        headers=normal_user_api_key_headers,
     )
 
     assert response.status_code == 201

backend/app/tests/api/routes/test_responses.py

@@ -16,9 +16,7 @@
 @patch("app.api.routes.responses.OpenAI")
 @patch("app.api.routes.responses.get_provider_credential")
 def test_responses_endpoint_success(
-    mock_get_credential,
-    mock_openai,
-    db,
+    mock_get_credential, mock_openai, db, normal_user_api_key_headers: dict[str, str]
 ):
     """Test the /responses endpoint for successful response creation."""
     # Setup mock credentials
@@ -44,17 +42,15 @@ def test_responses_endpoint_success(
     if not glific_project:
         pytest.skip("Glific project not found in the database")
 
-    # Use the original API key from seed data (not the encrypted one)
-    original_api_key = "ApiKey No3x47A5qoIGhm0kVKjQ77dhCqEdWRIQZlEPzzzh7i8"
-
-    headers = {"X-API-KEY": original_api_key}
     request_data = {
         "assistant_id": "assistant_glific",
         "question": "What is Glific?",
         "callback_url": "http://example.com/callback",
     }
 
-    response = client.post("/responses", json=request_data, headers=headers)
+    response = client.post(
+        "/responses", json=request_data, headers=normal_user_api_key_headers
+    )
     assert response.status_code == 200
     response_json = response.json()
     assert response_json["success"] is True
@@ -70,6 +66,7 @@ def test_responses_endpoint_without_vector_store(
     mock_get_credential,
     mock_openai,
     db,
+    normal_user_api_key_headers,
 ):
     """Test the /responses endpoint when assistant has no vector store configured."""
     # Setup mock credentials
@@ -103,17 +100,15 @@ def test_responses_endpoint_without_vector_store(
     if not glific_project:
         pytest.skip("Glific project not found in the database")
 
-    # Use the original API key from seed data
-    original_api_key = "ApiKey No3x47A5qoIGhm0kVKjQ77dhCqEdWRIQZlEPzzzh7i8"
-
-    headers = {"X-API-KEY": original_api_key}
     request_data = {
         "assistant_id": "assistant_123",
         "question": "What is Glific?",
         "callback_url": "http://example.com/callback",
     }
 
-    response = client.post("/responses", json=request_data, headers=headers)
+    response = client.post(
+        "/responses", json=request_data, headers=normal_user_api_key_headers
+    )
     assert response.status_code == 200
     response_json = response.json()
     assert response_json["success"] is True

backend/app/tests/conftest.py

@@ -1,6 +1,8 @@
 from collections.abc import Generator
 
 import pytest
+import time
+
 from fastapi.testclient import TestClient
 from sqlmodel import Session
 from sqlalchemy import event
@@ -10,7 +12,7 @@
 from app.api.deps import get_db
 from app.main import app
 from app.tests.utils.user import authentication_token_from_email
-from app.tests.utils.utils import get_superuser_token_headers
+from app.tests.utils.utils import get_superuser_token_headers, get_api_key_by_email
 from app.seed_data.seed_data import seed_database
 
 
@@ -59,3 +61,15 @@ def normal_user_token_headers(client: TestClient, db: Session) -> dict[str, str]
     return authentication_token_from_email(
         client=client, email=settings.EMAIL_TEST_USER, db=db
     )
+
+
+@pytest.fixture(scope="function")
+def superuser_api_key_headers(db: Session) -> dict[str, str]:
+    api_key = get_api_key_by_email(db, settings.FIRST_SUPERUSER)
+    return {"X-API-KEY": api_key}
+
+
+@pytest.fixture(scope="function")
+def normal_user_api_key_headers(db: Session) -> dict[str, str]:
+    api_key = get_api_key_by_email(db, settings.EMAIL_TEST_USER)
+    return {"X-API-KEY": api_key}

backend/app/tests/crud/test_api_key.py

@@ -98,3 +98,20 @@ def test_get_api_keys_by_project(db: Session) -> None:
     assert retrieved_key.id == created_key.id
     assert retrieved_key.project_id == project.id
     assert retrieved_key.key.startswith("ApiKey ")
+
+
+def test_get_api_key_by_user_id(db: Session) -> None:
+    user = create_random_user(db)
+    project = create_test_project(db)
+
+    created_key = api_key_crud.create_api_key(
+        db, project.organization_id, user.id, project.id
+    )
+
+    retrieved_key = api_key_crud.get_api_key_by_user_id(db, user.id)
+
+    assert retrieved_key is not None
+
+    assert retrieved_key.id == created_key.id
+    assert retrieved_key.user_id == user.id
+    assert retrieved_key.key.startswith("ApiKey ")

backend/app/tests/utils/document.py

@@ -113,18 +113,18 @@ def append(self, doc: Document, suffix: str = None):
 @dataclass
 class WebCrawler:
     client: TestClient
-    superuser_token_headers: dict[str, str]
+    normal_user_api_key_headers: dict[str, str]
 
     def get(self, route: Route):
         return self.client.get(
             str(route),
-            headers=self.superuser_token_headers,
+            headers=self.normal_user_api_key_headers,
         )
 
     def delete(self, route: Route):
         return self.client.delete(
             str(route),
-            headers=self.superuser_token_headers,
+            headers=self.normal_user_api_key_headers,
         )
 
 
@@ -158,5 +158,5 @@ def to_dict(self):
 
 
 @pytest.fixture
-def crawler(client: TestClient, superuser_token_headers: dict[str, str]):
-    return WebCrawler(client, superuser_token_headers)
+def crawler(client: TestClient, normal_user_api_key_headers: dict[str, str]):
+    return WebCrawler(client, normal_user_api_key_headers)