Spaces:
Running
Running
| """Unit tests for /analyze endpoint - Catalyst API performance and correctness.""" | |
| import pytest | |
| from core.scanner_service import ScannerService | |
| class TestStockCatalystsEndpoint: | |
| """Test stock catalysts API endpoint - ultra fast path.""" | |
| def scanner_service(self): | |
| """Create scanner service instance.""" | |
| return ScannerService() | |
| class TestCatalystStructure: | |
| """Test catalyst data structure.""" | |
| def test_catalyst_has_required_keys(self, scanner_service): | |
| """Verify catalyst response has all required keys (even when no catalyst).""" | |
| result = scanner_service.finviz.fetch_catalyst("AAPL") | |
| # Required keys - always present even if no catalyst | |
| assert "has_catalyst" in result | |
| assert "symbol" in result | |
| # When has_catalyst is True, these should be present | |
| if result.get("has_catalyst"): | |
| assert "headline" in result | |
| assert "summary" in result | |
| assert "source" in result | |
| assert "datetime" in result | |
| def test_catalyst_company_info_when_available(self, scanner_service): | |
| """Verify catalyst includes company info when catalyst exists.""" | |
| result = scanner_service.finviz.fetch_catalyst("AAPL") | |
| # Company info only present when catalyst exists | |
| if result.get("has_catalyst"): | |
| assert "sector" in result | |
| assert "industry" in result | |
| assert "country" in result | |
| def test_catalyst_sentiment_when_available(self, scanner_service): | |
| """Verify catalyst includes sentiment when catalyst exists.""" | |
| result = scanner_service.finviz.fetch_catalyst("AAPL") | |
| # Sentiment only present when catalyst exists | |
| if result.get("has_catalyst"): | |
| assert "sentiment" in result | |
| assert result["sentiment"] in ["positive", "negative", "neutral", None] | |
| class TestCatalystContent: | |
| """Test catalyst content validation.""" | |
| def test_headline_not_empty_when_has_catalyst(self, scanner_service): | |
| """If has_catalyst is True, headline should not be empty.""" | |
| result = scanner_service.finviz.fetch_catalyst("AAPL") | |
| if result.get("has_catalyst"): | |
| assert result.get("headline") | |
| assert len(result.get("headline", "")) > 0 | |
| def test_summary_not_truncated(self, scanner_service): | |
| """Summary should be complete, not artificially truncated.""" | |
| result = scanner_service.finviz.fetch_catalyst("AAPL") | |
| if result.get("summary"): | |
| # Summary should NOT end with "..." (frontend shows full text) | |
| assert not result["summary"].endswith("...") | |
| def test_datetime_is_valid_format(self, scanner_service): | |
| """Datetime should be parseable.""" | |
| result = scanner_service.finviz.fetch_catalyst("AAPL") | |
| if result.get("datetime"): | |
| from datetime import datetime | |
| # Should be parseable | |
| try: | |
| datetime.fromisoformat(result["datetime"].replace("Z", "+00:00")) | |
| except (ValueError, AttributeError): | |
| pytest.fail(f"Invalid datetime format: {result.get('datetime')}") | |
| class TestPerformance: | |
| """Test catalyst fetch performance - must be fast.""" | |
| def test_catalyst_fetch_under_5_seconds(self, scanner_service): | |
| """Catalyst fetch should complete in under 5 seconds.""" | |
| import time | |
| start = time.time() | |
| # Fetch both catalyst and fundamentals (like the endpoint does) | |
| _ = scanner_service.finviz.fetch_catalyst("AAPL") | |
| _ = scanner_service.finviz.fetch_fundamentals("AAPL") | |
| elapsed = time.time() - start | |
| # Must be fast (< 5 seconds for 2 Finviz calls) | |
| assert elapsed < 5.0, f"Finviz fetch took {elapsed:.2f}s, expected < 5s" | |
| def test_catalyst_fetch_typically_under_3_seconds(self, scanner_service): | |
| """Catalyst + fundamentals fetch should typically complete in under 3 seconds.""" | |
| import time | |
| start = time.time() | |
| # Fetch both catalyst and fundamentals (like the endpoint does) | |
| _ = scanner_service.finviz.fetch_catalyst("AAPL") | |
| _ = scanner_service.finviz.fetch_fundamentals("AAPL") | |
| elapsed = time.time() - start | |
| # Target: < 3 seconds (2x Finviz calls) | |
| # This is a soft requirement (may fail occasionally due to network) | |
| assert elapsed < 3.0, f"Finviz fetch took {elapsed:.2f}s, target < 3s" | |
| class TestMultipleSymbols: | |
| """Test catalyst fetch for multiple symbols.""" | |
| def test_catalyst_for_different_symbols(self, scanner_service, symbol): | |
| """Verify catalyst works for various symbols (may not have catalyst).""" | |
| result = scanner_service.finviz.fetch_catalyst(symbol) | |
| # Should have basic structure | |
| assert result is not None | |
| assert "has_catalyst" in result | |
| assert "symbol" in result | |
| # If has catalyst, verify structure | |
| if result.get("has_catalyst"): | |
| assert "headline" in result | |
| class TestEdgeCases: | |
| """Test edge cases and error handling.""" | |
| def test_invalid_symbol_returns_empty_catalyst(self, scanner_service): | |
| """Invalid symbol should return empty catalyst, not crash.""" | |
| result = scanner_service.finviz.fetch_catalyst("INVALIDXYZ123") | |
| # Should not crash, returns empty catalyst | |
| assert result is not None | |
| assert result.get("has_catalyst") is False or not result.get("headline") | |
| def test_empty_symbol_handling(self, scanner_service): | |
| """Empty symbol should be handled gracefully.""" | |
| # Empty string should not crash | |
| result = scanner_service.finviz.fetch_catalyst("") | |
| assert result is not None | |
| class TestAnalyzePageRequirements: | |
| """Test specific requirements for /analyze page.""" | |
| def test_catalyst_sufficient_for_display(self, scanner_service): | |
| """ | |
| Verify catalyst has all data needed for /analyze page. | |
| The page shows: | |
| - Symbol (from API response) | |
| - Headline (from catalyst, if exists) | |
| - Summary (from catalyst, if exists) | |
| - Source + Datetime (from catalyst, if exists) | |
| When no catalyst: shows "No catalyst" message | |
| """ | |
| result = scanner_service.finviz.fetch_catalyst("AAPL") | |
| # Symbol is always required | |
| assert "symbol" in result | |
| assert result["symbol"] == "AAPL" | |
| # has_catalyst determines what to show | |
| if result.get("has_catalyst"): | |
| # Must have all fields for display | |
| required_for_display = { | |
| "headline": str, | |
| "summary": str, | |
| "source": str, | |
| "datetime": str, | |
| } | |
| for field, field_type in required_for_display.items(): | |
| assert field in result, f"Missing required field: {field}" | |
| if result.get(field): | |
| assert isinstance(result[field], field_type), f"Field {field} should be {field_type}" | |
| else: | |
| # No catalyst - frontend shows "No catalyst" message | |
| pass | |
| def test_no_external_dependencies_for_catalyst(self, scanner_service): | |
| """ | |
| Catalyst endpoint should only call Finviz (not Yahoo, not SEC). | |
| This is a performance requirement - multiple external calls | |
| would make the page slow. | |
| Now fetches both catalyst + fundamentals from Finviz (2 calls total). | |
| """ | |
| import time | |
| # Measure time for catalyst + fundamentals fetch | |
| start = time.time() | |
| _ = scanner_service.finviz.fetch_catalyst("AAPL") | |
| _ = scanner_service.finviz.fetch_fundamentals("AAPL") | |
| elapsed = time.time() - start | |
| # If it takes > 5 seconds, likely making calls to other services | |
| # (2x Finviz calls should be < 3-4 seconds) | |
| assert elapsed < 5.0, f"Finviz fetch took {elapsed:.2f}s - may have unnecessary dependencies" | |