justinzeus/scholarr
1
0
Fork 0
Code Issues 1 Pull requests 18 Projects Releases Packages Wiki Activity Actions

removed old UI

Browse source
This commit is contained in:
Justin Visser 2026-02-17 15:21:08 +01:00
parent 4433d7d2c4
commit f71841e922
62 changed files with 411 additions and 5815 deletions
Download patch file Download diff file Expand all files Collapse all files

30
tests/integration/helpers.py
View file

@ -1,36 +1,29 @@
from __future__ import annotations
import re
from fastapi.testclient import TestClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from app.auth.security import PasswordService
CSRF_TOKEN_PATTERN = re.compile(r'name="csrf_token" value="([^"]+)"')
def extract_csrf_token(html: str) -> str:
match = CSRF_TOKEN_PATTERN.search(html)
assert match is not None
return match.group(1)
def login_user(client: TestClient, *, email: str, password: str) -> None:
login_page = client.get("/login")
csrf_token = extract_csrf_token(login_page.text)
bootstrap_response = client.get("/api/v1/auth/csrf")
assert bootstrap_response.status_code == 200
csrf_token = bootstrap_response.json()["data"]["csrf_token"]
assert isinstance(csrf_token, str) and csrf_token
response = client.post(
"/login",
data={
"/api/v1/auth/login",
json={
"email": email,
"password": password,
"csrf_token": csrf_token,
},
headers={"X-CSRF-Token": csrf_token},
follow_redirects=False,
)
assert response.status_code == 303
assert response.headers["location"] == "/"
assert response.status_code == 200
payload = response.json()
assert payload["data"]["authenticated"] is True
async def insert_user(
@ -60,4 +53,3 @@ async def insert_user(
user_id = int(result.scalar_one())
await db_session.commit()
return user_id

200
tests/integration/test_admin_user_management.py
View file

@ -1,200 +0,0 @@
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from app.main import app
from tests.integration.helpers import extract_csrf_token, insert_user, login_user
@pytest.mark.integration
@pytest.mark.db
@pytest.mark.asyncio
async def test_users_page_is_admin_only(db_session: AsyncSession) -> None:
await insert_user(
db_session,
email="member@example.com",
password="member-pass",
is_admin=False,
)
client = TestClient(app)
login_user(client, email="member@example.com", password="member-pass")
response = client.get("/users")
assert response.status_code == 403
assert response.json()["detail"] == "Admin access required."
@pytest.mark.integration
@pytest.mark.db
@pytest.mark.asyncio
async def test_non_admin_dashboard_hides_users_nav(db_session: AsyncSession) -> None:
await insert_user(
db_session,
email="member@example.com",
password="member-pass",
is_admin=False,
)
client = TestClient(app)
login_user(client, email="member@example.com", password="member-pass")
dashboard = client.get("/")
assert dashboard.status_code == 200
assert ">Users<" not in dashboard.text
@pytest.mark.integration
@pytest.mark.db
@pytest.mark.asyncio
async def test_admin_dashboard_shows_users_nav(db_session: AsyncSession) -> None:
await insert_user(
db_session,
email="admin@example.com",
password="admin-pass",
is_admin=True,
)
client = TestClient(app)
login_user(client, email="admin@example.com", password="admin-pass")
dashboard = client.get("/")
assert dashboard.status_code == 200
assert ">Users<" in dashboard.text
@pytest.mark.integration
@pytest.mark.db
@pytest.mark.asyncio
async def test_admin_can_create_and_deactivate_user(db_session: AsyncSession) -> None:
await insert_user(
db_session,
email="admin@example.com",
password="admin-pass",
is_admin=True,
)
client = TestClient(app)
login_user(client, email="admin@example.com", password="admin-pass")
users_page = client.get("/users")
csrf_token = extract_csrf_token(users_page.text)
create_response = client.post(
"/users",
data={
"email": "new-user@example.com",
"password": "new-user-pass",
"csrf_token": csrf_token,
},
follow_redirects=False,
)
assert create_response.status_code == 303
assert create_response.headers["location"].startswith("/users")
created_user_id_result = await db_session.execute(
text("SELECT id FROM users WHERE email = 'new-user@example.com'")
)
created_user_id = int(created_user_id_result.scalar_one())
users_page_after_create = client.get("/users")
csrf_after_create = extract_csrf_token(users_page_after_create.text)
toggle_response = client.post(
f"/users/{created_user_id}/toggle-active",
data={"csrf_token": csrf_after_create},
follow_redirects=False,
)
assert toggle_response.status_code == 303
status_result = await db_session.execute(
text("SELECT is_active FROM users WHERE id = :user_id"),
{"user_id": created_user_id},
)
assert status_result.scalar_one() is False
@pytest.mark.integration
@pytest.mark.db
@pytest.mark.asyncio
async def test_admin_can_reset_user_password(db_session: AsyncSession) -> None:
target_user_id = await insert_user(
db_session,
email="target@example.com",
password="old-password",
is_admin=False,
)
await insert_user(
db_session,
email="admin@example.com",
password="admin-pass",
is_admin=True,
)
client = TestClient(app)
login_user(client, email="admin@example.com", password="admin-pass")
users_page = client.get("/users")
csrf_token = extract_csrf_token(users_page.text)
reset_response = client.post(
f"/users/{target_user_id}/reset-password",
data={
"new_password": "new-password",
"csrf_token": csrf_token,
},
follow_redirects=False,
)
assert reset_response.status_code == 303
users_page_after_reset = client.get("/users")
logout_csrf = extract_csrf_token(users_page_after_reset.text)
client.post(
"/logout",
data={"csrf_token": logout_csrf},
follow_redirects=False,
)
failed_login_page = client.get("/login")
failed_login_csrf = extract_csrf_token(failed_login_page.text)
failed_login = client.post(
"/login",
data={
"email": "target@example.com",
"password": "old-password",
"csrf_token": failed_login_csrf,
},
follow_redirects=False,
)
assert failed_login.status_code == 401
login_user(client, email="target@example.com", password="new-password")
@pytest.mark.integration
@pytest.mark.db
@pytest.mark.asyncio
async def test_admin_cannot_deactivate_self(db_session: AsyncSession) -> None:
admin_user_id = await insert_user(
db_session,
email="admin@example.com",
password="admin-pass",
is_admin=True,
)
client = TestClient(app)
login_user(client, email="admin@example.com", password="admin-pass")
users_page = client.get("/users")
csrf_token = extract_csrf_token(users_page.text)
response = client.post(
f"/users/{admin_user_id}/toggle-active",
data={"csrf_token": csrf_token},
follow_redirects=False,
)
assert response.status_code == 303
assert response.headers["location"].startswith("/users")
result = await db_session.execute(
text("SELECT is_active FROM users WHERE id = :user_id"),
{"user_id": admin_user_id},
)
assert result.scalar_one() is True

109
tests/integration/test_auth_flow.py
View file

@ -1,109 +0,0 @@
import re
import pytest
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi.testclient import TestClient
from app.auth.security import PasswordService
from app.main import app
CSRF_TOKEN_PATTERN = re.compile(r'name="csrf_token" value="([^"]+)"')
def _extract_csrf_token(html: str) -> str:
match = CSRF_TOKEN_PATTERN.search(html)
assert match is not None
return match.group(1)
@pytest.mark.integration
def test_dashboard_requires_authentication() -> None:
client = TestClient(app)
response = client.get("/", follow_redirects=False)
assert response.status_code == 303
assert response.headers["location"] == "/login"
@pytest.mark.integration
@pytest.mark.db
@pytest.mark.asyncio
async def test_login_with_valid_credentials_allows_access(db_session: AsyncSession) -> None:
password_service = PasswordService()
await db_session.execute(
text(
"""
INSERT INTO users (email, password_hash, is_active, is_admin)
VALUES (:email, :password_hash, :is_active, :is_admin)
"""
),
{
"email": "reader@example.com",
"password_hash": password_service.hash_password("correct-password"),
"is_active": True,
"is_admin": False,
},
)
await db_session.commit()
client = TestClient(app)
login_page = client.get("/login")
csrf_token = _extract_csrf_token(login_page.text)
login_response = client.post(
"/login",
data={
"email": "reader@example.com",
"password": "correct-password",
"csrf_token": csrf_token,
},
follow_redirects=False,
)
assert login_response.status_code == 303
assert login_response.headers["location"] == "/"
dashboard_response = client.get("/")
assert dashboard_response.status_code == 200
assert "reader@example.com" in dashboard_response.text
assert 'data-test="home-hero"' in dashboard_response.text
@pytest.mark.integration
@pytest.mark.db
@pytest.mark.asyncio
async def test_login_rejects_inactive_user(db_session: AsyncSession) -> None:
password_service = PasswordService()
await db_session.execute(
text(
"""
INSERT INTO users (email, password_hash, is_active, is_admin)
VALUES (:email, :password_hash, :is_active, :is_admin)
"""
),
{
"email": "inactive@example.com",
"password_hash": password_service.hash_password("correct-password"),
"is_active": False,
"is_admin": False,
},
)
await db_session.commit()
client = TestClient(app)
login_page = client.get("/login")
csrf_token = _extract_csrf_token(login_page.text)
login_response = client.post(
"/login",
data={
"email": "inactive@example.com",
"password": "correct-password",
"csrf_token": csrf_token,
},
follow_redirects=False,
)
assert login_response.status_code == 401
assert "Invalid email or password." in login_response.text

1151
tests/integration/test_manual_ingestion_flow.py
View file

File diff suppressed because it is too large Load diff

227
tests/integration/test_queue_ui.py
View file

@ -1,227 +0,0 @@
from __future__ import annotations
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from app.main import app
from tests.integration.helpers import extract_csrf_token, insert_user, login_user
@pytest.mark.integration
@pytest.mark.db
@pytest.mark.asyncio
async def test_runs_page_queue_actions_lifecycle(db_session: AsyncSession) -> None:
user_id = await insert_user(
db_session,
email="queue-ui@example.com",
password="queue-ui-password",
)
scholar_result = await db_session.execute(
text(
"""
INSERT INTO scholar_profiles (user_id, scholar_id, display_name, is_enabled)
VALUES (:user_id, :scholar_id, :display_name, true)
RETURNING id
"""
),
{
"user_id": user_id,
"scholar_id": "abcDEF123456",
"display_name": "Queue UI Scholar",
},
)
scholar_profile_id = int(scholar_result.scalar_one())
queue_result = await db_session.execute(
text(
"""
INSERT INTO ingestion_queue_items (
user_id,
scholar_profile_id,
resume_cstart,
reason,
status,
attempt_count,
next_attempt_dt,
last_error,
dropped_reason,
dropped_at
)
VALUES (
:user_id,
:scholar_profile_id,
200,
'dropped',
'dropped',
3,
NOW() + INTERVAL '30 minutes',
'captcha challenge',
'max_attempts_after_run',
NOW() - INTERVAL '1 minute'
)
RETURNING id
"""
),
{
"user_id": user_id,
"scholar_profile_id": scholar_profile_id,
},
)
queue_item_id = int(queue_result.scalar_one())
await db_session.commit()
client = TestClient(app)
login_user(client, email="queue-ui@example.com", password="queue-ui-password")
runs_page = client.get("/runs")
assert runs_page.status_code == 200
assert "Continuation Queue" in runs_page.text
assert "Queue UI Scholar" in runs_page.text
assert "dropped" in runs_page.text
assert "max_attempts_after_run" in runs_page.text
csrf_retry = extract_csrf_token(runs_page.text)
retry_response = client.post(
f"/runs/queue/{queue_item_id}/retry",
data={"csrf_token": csrf_retry},
follow_redirects=False,
)
assert retry_response.status_code == 303
queue_after_retry = await db_session.execute(
text(
"""
SELECT status, reason, attempt_count
FROM ingestion_queue_items
WHERE id = :queue_item_id
"""
),
{"queue_item_id": queue_item_id},
)
assert queue_after_retry.one() == ("queued", "manual_retry", 0)
runs_page_after_retry = client.get("/runs")
csrf_drop = extract_csrf_token(runs_page_after_retry.text)
drop_response = client.post(
f"/runs/queue/{queue_item_id}/drop",
data={"csrf_token": csrf_drop},
follow_redirects=False,
)
assert drop_response.status_code == 303
queue_after_drop = await db_session.execute(
text(
"""
SELECT status, reason, dropped_reason
FROM ingestion_queue_items
WHERE id = :queue_item_id
"""
),
{"queue_item_id": queue_item_id},
)
assert queue_after_drop.one() == ("dropped", "dropped", "manual_drop")
runs_page_after_drop = client.get("/runs")
csrf_clear = extract_csrf_token(runs_page_after_drop.text)
clear_response = client.post(
f"/runs/queue/{queue_item_id}/clear",
data={"csrf_token": csrf_clear},
follow_redirects=False,
)
assert clear_response.status_code == 303
queue_after_clear = await db_session.execute(
text("SELECT COUNT(*) FROM ingestion_queue_items WHERE id = :queue_item_id"),
{"queue_item_id": queue_item_id},
)
assert queue_after_clear.scalar_one() == 0
@pytest.mark.integration
@pytest.mark.db
@pytest.mark.asyncio
async def test_runs_queue_actions_are_tenant_scoped(db_session: AsyncSession) -> None:
user_a_id = await insert_user(
db_session,
email="queue-owner-a@example.com",
password="queue-owner-a-password",
)
user_b_id = await insert_user(
db_session,
email="queue-owner-b@example.com",
password="queue-owner-b-password",
)
scholar_result = await db_session.execute(
text(
"""
INSERT INTO scholar_profiles (user_id, scholar_id, display_name, is_enabled)
VALUES (:user_id, :scholar_id, :display_name, true)
RETURNING id
"""
),
{
"user_id": user_b_id,
"scholar_id": "zxyWVU654321",
"display_name": "Owner B Queue Scholar",
},
)
scholar_profile_id = int(scholar_result.scalar_one())
queue_result = await db_session.execute(
text(
"""
INSERT INTO ingestion_queue_items (
user_id,
scholar_profile_id,
resume_cstart,
reason,
status,
attempt_count,
next_attempt_dt
)
VALUES (
:user_id,
:scholar_profile_id,
0,
'max_pages_reached',
'queued',
0,
NOW()
)
RETURNING id
"""
),
{
"user_id": user_b_id,
"scholar_profile_id": scholar_profile_id,
},
)
queue_item_id = int(queue_result.scalar_one())
await db_session.commit()
client = TestClient(app)
login_user(client, email="queue-owner-a@example.com", password="queue-owner-a-password")
runs_page = client.get("/runs")
csrf_token = extract_csrf_token(runs_page.text)
forbidden_retry = client.post(
f"/runs/queue/{queue_item_id}/retry",
data={"csrf_token": csrf_token},
follow_redirects=False,
)
assert forbidden_retry.status_code == 404
unchanged = await db_session.execute(
text(
"""
SELECT status, reason, user_id
FROM ingestion_queue_items
WHERE id = :queue_item_id
"""
),
{"queue_item_id": queue_item_id},
)
assert unchanged.one() == ("queued", "max_pages_reached", user_b_id)
assert user_a_id != user_b_id

197
tests/integration/test_tenant_routes.py
View file

@ -1,197 +0,0 @@
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from app.main import app
from tests.integration.helpers import extract_csrf_token, insert_user, login_user
@pytest.mark.integration
@pytest.mark.db
@pytest.mark.asyncio
async def test_user_can_change_own_password(db_session: AsyncSession) -> None:
await insert_user(
db_session,
email="reader@example.com",
password="old-password",
)
client = TestClient(app)
login_user(client, email="reader@example.com", password="old-password")
account_page = client.get("/account/password")
csrf_token = extract_csrf_token(account_page.text)
wrong_current = client.post(
"/account/password",
data={
"current_password": "wrong-password",
"new_password": "new-password",
"confirm_password": "new-password",
"csrf_token": csrf_token,
},
follow_redirects=False,
)
assert wrong_current.status_code == 303
assert wrong_current.headers["location"].startswith("/account/password")
account_page_retry = client.get("/account/password")
retry_csrf = extract_csrf_token(account_page_retry.text)
success = client.post(
"/account/password",
data={
"current_password": "old-password",
"new_password": "new-password",
"confirm_password": "new-password",
"csrf_token": retry_csrf,
},
follow_redirects=False,
)
assert success.status_code == 303
account_page_after = client.get("/account/password")
logout_csrf = extract_csrf_token(account_page_after.text)
client.post("/logout", data={"csrf_token": logout_csrf}, follow_redirects=False)
failed_login_page = client.get("/login")
failed_login_csrf = extract_csrf_token(failed_login_page.text)
failed_login = client.post(
"/login",
data={
"email": "reader@example.com",
"password": "old-password",
"csrf_token": failed_login_csrf,
},
follow_redirects=False,
)
assert failed_login.status_code == 401
login_user(client, email="reader@example.com", password="new-password")
@pytest.mark.integration
@pytest.mark.db
@pytest.mark.asyncio
async def test_scholar_routes_are_tenant_scoped(db_session: AsyncSession) -> None:
user_a_id = await insert_user(
db_session,
email="owner-a@example.com",
password="owner-a-password",
)
user_b_id = await insert_user(
db_session,
email="owner-b@example.com",
password="owner-b-password",
)
client = TestClient(app)
login_user(client, email="owner-a@example.com", password="owner-a-password")
scholars_page = client.get("/scholars")
csrf_token = extract_csrf_token(scholars_page.text)
create_response = client.post(
"/scholars",
data={
"scholar_id": "abcDEF123456",
"display_name": "Owner A Scholar",
"csrf_token": csrf_token,
},
follow_redirects=False,
)
assert create_response.status_code == 303
owner_a_row = await db_session.execute(
text(
"""
SELECT user_id
FROM scholar_profiles
WHERE scholar_id = 'abcDEF123456'
"""
)
)
assert owner_a_row.scalar_one() == user_a_id
owner_b_profile = await db_session.execute(
text(
"""
INSERT INTO scholar_profiles (user_id, scholar_id, display_name, is_enabled)
VALUES (:user_id, :scholar_id, :display_name, :is_enabled)
RETURNING id
"""
),
{
"user_id": user_b_id,
"scholar_id": "zxcvbn654321",
"display_name": "Owner B Scholar",
"is_enabled": True,
},
)
owner_b_profile_id = int(owner_b_profile.scalar_one())
await db_session.commit()
scholars_page_after_insert = client.get("/scholars")
csrf_after_insert = extract_csrf_token(scholars_page_after_insert.text)
forbidden_toggle = client.post(
f"/scholars/{owner_b_profile_id}/toggle",
data={"csrf_token": csrf_after_insert},
follow_redirects=False,
)
assert forbidden_toggle.status_code == 404
owner_b_status = await db_session.execute(
text("SELECT is_enabled FROM scholar_profiles WHERE id = :profile_id"),
{"profile_id": owner_b_profile_id},
)
assert owner_b_status.scalar_one() is True
@pytest.mark.integration
@pytest.mark.db
@pytest.mark.asyncio
async def test_settings_updates_are_user_scoped(db_session: AsyncSession) -> None:
user_a_id = await insert_user(
db_session,
email="settings-a@example.com",
password="settings-a-password",
)
user_b_id = await insert_user(
db_session,
email="settings-b@example.com",
password="settings-b-password",
)
client = TestClient(app)
login_user(client, email="settings-a@example.com", password="settings-a-password")
settings_page = client.get("/settings")
csrf_token = extract_csrf_token(settings_page.text)
response = client.post(
"/settings",
data={
"auto_run_enabled": "on",
"run_interval_minutes": "45",
"request_delay_seconds": "7",
"csrf_token": csrf_token,
},
follow_redirects=False,
)
assert response.status_code == 303
user_a_settings = await db_session.execute(
text(
"""
SELECT auto_run_enabled, run_interval_minutes, request_delay_seconds
FROM user_settings
WHERE user_id = :user_id
"""
),
{"user_id": user_a_id},
)
assert user_a_settings.one() == (True, 45, 7)
user_b_settings = await db_session.execute(
text("SELECT COUNT(*) FROM user_settings WHERE user_id = :user_id"),
{"user_id": user_b_id},
)
assert user_b_settings.scalar_one() == 0

204
tests/unit/test_auth.py
View file

@ -1,204 +0,0 @@
import re
from collections.abc import AsyncIterator
from types import SimpleNamespace
from unittest.mock import AsyncMock
from fastapi.testclient import TestClient
import pytest
from app.auth.deps import get_auth_service, get_login_rate_limiter
from app.auth.rate_limit import SlidingWindowRateLimiter
from app.db.session import get_db_session
from app.main import app
CSRF_TOKEN_PATTERN = re.compile(r'name="csrf_token" value="([^"]+)"')
class StubAuthService:
def __init__(self, *, user: object | None) -> None:
self._user = user
async def authenticate_user(self, _db_session, *, email: str, password: str):
if self._user is None:
return None
if email.strip().lower() != str(self._user.email):
return None
if password != "correct-password":
return None
return self._user
def _extract_csrf_token(html: str) -> str:
match = CSRF_TOKEN_PATTERN.search(html)
assert match is not None
return match.group(1)
async def _override_db_session() -> AsyncIterator[object]:
yield object()
@pytest.fixture(autouse=True)
def clear_dependency_overrides() -> AsyncIterator[None]:
app.dependency_overrides.clear()
yield
app.dependency_overrides.clear()
def test_login_requires_csrf_token() -> None:
client = TestClient(app)
response = client.post(
"/login",
data={"email": "user@example.com", "password": "correct-password"},
follow_redirects=False,
)
assert response.status_code == 403
assert response.text == "CSRF token missing."
def test_successful_login_creates_session_and_allows_dashboard(monkeypatch) -> None:
limiter = SlidingWindowRateLimiter(max_attempts=5, window_seconds=60)
app.dependency_overrides[get_db_session] = _override_db_session
app.dependency_overrides[get_auth_service] = lambda: StubAuthService(
user=SimpleNamespace(id=1, email="user@example.com", is_admin=False)
)
app.dependency_overrides[get_login_rate_limiter] = lambda: limiter
client = TestClient(app)
monkeypatch.setattr(
"app.web.common.get_authenticated_user",
AsyncMock(
return_value=SimpleNamespace(
id=1,
email="user@example.com",
is_admin=False,
)
),
)
monkeypatch.setattr(
"app.web.routers.dashboard.publication_service.list_new_for_latest_run_for_user",
AsyncMock(return_value=[]),
)
monkeypatch.setattr(
"app.web.routers.dashboard.run_service.list_recent_runs_for_user",
AsyncMock(return_value=[]),
)
monkeypatch.setattr(
"app.web.routers.dashboard.run_service.queue_status_counts_for_user",
AsyncMock(return_value={"queued": 0, "retrying": 0, "dropped": 0}),
)
monkeypatch.setattr(
"app.web.routers.dashboard.user_settings_service.get_or_create_settings",
AsyncMock(return_value=SimpleNamespace(request_delay_seconds=0)),
)
login_page = client.get("/login")
csrf_token = _extract_csrf_token(login_page.text)
login_response = client.post(
"/login",
data={
"email": "user@example.com",
"password": "correct-password",
"csrf_token": csrf_token,
},
follow_redirects=False,
)
assert login_response.status_code == 303
assert login_response.headers["location"] == "/"
dashboard_response = client.get("/")
assert dashboard_response.status_code == 200
assert 'data-test="home-hero"' in dashboard_response.text
assert 'data-test="session-user"' in dashboard_response.text
assert "user@example.com" in dashboard_response.text
def test_login_rate_limiting_returns_429_after_threshold() -> None:
limiter = SlidingWindowRateLimiter(max_attempts=2, window_seconds=60)
app.dependency_overrides[get_db_session] = _override_db_session
app.dependency_overrides[get_auth_service] = lambda: StubAuthService(user=None)
app.dependency_overrides[get_login_rate_limiter] = lambda: limiter
client = TestClient(app)
login_page = client.get("/login")
csrf_token = _extract_csrf_token(login_page.text)
payload = {
"email": "user@example.com",
"password": "wrong-password",
"csrf_token": csrf_token,
}
first = client.post("/login", data=payload, follow_redirects=False)
second = client.post("/login", data=payload, follow_redirects=False)
third = client.post("/login", data=payload, follow_redirects=False)
assert first.status_code == 401
assert second.status_code == 401
assert third.status_code == 429
assert third.headers["Retry-After"] == "60"
def test_logout_requires_csrf_token_and_clears_session(monkeypatch) -> None:
limiter = SlidingWindowRateLimiter(max_attempts=5, window_seconds=60)
app.dependency_overrides[get_db_session] = _override_db_session
app.dependency_overrides[get_auth_service] = lambda: StubAuthService(
user=SimpleNamespace(id=1, email="user@example.com", is_admin=False)
)
app.dependency_overrides[get_login_rate_limiter] = lambda: limiter
client = TestClient(app)
monkeypatch.setattr(
"app.web.common.get_authenticated_user",
AsyncMock(
side_effect=[
SimpleNamespace(id=1, email="user@example.com", is_admin=False),
None,
]
),
)
monkeypatch.setattr(
"app.web.routers.dashboard.publication_service.list_new_for_latest_run_for_user",
AsyncMock(return_value=[]),
)
monkeypatch.setattr(
"app.web.routers.dashboard.run_service.list_recent_runs_for_user",
AsyncMock(return_value=[]),
)
monkeypatch.setattr(
"app.web.routers.dashboard.run_service.queue_status_counts_for_user",
AsyncMock(return_value={"queued": 0, "retrying": 0, "dropped": 0}),
)
monkeypatch.setattr(
"app.web.routers.dashboard.user_settings_service.get_or_create_settings",
AsyncMock(return_value=SimpleNamespace(request_delay_seconds=0)),
)
login_page = client.get("/login")
csrf_token = _extract_csrf_token(login_page.text)
client.post(
"/login",
data={
"email": "user@example.com",
"password": "correct-password",
"csrf_token": csrf_token,
},
follow_redirects=False,
)
failed_logout = client.post("/logout", data={}, follow_redirects=False)
assert failed_logout.status_code == 403
assert failed_logout.text == "CSRF token invalid."
dashboard_page = client.get("/")
logout_token = _extract_csrf_token(dashboard_page.text)
successful_logout = client.post(
"/logout",
data={"csrf_token": logout_token},
follow_redirects=False,
)
assert successful_logout.status_code == 303
assert successful_logout.headers["location"] == "/login"
assert client.get("/", follow_redirects=False).headers["location"] == "/login"

88
tests/unit/test_dashboard_view_model.py
View file

@ -1,88 +0,0 @@
from __future__ import annotations
from datetime import datetime, timezone
from app.db.models import CrawlRun, RunStatus, RunTriggerType
from app.presentation.dashboard import SECTION_TEMPLATES, build_dashboard_view_model
from app.services.publications import UnreadPublicationItem
def test_build_dashboard_view_model_maps_sections_and_unread_items() -> None:
unread_items = [
UnreadPublicationItem(
publication_id=10,
scholar_profile_id=3,
scholar_label="Ada Lovelace",
title="Analytical Engine Notes",
year=None,
citation_count=42,
venue_text="Computing Letters",
pub_url="https://example.test/pub-10",
)
]
runs = [
CrawlRun(
id=77,
user_id=1,
trigger_type=RunTriggerType.MANUAL,
status=RunStatus.SUCCESS,
start_dt=datetime(2026, 2, 16, 18, 0, tzinfo=timezone.utc),
scholar_count=1,
new_pub_count=1,
error_log={},
)
]
vm = build_dashboard_view_model(
unread_publications=unread_items,
recent_runs=runs,
request_delay_seconds=7,
queue_counts={"queued": 2, "retrying": 1, "dropped": 3},
)
assert vm.section_templates == SECTION_TEMPLATES
assert vm.run_controls.request_delay_seconds == 7
assert vm.run_controls.queue_queued_count == 2
assert vm.run_controls.queue_retrying_count == 1
assert vm.run_controls.queue_dropped_count == 3
assert vm.unread_publications[0].title == "Analytical Engine Notes"
assert vm.unread_publications[0].year_display == "-"
assert vm.unread_publications[0].citation_count == 42
assert vm.run_history[0].status_label == "success"
assert vm.run_history[0].status_badge == "ok"
assert vm.run_history[0].started_at_display == "2026-02-16 18:00 UTC"
assert vm.run_history[0].detail_url == "/runs/77"
def test_build_dashboard_view_model_maps_failed_and_partial_statuses() -> None:
runs = [
CrawlRun(
id=11,
user_id=1,
trigger_type=RunTriggerType.MANUAL,
status=RunStatus.FAILED,
start_dt=datetime(2026, 2, 16, 19, 0, tzinfo=timezone.utc),
scholar_count=2,
new_pub_count=0,
error_log={},
),
CrawlRun(
id=12,
user_id=1,
trigger_type=RunTriggerType.MANUAL,
status=RunStatus.PARTIAL_FAILURE,
start_dt=datetime(2026, 2, 16, 20, 0, tzinfo=timezone.utc),
scholar_count=2,
new_pub_count=1,
error_log={},
),
]
vm = build_dashboard_view_model(
unread_publications=[],
recent_runs=runs,
request_delay_seconds=1,
)
assert [item.status_badge for item in vm.run_history] == ["danger", "warn"]
assert [item.status_label for item in vm.run_history] == ["failed", "partial_failure"]

4
tests/unit/test_healthz.py
View file

@ -6,7 +6,7 @@ from app.main import app
def test_healthz_returns_200_when_database_is_available(monkeypatch) -> None:
monkeypatch.setattr("app.web.routers.health.check_database", AsyncMock(return_value=True))
monkeypatch.setattr("app.main.check_database", AsyncMock(return_value=True))
client = TestClient(app)
response = client.get("/healthz")
@ -16,7 +16,7 @@ def test_healthz_returns_200_when_database_is_available(monkeypatch) -> None:
def test_healthz_returns_500_when_database_is_unavailable(monkeypatch) -> None:
monkeypatch.setattr("app.web.routers.health.check_database", AsyncMock(return_value=False))
monkeypatch.setattr("app.main.check_database", AsyncMock(return_value=False))
client = TestClient(app)
response = client.get("/healthz")

15
tests/unit/test_logging.py
View file

@ -3,12 +3,13 @@ from __future__ import annotations
import json
import logging
import re
from unittest.mock import AsyncMock
from fastapi.testclient import TestClient
from app.logging_config import JsonLogFormatter, parse_redact_fields
from app.main import app
from app.web.middleware import REQUEST_ID_HEADER, parse_skip_paths
from app.http.middleware import REQUEST_ID_HEADER, parse_skip_paths
def test_json_log_formatter_redacts_sensitive_fields() -> None:
@ -39,14 +40,16 @@ def test_json_log_formatter_redacts_sensitive_fields() -> None:
assert "color_message" not in payload
def test_request_logging_middleware_sets_request_id_header() -> None:
def test_request_logging_middleware_sets_request_id_header(monkeypatch) -> None:
monkeypatch.setattr("app.main.check_database", AsyncMock(return_value=True))
client = TestClient(app)
response = client.get("/login", headers={REQUEST_ID_HEADER: "request-123"})
response = client.get("/healthz", headers={REQUEST_ID_HEADER: "request-123"})
assert response.status_code == 200
assert response.headers[REQUEST_ID_HEADER] == "request-123"
def test_parse_skip_paths_trims_and_discards_empty_segments() -> None:
assert parse_skip_paths(" /healthz , , /static/ ") == ("/healthz", "/static/")
assert parse_skip_paths(" /healthz , , /api/v1/metrics ") == (
"/healthz",
"/api/v1/metrics",
)

54
tests/unit/test_pages.py
View file

@ -1,54 +0,0 @@
from fastapi.testclient import TestClient
from app.main import app
def test_home_page_redirects_to_login_when_unauthenticated() -> None:
client = TestClient(app)
response = client.get("/", follow_redirects=False)
assert response.status_code == 303
assert response.headers["location"] == "/login"
def test_dev_ui_page_is_removed() -> None:
client = TestClient(app)
response = client.get("/dev/ui", follow_redirects=False)
assert response.status_code == 404
def test_login_page_renders_html() -> None:
client = TestClient(app)
response = client.get("/login")
assert response.status_code == 200
assert response.headers["content-type"].startswith("text/html")
assert "scholarr" in response.text
assert 'data-test="login-form"' in response.text
assert 'name="csrf_token"' in response.text
assert 'data-theme-control' in response.text
set_cookie = response.headers["set-cookie"].lower()
assert "httponly" in set_cookie
assert "samesite=lax" in set_cookie
def test_theme_query_parameter_accepts_supported_theme() -> None:
client = TestClient(app)
response = client.get("/login?theme=spruce")
assert response.status_code == 200
assert 'data-theme="spruce"' in response.text
def test_theme_query_parameter_falls_back_for_unknown_theme() -> None:
client = TestClient(app)
response = client.get("/login?theme=not-a-theme")
assert response.status_code == 200
assert 'data-theme="terracotta"' in response.text

20
tests/unit/test_phase2_pages.py
View file

@ -1,20 +0,0 @@
from fastapi.testclient import TestClient
from app.main import app
def test_phase2_pages_redirect_to_login_when_unauthenticated() -> None:
client = TestClient(app)
for path in (
"/users",
"/runs",
"/runs/1",
"/publications",
"/scholars",
"/settings",
"/account/password",
):
response = client.get(path, follow_redirects=False)
assert response.status_code == 303
assert response.headers["location"] == "/login"