Compare commits

...

6 Commits

Author SHA1 Message Date
agent-company a2f81b0396 Add test coverage for analyze_single_patent auto-download path
7 test cases covering:
- PDF on disk analyzed directly (no download)
- Auto-download from cached metadata link when PDF missing
- FileNotFoundError when no cached link available
- Cached patent without pdf_link raises FileNotFoundError
- Analysis pipeline failure returns error string gracefully
- Model override parameter forwarded to LLM
- FileNotFoundError during parsing re-raised (not swallowed)

Closes leeworks-agents/SPARC#1661

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-20 19:21:53 +00:00
AI-Manager a07a0c7fbe Merge pull request 'Fix remaining dark mode issue in Analysis page prose block' (#1628) from feature/1605-dark-mode into main
Fix remaining dark mode issue in Analysis page prose block (#1628)
2026-04-20 06:41:59 +00:00
AI-Manager 43fd2c9575 Merge pull request 'Expand JWT auth integration tests to 33 cases' (#1627) from feature/1624-jwt-auth-tests into main
Expand JWT auth integration tests to 33 cases (#1627)
2026-04-20 06:41:47 +00:00
agent-company d4d43cf9b8 Fix prose-invert to only apply in dark mode on Analysis page
The prose-invert class was applied unconditionally, causing inverted
(light) text in light mode within the AI analysis results section.
Changed to dark:prose-invert so it only activates when dark mode is
enabled.

Note: The broader dark mode feature (issue #1605) is already fully
implemented -- ThemeContext, toggle button, CSS variables, dark:
variants across all pages. This fix addresses the only remaining
unstyled element.

Closes leeworks-agents/SPARC#1605

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-20 06:08:02 +00:00
agent-company 2f2b6382fa Expand JWT auth integration tests from 17 to 33 cases
Add comprehensive edge-case coverage for issue #1624:

- Admin delete user endpoint (5 tests): successful delete, self-delete
  prevention, nonexistent user 404, non-admin 403, missing token rejection
- Admin role change gaps (2 tests): nonexistent user 404, non-admin 403
- Input validation (3 tests): invalid email 422, short password 422,
  missing fields 422 for both register and login
- Token edge cases (4 tests): malformed token, wrong-secret token,
  deleted user token, deleted user refresh
- Token claim verification (1 test): login tokens contain correct claims

All tests use mocked DB fixtures and require no live database.

Closes leeworks-agents/SPARC#1624

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-20 06:05:54 +00:00
AI-Manager 1319530f04 Merge pull request 'ci: enable ruff linting and pytest in CI pipeline' (#1568) from feature/1559-1560-enable-ci-linting-and-tests into main
Merge PR #1568: ci: enable ruff linting and pytest in CI pipeline

Closes #1559
Closes #1560
2026-04-19 23:08:07 +00:00
3 changed files with 421 additions and 11 deletions
+1 -1
View File
@@ -159,7 +159,7 @@ export function Analysis() {
</button>
</div>
</div>
<div className="prose prose-invert max-w-none">
<div className="prose dark:prose-invert max-w-none">
<div className="text-text-primary whitespace-pre-wrap leading-relaxed">
{result.analysis}
</div>
+211
View File
@@ -0,0 +1,211 @@
"""Tests for analyze_single_patent auto-download path.
Covers issue #1661:
- PDF exists on disk: direct analysis (happy path)
- PDF not on disk, cached link exists: auto-download and analyze
- PDF not on disk, no cached link: FileNotFoundError
- Analysis failure after PDF found: graceful error message
- Model override parameter passthrough
"""
import os
from unittest.mock import MagicMock, patch
import pytest
from SPARC.analyzer import CompanyAnalyzer
from SPARC.types import Patent
@pytest.fixture(autouse=True)
def mock_db(mocker):
"""Mock DatabaseClient so no real DB is needed."""
mock_db_cls = mocker.patch("SPARC.analyzer.DatabaseClient")
mock_db_instance = MagicMock()
mock_db_instance.get_cached_patent.return_value = None
mock_db_instance.get_cached_serp_query.return_value = None
mock_db_cls.return_value = mock_db_instance
return mock_db_instance
@pytest.fixture
def analyzer(mocker, mock_db):
"""Create a CompanyAnalyzer with mocked LLM and DB."""
mocker.patch("SPARC.analyzer.LLMAnalyzer")
return CompanyAnalyzer(openrouter_api_key="test-key")
class TestAnalyzeSinglePatentAutoDownload:
"""Test the auto-download logic in analyze_single_patent."""
def test_pdf_on_disk_analyzed_directly(self, analyzer, mocker, tmp_path):
"""When PDF exists on disk, it is analyzed directly without download."""
patent_id = "US-11234567-B2"
# Create the patents dir and PDF file
patents_dir = tmp_path / "patents"
patents_dir.mkdir()
pdf_path = patents_dir / f"{patent_id}.pdf"
pdf_path.write_bytes(b"fake PDF content")
mock_parse = mocker.patch("SPARC.analyzer.SERP.parse_patent_pdf")
mock_minimize = mocker.patch("SPARC.analyzer.SERP.minimize_patent_for_llm")
mock_parse.return_value = {"abstract": "test", "claims": "test claims"}
mock_minimize.return_value = "minimized content"
analyzer.llm_analyzer.analyze_patent_content.return_value = "Good patent."
# Change cwd so patents/{patent_id}.pdf resolves to our tmp_path
original_cwd = os.getcwd()
os.chdir(tmp_path)
try:
result = analyzer.analyze_single_patent(patent_id, "TestCo")
finally:
os.chdir(original_cwd)
assert result == "Good patent."
# DB cache should not have been queried since file existed
analyzer.db.get_cached_patent.assert_not_called()
def test_auto_download_from_cached_link(self, analyzer, mocker, tmp_path):
"""When PDF is not on disk but link is cached, auto-download occurs."""
patent_id = "US-99887766-A1"
# No patents dir exists (PDF not on disk)
mock_save = mocker.patch("SPARC.analyzer.SERP.save_patents")
downloaded_patent = Patent(patent_id=patent_id, pdf_link="https://example.com/patent.pdf")
downloaded_patent.pdf_path = f"patents/{patent_id}.pdf"
mock_save.return_value = downloaded_patent
# Cached patent has a PDF link
analyzer.db.get_cached_patent.return_value = {
"patent_id": patent_id,
"pdf_link": "https://example.com/patent.pdf",
}
# Mock the rest of the analysis pipeline
mock_parse = mocker.patch("SPARC.analyzer.SERP.parse_patent_pdf")
mock_minimize = mocker.patch("SPARC.analyzer.SERP.minimize_patent_for_llm")
mock_parse.return_value = {"abstract": "test abstract"}
mock_minimize.return_value = "minimized content"
analyzer.llm_analyzer.analyze_patent_content.return_value = "Strong innovation."
# Change cwd so patents/{patent_id}.pdf does NOT exist
original_cwd = os.getcwd()
os.chdir(tmp_path)
try:
result = analyzer.analyze_single_patent(patent_id, "DownloadCo")
finally:
os.chdir(original_cwd)
assert result == "Strong innovation."
analyzer.db.get_cached_patent.assert_called_once_with(patent_id)
mock_save.assert_called_once()
# Verify the Patent passed to save_patents has the correct ID and link
saved_patent = mock_save.call_args[0][0]
assert saved_patent.patent_id == patent_id
assert saved_patent.pdf_link == "https://example.com/patent.pdf"
def test_no_cached_link_raises_file_not_found(self, analyzer, mocker, tmp_path):
"""When PDF is not on disk and no cached link, FileNotFoundError raised."""
patent_id = "US-00000000-X1"
analyzer.db.get_cached_patent.return_value = None
original_cwd = os.getcwd()
os.chdir(tmp_path)
try:
with pytest.raises(FileNotFoundError, match="no download link is cached"):
analyzer.analyze_single_patent(patent_id, "MissingCo")
finally:
os.chdir(original_cwd)
def test_cached_patent_without_pdf_link_raises(self, analyzer, mocker, tmp_path):
"""When cached patent exists but has no pdf_link, FileNotFoundError raised."""
patent_id = "US-11111111-B1"
analyzer.db.get_cached_patent.return_value = {
"patent_id": patent_id,
"pdf_link": None,
}
original_cwd = os.getcwd()
os.chdir(tmp_path)
try:
with pytest.raises(FileNotFoundError, match="no download link is cached"):
analyzer.analyze_single_patent(patent_id, "NoPDFCo")
finally:
os.chdir(original_cwd)
def test_analysis_exception_returns_error_message(self, analyzer, mocker, tmp_path):
"""When analysis pipeline fails, returns error string instead of raising."""
patent_id = "US-22222222-A2"
# Create the PDF on disk so it skips download
patents_dir = tmp_path / "patents"
patents_dir.mkdir()
(patents_dir / f"{patent_id}.pdf").write_bytes(b"fake PDF")
# Parse fails
mocker.patch(
"SPARC.analyzer.SERP.parse_patent_pdf",
side_effect=ValueError("Corrupt PDF"),
)
original_cwd = os.getcwd()
os.chdir(tmp_path)
try:
result = analyzer.analyze_single_patent(patent_id, "ErrorCo")
finally:
os.chdir(original_cwd)
assert "Failed to analyze patent" in result
assert "Corrupt PDF" in result
def test_model_override_passed_to_llm(self, analyzer, mocker, tmp_path):
"""The model parameter is forwarded to the LLM analyzer."""
patent_id = "US-33333333-B2"
patents_dir = tmp_path / "patents"
patents_dir.mkdir()
(patents_dir / f"{patent_id}.pdf").write_bytes(b"fake PDF")
mocker.patch("SPARC.analyzer.SERP.parse_patent_pdf", return_value={"abstract": "test"})
mocker.patch("SPARC.analyzer.SERP.minimize_patent_for_llm", return_value="content")
analyzer.llm_analyzer.analyze_patent_content.return_value = "Analysis result."
original_cwd = os.getcwd()
os.chdir(tmp_path)
try:
result = analyzer.analyze_single_patent(
patent_id, "ModelCo", model="openai/gpt-4o"
)
finally:
os.chdir(original_cwd)
assert result == "Analysis result."
analyzer.llm_analyzer.analyze_patent_content.assert_called_once_with(
patent_content="content",
company_name="ModelCo",
model="openai/gpt-4o",
)
def test_file_not_found_during_parse_re_raised(self, analyzer, mocker, tmp_path):
"""FileNotFoundError during parsing is re-raised, not caught."""
patent_id = "US-44444444-C1"
patents_dir = tmp_path / "patents"
patents_dir.mkdir()
(patents_dir / f"{patent_id}.pdf").write_bytes(b"fake PDF")
mocker.patch(
"SPARC.analyzer.SERP.parse_patent_pdf",
side_effect=FileNotFoundError("PDF file vanished"),
)
original_cwd = os.getcwd()
os.chdir(tmp_path)
try:
with pytest.raises(FileNotFoundError, match="PDF file vanished"):
analyzer.analyze_single_patent(patent_id, "VanishCo")
finally:
os.chdir(original_cwd)
+209 -10
View File
@@ -1,13 +1,29 @@
"""Tests for JWT authentication flow: register, login, protected routes, refresh, admin access."""
"""Tests for JWT authentication flow: register, login, protected routes, refresh, admin access.
from datetime import datetime, timezone
Covers all five scenarios required by issue #1624:
1. Registration (POST /auth/register)
2. Login (POST /auth/login)
3. Protected route access (GET /auth/me) -- valid, missing, expired, wrong-type tokens
4. Token refresh (POST /auth/refresh)
5. Admin-only endpoints (GET /admin/users, PATCH role, DELETE user)
All tests use mocked DB fixtures and require no live database.
"""
from datetime import datetime, timedelta, timezone
from unittest.mock import MagicMock, patch
import jwt as pyjwt
import pytest
from fastapi.testclient import TestClient
from SPARC.api import app
from SPARC.auth import create_access_token, create_refresh_token
from SPARC.auth import (
JWT_ALGORITHM,
JWT_SECRET,
create_access_token,
create_refresh_token,
)
@pytest.fixture
@@ -171,13 +187,6 @@ class TestGetMe:
def test_expired_token_returns_401(self, client, mock_db):
"""An expired token should return 401."""
# Create a token that has already expired
from datetime import timedelta
import jwt as pyjwt
from SPARC.auth import JWT_ALGORITHM, JWT_SECRET
payload = {
"sub": "1",
"email": "user@test.com",
@@ -301,3 +310,193 @@ class TestAdminUsers:
assert response.status_code == 400
assert "own role" in response.json()["detail"].lower()
def test_role_change_nonexistent_user_returns_404(self, client, mock_db):
"""Changing role for a user that does not exist should return 404."""
admin = _make_admin_user()
mock_db.get_user_by_id.return_value = admin
mock_db.update_user_role.return_value = None
response = client.patch(
"/admin/users/999/role",
json={"role": "admin"},
headers=_auth_header(admin),
)
assert response.status_code == 404
assert "not found" in response.json()["detail"].lower()
def test_regular_user_cannot_change_role(self, client, mock_db):
"""Non-admin user should receive 403 when trying to change roles."""
user = _make_regular_user()
mock_db.get_user_by_id.return_value = user
response = client.patch(
"/admin/users/1/role",
json={"role": "admin"},
headers=_auth_header(user),
)
assert response.status_code == 403
class TestAdminDeleteUser:
"""DELETE /admin/users/{user_id}"""
def test_admin_can_delete_user(self, client, mock_db):
"""Admin should be able to delete another user."""
admin = _make_admin_user()
mock_db.get_user_by_id.return_value = admin
mock_db.delete_user.return_value = True
response = client.delete(
"/admin/users/2",
headers=_auth_header(admin),
)
assert response.status_code == 200
assert "deleted" in response.json()["message"].lower()
mock_db.delete_user.assert_called_once_with(2)
def test_admin_cannot_delete_self(self, client, mock_db):
"""Admin should not be able to delete themselves."""
admin = _make_admin_user()
mock_db.get_user_by_id.return_value = admin
response = client.delete(
"/admin/users/1",
headers=_auth_header(admin),
)
assert response.status_code == 400
assert "yourself" in response.json()["detail"].lower()
def test_delete_nonexistent_user_returns_404(self, client, mock_db):
"""Deleting a user that does not exist should return 404."""
admin = _make_admin_user()
mock_db.get_user_by_id.return_value = admin
mock_db.delete_user.return_value = False
response = client.delete(
"/admin/users/999",
headers=_auth_header(admin),
)
assert response.status_code == 404
assert "not found" in response.json()["detail"].lower()
def test_regular_user_cannot_delete_user(self, client, mock_db):
"""Non-admin user should receive 403 when trying to delete users."""
user = _make_regular_user()
mock_db.get_user_by_id.return_value = user
response = client.delete(
"/admin/users/1",
headers=_auth_header(user),
)
assert response.status_code == 403
def test_no_token_cannot_delete_user(self, client):
"""Missing token should be rejected for delete endpoint."""
response = client.delete("/admin/users/1")
assert response.status_code in (401, 403)
class TestEdgeCases:
"""Additional edge-case tests for auth robustness."""
def test_register_invalid_email_returns_422(self, client, mock_db):
"""Registration with an invalid email format should return 422."""
response = client.post(
"/auth/register",
json={"email": "not-an-email", "password": "securepass123"},
)
assert response.status_code == 422
def test_register_short_password_returns_422(self, client, mock_db):
"""Registration with a password shorter than 8 chars should return 422."""
response = client.post(
"/auth/register",
json={"email": "user@test.com", "password": "short"},
)
assert response.status_code == 422
def test_register_missing_fields_returns_422(self, client, mock_db):
"""Registration with missing fields should return 422."""
response = client.post("/auth/register", json={})
assert response.status_code == 422
def test_login_missing_fields_returns_422(self, client, mock_db):
"""Login with missing fields should return 422."""
response = client.post("/auth/login", json={"email": "user@test.com"})
assert response.status_code == 422
def test_malformed_token_returns_401(self, client, mock_db):
"""A completely malformed token string should return 401."""
response = client.get(
"/auth/me",
headers={"Authorization": "Bearer not.a.valid.jwt.token"},
)
assert response.status_code == 401
def test_token_with_wrong_secret_returns_401(self, client, mock_db):
"""A token signed with a different secret should return 401."""
payload = {
"sub": "1",
"email": "user@test.com",
"role": "user",
"exp": datetime.now(timezone.utc) + timedelta(hours=1),
"type": "access",
}
wrong_secret_token = pyjwt.encode(payload, "wrong-secret", algorithm=JWT_ALGORITHM)
response = client.get(
"/auth/me",
headers={"Authorization": f"Bearer {wrong_secret_token}"},
)
assert response.status_code == 401
def test_token_for_deleted_user_returns_401(self, client, mock_db):
"""A valid token for a user no longer in the DB should return 401."""
user = _make_regular_user()
mock_db.get_user_by_id.return_value = None # user was deleted
response = client.get("/auth/me", headers=_auth_header(user))
assert response.status_code == 401
def test_refresh_for_deleted_user_returns_401(self, client, mock_db):
"""Refreshing a token for a deleted user should return 401."""
user = _make_regular_user()
mock_db.get_user_by_id.return_value = None
refresh = create_refresh_token(user["id"], user["email"], user["role"])
response = client.post(
"/auth/refresh", json={"refresh_token": refresh}
)
assert response.status_code == 401
def test_login_returns_decodable_tokens(self, client, mock_db):
"""Tokens returned by login should be decodable and contain expected claims."""
user = _make_regular_user()
mock_db.authenticate_user.return_value = user
response = client.post(
"/auth/login",
json={"email": "user@test.com", "password": "correctpassword"},
)
data = response.json()
access_payload = pyjwt.decode(
data["access_token"], JWT_SECRET, algorithms=[JWT_ALGORITHM]
)
assert access_payload["sub"] == str(user["id"])
assert access_payload["email"] == user["email"]
assert access_payload["type"] == "access"
refresh_payload = pyjwt.decode(
data["refresh_token"], JWT_SECRET, algorithms=[JWT_ALGORITHM]
)
assert refresh_payload["type"] == "refresh"