0xWheatyz
a91c3badab
Created CompanyAnalyzer class that orchestrates the complete pipeline: 1. Retrieves patents via SERP API 2. Downloads and parses PDFs 3. Minimizes content (removes bloat) 4. Analyzes portfolio with LLM 5. Returns performance estimation Features: - Full company portfolio analysis - Single patent analysis support - Robust error handling (continues on partial failures) - Progress logging for user visibility Updated main.py with clean example usage demonstrating the high-level API. Added comprehensive test suite (7 tests) covering: - Full pipeline integration - Error handling at each stage - Single patent analysis - Edge cases (no patents, all failures) All 26 tests passing. This completes the core functionality for patent-based company performance estimation. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
179 lines
7.1 KiB
Python
179 lines
7.1 KiB
Python
"""Tests for the high-level company analyzer orchestration."""
|
|
|
|
import pytest
|
|
from unittest.mock import Mock, patch
|
|
from SPARC.analyzer import CompanyAnalyzer
|
|
from SPARC.types import Patent, Patents
|
|
|
|
|
|
class TestCompanyAnalyzer:
|
|
"""Test the CompanyAnalyzer orchestration logic."""
|
|
|
|
def test_analyzer_initialization(self, mocker):
|
|
"""Test analyzer initialization with API key."""
|
|
mock_llm = mocker.patch("SPARC.analyzer.LLMAnalyzer")
|
|
|
|
analyzer = CompanyAnalyzer(anthropic_api_key="test-key")
|
|
|
|
mock_llm.assert_called_once_with(api_key="test-key")
|
|
|
|
def test_analyze_company_full_pipeline(self, mocker):
|
|
"""Test complete company analysis pipeline."""
|
|
# Mock all the dependencies
|
|
mock_query = mocker.patch("SPARC.analyzer.SERP.query")
|
|
mock_save = mocker.patch("SPARC.analyzer.SERP.save_patents")
|
|
mock_parse = mocker.patch("SPARC.analyzer.SERP.parse_patent_pdf")
|
|
mock_minimize = mocker.patch("SPARC.analyzer.SERP.minimize_patent_for_llm")
|
|
mock_llm = mocker.patch("SPARC.analyzer.LLMAnalyzer")
|
|
|
|
# Setup mock return values
|
|
test_patent = Patent(
|
|
patent_id="US123", pdf_link="http://example.com/test.pdf"
|
|
)
|
|
mock_query.return_value = Patents(patents=[test_patent])
|
|
|
|
test_patent.pdf_path = "patents/US123.pdf"
|
|
mock_save.return_value = test_patent
|
|
|
|
mock_parse.return_value = {
|
|
"abstract": "Test abstract",
|
|
"claims": "Test claims",
|
|
}
|
|
|
|
mock_minimize.return_value = "Minimized content"
|
|
|
|
mock_llm_instance = Mock()
|
|
mock_llm_instance.analyze_patent_portfolio.return_value = (
|
|
"Strong innovation portfolio"
|
|
)
|
|
mock_llm.return_value = mock_llm_instance
|
|
|
|
# Run the analysis
|
|
analyzer = CompanyAnalyzer()
|
|
result = analyzer.analyze_company("TestCorp")
|
|
|
|
# Verify the pipeline executed correctly
|
|
assert result == "Strong innovation portfolio"
|
|
mock_query.assert_called_once_with("TestCorp")
|
|
mock_save.assert_called_once()
|
|
mock_parse.assert_called_once_with("patents/US123.pdf")
|
|
mock_minimize.assert_called_once()
|
|
mock_llm_instance.analyze_patent_portfolio.assert_called_once()
|
|
|
|
# Verify the data passed to LLM
|
|
llm_call_args = mock_llm_instance.analyze_patent_portfolio.call_args
|
|
patents_data = llm_call_args[1]["patents_data"]
|
|
assert len(patents_data) == 1
|
|
assert patents_data[0]["patent_id"] == "US123"
|
|
assert patents_data[0]["content"] == "Minimized content"
|
|
|
|
def test_analyze_company_no_patents_found(self, mocker):
|
|
"""Test handling when no patents are found for a company."""
|
|
mock_query = mocker.patch("SPARC.analyzer.SERP.query")
|
|
mock_query.return_value = Patents(patents=[])
|
|
mocker.patch("SPARC.analyzer.LLMAnalyzer")
|
|
|
|
analyzer = CompanyAnalyzer()
|
|
result = analyzer.analyze_company("UnknownCorp")
|
|
|
|
assert result == "No patents found for UnknownCorp"
|
|
|
|
def test_analyze_company_handles_processing_errors(self, mocker):
|
|
"""Test that analysis continues even if some patents fail to process."""
|
|
mock_query = mocker.patch("SPARC.analyzer.SERP.query")
|
|
mock_save = mocker.patch("SPARC.analyzer.SERP.save_patents")
|
|
mock_parse = mocker.patch("SPARC.analyzer.SERP.parse_patent_pdf")
|
|
mock_minimize = mocker.patch("SPARC.analyzer.SERP.minimize_patent_for_llm")
|
|
mock_llm = mocker.patch("SPARC.analyzer.LLMAnalyzer")
|
|
|
|
# Create two test patents
|
|
patent1 = Patent(patent_id="US123", pdf_link="http://example.com/1.pdf")
|
|
patent2 = Patent(patent_id="US456", pdf_link="http://example.com/2.pdf")
|
|
mock_query.return_value = Patents(patents=[patent1, patent2])
|
|
|
|
# First patent processes successfully
|
|
patent1.pdf_path = "patents/US123.pdf"
|
|
|
|
# Second patent raises an error
|
|
def save_side_effect(p):
|
|
if p.patent_id == "US123":
|
|
p.pdf_path = "patents/US123.pdf"
|
|
return p
|
|
else:
|
|
raise Exception("Download failed")
|
|
|
|
mock_save.side_effect = save_side_effect
|
|
|
|
mock_parse.return_value = {"abstract": "Test"}
|
|
mock_minimize.return_value = "Content"
|
|
|
|
mock_llm_instance = Mock()
|
|
mock_llm_instance.analyze_patent_portfolio.return_value = "Analysis result"
|
|
mock_llm.return_value = mock_llm_instance
|
|
|
|
analyzer = CompanyAnalyzer()
|
|
result = analyzer.analyze_company("TestCorp")
|
|
|
|
# Should still succeed with the one patent that worked
|
|
assert result == "Analysis result"
|
|
|
|
# Verify only one patent was analyzed
|
|
llm_call_args = mock_llm_instance.analyze_patent_portfolio.call_args
|
|
patents_data = llm_call_args[1]["patents_data"]
|
|
assert len(patents_data) == 1
|
|
assert patents_data[0]["patent_id"] == "US123"
|
|
|
|
def test_analyze_company_all_patents_fail(self, mocker):
|
|
"""Test handling when all patents fail to process."""
|
|
mock_query = mocker.patch("SPARC.analyzer.SERP.query")
|
|
mock_save = mocker.patch("SPARC.analyzer.SERP.save_patents")
|
|
mocker.patch("SPARC.analyzer.LLMAnalyzer")
|
|
|
|
patent = Patent(patent_id="US123", pdf_link="http://example.com/1.pdf")
|
|
mock_query.return_value = Patents(patents=[patent])
|
|
|
|
# Make processing fail
|
|
mock_save.side_effect = Exception("Processing error")
|
|
|
|
analyzer = CompanyAnalyzer()
|
|
result = analyzer.analyze_company("TestCorp")
|
|
|
|
assert result == "Failed to process any patents for TestCorp"
|
|
|
|
def test_analyze_single_patent(self, mocker):
|
|
"""Test single patent analysis."""
|
|
mock_parse = mocker.patch("SPARC.analyzer.SERP.parse_patent_pdf")
|
|
mock_minimize = mocker.patch("SPARC.analyzer.SERP.minimize_patent_for_llm")
|
|
mock_llm = mocker.patch("SPARC.analyzer.LLMAnalyzer")
|
|
|
|
mock_parse.return_value = {"abstract": "Test abstract"}
|
|
mock_minimize.return_value = "Minimized content"
|
|
|
|
mock_llm_instance = Mock()
|
|
mock_llm_instance.analyze_patent_content.return_value = (
|
|
"Innovative patent analysis"
|
|
)
|
|
mock_llm.return_value = mock_llm_instance
|
|
|
|
analyzer = CompanyAnalyzer()
|
|
result = analyzer.analyze_single_patent("US123", "TestCorp")
|
|
|
|
assert result == "Innovative patent analysis"
|
|
mock_parse.assert_called_once_with("patents/US123.pdf")
|
|
mock_llm_instance.analyze_patent_content.assert_called_once_with(
|
|
patent_content="Minimized content", company_name="TestCorp"
|
|
)
|
|
|
|
def test_analyze_single_patent_error_handling(self, mocker):
|
|
"""Test single patent analysis with processing error."""
|
|
mock_parse = mocker.patch("SPARC.analyzer.SERP.parse_patent_pdf")
|
|
mocker.patch("SPARC.analyzer.LLMAnalyzer")
|
|
|
|
mock_parse.side_effect = FileNotFoundError("PDF not found")
|
|
|
|
analyzer = CompanyAnalyzer()
|
|
result = analyzer.analyze_single_patent("US999", "TestCorp")
|
|
|
|
assert "Failed to analyze patent US999" in result
|
|
assert "PDF not found" in result
|