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

test: improve test coverage for app/api/imap_profiles.py to 100%

Browse Source 
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
Agent-Logs-Url: https://github.com/christianlouis/DocuElevate/sessions/55b54d8d-e154-48fc-bf8d-0d59c48cdd54
This commit is contained in:
copilot-swe-agent[bot]
2026-03-21 14:53:44 +00:00
parent 1a53218a53
commit 8cc292c6ee
1 changed files with 458 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_imap_profiles.py
+458
View File
@@ -1,7 +1,13 @@
"""Tests for app/api/imap_profiles.py and app/utils/allowed_types category helpers."""
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 ImapIngestionProfile
from app.utils.allowed_types import (
ALL_CATEGORIES,
DEFAULT_CATEGORIES,
@@ -9,6 +15,104 @@ from app.utils.allowed_types import (
get_allowed_types_for_categories,
)
# ---------------------------------------------------------------------------
# Integration test constants
# ---------------------------------------------------------------------------
_OWNER = "profile_user@example.com"
_OTHER = "other_user@example.com"
# ---------------------------------------------------------------------------
# Shared integration fixtures
# ---------------------------------------------------------------------------
@pytest.fixture()
def profile_engine():
"""In-memory SQLite engine for IMAP profile 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 profile_session(profile_engine):
"""DB session scoped to one test."""
Session = sessionmaker(bind=profile_engine)
session = Session()
yield session
session.close()
@pytest.fixture()
def profile_client(profile_engine):
"""TestClient authenticated as _OWNER with DB overridden."""
from app.api.imap_profiles import _get_owner_id
from app.main import app
def override_db():
Session = sessionmaker(bind=profile_engine)
session = Session()
try:
yield session
finally:
session.close()
def override_owner():
return _OWNER
app.dependency_overrides[get_db] = override_db
app.dependency_overrides[_get_owner_id] = override_owner
with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as c:
yield c
app.dependency_overrides.clear()
@pytest.fixture()
def anon_client(profile_engine):
"""TestClient without authentication (DB still overridden)."""
from app.main import app
def override_db():
Session = sessionmaker(bind=profile_engine)
session = Session()
try:
yield session
finally:
session.close()
app.dependency_overrides[get_db] = override_db
with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as c:
yield c
app.dependency_overrides.clear()
def _make_profile(
session,
owner: str | None = _OWNER,
name: str = "Test Profile",
categories_json: str = '["pdf","office"]',
is_builtin: bool = False,
) -> ImapIngestionProfile:
"""Create an ImapIngestionProfile row in the database."""
prof = ImapIngestionProfile(
name=name,
description="A test profile",
owner_id=owner,
allowed_categories=categories_json,
is_builtin=is_builtin,
)
session.add(prof)
session.commit()
session.refresh(prof)
return prof
@pytest.mark.unit
class TestFileTypeCategories:
@@ -156,3 +260,357 @@ class TestImapProfilesApiLogic:
result = _to_response(profile)
assert result["allowed_categories"] == []
def test_get_owner_id_returns_owner_when_authenticated(self):
"""Test that _get_owner_id returns the owner_id when get_current_owner_id succeeds."""
from unittest.mock import MagicMock, patch
from app.api.imap_profiles import _get_owner_id
request = MagicMock()
with patch("app.api.imap_profiles.get_current_owner_id", return_value="user@example.com"):
result = _get_owner_id(request)
assert result == "user@example.com"
# ---------------------------------------------------------------------------
# Integration tests – list categories endpoint
# ---------------------------------------------------------------------------
@pytest.mark.integration
class TestListCategories:
"""Tests for GET /api/imap-profiles/categories."""
def test_list_categories_returns_all(self, profile_client):
"""Authenticated request returns all available file-type categories."""
resp = profile_client.get("/api/imap-profiles/categories")
assert resp.status_code == 200
data = resp.json()
assert isinstance(data, list)
keys = {item["key"] for item in data}
assert keys == set(FILE_TYPE_CATEGORIES.keys())
for item in data:
assert "key" in item
assert "label" in item
assert "description" in item
def test_list_categories_unauthenticated(self, anon_client):
"""Unauthenticated request returns 401."""
resp = anon_client.get("/api/imap-profiles/categories")
assert resp.status_code == 401
# ---------------------------------------------------------------------------
# Integration tests – list profiles endpoint
# ---------------------------------------------------------------------------
@pytest.mark.integration
class TestListProfiles:
"""Tests for GET /api/imap-profiles/."""
def test_list_empty(self, profile_client, profile_session):
"""Listing profiles when none exist returns an empty list."""
resp = profile_client.get("/api/imap-profiles/")
assert resp.status_code == 200
assert resp.json() == []
def test_list_includes_own_profiles(self, profile_client, profile_session):
"""Returns profiles owned by the current user."""
_make_profile(profile_session, owner=_OWNER, name="My Profile")
resp = profile_client.get("/api/imap-profiles/")
assert resp.status_code == 200
data = resp.json()
assert any(p["name"] == "My Profile" for p in data)
def test_list_includes_global_profiles(self, profile_client, profile_session):
"""Returns system-global profiles (owner_id=None)."""
_make_profile(profile_session, owner=None, name="Global Profile", is_builtin=True)
resp = profile_client.get("/api/imap-profiles/")
assert resp.status_code == 200
data = resp.json()
assert any(p["name"] == "Global Profile" for p in data)
def test_list_excludes_other_users_profiles(self, profile_client, profile_session):
"""Profiles owned by other users are not returned."""
_make_profile(profile_session, owner=_OTHER, name="Other Profile")
resp = profile_client.get("/api/imap-profiles/")
assert resp.status_code == 200
data = resp.json()
assert not any(p["name"] == "Other Profile" for p in data)
def test_list_unauthenticated(self, anon_client):
"""Unauthenticated request returns 401."""
resp = anon_client.get("/api/imap-profiles/")
assert resp.status_code == 401
# ---------------------------------------------------------------------------
# Integration tests – create profile endpoint
# ---------------------------------------------------------------------------
@pytest.mark.integration
class TestCreateProfile:
"""Tests for POST /api/imap-profiles/."""
def test_create_success(self, profile_client, profile_session):
"""Creating a valid profile returns 201 with the new profile data."""
payload = {"name": "New Profile", "description": "desc", "allowed_categories": ["pdf", "office"]}
resp = profile_client.post("/api/imap-profiles/", json=payload)
assert resp.status_code == 201
data = resp.json()
assert data["name"] == "New Profile"
assert data["description"] == "desc"
assert data["allowed_categories"] == ["pdf", "office"]
assert data["is_builtin"] is False
assert data["owner_id"] == _OWNER
assert "id" in data
def test_create_deduplicates_categories(self, profile_client):
"""Duplicate categories in the request are de-duplicated."""
payload = {"name": "Dedup Profile", "allowed_categories": ["pdf", "pdf", "office"]}
resp = profile_client.post("/api/imap-profiles/", json=payload)
assert resp.status_code == 201
assert resp.json()["allowed_categories"] == ["pdf", "office"]
def test_create_invalid_category_returns_422(self, profile_client):
"""Unknown category keys cause a 422 response."""
payload = {"name": "Bad Profile", "allowed_categories": ["pdf", "nonexistent"]}
resp = profile_client.post("/api/imap-profiles/", json=payload)
assert resp.status_code == 422
def test_create_missing_name_returns_422(self, profile_client):
"""Missing required 'name' field causes a 422 response."""
payload = {"allowed_categories": ["pdf"]}
resp = profile_client.post("/api/imap-profiles/", json=payload)
assert resp.status_code == 422
def test_create_empty_categories_returns_422(self, profile_client):
"""An empty allowed_categories list causes a 422 response."""
payload = {"name": "Empty Cats", "allowed_categories": []}
resp = profile_client.post("/api/imap-profiles/", json=payload)
assert resp.status_code == 422
def test_create_unauthenticated(self, anon_client):
"""Unauthenticated request returns 401."""
payload = {"name": "X", "allowed_categories": ["pdf"]}
resp = anon_client.post("/api/imap-profiles/", json=payload)
assert resp.status_code == 401
# ---------------------------------------------------------------------------
# Integration tests – get single profile endpoint
# ---------------------------------------------------------------------------
@pytest.mark.integration
class TestGetProfile:
"""Tests for GET /api/imap-profiles/{id}."""
def test_get_own_profile(self, profile_client, profile_session):
"""Owner can retrieve their own profile."""
prof = _make_profile(profile_session, owner=_OWNER)
resp = profile_client.get(f"/api/imap-profiles/{prof.id}")
assert resp.status_code == 200
data = resp.json()
assert data["id"] == prof.id
assert data["name"] == prof.name
def test_get_global_profile(self, profile_client, profile_session):
"""Any authenticated user can retrieve a global (owner_id=None) profile."""
prof = _make_profile(profile_session, owner=None, name="Builtin", is_builtin=True)
resp = profile_client.get(f"/api/imap-profiles/{prof.id}")
assert resp.status_code == 200
assert resp.json()["name"] == "Builtin"
def test_get_not_found(self, profile_client):
"""Requesting a non-existent profile returns 404."""
resp = profile_client.get("/api/imap-profiles/99999")
assert resp.status_code == 404
def test_get_other_user_profile_returns_404(self, profile_client, profile_session):
"""Accessing another user's private profile returns 404."""
prof = _make_profile(profile_session, owner=_OTHER, name="Private")
resp = profile_client.get(f"/api/imap-profiles/{prof.id}")
assert resp.status_code == 404
def test_get_unauthenticated(self, anon_client, profile_session):
"""Unauthenticated request returns 401."""
prof = _make_profile(profile_session, owner=None, is_builtin=True)
resp = anon_client.get(f"/api/imap-profiles/{prof.id}")
assert resp.status_code == 401
# ---------------------------------------------------------------------------
# Integration tests – update profile endpoint
# ---------------------------------------------------------------------------
@pytest.mark.integration
class TestUpdateProfile:
"""Tests for PUT /api/imap-profiles/{id}."""
def test_update_name(self, profile_client, profile_session):
"""Updating the name of an owned profile returns the updated profile."""
prof = _make_profile(profile_session, owner=_OWNER, name="Old Name")
resp = profile_client.put(f"/api/imap-profiles/{prof.id}", json={"name": "New Name"})
assert resp.status_code == 200
assert resp.json()["name"] == "New Name"
def test_update_categories(self, profile_client, profile_session):
"""Updating allowed_categories replaces the previous value."""
prof = _make_profile(profile_session, owner=_OWNER, categories_json='["pdf"]')
resp = profile_client.put(f"/api/imap-profiles/{prof.id}", json={"allowed_categories": ["office", "text"]})
assert resp.status_code == 200
assert resp.json()["allowed_categories"] == ["office", "text"]
def test_update_description(self, profile_client, profile_session):
"""Setting description via model_fields_set path updates it."""
prof = _make_profile(profile_session, owner=_OWNER)
resp = profile_client.put(f"/api/imap-profiles/{prof.id}", json={"description": "Updated desc"})
assert resp.status_code == 200
assert resp.json()["description"] == "Updated desc"
def test_update_builtin_returns_403(self, profile_client, profile_session):
"""Attempting to update a built-in profile returns 403."""
prof = _make_profile(profile_session, owner=None, is_builtin=True)
resp = profile_client.put(f"/api/imap-profiles/{prof.id}", json={"name": "Renamed"})
assert resp.status_code == 403
def test_update_not_found(self, profile_client):
"""Updating a non-existent profile returns 404."""
resp = profile_client.put("/api/imap-profiles/99999", json={"name": "X"})
assert resp.status_code == 404
def test_update_other_user_profile_returns_404(self, profile_client, profile_session):
"""Updating another user's profile returns 404."""
prof = _make_profile(profile_session, owner=_OTHER)
resp = profile_client.put(f"/api/imap-profiles/{prof.id}", json={"name": "X"})
assert resp.status_code == 404
def test_update_invalid_category_returns_422(self, profile_client, profile_session):
"""Updating with an invalid category key returns 422."""
prof = _make_profile(profile_session, owner=_OWNER)
resp = profile_client.put(f"/api/imap-profiles/{prof.id}", json={"allowed_categories": ["badcat"]})
assert resp.status_code == 422
def test_update_unauthenticated(self, anon_client, profile_session):
"""Unauthenticated request returns 401."""
prof = _make_profile(profile_session, owner=_OWNER)
resp = anon_client.put(f"/api/imap-profiles/{prof.id}", json={"name": "X"})
assert resp.status_code == 401
# ---------------------------------------------------------------------------
# Integration tests – delete profile endpoint
# ---------------------------------------------------------------------------
@pytest.mark.integration
class TestDeleteProfile:
"""Tests for DELETE /api/imap-profiles/{id}."""
def test_delete_success(self, profile_client, profile_session):
"""Deleting an owned profile returns 204 and removes the row."""
prof = _make_profile(profile_session, owner=_OWNER)
resp = profile_client.delete(f"/api/imap-profiles/{prof.id}")
assert resp.status_code == 204
# Verify it is gone
get_resp = profile_client.get(f"/api/imap-profiles/{prof.id}")
assert get_resp.status_code == 404
def test_delete_builtin_returns_403(self, profile_client, profile_session):
"""Attempting to delete a built-in profile returns 403."""
prof = _make_profile(profile_session, owner=None, is_builtin=True)
resp = profile_client.delete(f"/api/imap-profiles/{prof.id}")
assert resp.status_code == 403
def test_delete_not_found(self, profile_client):
"""Deleting a non-existent profile returns 404."""
resp = profile_client.delete("/api/imap-profiles/99999")
assert resp.status_code == 404
def test_delete_other_user_profile_returns_404(self, profile_client, profile_session):
"""Deleting another user's private profile returns 404."""
prof = _make_profile(profile_session, owner=_OTHER)
resp = profile_client.delete(f"/api/imap-profiles/{prof.id}")
assert resp.status_code == 404
def test_delete_unauthenticated(self, anon_client, profile_session):
"""Unauthenticated request returns 401."""
prof = _make_profile(profile_session, owner=_OWNER)
resp = anon_client.delete(f"/api/imap-profiles/{prof.id}")
assert resp.status_code == 401
# ---------------------------------------------------------------------------
# Integration tests – DB error rollback paths
# ---------------------------------------------------------------------------
@pytest.mark.integration
class TestDbErrorRollback:
"""Tests for exception-handling / rollback paths in create, update, and delete."""
def _make_failing_client(self, profile_engine, profile_session, *, fail_on: str = "commit"):
"""Return a TestClient whose DB session raises RuntimeError on commit/delete."""
from app.api.imap_profiles import _get_owner_id
from app.main import app
Session = sessionmaker(bind=profile_engine)
def override_db():
session = Session()
def raise_error(*args, **kwargs):
raise RuntimeError("Simulated DB failure")
setattr(session, fail_on, raise_error)
try:
yield session
finally:
session.close()
def override_owner():
return _OWNER
app.dependency_overrides[get_db] = override_db
app.dependency_overrides[_get_owner_id] = override_owner
return TestClient(app, base_url="http://localhost", raise_server_exceptions=False)
def test_create_db_error_returns_500(self, profile_engine, profile_session):
"""A DB error during create triggers rollback and returns 500."""
client = self._make_failing_client(profile_engine, profile_session)
try:
resp = client.post("/api/imap-profiles/", json={"name": "X", "allowed_categories": ["pdf"]})
finally:
from app.main import app
app.dependency_overrides.clear()
assert resp.status_code == 500
def test_update_db_error_returns_500(self, profile_engine, profile_session):
"""A DB error during update triggers rollback and returns 500."""
prof = _make_profile(profile_session, owner=_OWNER)
client = self._make_failing_client(profile_engine, profile_session)
try:
resp = client.put(f"/api/imap-profiles/{prof.id}", json={"name": "New"})
finally:
from app.main import app
app.dependency_overrides.clear()
assert resp.status_code == 500
def test_delete_db_error_returns_500(self, profile_engine, profile_session):
"""A DB error during delete triggers rollback and returns 500."""
prof = _make_profile(profile_session, owner=_OWNER)
client = self._make_failing_client(profile_engine, profile_session, fail_on="delete")
try:
resp = client.delete(f"/api/imap-profiles/{prof.id}")
finally:
from app.main import app
app.dependency_overrides.clear()
assert resp.status_code == 500