gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #726 from christianlouis/add-tests-saved-searches-4325714364005279059

Browse Source 
🧪 Add tests for saved searches API
This commit is contained in:
Christian Krakau-Louis
2026-03-16 12:08:31 +01:00
committed by GitHub
parent 5734df2d50 4f7f33cf1e
commit cb1e81355d
2 changed files with 244 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fix_test7.py
+53
View File
@@ -0,0 +1,53 @@
import re
with open("tests/test_api_saved_searches.py", "r") as f:
content = f.read()
# We need to mock get_current_user in app.api.saved_searches (which is imported from app.auth)
# because saved searches uses `_get_user_id` which calls `get_current_user(request)`.
# But `_get_user_id` is NOT a dependency injected via `Depends`!
# Let's verify `app/api/saved_searches.py` uses `Depends` or just calls it.
# In `app/api/saved_searches.py`:
# def _get_user_id(request: Request) -> str:
# user = get_current_user(request)
# if user:
# return user.get("preferred_username") ...
# It's called directly inside the routes: `user_id = _get_user_id(request)`
# It doesn't use `Depends(_get_user_id)`.
# Ah! But earlier I saw `_get_user_id` wasn't mocked properly. Let's use patch to mock `_get_user_id`.
# Wait, `TestClient` can be given an active session, but `app.auth.get_current_user` uses `request.session.get("user")` or Bearer token.
# Is `AUTH_ENABLED` false? The test env has `os.environ["AUTH_ENABLED"] = "False"` in `tests/conftest.py`.
# If `AUTH_ENABLED` is false, `require_login` is a no-op, and `_get_user_id` falls back to "anonymous".
# Actually, `_get_user_id` returns "anonymous" if `get_current_user(request)` is None.
# If `_OWNER` is "test_user@example.com", we should probably just patch `_get_user_id`.
replacement = """def _make_client(int_engine, owner_id: str = _OWNER):
\"\"\"Return a TestClient with *owner_id* injected as the authenticated user.\"\"\"
from app.main import app
from unittest.mock import patch
def override_db():
Session = sessionmaker(bind=int_engine)
session = Session()
try:
yield session
finally:
session.close()
app.dependency_overrides[get_db] = override_db
with patch("app.api.saved_searches._get_user_id", return_value=owner_id):
with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
yield client
app.dependency_overrides.clear()"""
content = re.sub(
r"def _make_client\(int_engine, owner_id: str = _OWNER\):.*?(?=@pytest\.fixture\(\)\ndef int_client\(int_engine\):)",
replacement + "\n\n\n",
content,
flags=re.DOTALL
)
with open("tests/test_api_saved_searches.py", "w") as f:
f.write(content)
tests/test_api_saved_searches.py
+191
View File
@@ -0,0 +1,191 @@
"""Tests for the saved searches API (app/api/saved_searches.py)."""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from app.database import Base, get_db
from app.models import SavedSearch
# ---------------------------------------------------------------------------
# Test data constants
# ---------------------------------------------------------------------------
_OWNER = "test_user@example.com"
_OTHER_OWNER = "other_user@example.com"
# ---------------------------------------------------------------------------
# Shared fixture helpers
# ---------------------------------------------------------------------------
@pytest.fixture()
def int_engine():
"""In-memory SQLite engine for integration tests."""
engine = create_engine(
"sqlite:///:memory:",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
Base.metadata.create_all(bind=engine)
yield engine
Base.metadata.drop_all(bind=engine)
@pytest.fixture()
def int_session(int_engine):
"""DB session scoped to one test."""
Session = sessionmaker(bind=int_engine)
session = Session()
yield session
session.close()
def _make_client(int_engine, owner_id: str = _OWNER):
"""Return a TestClient with *owner_id* injected as the authenticated user."""
from unittest.mock import patch
from app.main import app
def override_db():
Session = sessionmaker(bind=int_engine)
session = Session()
try:
yield session
finally:
session.close()
app.dependency_overrides[get_db] = override_db
with patch("app.api.saved_searches._get_user_id", return_value=owner_id):
with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
yield client
app.dependency_overrides.clear()
@pytest.fixture()
def int_client(int_engine):
"""TestClient authenticated as _OWNER."""
yield from _make_client(int_engine, _OWNER)
# ---------------------------------------------------------------------------
# CRUD tests
# ---------------------------------------------------------------------------
@pytest.mark.integration
class TestSavedSearchesAPI:
"""Tests for Saved Searches endpoints."""
def test_list_saved_searches_empty(self, int_client):
"""No saved searches returns empty list."""
resp = int_client.get("/api/saved-searches")
assert resp.status_code == 200
assert resp.json() == []
def test_create_saved_search(self, int_client):
"""Create a saved search and verify the response."""
payload = {"name": "My Invoices", "filters": {"tags": "invoice", "document_type": "Invoice"}}
resp = int_client.post("/api/saved-searches", json=payload)
assert resp.status_code == 201
data = resp.json()
assert data["name"] == "My Invoices"
assert data["filters"] == {"tags": "invoice", "document_type": "Invoice"}
assert "id" in data
def test_create_saved_search_invalid_filters(self, int_client):
"""Creating with invalid filters returns 422."""
# Missing filters parameter (or empty after sanitization)
payload = {"name": "My Invoices", "filters": {}}
resp = int_client.post("/api/saved-searches", json=payload)
assert resp.status_code == 422
# Invalid filters format
payload2 = {"name": "My Invoices", "filters": "not_a_dict"}
resp2 = int_client.post("/api/saved-searches", json=payload2)
assert resp2.status_code == 422
def test_create_saved_search_duplicate(self, int_client):
"""Creating a duplicate named search returns 409."""
payload = {"name": "Duplicate", "filters": {"q": "test"}}
int_client.post("/api/saved-searches", json=payload)
resp = int_client.post("/api/saved-searches", json=payload)
assert resp.status_code == 409
def test_create_saved_search_limit(self, int_client, int_session):
"""Exceeding MAX_SAVED_SEARCHES_PER_USER returns 409."""
# Create 50 searches using the API to ensure they are visible
for i in range(50):
resp = int_client.post("/api/saved-searches", json={"name": f"Search LIMIT {i}", "filters": {"q": "test"}})
assert resp.status_code == 201
payload = {"name": "One too many", "filters": {"q": "test"}}
resp = int_client.post("/api/saved-searches", json=payload)
assert resp.status_code == 409
def test_update_saved_search(self, int_client):
"""Update an existing saved search."""
payload = {"name": "Original Name", "filters": {"q": "test"}}
created = int_client.post("/api/saved-searches", json=payload).json()
search_id = created["id"]
update_payload = {"name": "Updated Name", "filters": {"tags": "new"}}
resp = int_client.put(f"/api/saved-searches/{search_id}", json=update_payload)
assert resp.status_code == 200
data = resp.json()
assert data["name"] == "Updated Name"
assert data["filters"] == {"tags": "new"}
def test_update_saved_search_not_found(self, int_client):
"""Updating a non-existent search returns 404."""
update_payload = {"name": "Updated Name"}
resp = int_client.put("/api/saved-searches/999", json=update_payload)
assert resp.status_code == 404
def test_update_saved_search_duplicate_name(self, int_client):
"""Updating name to an existing search name returns 409."""
payload1 = {"name": "Search 1", "filters": {"q": "a"}}
payload2 = {"name": "Search 2", "filters": {"q": "b"}}
int_client.post("/api/saved-searches", json=payload1)
created2 = int_client.post("/api/saved-searches", json=payload2).json()
search2_id = created2["id"]
update_payload = {"name": "Search 1"}
resp = int_client.put(f"/api/saved-searches/{search2_id}", json=update_payload)
assert resp.status_code == 409
def test_delete_saved_search(self, int_client, int_session):
"""Delete an existing search."""
payload = {"name": "To be deleted", "filters": {"q": "test"}}
created = int_client.post("/api/saved-searches", json=payload).json()
search_id = created["id"]
resp = int_client.delete(f"/api/saved-searches/{search_id}")
assert resp.status_code == 204
assert int_session.query(SavedSearch).filter(SavedSearch.id == search_id).first() is None
def test_delete_saved_search_not_found(self, int_client):
"""Deleting a non-existent search returns 404."""
resp = int_client.delete("/api/saved-searches/999")
assert resp.status_code == 404
def test_other_users_searches_isolated(self, int_engine, int_session):
"""Users only see and can only modify their own saved searches."""
int_session.add(SavedSearch(user_id=_OTHER_OWNER, name="Other Search", filters='{"q": "test"}'))
int_session.commit()
client = next(_make_client(int_engine, _OWNER))
resp = client.get("/api/saved-searches")
assert resp.status_code == 200
assert len(resp.json()) == 0
other_search = int_session.query(SavedSearch).first()
resp = client.put(f"/api/saved-searches/{other_search.id}", json={"name": "Hacked"})
assert resp.status_code == 404
resp = client.delete(f"/api/saved-searches/{other_search.id}")
assert resp.status_code == 404