"""Tests for historical analysis diff endpoint.""" from datetime import datetime from unittest.mock import MagicMock, patch import pytest from fastapi.testclient import TestClient from SPARC.api import AnalysisDiffResponse, _compute_analysis_diff, _extract_patent_ids, app from SPARC.auth import UserResponse, get_current_user # ---------- helpers ---------- def _mock_user(): """Return a fake authenticated user for dependency override.""" return UserResponse( id=1, email="test@example.com", role="user", created_at=datetime(2025, 1, 1), ) @pytest.fixture def auth_client(): """TestClient with auth dependency overridden.""" app.dependency_overrides[get_current_user] = _mock_user client = TestClient(app, raise_server_exceptions=False) yield client app.dependency_overrides.clear() # ---------- unit tests for helpers ---------- class TestExtractPatentIds: """Test _extract_patent_ids utility.""" def test_extracts_standard_ids(self): text = "Patent US-12345678-B2 covers the device. Also see US-9876543-A1." ids = _extract_patent_ids(text) assert "US-12345678-B2" in ids assert "US-9876543-A1" in ids def test_empty_text(self): assert _extract_patent_ids("") == set() assert _extract_patent_ids(None) == set() # type: ignore[arg-type] class TestComputeAnalysisDiff: """Test _compute_analysis_diff logic.""" def test_identical_analyses(self): rec = { "id": 1, "company_name": "nvidia", "analysis_type": "portfolio", "model": "openai/gpt-4o", "response": "Patent US-12345678-B2 is notable.", "timestamp": datetime(2025, 5, 1), } diff = _compute_analysis_diff(rec, dict(rec, id=2, timestamp=datetime(2025, 5, 2))) assert diff.patent_count_delta == 0 assert diff.added_patents == [] assert diff.removed_patents == [] def test_added_and_removed_patents(self): from_rec = { "id": 1, "company_name": "nvidia", "analysis_type": "portfolio", "model": "openai/gpt-4o", "response": "Patent US-12345678-B2 and US-11111111-A1.", "timestamp": datetime(2025, 5, 1), } to_rec = { "id": 2, "company_name": "nvidia", "analysis_type": "portfolio", "model": "openai/gpt-4o", "response": "Patent US-12345678-B2 and US-99999999-B1.", "timestamp": datetime(2025, 5, 2), } diff = _compute_analysis_diff(from_rec, to_rec) assert "US-99999999-B1" in diff.added_patents assert "US-11111111-A1" in diff.removed_patents assert diff.patent_count_delta == 0 # one added, one removed def test_model_change_detected(self): from_rec = { "id": 1, "company_name": "nvidia", "analysis_type": "portfolio", "model": "openai/gpt-4o", "response": "", "timestamp": datetime(2025, 5, 1), } to_rec = { "id": 2, "company_name": "nvidia", "analysis_type": "portfolio", "model": "anthropic/claude-3.5-sonnet", "response": "", "timestamp": datetime(2025, 5, 2), } diff = _compute_analysis_diff(from_rec, to_rec) assert "model" in diff.changed_fields assert diff.changed_fields["model"]["from"] == "openai/gpt-4o" assert diff.changed_fields["model"]["to"] == "anthropic/claude-3.5-sonnet" # ---------- API endpoint tests ---------- class TestDiffEndpoint: """Test GET /analyze/{company_name}/diff.""" @patch("SPARC.api._get_job_db") def test_happy_path(self, mock_get_db, auth_client): """Diff returns structured response when both IDs exist.""" db = MagicMock() mock_get_db.return_value = db from_rec = { "id": 10, "company_name": "nvidia", "analysis_type": "portfolio", "model": "openai/gpt-4o", "response": "Patent US-12345678-B2 found.", "timestamp": datetime(2025, 5, 1), } to_rec = { "id": 20, "company_name": "nvidia", "analysis_type": "portfolio", "model": "openai/gpt-4o", "response": "Patent US-12345678-B2 and US-99999999-A1 found.", "timestamp": datetime(2025, 5, 10), } db.get_analysis_by_id.side_effect = lambda aid: from_rec if aid == 10 else to_rec response = auth_client.get("/analyze/nvidia/diff?from=10&to=20") assert response.status_code == 200 data = response.json() assert data["company_name"] == "nvidia" assert data["from_id"] == 10 assert data["to_id"] == 20 assert "US-99999999-A1" in data["added_patents"] assert data["patent_count_delta"] == 1 @patch("SPARC.api._get_job_db") def test_from_id_not_found(self, mock_get_db, auth_client): """Returns 404 when 'from' analysis ID doesn't exist.""" db = MagicMock() mock_get_db.return_value = db db.get_analysis_by_id.return_value = None response = auth_client.get("/analyze/nvidia/diff?from=999&to=1000") assert response.status_code == 404 assert "999" in response.json()["detail"] @patch("SPARC.api._get_job_db") def test_to_id_not_found(self, mock_get_db, auth_client): """Returns 404 when 'to' analysis ID doesn't exist.""" db = MagicMock() mock_get_db.return_value = db from_rec = { "id": 10, "company_name": "nvidia", "analysis_type": "portfolio", "model": "openai/gpt-4o", "response": "", "timestamp": datetime(2025, 5, 1), } db.get_analysis_by_id.side_effect = lambda aid: from_rec if aid == 10 else None response = auth_client.get("/analyze/nvidia/diff?from=10&to=999") assert response.status_code == 404 assert "999" in response.json()["detail"] @patch("SPARC.api._get_job_db") def test_company_mismatch(self, mock_get_db, auth_client): """Returns 404 when analysis belongs to a different company.""" db = MagicMock() mock_get_db.return_value = db rec = { "id": 10, "company_name": "intel", "analysis_type": "portfolio", "model": "openai/gpt-4o", "response": "", "timestamp": datetime(2025, 5, 1), } db.get_analysis_by_id.return_value = rec response = auth_client.get("/analyze/nvidia/diff?from=10&to=20") assert response.status_code == 404 class TestHistoryEndpoint: """Test GET /analyze/{company_name}/history.""" @patch("SPARC.api._get_job_db") def test_returns_history_list(self, mock_get_db, auth_client): """History endpoint returns list of past analysis runs.""" db = MagicMock() mock_get_db.return_value = db db.list_company_analyses.return_value = [ { "id": 20, "company_name": "nvidia", "analysis_type": "portfolio", "model": "openai/gpt-4o", "response": "...", "timestamp": datetime(2025, 5, 10), }, { "id": 10, "company_name": "nvidia", "analysis_type": "portfolio", "model": "openai/gpt-4o", "response": "...", "timestamp": datetime(2025, 5, 1), }, ] response = auth_client.get("/analyze/nvidia/history") assert response.status_code == 200 data = response.json() assert len(data) == 2 assert data[0]["id"] == 20 assert data[1]["id"] == 10 @patch("SPARC.api._get_job_db") def test_empty_history(self, mock_get_db, auth_client): """History endpoint returns empty list when no analyses exist.""" db = MagicMock() mock_get_db.return_value = db db.list_company_analyses.return_value = [] response = auth_client.get("/analyze/nvidia/history") assert response.status_code == 200 assert response.json() == []