Compare commits

...

5 Commits

Author SHA1 Message Date
agent-company 857b3444df Add cursor-based pagination to GET /analyze/batch and update /jobs defaults
Add a new GET /analyze/batch endpoint that returns stored analysis results
with cursor-based pagination (default limit 50, max 200). Also update the
existing /jobs endpoint defaults from limit=10/max=100 to limit=50/max=200
for consistency.

The database layer gains a list_analyses() method with cursor support using
(timestamp, id) ordering, matching the existing list_jobs() pattern.

Includes tests for pagination behavior, boundary limits, cursor forwarding,
company name filtering, and empty result sets.

Closes leeworks-agents/SPARC#1669

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-05-18 21:49:22 +00:00
AI-Manager 4c411e1e0b Merge pull request 'Add tests for tracked company admin endpoints and scheduler' (#1667) from feature/1656-tracked-company-admin-tests into main
Merge: Add tests for tracked company admin endpoints and scheduler integration

Closes #1656
2026-04-20 23:05:57 +00:00
agent-company 6165d66760 Fix scheduler tests to use get_db_client after scheduler refactor
The scheduler was refactored (PR #1665) to use the pooled
get_db_client() from SPARC.auth instead of creating its own
DatabaseClient. Update test mocks accordingly and remove the
db.close() assertion since the pooled client is no longer closed
by the scheduler.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-20 23:05:42 +00:00
agent-company e610dea9a9 Merge remote-tracking branch 'origin/main' into feature/1656-tracked-company-admin-tests 2026-04-20 23:04:59 +00:00
agent-company fc942b2aa4 Add tests for tracked company admin endpoints and scheduler integration
20 test cases covering:
- GET/POST/DELETE /admin/tracked endpoints with admin auth enforcement
- GET /admin/alerts with limit parameter and auth
- scheduler.run_scheduled_analysis() for multi-company analysis, alert
  triggering on significant patent count changes, graceful failure handling

Closes leeworks-agents/SPARC#1656

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-20 19:14:29 +00:00
4 changed files with 669 additions and 1 deletions
+71 -1
View File
@@ -96,6 +96,24 @@ class JobStatus(BaseModel):
error: str | None = None
class AnalysisRecord(BaseModel):
"""A single stored analysis result."""
id: int
company_name: str | None = None
analysis_type: str | None = None
model: str | None = None
response: str | None = None
timestamp: datetime | None = None
class PaginatedAnalysisResponse(BaseModel):
"""Paginated response for analysis result listings."""
items: list[AnalysisRecord]
next_cursor: str | None = None
class PaginatedJobsResponse(BaseModel):
"""Paginated response for job listings."""
@@ -872,6 +890,58 @@ async def analyze_single_patent(
raise HTTPException(status_code=404, detail=str(e))
@app.get(
"/analyze/batch",
response_model=PaginatedAnalysisResponse,
tags=["Analysis"],
)
async def list_analysis_results(
company_name: Annotated[
str | None,
Query(description="Filter results by company name"),
] = None,
limit: Annotated[int, Query(ge=1, le=200)] = 50,
cursor: Annotated[
str | None,
Query(description="Opaque cursor from a previous response's next_cursor field"),
] = None,
_: UserResponse = Depends(get_current_user),
):
"""List stored analysis results with cursor-based pagination.
Returns past analysis results ordered by timestamp descending. Use
``limit`` to control page size (default 50, max 200). The response
includes a ``next_cursor`` field; pass it back as the ``cursor`` query
parameter to fetch the next page. When ``next_cursor`` is ``null``,
there are no more results.
Args:
company_name: Optional filter by company name
limit: Maximum number of results to return (default 50, max 200)
cursor: Opaque pagination cursor from a previous response
Returns:
Paginated list of analysis results
"""
db = _get_job_db()
rows = db.list_analyses(company_name=company_name, limit=limit + 1, cursor=cursor)
has_next = len(rows) > limit
if has_next:
rows = rows[:limit]
items = [AnalysisRecord(**row) for row in rows]
next_cursor = None
if has_next and rows:
last = rows[-1]
ts = last["timestamp"]
ts_str = ts.isoformat() if hasattr(ts, "isoformat") else str(ts)
next_cursor = f"{ts_str}|{last['id']}"
return PaginatedAnalysisResponse(items=items, next_cursor=next_cursor)
@app.post(
"/analyze/batch",
response_model=BatchAnalysisResponse,
@@ -1047,7 +1117,7 @@ async def list_jobs(
str | None,
Query(description="Filter by status: pending, running, completed, failed"),
] = None,
limit: Annotated[int, Query(ge=1, le=100)] = 10,
limit: Annotated[int, Query(ge=1, le=200)] = 50,
cursor: Annotated[
str | None,
Query(description="Opaque cursor from a previous response's next_cursor field"),
+42
View File
@@ -371,6 +371,48 @@ class DatabaseClient:
cursor.execute(query, params)
return [dict(row) for row in cursor.fetchall()]
def list_analyses(
self,
company_name: Optional[str] = None,
limit: int = 50,
cursor: Optional[str] = None,
) -> List[Dict]:
"""List analysis results with cursor-based pagination.
Args:
company_name: Optional filter by company name.
limit: Maximum number of records to return.
cursor: Opaque cursor (``timestamp|id``) from a previous response.
Returns:
List of analysis dicts ordered by timestamp descending.
"""
conditions: list[str] = ["is_cached = FALSE"]
params: list = []
if company_name:
conditions.append("LOWER(company_name) = LOWER(%s)")
params.append(company_name)
if cursor:
try:
ts_str, cursor_id = cursor.rsplit("|", 1)
conditions.append("(timestamp, id) < (%s, %s)")
params.extend([ts_str, int(cursor_id)])
except (ValueError, TypeError):
pass # Ignore malformed cursors; return from start
query = "SELECT id, company_name, analysis_type, model, response, timestamp FROM llm_messages"
if conditions:
query += " WHERE " + " AND ".join(conditions)
query += " ORDER BY timestamp DESC, id DESC LIMIT %s"
params.append(limit)
with self.get_conn() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(query, params)
return [dict(row) for row in cur.fetchall()]
def get_analytics(self, days: int = 30) -> Dict:
"""Get analytics on message usage.
+169
View File
@@ -0,0 +1,169 @@
"""Tests for cursor-based pagination on /analyze/batch GET and /jobs endpoints."""
from datetime import datetime, timedelta
from unittest.mock import Mock, patch
import pytest
from fastapi.testclient import TestClient
from SPARC.api import app
@pytest.fixture
def client():
"""Create test client."""
return TestClient(app)
def _make_analysis_row(id_: int, minutes_ago: int = 0, company: str = "nvidia"):
"""Create a fake analysis row dict."""
ts = datetime.now() - timedelta(minutes=minutes_ago)
return {
"id": id_,
"company_name": company,
"analysis_type": "patent_portfolio",
"model": "openai/gpt-4o",
"response": f"Analysis for {company}",
"timestamp": ts,
}
def _make_job_row(job_id: str, minutes_ago: int = 0, status: str = "completed"):
"""Create a fake job row dict."""
ts = datetime.now() - timedelta(minutes=minutes_ago)
return {
"job_id": job_id,
"status": status,
"progress": 100 if status == "completed" else 0,
"total_companies": 1,
"completed_companies": 1 if status == "completed" else 0,
"result": None,
"error": None,
"created_at": ts,
}
class TestAnalyzeBatchGetPagination:
"""Test cursor-based pagination on GET /analyze/batch."""
@patch("SPARC.api._get_job_db")
def test_returns_items_and_no_cursor_when_less_than_limit(self, mock_get_db, client):
"""When fewer results than limit, next_cursor should be null."""
db = Mock()
db.list_analyses.return_value = [
_make_analysis_row(1, minutes_ago=10),
_make_analysis_row(2, minutes_ago=20),
]
mock_get_db.return_value = db
response = client.get("/analyze/batch?limit=10")
assert response.status_code == 200
data = response.json()
assert len(data["items"]) == 2
assert data["next_cursor"] is None
@patch("SPARC.api._get_job_db")
def test_returns_cursor_when_more_results_exist(self, mock_get_db, client):
"""When more results exist than limit, next_cursor should be set."""
db = Mock()
# Return limit+1 rows to simulate more data
rows = [_make_analysis_row(i, minutes_ago=i) for i in range(4)]
db.list_analyses.return_value = rows
mock_get_db.return_value = db
response = client.get("/analyze/batch?limit=3")
assert response.status_code == 200
data = response.json()
assert len(data["items"]) == 3
assert data["next_cursor"] is not None
@patch("SPARC.api._get_job_db")
def test_cursor_passed_to_db(self, mock_get_db, client):
"""The cursor query param should be forwarded to the database layer."""
db = Mock()
db.list_analyses.return_value = []
mock_get_db.return_value = db
client.get("/analyze/batch?cursor=2025-01-01T00:00:00|42")
db.list_analyses.assert_called_once()
call_kwargs = db.list_analyses.call_args
assert call_kwargs.kwargs.get("cursor") == "2025-01-01T00:00:00|42" or \
(call_kwargs[1].get("cursor") == "2025-01-01T00:00:00|42" if len(call_kwargs) > 1 else False)
@patch("SPARC.api._get_job_db")
def test_default_limit_is_50(self, mock_get_db, client):
"""Default limit should be 50."""
db = Mock()
db.list_analyses.return_value = []
mock_get_db.return_value = db
client.get("/analyze/batch")
call_kwargs = db.list_analyses.call_args
# The endpoint requests limit+1 from DB, so 51
assert 51 in call_kwargs.args or call_kwargs.kwargs.get("limit") == 51
def test_limit_over_200_rejected(self, client):
"""Limit > 200 should be rejected with 422."""
response = client.get("/analyze/batch?limit=201")
assert response.status_code == 422
def test_limit_zero_rejected(self, client):
"""Limit < 1 should be rejected with 422."""
response = client.get("/analyze/batch?limit=0")
assert response.status_code == 422
@patch("SPARC.api._get_job_db")
def test_company_name_filter(self, mock_get_db, client):
"""The company_name filter should be forwarded to the database."""
db = Mock()
db.list_analyses.return_value = []
mock_get_db.return_value = db
client.get("/analyze/batch?company_name=intel")
call_kwargs = db.list_analyses.call_args
assert call_kwargs.kwargs.get("company_name") == "intel" or \
"intel" in (call_kwargs.args if call_kwargs.args else [])
@patch("SPARC.api._get_job_db")
def test_empty_result_set(self, mock_get_db, client):
"""Empty result set returns empty items and null cursor."""
db = Mock()
db.list_analyses.return_value = []
mock_get_db.return_value = db
response = client.get("/analyze/batch")
assert response.status_code == 200
data = response.json()
assert data["items"] == []
assert data["next_cursor"] is None
class TestJobsPaginationDefaults:
"""Test that /jobs endpoint uses updated defaults."""
@patch("SPARC.api._get_job_db")
def test_default_limit_is_50(self, mock_get_db, client):
"""Default limit should now be 50."""
db = Mock()
db.list_jobs.return_value = []
mock_get_db.return_value = db
client.get("/jobs")
call_kwargs = db.list_jobs.call_args
# Endpoint requests limit+1 from DB, so 51
assert 51 in call_kwargs.args or call_kwargs.kwargs.get("limit") == 51
def test_limit_over_200_rejected(self, client):
"""Limit > 200 should be rejected with 422."""
response = client.get("/jobs?limit=201")
assert response.status_code == 422
@patch("SPARC.api._get_job_db")
def test_limit_200_accepted(self, mock_get_db, client):
"""Limit of exactly 200 should be accepted."""
db = Mock()
db.list_jobs.return_value = []
mock_get_db.return_value = db
response = client.get("/jobs?limit=200")
assert response.status_code == 200
+387
View File
@@ -0,0 +1,387 @@
"""Tests for tracked company admin endpoints and scheduler integration.
Covers issue #1656:
- GET /admin/tracked (list tracked companies)
- POST /admin/tracked (add a tracked company)
- DELETE /admin/tracked/{company_name} (remove a tracked company)
- GET /admin/alerts (list alerts)
- scheduler.run_scheduled_analysis() integration
All tests mock the database layer and use JWT auth fixtures.
"""
from datetime import datetime, timezone
from unittest.mock import MagicMock, patch, call
import pytest
from fastapi.testclient import TestClient
from SPARC.api import app
from SPARC.auth import create_access_token
@pytest.fixture
def client():
"""Create test client."""
return TestClient(app)
@pytest.fixture(autouse=True)
def mock_db():
"""Mock the database client used by admin and auth endpoints."""
db = MagicMock()
# Default admin user for auth
db.get_user_by_id.return_value = {
"id": 1,
"email": "admin@test.com",
"role": "admin",
"created_at": datetime(2025, 1, 1, tzinfo=timezone.utc),
}
with patch("SPARC.api.get_db_client", return_value=db), \
patch("SPARC.auth.get_db_client", return_value=db):
yield db
def _admin_header():
"""Create an Authorization header with a valid admin access token."""
token = create_access_token(1, "admin@test.com", "admin")
return {"Authorization": f"Bearer {token}"}
def _user_header():
"""Create an Authorization header with a regular user access token."""
token = create_access_token(2, "user@test.com", "user")
return {"Authorization": f"Bearer {token}"}
# ---------- GET /admin/tracked ----------
class TestListTrackedCompanies:
"""GET /admin/tracked"""
def test_list_tracked_returns_companies(self, client, mock_db):
"""Admin can list tracked companies."""
mock_db.list_tracked_companies.return_value = [
{"company_name": "NVIDIA", "last_patent_count": 120, "last_analyzed": "2025-06-15"},
{"company_name": "AMD", "last_patent_count": 80, "last_analyzed": "2025-06-14"},
]
response = client.get("/admin/tracked", headers=_admin_header())
assert response.status_code == 200
data = response.json()
assert len(data) == 2
assert data[0]["company_name"] == "NVIDIA"
def test_list_tracked_empty(self, client, mock_db):
"""Returns empty list when no companies are tracked."""
mock_db.list_tracked_companies.return_value = []
response = client.get("/admin/tracked", headers=_admin_header())
assert response.status_code == 200
assert response.json() == []
def test_list_tracked_requires_admin(self, client, mock_db):
"""Regular user cannot access tracked companies list."""
mock_db.get_user_by_id.return_value = {
"id": 2,
"email": "user@test.com",
"role": "user",
"created_at": datetime(2025, 1, 1, tzinfo=timezone.utc),
}
response = client.get("/admin/tracked", headers=_user_header())
assert response.status_code == 403
def test_list_tracked_unauthenticated(self, client):
"""Unauthenticated request returns 401."""
response = client.get("/admin/tracked")
assert response.status_code == 401
# ---------- POST /admin/tracked ----------
class TestAddTrackedCompany:
"""POST /admin/tracked"""
def test_add_tracked_company_success(self, client, mock_db):
"""Admin can add a company to tracking."""
mock_db.add_tracked_company.return_value = {
"company_name": "Intel",
"last_patent_count": 0,
"last_analyzed": None,
}
response = client.post(
"/admin/tracked",
json={"company_name": "Intel"},
headers=_admin_header(),
)
assert response.status_code == 200
data = response.json()
assert data["company_name"] == "Intel"
mock_db.add_tracked_company.assert_called_once_with("Intel")
def test_add_duplicate_returns_409(self, client, mock_db):
"""Adding an already-tracked company returns 409."""
mock_db.add_tracked_company.return_value = None
response = client.post(
"/admin/tracked",
json={"company_name": "NVIDIA"},
headers=_admin_header(),
)
assert response.status_code == 409
assert "already tracked" in response.json()["detail"].lower()
def test_add_tracked_requires_admin(self, client, mock_db):
"""Regular user cannot add tracked companies."""
mock_db.get_user_by_id.return_value = {
"id": 2,
"email": "user@test.com",
"role": "user",
"created_at": datetime(2025, 1, 1, tzinfo=timezone.utc),
}
response = client.post(
"/admin/tracked",
json={"company_name": "Intel"},
headers=_user_header(),
)
assert response.status_code == 403
def test_add_tracked_empty_name_rejected(self, client):
"""Empty company name is rejected by validation."""
response = client.post(
"/admin/tracked",
json={"company_name": ""},
headers=_admin_header(),
)
assert response.status_code == 422 # Pydantic validation error
# ---------- DELETE /admin/tracked/{company_name} ----------
class TestRemoveTrackedCompany:
"""DELETE /admin/tracked/{company_name}"""
def test_remove_tracked_company_success(self, client, mock_db):
"""Admin can remove a tracked company."""
mock_db.remove_tracked_company.return_value = True
response = client.delete(
"/admin/tracked/NVIDIA",
headers=_admin_header(),
)
assert response.status_code == 200
assert "Stopped tracking" in response.json()["message"]
mock_db.remove_tracked_company.assert_called_once_with("NVIDIA")
def test_remove_nonexistent_returns_404(self, client, mock_db):
"""Removing a non-tracked company returns 404."""
mock_db.remove_tracked_company.return_value = False
response = client.delete(
"/admin/tracked/UnknownCorp",
headers=_admin_header(),
)
assert response.status_code == 404
assert "not found" in response.json()["detail"].lower()
def test_remove_tracked_requires_admin(self, client, mock_db):
"""Regular user cannot remove tracked companies."""
mock_db.get_user_by_id.return_value = {
"id": 2,
"email": "user@test.com",
"role": "user",
"created_at": datetime(2025, 1, 1, tzinfo=timezone.utc),
}
response = client.delete(
"/admin/tracked/NVIDIA",
headers=_user_header(),
)
assert response.status_code == 403
# ---------- GET /admin/alerts ----------
class TestListAlerts:
"""GET /admin/alerts"""
def test_list_alerts_returns_data(self, client, mock_db):
"""Admin can list alerts."""
mock_db.list_alerts.return_value = [
{
"id": 1,
"company_name": "NVIDIA",
"alert_type": "patent_count_change",
"message": "Patent count increased by 25%",
"created_at": "2025-06-15T10:00:00Z",
},
]
response = client.get("/admin/alerts", headers=_admin_header())
assert response.status_code == 200
data = response.json()
assert len(data) == 1
assert data[0]["alert_type"] == "patent_count_change"
def test_list_alerts_with_limit(self, client, mock_db):
"""Custom limit parameter is passed to the database."""
mock_db.list_alerts.return_value = []
response = client.get("/admin/alerts?limit=10", headers=_admin_header())
assert response.status_code == 200
mock_db.list_alerts.assert_called_once_with(limit=10)
def test_list_alerts_requires_admin(self, client, mock_db):
"""Regular user cannot access alerts."""
mock_db.get_user_by_id.return_value = {
"id": 2,
"email": "user@test.com",
"role": "user",
"created_at": datetime(2025, 1, 1, tzinfo=timezone.utc),
}
response = client.get("/admin/alerts", headers=_user_header())
assert response.status_code == 403
# ---------- Scheduler integration ----------
class TestSchedulerIntegration:
"""Tests for scheduler.run_scheduled_analysis()."""
def test_no_tracked_companies_skips_analysis(self):
"""Scheduler does nothing when no companies are tracked."""
mock_db = MagicMock()
mock_db.list_tracked_companies.return_value = []
with patch("SPARC.scheduler.get_db_client", return_value=mock_db), \
patch("SPARC.scheduler.CompanyAnalyzer") as mock_analyzer_cls:
from SPARC.scheduler import run_scheduled_analysis
run_scheduled_analysis()
mock_analyzer_cls.assert_not_called()
def test_scheduler_analyzes_each_tracked_company(self):
"""Scheduler runs analysis for every tracked company."""
mock_db = MagicMock()
mock_db.list_tracked_companies.return_value = [
{"company_name": "NVIDIA", "last_patent_count": 100},
{"company_name": "AMD", "last_patent_count": 50},
]
mock_result_nvidia = MagicMock(success=True, patent_count=110)
mock_result_amd = MagicMock(success=True, patent_count=55)
mock_analyzer = MagicMock()
mock_analyzer._analyze_company_safe.side_effect = [mock_result_nvidia, mock_result_amd]
with patch("SPARC.scheduler.get_db_client", return_value=mock_db), \
patch("SPARC.scheduler.CompanyAnalyzer", return_value=mock_analyzer):
from SPARC.scheduler import run_scheduled_analysis
run_scheduled_analysis()
assert mock_analyzer._analyze_company_safe.call_count == 2
mock_db.update_tracked_company.assert_any_call("NVIDIA", 110)
mock_db.update_tracked_company.assert_any_call("AMD", 55)
def test_scheduler_triggers_alert_on_significant_change(self):
"""Scheduler stores an alert when patent count changes significantly."""
mock_db = MagicMock()
mock_db.list_tracked_companies.return_value = [
{"company_name": "Tesla", "last_patent_count": 100},
]
mock_result = MagicMock(success=True, patent_count=130) # 30% increase
mock_analyzer = MagicMock()
mock_analyzer._analyze_company_safe.return_value = mock_result
with patch("SPARC.scheduler.get_db_client", return_value=mock_db), \
patch("SPARC.scheduler.CompanyAnalyzer", return_value=mock_analyzer):
from SPARC.scheduler import run_scheduled_analysis
run_scheduled_analysis()
mock_db.store_alert.assert_called_once()
alert_kwargs = mock_db.store_alert.call_args
assert alert_kwargs[1]["company_name"] == "Tesla"
assert alert_kwargs[1]["alert_type"] == "patent_count_change"
assert alert_kwargs[1]["old_value"] == 100
assert alert_kwargs[1]["new_value"] == 130
def test_scheduler_no_alert_for_small_change(self):
"""Scheduler does not alert when change is below threshold."""
mock_db = MagicMock()
mock_db.list_tracked_companies.return_value = [
{"company_name": "Intel", "last_patent_count": 100},
]
mock_result = MagicMock(success=True, patent_count=105) # 5% increase
mock_analyzer = MagicMock()
mock_analyzer._analyze_company_safe.return_value = mock_result
with patch("SPARC.scheduler.get_db_client", return_value=mock_db), \
patch("SPARC.scheduler.CompanyAnalyzer", return_value=mock_analyzer):
from SPARC.scheduler import run_scheduled_analysis
run_scheduled_analysis()
mock_db.store_alert.assert_not_called()
def test_scheduler_handles_analysis_failure(self):
"""Scheduler continues when one company fails analysis."""
mock_db = MagicMock()
mock_db.list_tracked_companies.return_value = [
{"company_name": "FailCo", "last_patent_count": 50},
{"company_name": "SuccessCo", "last_patent_count": 30},
]
mock_fail_result = MagicMock(success=False, error="API timeout")
mock_ok_result = MagicMock(success=True, patent_count=35)
mock_analyzer = MagicMock()
mock_analyzer._analyze_company_safe.side_effect = [mock_fail_result, mock_ok_result]
with patch("SPARC.scheduler.get_db_client", return_value=mock_db), \
patch("SPARC.scheduler.CompanyAnalyzer", return_value=mock_analyzer):
from SPARC.scheduler import run_scheduled_analysis
run_scheduled_analysis()
# FailCo should not get updated, SuccessCo should
mock_db.update_tracked_company.assert_called_once_with("SuccessCo", 35)
def test_scheduler_handles_exception_in_analysis(self):
"""Scheduler continues even when analysis raises an exception."""
mock_db = MagicMock()
mock_db.list_tracked_companies.return_value = [
{"company_name": "CrashCo", "last_patent_count": 10},
{"company_name": "OKCo", "last_patent_count": 20},
]
mock_ok_result = MagicMock(success=True, patent_count=22)
mock_analyzer = MagicMock()
mock_analyzer._analyze_company_safe.side_effect = [
RuntimeError("unexpected error"),
mock_ok_result,
]
with patch("SPARC.scheduler.get_db_client", return_value=mock_db), \
patch("SPARC.scheduler.CompanyAnalyzer", return_value=mock_analyzer):
from SPARC.scheduler import run_scheduled_analysis
run_scheduled_analysis()
# OKCo should still be processed
mock_db.update_tracked_company.assert_called_once_with("OKCo", 22)