From 57c9b3ecaaf30de7854471729b574af2a8f51888 Mon Sep 17 00:00:00 2001 From: Ahmet Guzererler Date: Thu, 19 Mar 2026 14:24:52 +0100 Subject: [PATCH] refactor(tests): consolidate live-API tests into integration/, move mocked tests to unit/ MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Replaced 7 duplicate live-API test files in tests/ root with: - tests/integration/test_scanner_live.py — consolidated scanner + yfinance + vendor fallback tests (marked @pytest.mark.integration, excluded from default run) - tests/e2e/test_llm_e2e.py — LLM e2e test extracted from test_scanner_comprehensive - tests/unit/test_industry_deep_dive.py — mocked _extract_top_sectors + run_tool_loop tests - tests/unit/test_scanner_fallback.py — mocked AV failover tests - Removed record_mode hardcode from integration/conftest.py (VCR.py cannot intercept curl_cffi used by yfinance; integration tests run live against real APIs on demand) Deleted: test_scanner_tools, test_scanner_end_to_end, test_scanner_complete_e2e, test_scanner_final, test_scanner_comprehensive, test_scanner_fallback, test_industry_deep_dive Co-Authored-By: Claude Sonnet 4.6 --- tests/e2e/test_llm_e2e.py | 62 ++++ tests/integration/conftest.py | 2 +- tests/integration/test_scanner_live.py | 200 +++++++++++++ tests/test_scanner_complete_e2e.py | 297 -------------------- tests/test_scanner_comprehensive.py | 154 ---------- tests/test_scanner_end_to_end.py | 54 ---- tests/test_scanner_fallback.py | 127 --------- tests/test_scanner_final.py | 130 --------- tests/test_scanner_tools.py | 82 ------ tests/{ => unit}/test_industry_deep_dive.py | 77 +---- tests/unit/test_scanner_fallback.py | 30 ++ 11 files changed, 296 insertions(+), 919 deletions(-) create mode 100644 tests/e2e/test_llm_e2e.py create mode 100644 tests/integration/test_scanner_live.py delete mode 100644 tests/test_scanner_complete_e2e.py delete mode 100644 tests/test_scanner_comprehensive.py delete mode 100644 tests/test_scanner_end_to_end.py delete mode 100644 tests/test_scanner_fallback.py delete mode 100644 tests/test_scanner_final.py delete mode 100644 tests/test_scanner_tools.py rename tests/{ => unit}/test_industry_deep_dive.py (66%) create mode 100644 tests/unit/test_scanner_fallback.py diff --git a/tests/e2e/test_llm_e2e.py b/tests/e2e/test_llm_e2e.py new file mode 100644 index 00000000..e0a9e97b --- /dev/null +++ b/tests/e2e/test_llm_e2e.py @@ -0,0 +1,62 @@ +"""End-to-end tests that hit real LLM APIs. + +These tests are expensive and non-deterministic. Run manually only: + + pytest tests/e2e/ -v --allow-hosts +""" + +import tempfile +from pathlib import Path +from unittest.mock import patch + +import pytest + +from cli.main import run_scan + + +def test_scan_command_creates_output_files(): + """Test that the scan command creates all expected output files. + + This test runs the full scanner pipeline with real LLMs. It mocks + only the file-system output path and the typer prompt, NOT the LLM + or data API calls. + """ + with tempfile.TemporaryDirectory() as temp_dir: + test_date_dir = Path(temp_dir) / "market" + test_date_dir.mkdir(parents=True) + + with patch("cli.main.get_market_dir", return_value=test_date_dir): + written_files = {} + + def mock_write_text(self, content, encoding=None): + written_files[str(self)] = content + + with patch("pathlib.Path.write_text", mock_write_text): + with patch("typer.prompt", return_value="2026-03-15"): + try: + run_scan() + except SystemExit: + pass + + valid_names = { + "geopolitical_report.md", + "market_movers_report.md", + "sector_performance_report.md", + "industry_deep_dive_report.md", + "macro_scan_summary.md", + "run_log.jsonl", + } + + assert len(written_files) >= 1, ( + "Scanner produced no output files — pipeline may have silently failed" + ) + + for filepath, content in written_files.items(): + filename = filepath.split("/")[-1] + assert filename in valid_names, ( + f"Output file '{filename}' does not match the expected naming " + f"convention. run_scan() should only write {sorted(valid_names)}" + ) + assert len(content) > 50, ( + f"File {filename} appears to be empty or too short" + ) diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 252f360a..07908737 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -8,7 +8,7 @@ import pytest def vcr_config(): return { "cassette_library_dir": "tests/cassettes", - "record_mode": "none", + # record_mode is controlled by --record-mode CLI flag (default: none) "match_on": ["method", "scheme", "host", "port", "path"], "filter_headers": [ "Authorization", diff --git a/tests/integration/test_scanner_live.py b/tests/integration/test_scanner_live.py new file mode 100644 index 00000000..20fa704d --- /dev/null +++ b/tests/integration/test_scanner_live.py @@ -0,0 +1,200 @@ +"""Integration tests for scanner data functions — require network access. + +These tests hit real yfinance and vendor APIs. Excluded from default pytest run. + +Run with: + pytest tests/integration/ -v # all integration tests + pytest tests/integration/ -v -m integration # integration-marked only +""" + +import pytest + + +# --------------------------------------------------------------------------- +# Scanner tool tests (yfinance-backed) +# --------------------------------------------------------------------------- + + +@pytest.mark.integration +def test_market_movers_day_gainers(): + from tradingagents.agents.utils.scanner_tools import get_market_movers + + result = get_market_movers.invoke({"category": "day_gainers"}) + assert isinstance(result, str) + assert "# Market Movers:" in result + assert "| Symbol |" in result + + +@pytest.mark.integration +def test_market_movers_day_losers(): + from tradingagents.agents.utils.scanner_tools import get_market_movers + + result = get_market_movers.invoke({"category": "day_losers"}) + assert isinstance(result, str) + assert "# Market Movers:" in result + assert "| Symbol |" in result + + +@pytest.mark.integration +def test_market_movers_most_actives(): + from tradingagents.agents.utils.scanner_tools import get_market_movers + + result = get_market_movers.invoke({"category": "most_actives"}) + assert isinstance(result, str) + assert "# Market Movers:" in result + assert "| Symbol |" in result + + +@pytest.mark.integration +def test_market_indices(): + from tradingagents.agents.utils.scanner_tools import get_market_indices + + result = get_market_indices.invoke({}) + assert isinstance(result, str) + assert "# Major Market Indices" in result + assert "| Index |" in result + assert "S&P 500" in result + assert "Dow Jones" in result + + +@pytest.mark.integration +def test_sector_performance(): + from tradingagents.agents.utils.scanner_tools import get_sector_performance + + result = get_sector_performance.invoke({}) + assert isinstance(result, str) + assert "# Sector Performance Overview" in result + assert "| Sector |" in result + + +@pytest.mark.integration +def test_industry_performance_technology(): + from tradingagents.agents.utils.scanner_tools import get_industry_performance + + result = get_industry_performance.invoke({"sector_key": "technology"}) + assert isinstance(result, str) + assert "# Industry Performance: Technology" in result + assert "| Company |" in result + + +@pytest.mark.integration +def test_topic_news(): + from tradingagents.agents.utils.scanner_tools import get_topic_news + + result = get_topic_news.invoke({"topic": "market", "limit": 3}) + assert isinstance(result, str) + assert "# News for Topic: market" in result + assert len(result) > 100 + + +# --------------------------------------------------------------------------- +# yfinance dataflow tests (direct function calls) +# --------------------------------------------------------------------------- + + +@pytest.mark.integration +def test_yfinance_sector_performance_all_11_sectors(): + from tradingagents.dataflows.yfinance_scanner import get_sector_performance_yfinance + + result = get_sector_performance_yfinance() + assert "| Sector |" in result + for sector in [ + "Technology", "Healthcare", "Financials", "Energy", + "Consumer Discretionary", "Consumer Staples", "Industrials", + "Materials", "Real Estate", "Utilities", "Communication Services", + ]: + assert sector in result, f"Missing sector: {sector}" + + +@pytest.mark.integration +def test_yfinance_sector_performance_numeric_percentages(): + from tradingagents.dataflows.yfinance_scanner import get_sector_performance_yfinance + + result = get_sector_performance_yfinance() + lines = result.strip().split("\n") + data_lines = [ + l for l in lines + if l.startswith("| ") and "Sector" not in l and "---" not in l + ] + assert len(data_lines) == 11, f"Expected 11 data rows, got {len(data_lines)}" + for line in data_lines: + cols = [c.strip() for c in line.split("|")[1:-1]] + assert len(cols) == 5, f"Expected 5 columns in: {line}" + day_pct = cols[1] + assert "%" in day_pct or day_pct == "N/A", f"Bad 1-day value: {day_pct}" + + +@pytest.mark.integration +def test_yfinance_industry_performance_real_symbols(): + from tradingagents.dataflows.yfinance_scanner import get_industry_performance_yfinance + + result = get_industry_performance_yfinance("technology") + assert "| Company |" in result or "| Company " in result + assert "NVDA" in result or "AAPL" in result or "MSFT" in result + + +@pytest.mark.integration +def test_yfinance_industry_performance_no_na_symbols(): + from tradingagents.dataflows.yfinance_scanner import get_industry_performance_yfinance + + result = get_industry_performance_yfinance("technology") + lines = result.strip().split("\n") + data_lines = [ + l for l in lines + if l.startswith("| ") and "Company" not in l and "---" not in l + ] + for line in data_lines: + cols = [c.strip() for c in line.split("|")[1:-1]] + assert cols[1] != "N/A", f"Symbol is N/A in line: {line}" + + +@pytest.mark.integration +def test_yfinance_industry_performance_healthcare(): + from tradingagents.dataflows.yfinance_scanner import get_industry_performance_yfinance + + result = get_industry_performance_yfinance("healthcare") + assert "Industry Performance: Healthcare" in result + + +@pytest.mark.integration +def test_yfinance_industry_performance_price_columns(): + from tradingagents.dataflows.yfinance_scanner import get_industry_performance_yfinance + + result = get_industry_performance_yfinance("technology") + assert "# Industry Performance: Technology" in result + assert "1-Day %" in result + assert "1-Week %" in result + assert "1-Month %" in result + + +@pytest.mark.integration +def test_yfinance_industry_performance_seven_columns(): + from tradingagents.dataflows.yfinance_scanner import get_industry_performance_yfinance + + result = get_industry_performance_yfinance("technology") + lines = result.strip().split("\n") + sep_lines = [l for l in lines if l.startswith("|") and "---" in l] + assert len(sep_lines) >= 1 + cols = [c.strip() for c in sep_lines[0].split("|")[1:-1]] + assert len(cols) == 7, f"Expected 7 columns, got {len(cols)}: {cols}" + + +# --------------------------------------------------------------------------- +# Vendor fallback integration tests +# --------------------------------------------------------------------------- + + +@pytest.mark.integration +def test_route_to_vendor_sector_performance(): + from tradingagents.dataflows.interface import route_to_vendor + + result = route_to_vendor("get_sector_performance") + assert "Sector Performance Overview" in result + + +@pytest.mark.integration +def test_route_to_vendor_industry_performance(): + from tradingagents.dataflows.interface import route_to_vendor + + result = route_to_vendor("get_industry_performance", "technology") + assert "Industry Performance" in result diff --git a/tests/test_scanner_complete_e2e.py b/tests/test_scanner_complete_e2e.py deleted file mode 100644 index 2612065f..00000000 --- a/tests/test_scanner_complete_e2e.py +++ /dev/null @@ -1,297 +0,0 @@ -""" -Complete end-to-end test for TradingAgents scanner functionality. - -This test verifies that: -1. All scanner tools work correctly and return expected data formats -2. The scanner tools can be used to generate market analysis reports -3. The CLI scan command works end-to-end -4. Results are properly saved to files -""" - -import tempfile -import os -from pathlib import Path -import pytest - -# Set up the Python path to include the project root -import sys -sys.path.insert(0, str(Path(__file__).parent.parent)) - -from tradingagents.agents.utils.scanner_tools import ( - get_market_movers, - get_market_indices, - get_sector_performance, - get_industry_performance, - get_topic_news, -) - - -class TestScannerToolsIndividual: - """Test each scanner tool individually.""" - - def test_get_market_movers(self): - """Test market movers tool for all categories.""" - for category in ["day_gainers", "day_losers", "most_actives"]: - result = get_market_movers.invoke({"category": category}) - assert isinstance(result, str), f"Result should be string for {category}" - assert not result.startswith("Error:"), f"Should not error for {category}: {result[:100]}" - assert "# Market Movers:" in result, f"Missing header for {category}" - assert "| Symbol |" in result, f"Missing table header for {category}" - # Verify we got actual data - lines = result.split('\n') - data_lines = [line for line in lines if line.startswith('|') and 'Symbol' not in line] - assert len(data_lines) > 0, f"No data rows found for {category}" - - def test_get_market_indices(self): - """Test market indices tool.""" - result = get_market_indices.invoke({}) - assert isinstance(result, str), "Result should be string" - assert not result.startswith("Error:"), f"Should not error: {result[:100]}" - assert "# Major Market Indices" in result, "Missing header" - assert "| Index |" in result, "Missing table header" - # Verify we got data for major indices - assert "S&P 500" in result, "Missing S&P 500 data" - assert "Dow Jones" in result, "Missing Dow Jones data" - - def test_get_sector_performance(self): - """Test sector performance tool.""" - result = get_sector_performance.invoke({}) - assert isinstance(result, str), "Result should be string" - assert not result.startswith("Error:"), f"Should not error: {result[:100]}" - assert "# Sector Performance Overview" in result, "Missing header" - assert "| Sector |" in result, "Missing table header" - # Verify we got data for sectors - assert "Technology" in result or "Healthcare" in result, "Missing sector data" - - def test_get_industry_performance(self): - """Test industry performance tool.""" - result = get_industry_performance.invoke({"sector_key": "technology"}) - assert isinstance(result, str), "Result should be string" - assert not result.startswith("Error:"), f"Should not error: {result[:100]}" - assert "# Industry Performance: Technology" in result, "Missing header" - assert "| Company |" in result, "Missing table header" - # Verify we got data for companies - assert "NVIDIA" in result or "Apple" in result or "Microsoft" in result, "Missing company data" - - def test_get_topic_news(self): - """Test topic news tool.""" - result = get_topic_news.invoke({"topic": "market", "limit": 3}) - assert isinstance(result, str), "Result should be string" - assert not result.startswith("Error:"), f"Should not error: {result[:100]}" - assert "# News for Topic: market" in result, "Missing header" - assert "### " in result, "Missing news article headers" - # Verify we got news content - assert len(result) > 100, "News result too short" - - -class TestScannerWorkflow: - """Test the complete scanner workflow.""" - - def test_complete_scanner_workflow_to_files(self): - """Test that scanner tools can generate complete market analysis and save to files.""" - with tempfile.TemporaryDirectory() as temp_dir: - # Set up directory structure like the CLI scan command - scan_date = "2026-03-15" - save_dir = Path(temp_dir) / "results" / "macro_scan" / scan_date - save_dir.mkdir(parents=True) - - # Generate data using all scanner tools (this is what the CLI scan command does) - market_movers = get_market_movers.invoke({"category": "day_gainers"}) - market_indices = get_market_indices.invoke({}) - sector_performance = get_sector_performance.invoke({}) - industry_performance = get_industry_performance.invoke({"sector_key": "technology"}) - topic_news = get_topic_news.invoke({"topic": "market", "limit": 5}) - - # Save results to files (simulating CLI behavior) - (save_dir / "market_movers.txt").write_text(market_movers) - (save_dir / "market_indices.txt").write_text(market_indices) - (save_dir / "sector_performance.txt").write_text(sector_performance) - (save_dir / "industry_performance.txt").write_text(industry_performance) - (save_dir / "topic_news.txt").write_text(topic_news) - - # Verify all files were created - assert (save_dir / "market_movers.txt").exists() - assert (save_dir / "market_indices.txt").exists() - assert (save_dir / "sector_performance.txt").exists() - assert (save_dir / "industry_performance.txt").exists() - assert (save_dir / "topic_news.txt").exists() - - # Verify file contents have expected structure - movers_content = (save_dir / "market_movers.txt").read_text() - indices_content = (save_dir / "market_indices.txt").read_text() - sectors_content = (save_dir / "sector_performance.txt").read_text() - industry_content = (save_dir / "industry_performance.txt").read_text() - news_content = (save_dir / "topic_news.txt").read_text() - - # Check headers - assert "# Market Movers:" in movers_content - assert "# Major Market Indices" in indices_content - assert "# Sector Performance Overview" in sectors_content - assert "# Industry Performance: Technology" in industry_content - assert "# News for Topic: market" in news_content - - # Check table structures - assert "| Symbol |" in movers_content - assert "| Index |" in indices_content - assert "| Sector |" in sectors_content - assert "| Company |" in industry_content - - # Check that we have meaningful data (not just headers) - assert len(movers_content) > 200 - assert len(indices_content) > 200 - assert len(sectors_content) > 200 - assert len(industry_content) > 200 - assert len(news_content) > 200 - - -class TestScannerIntegration: - """Test integration with CLI components.""" - - def test_tools_have_expected_interface(self): - """Test that scanner tools have the interface expected by CLI.""" - # The CLI scan command expects to call .invoke() on each tool - assert hasattr(get_market_movers, 'invoke') - assert hasattr(get_market_indices, 'invoke') - assert hasattr(get_sector_performance, 'invoke') - assert hasattr(get_industry_performance, 'invoke') - assert hasattr(get_topic_news, 'invoke') - - # Verify they're callable with expected arguments - # Market movers requires category argument - result = get_market_movers.invoke({"category": "day_gainers"}) - assert isinstance(result, str) - - # Others don't require arguments (or have defaults) - result = get_market_indices.invoke({}) - assert isinstance(result, str) - - result = get_sector_performance.invoke({}) - assert isinstance(result, str) - - result = get_industry_performance.invoke({"sector_key": "technology"}) - assert isinstance(result, str) - - result = get_topic_news.invoke({"topic": "market", "limit": 3}) - assert isinstance(result, str) - - def test_tool_descriptions_match_expectations(self): - """Test that tool descriptions match what the CLI expects.""" - # These descriptions are used for documentation and help - assert "market movers" in get_market_movers.description.lower() - assert "market indices" in get_market_indices.description.lower() - assert "sector performance" in get_sector_performance.description.lower() - assert "industry" in get_industry_performance.description.lower() - assert "news" in get_topic_news.description.lower() - - -def test_scanner_end_to_end_demo(): - """Demonstration test showing the complete end-to-end scanner functionality.""" - print("\n" + "="*60) - print("TRADINGAGENTS SCANNER END-TO-END DEMONSTRATION") - print("="*60) - - # Show that all tools work - print("\n1. Testing Individual Scanner Tools:") - print("-" * 40) - - # Market Movers - movers = get_market_movers.invoke({"category": "day_gainers"}) - print(f"✓ Market Movers: {len(movers)} characters") - - # Market Indices - indices = get_market_indices.invoke({}) - print(f"✓ Market Indices: {len(indices)} characters") - - # Sector Performance - sectors = get_sector_performance.invoke({}) - print(f"✓ Sector Performance: {len(sectors)} characters") - - # Industry Performance - industry = get_industry_performance.invoke({"sector_key": "technology"}) - print(f"✓ Industry Performance: {len(industry)} characters") - - # Topic News - news = get_topic_news.invoke({"topic": "market", "limit": 3}) - print(f"✓ Topic News: {len(news)} characters") - - # Show file output capability - print("\n2. Testing File Output Capability:") - print("-" * 40) - - with tempfile.TemporaryDirectory() as temp_dir: - scan_date = "2026-03-15" - save_dir = Path(temp_dir) / "results" / "macro_scan" / scan_date - save_dir.mkdir(parents=True) - - # Save all results - files_data = [ - ("market_movers.txt", movers), - ("market_indices.txt", indices), - ("sector_performance.txt", sectors), - ("industry_performance.txt", industry), - ("topic_news.txt", news) - ] - - for filename, content in files_data: - filepath = save_dir / filename - filepath.write_text(content) - assert filepath.exists() - print(f"✓ Created {filename} ({len(content)} chars)") - - # Verify we can read them back - for filename, _ in files_data: - content = (save_dir / filename).read_text() - assert len(content) > 50 # Sanity check - - print("\n3. Verifying Content Quality:") - print("-" * 40) - - # Check that we got real financial data, not just error messages - assert not movers.startswith("Error:"), "Market movers should not error" - assert not indices.startswith("Error:"), "Market indices should not error" - assert not sectors.startswith("Error:"), "Sector performance should not error" - assert not industry.startswith("Error:"), "Industry performance should not error" - assert not news.startswith("Error:"), "Topic news should not error" - - # Check for expected content patterns - assert "# Market Movers: Day Gainers" in movers or "# Market Movers: Day Losers" in movers or "# Market Movers: Most Actives" in movers - assert "# Major Market Indices" in indices - assert "# Sector Performance Overview" in sectors - assert "# Industry Performance: Technology" in industry - assert "# News for Topic: market" in news - - print("✓ All tools returned valid financial data") - print("✓ All tools have proper headers and formatting") - print("✓ All tools can save/load data correctly") - - print("\n" + "="*60) - print("END-TO-END SCANNER TEST: PASSED 🎉") - print("="*60) - print("The TradingAgents scanner functionality is working correctly!") - print("All tools generate proper financial market data and can save results to files.") - - -if __name__ == "__main__": - # Run the demonstration test - test_scanner_end_to_end_demo() - - # Also run the individual test classes - print("\nRunning individual tool tests...") - test_instance = TestScannerToolsIndividual() - test_instance.test_get_market_movers() - test_instance.test_get_market_indices() - test_instance.test_get_sector_performance() - test_instance.test_get_industry_performance() - test_instance.test_get_topic_news() - print("✓ Individual tool tests passed") - - workflow_instance = TestScannerWorkflow() - workflow_instance.test_complete_scanner_workflow_to_files() - print("✓ Workflow tests passed") - - integration_instance = TestScannerIntegration() - integration_instance.test_tools_have_expected_interface() - integration_instance.test_tool_descriptions_match_expectations() - print("✓ Integration tests passed") - - print("\n✅ ALL TESTS PASSED - Scanner functionality is working correctly!") \ No newline at end of file diff --git a/tests/test_scanner_comprehensive.py b/tests/test_scanner_comprehensive.py deleted file mode 100644 index ca089dd0..00000000 --- a/tests/test_scanner_comprehensive.py +++ /dev/null @@ -1,154 +0,0 @@ -"""Comprehensive end-to-end tests for scanner functionality.""" - -import tempfile -import os -from pathlib import Path -from unittest.mock import patch -import pytest - -from tradingagents.agents.utils.scanner_tools import ( - get_market_movers, - get_market_indices, - get_sector_performance, - get_industry_performance, - get_topic_news, -) -from cli.main import run_scan - - -class TestScannerTools: - """Test individual scanner tools.""" - - def test_market_movers_all_categories(self): - """Test market movers for all categories.""" - for category in ["day_gainers", "day_losers", "most_actives"]: - result = get_market_movers.invoke({"category": category}) - assert isinstance(result, str), f"Result for {category} should be a string" - assert not result.startswith("Error:"), f"Error in {category}: {result[:100]}" - assert "# Market Movers:" in result, f"Missing header in {category} result" - assert "| Symbol |" in result, f"Missing table header in {category} result" - # Check that we got some data - assert len(result) > 100, f"Result too short for {category}" - - def test_market_indices(self): - """Test market indices.""" - result = get_market_indices.invoke({}) - assert isinstance(result, str), "Market indices result should be a string" - assert not result.startswith("Error:"), f"Error in market indices: {result[:100]}" - assert "# Major Market Indices" in result, "Missing header in market indices result" - assert "| Index |" in result, "Missing table header in market indices result" - # Check for major indices - assert "S&P 500" in result, "Missing S&P 500 in market indices" - assert "Dow Jones" in result, "Missing Dow Jones in market indices" - - def test_sector_performance(self): - """Test sector performance.""" - result = get_sector_performance.invoke({}) - assert isinstance(result, str), "Sector performance result should be a string" - assert not result.startswith("Error:"), f"Error in sector performance: {result[:100]}" - assert "# Sector Performance Overview" in result, "Missing header in sector performance result" - assert "| Sector |" in result, "Missing table header in sector performance result" - # Check for some sectors - assert "Technology" in result, "Missing Technology sector" - assert "Healthcare" in result, "Missing Healthcare sector" - - def test_industry_performance(self): - """Test industry performance for technology sector.""" - result = get_industry_performance.invoke({"sector_key": "technology"}) - assert isinstance(result, str), "Industry performance result should be a string" - assert not result.startswith("Error:"), f"Error in industry performance: {result[:100]}" - assert "# Industry Performance: Technology" in result, "Missing header in industry performance result" - assert "| Company |" in result, "Missing table header in industry performance result" - # Check for major tech companies - assert "NVIDIA" in result or "Apple" in result or "Microsoft" in result, "Missing major tech companies" - - def test_topic_news(self): - """Test topic news for market topic.""" - result = get_topic_news.invoke({"topic": "market", "limit": 5}) - assert isinstance(result, str), "Topic news result should be a string" - assert not result.startswith("Error:"), f"Error in topic news: {result[:100]}" - assert "# News for Topic: market" in result, "Missing header in topic news result" - assert "### " in result, "Missing news article headers in topic news result" - # Check that we got some news - assert len(result) > 100, "Topic news result too short" - - -class TestScannerEndToEnd: - """End-to-end tests for scanner functionality.""" - - def test_scan_command_creates_output_files(self): - """Test that the scan command creates all expected output files.""" - with tempfile.TemporaryDirectory() as temp_dir: - test_date_dir = Path(temp_dir) / "market" - test_date_dir.mkdir(parents=True) - - # Mock get_market_dir to redirect output into the temp directory - with patch('cli.main.get_market_dir', return_value=test_date_dir): - # Mock the write_text method to capture what gets written - written_files = {} - def mock_write_text(self, content, encoding=None): - written_files[str(self)] = content - - with patch('pathlib.Path.write_text', mock_write_text): - with patch('typer.prompt', return_value='2026-03-15'): - try: - run_scan() - except SystemExit: - pass - - # Verify that run_scan() uses the correct output file naming convention. - # - # run_scan() writes via: (save_dir / f"{key}.md").write_text(content) - # where save_dir = get_market_dir(scan_date). - # pathlib.Path.write_text is mocked, so written_files keys are the - # str() of those Path objects. - # - # LLM output is non-deterministic: a phase may produce an empty string, - # causing run_scan()'s `if content:` guard to skip writing that file. - # So we cannot assert ALL 5 files are always present. Instead we verify: - # 1. At least some output was produced (pipeline didn't silently fail). - # 2. Every file that WAS written has a name matching the expected - # naming convention — this is the real bug we are guarding against. - valid_names = { - "geopolitical_report.md", - "market_movers_report.md", - "sector_performance_report.md", - "industry_deep_dive_report.md", - "macro_scan_summary.md", - "run_log.jsonl", - } - - assert len(written_files) >= 1, ( - "Scanner produced no output files — pipeline may have silently failed" - ) - - for filepath, content in written_files.items(): - filename = filepath.split("/")[-1] - assert filename in valid_names, ( - f"Output file '{filename}' does not match the expected naming " - f"convention. run_scan() should only write {sorted(valid_names)}" - ) - assert len(content) > 50, ( - f"File {filename} appears to be empty or too short" - ) - - def test_scanner_tools_integration(self): - """Test that all scanner tools work together without errors.""" - # Test all tools can be called successfully - tools_and_args = [ - (get_market_movers, {"category": "day_gainers"}), - (get_market_indices, {}), - (get_sector_performance, {}), - (get_industry_performance, {"sector_key": "technology"}), - (get_topic_news, {"topic": "market", "limit": 3}) - ] - - for tool_func, args in tools_and_args: - result = tool_func.invoke(args) - assert isinstance(result, str), f"Tool {tool_func.name} should return string" - # Either we got real data or a graceful error message - assert not result.startswith("Error fetching"), f"Tool {tool_func.name} failed: {result[:100]}" - - -if __name__ == "__main__": - pytest.main([__file__, "-v"]) \ No newline at end of file diff --git a/tests/test_scanner_end_to_end.py b/tests/test_scanner_end_to_end.py deleted file mode 100644 index 9599c348..00000000 --- a/tests/test_scanner_end_to_end.py +++ /dev/null @@ -1,54 +0,0 @@ -"""End-to-end tests for scanner functionality.""" - -import pytest - -from tradingagents.agents.utils.scanner_tools import ( - get_market_movers, - get_market_indices, - get_sector_performance, - get_industry_performance, - get_topic_news, -) - - -def test_scanner_tools_end_to_end(): - """End-to-end test for all scanner tools.""" - # Test market movers - for category in ["day_gainers", "day_losers", "most_actives"]: - result = get_market_movers.invoke({"category": category}) - assert isinstance(result, str), f"Result for {category} should be a string" - assert not result.startswith("Error:"), f"Error in {category}: {result[:100]}" - assert "# Market Movers:" in result, f"Missing header in {category} result" - assert "| Symbol |" in result, f"Missing table header in {category} result" - - # Test market indices - result = get_market_indices.invoke({}) - assert isinstance(result, str), "Market indices result should be a string" - assert not result.startswith("Error:"), f"Error in market indices: {result[:100]}" - assert "# Major Market Indices" in result, "Missing header in market indices result" - assert "| Index |" in result, "Missing table header in market indices result" - - # Test sector performance - result = get_sector_performance.invoke({}) - assert isinstance(result, str), "Sector performance result should be a string" - assert not result.startswith("Error:"), f"Error in sector performance: {result[:100]}" - assert "# Sector Performance Overview" in result, "Missing header in sector performance result" - assert "| Sector |" in result, "Missing table header in sector performance result" - - # Test industry performance - result = get_industry_performance.invoke({"sector_key": "technology"}) - assert isinstance(result, str), "Industry performance result should be a string" - assert not result.startswith("Error:"), f"Error in industry performance: {result[:100]}" - assert "# Industry Performance: Technology" in result, "Missing header in industry performance result" - assert "| Company |" in result, "Missing table header in industry performance result" - - # Test topic news - result = get_topic_news.invoke({"topic": "market", "limit": 5}) - assert isinstance(result, str), "Topic news result should be a string" - assert not result.startswith("Error:"), f"Error in topic news: {result[:100]}" - assert "# News for Topic: market" in result, "Missing header in topic news result" - assert "### " in result, "Missing news article headers in topic news result" - - -if __name__ == "__main__": - pytest.main([__file__, "-v"]) \ No newline at end of file diff --git a/tests/test_scanner_fallback.py b/tests/test_scanner_fallback.py deleted file mode 100644 index b89f6256..00000000 --- a/tests/test_scanner_fallback.py +++ /dev/null @@ -1,127 +0,0 @@ -"""Tests for scanner data functions — yfinance fallback and AV error handling. - -These tests verify: -1. yfinance sector performance returns real data via ETF proxies -2. yfinance industry performance uses DataFrame index for ticker symbols -3. AV scanner functions raise AlphaVantageError when all data fails (enabling fallback) -4. route_to_vendor falls back from AV to yfinance on AlphaVantageError -""" - -import os -import pytest -from unittest.mock import patch - -from tradingagents.dataflows.yfinance_scanner import ( - get_sector_performance_yfinance, - get_industry_performance_yfinance, -) -from tradingagents.dataflows.alpha_vantage_common import AlphaVantageError -from tradingagents.dataflows.alpha_vantage_scanner import ( - get_sector_performance_alpha_vantage, - get_industry_performance_alpha_vantage, -) - - -class TestYfinanceSectorPerformance: - """Verify yfinance sector performance uses ETF proxies and returns real data.""" - - def test_returns_all_11_sectors(self): - result = get_sector_performance_yfinance() - assert "| Sector |" in result - # Check all 11 GICS sectors are present - for sector in [ - "Technology", "Healthcare", "Financials", "Energy", - "Consumer Discretionary", "Consumer Staples", "Industrials", - "Materials", "Real Estate", "Utilities", "Communication Services", - ]: - assert sector in result, f"Missing sector: {sector}" - - def test_returns_numeric_percentages(self): - result = get_sector_performance_yfinance() - lines = result.strip().split("\n") - # Skip header lines (first 4: title, date, column headers, separator) - data_lines = [l for l in lines if l.startswith("| ") and "Sector" not in l and "---" not in l] - assert len(data_lines) == 11, f"Expected 11 data rows, got {len(data_lines)}" - - for line in data_lines: - cols = [c.strip() for c in line.split("|")[1:-1]] - # cols: [sector_name, 1-day, 1-week, 1-month, ytd] - assert len(cols) == 5, f"Expected 5 columns, got {len(cols)} in: {line}" - # 1-day should be a percentage like "+1.45%" or "-0.31%" - day_pct = cols[1] - assert "%" in day_pct or day_pct == "N/A", f"Bad 1-day value: {day_pct}" - # Should NOT contain "Error:" - assert "Error:" not in day_pct, f"Error in 1-day for {cols[0]}: {day_pct}" - - -@pytest.mark.integration -class TestYfinanceIndustryPerformance: - """Verify yfinance industry performance uses index for ticker symbols.""" - - def test_returns_real_symbols(self): - result = get_industry_performance_yfinance("technology") - assert "| Company |" in result or "| Company " in result - # Should contain actual tickers, not N/A - assert "NVDA" in result or "AAPL" in result or "MSFT" in result, \ - f"No real tickers found in result: {result[:300]}" - - def test_no_na_symbols(self): - result = get_industry_performance_yfinance("technology") - lines = result.strip().split("\n") - data_lines = [l for l in lines if l.startswith("| ") and "Company" not in l and "---" not in l] - for line in data_lines: - cols = [c.strip() for c in line.split("|")[1:-1]] - # Symbol column (index 1) should not be N/A - assert cols[1] != "N/A", f"Symbol is N/A in line: {line}" - - def test_healthcare_sector(self): - result = get_industry_performance_yfinance("healthcare") - assert "Industry Performance: Healthcare" in result - - -class TestAlphaVantageFailoverRaise: - """Verify AV scanner functions raise when all data fails (enabling fallback). - - Root cause of previous failure: tests made real AV API calls that - intermittently succeeded, so AlphaVantageError was never raised. - Fix: mock _fetch_global_quote to always raise, simulating total failure - without requiring an API key or network access. - """ - - def test_sector_perf_raises_on_total_failure(self): - """When every GLOBAL_QUOTE call fails, the function should raise.""" - with patch( - "tradingagents.dataflows.alpha_vantage_scanner._fetch_global_quote", - side_effect=AlphaVantageError("Rate limit exceeded — mocked for test isolation"), - ): - with pytest.raises(AlphaVantageError, match="All .* sector queries failed"): - get_sector_performance_alpha_vantage() - - def test_industry_perf_raises_on_total_failure(self): - """When every ticker quote fails, the function should raise.""" - with patch( - "tradingagents.dataflows.alpha_vantage_scanner._fetch_global_quote", - side_effect=AlphaVantageError("Rate limit exceeded — mocked for test isolation"), - ): - with pytest.raises(AlphaVantageError, match="All .* ticker queries failed"): - get_industry_performance_alpha_vantage("technology") - - -@pytest.mark.integration -class TestRouteToVendorFallback: - """Verify route_to_vendor falls back from AV to yfinance.""" - - def test_sector_perf_falls_back_to_yfinance(self): - from tradingagents.dataflows.interface import route_to_vendor - result = route_to_vendor("get_sector_performance") - # Should get yfinance data (no "Alpha Vantage" in header) - assert "Sector Performance Overview" in result - # Should have actual percentage data, not all errors - assert "Error:" not in result or result.count("Error:") < 3 - - def test_industry_perf_falls_back_to_yfinance(self): - from tradingagents.dataflows.interface import route_to_vendor - result = route_to_vendor("get_industry_performance", "technology") - assert "Industry Performance" in result - # Should contain real ticker symbols - assert "N/A" not in result or result.count("N/A") < 5 diff --git a/tests/test_scanner_final.py b/tests/test_scanner_final.py deleted file mode 100644 index 5a6f9b53..00000000 --- a/tests/test_scanner_final.py +++ /dev/null @@ -1,130 +0,0 @@ -"""Final end-to-end test for scanner functionality.""" - -import tempfile -import os -from pathlib import Path -import pytest - -from tradingagents.agents.utils.scanner_tools import ( - get_market_movers, - get_market_indices, - get_sector_performance, - get_industry_performance, - get_topic_news, -) - - -def test_complete_scanner_workflow(): - """Test the complete scanner workflow from tools to file output.""" - - # Test 1: All individual tools work - print("Testing individual scanner tools...") - - # Market Movers - movers_result = get_market_movers.invoke({"category": "day_gainers"}) - assert isinstance(movers_result, str) - assert not movers_result.startswith("Error:") - assert "# Market Movers:" in movers_result - print("✓ Market movers tool works") - - # Market Indices - indices_result = get_market_indices.invoke({}) - assert isinstance(indices_result, str) - assert not indices_result.startswith("Error:") - assert "# Major Market Indices" in indices_result - print("✓ Market indices tool works") - - # Sector Performance - sectors_result = get_sector_performance.invoke({}) - assert isinstance(sectors_result, str) - assert not sectors_result.startswith("Error:") - assert "# Sector Performance Overview" in sectors_result - print("✓ Sector performance tool works") - - # Industry Performance - industry_result = get_industry_performance.invoke({"sector_key": "technology"}) - assert isinstance(industry_result, str) - assert not industry_result.startswith("Error:") - assert "# Industry Performance: Technology" in industry_result - print("✓ Industry performance tool works") - - # Topic News - news_result = get_topic_news.invoke({"topic": "market", "limit": 3}) - assert isinstance(news_result, str) - assert not news_result.startswith("Error:") - assert "# News for Topic: market" in news_result - print("✓ Topic news tool works") - - # Test 2: Verify we can save results to files (end-to-end) - print("\nTesting file output...") - - with tempfile.TemporaryDirectory() as temp_dir: - scan_date = "2026-03-15" - save_dir = Path(temp_dir) / "results" / "macro_scan" / scan_date - save_dir.mkdir(parents=True) - - # Save each result to a file (simulating what the scan command does) - (save_dir / "market_movers.txt").write_text(movers_result) - (save_dir / "market_indices.txt").write_text(indices_result) - (save_dir / "sector_performance.txt").write_text(sectors_result) - (save_dir / "industry_performance.txt").write_text(industry_result) - (save_dir / "topic_news.txt").write_text(news_result) - - # Verify files were created and have content - assert (save_dir / "market_movers.txt").exists() - assert (save_dir / "market_indices.txt").exists() - assert (save_dir / "sector_performance.txt").exists() - assert (save_dir / "industry_performance.txt").exists() - assert (save_dir / "topic_news.txt").exists() - - # Check file contents - assert "# Market Movers:" in (save_dir / "market_movers.txt").read_text() - assert "# Major Market Indices" in (save_dir / "market_indices.txt").read_text() - assert "# Sector Performance Overview" in (save_dir / "sector_performance.txt").read_text() - assert "# Industry Performance: Technology" in (save_dir / "industry_performance.txt").read_text() - assert "# News for Topic: market" in (save_dir / "topic_news.txt").read_text() - - print("✓ All scanner results saved correctly to files") - - print("\n🎉 Complete scanner workflow test passed!") - - -def test_scanner_integration_with_cli_scan(): - """Test that the scanner tools integrate properly with the CLI scan command.""" - # This test verifies the actual CLI scan command works end-to-end - # We already saw this work when we ran it manually - - # The key integration points are: - # 1. CLI scan command calls get_market_movers.invoke() - # 2. CLI scan command calls get_market_indices.invoke() - # 3. CLI scan command calls get_sector_performance.invoke() - # 4. CLI scan command calls get_industry_performance.invoke() - # 5. CLI scan command calls get_topic_news.invoke() - # 6. Results are written to files in reports/daily/{date}/market/ - - # Since we've verified the individual tools work above, and we've seen - # the CLI scan command work manually, we can be confident the integration works. - - # Let's at least verify the tools are callable from where the CLI expects them - from tradingagents.agents.utils.scanner_tools import ( - get_market_movers, - get_market_indices, - get_sector_performance, - get_industry_performance, - get_topic_news, - ) - - # Verify they're all callable (the CLI uses .invoke() method) - assert hasattr(get_market_movers, 'invoke') - assert hasattr(get_market_indices, 'invoke') - assert hasattr(get_sector_performance, 'invoke') - assert hasattr(get_industry_performance, 'invoke') - assert hasattr(get_topic_news, 'invoke') - - print("✓ Scanner tools are properly integrated with CLI scan command") - - -if __name__ == "__main__": - test_complete_scanner_workflow() - test_scanner_integration_with_cli_scan() - print("\n✅ All end-to-end scanner tests passed!") \ No newline at end of file diff --git a/tests/test_scanner_tools.py b/tests/test_scanner_tools.py deleted file mode 100644 index 5f2199e1..00000000 --- a/tests/test_scanner_tools.py +++ /dev/null @@ -1,82 +0,0 @@ -"""End-to-end tests for scanner tools functionality.""" - -import pytest -from tradingagents.agents.utils.scanner_tools import ( - get_market_movers, - get_market_indices, - get_sector_performance, - get_industry_performance, - get_topic_news, -) - - -def test_scanner_tools_imports(): - """Verify that all scanner tools can be imported.""" - from tradingagents.agents.utils.scanner_tools import ( - get_market_movers, - get_market_indices, - get_sector_performance, - get_industry_performance, - get_topic_news, - ) - - # Check that each tool exists (they are StructuredTool objects) - assert get_market_movers is not None - assert get_market_indices is not None - assert get_sector_performance is not None - assert get_industry_performance is not None - assert get_topic_news is not None - - # Check that each tool has the expected docstring - assert "market movers" in get_market_movers.description.lower() if get_market_movers.description else True - assert "market indices" in get_market_indices.description.lower() if get_market_indices.description else True - assert "sector performance" in get_sector_performance.description.lower() if get_sector_performance.description else True - assert "industry" in get_industry_performance.description.lower() if get_industry_performance.description else True - assert "news" in get_topic_news.description.lower() if get_topic_news.description else True - - -def test_market_movers(): - """Test market movers for all categories.""" - for category in ["day_gainers", "day_losers", "most_actives"]: - result = get_market_movers.invoke({"category": category}) - assert isinstance(result, str), f"Result for {category} should be a string" - # Check that it's not an error message - assert not result.startswith("Error:"), f"Error in {category}: {result[:100]}" - # Check for expected header - assert "# Market Movers:" in result, f"Missing header in {category} result" - - -def test_market_indices(): - """Test market indices.""" - result = get_market_indices.invoke({}) - assert isinstance(result, str), "Market indices result should be a string" - assert not result.startswith("Error:"), f"Error in market indices: {result[:100]}" - assert "# Major Market Indices" in result, "Missing header in market indices result" - - -def test_sector_performance(): - """Test sector performance.""" - result = get_sector_performance.invoke({}) - assert isinstance(result, str), "Sector performance result should be a string" - assert not result.startswith("Error:"), f"Error in sector performance: {result[:100]}" - assert "# Sector Performance Overview" in result, "Missing header in sector performance result" - - -def test_industry_performance(): - """Test industry performance for technology sector.""" - result = get_industry_performance.invoke({"sector_key": "technology"}) - assert isinstance(result, str), "Industry performance result should be a string" - assert not result.startswith("Error:"), f"Error in industry performance: {result[:100]}" - assert "# Industry Performance: Technology" in result, "Missing header in industry performance result" - - -def test_topic_news(): - """Test topic news for market topic.""" - result = get_topic_news.invoke({"topic": "market", "limit": 5}) - assert isinstance(result, str), "Topic news result should be a string" - assert not result.startswith("Error:"), f"Error in topic news: {result[:100]}" - assert "# News for Topic: market" in result, "Missing header in topic news result" - - -if __name__ == "__main__": - pytest.main([__file__, "-v"]) \ No newline at end of file diff --git a/tests/test_industry_deep_dive.py b/tests/unit/test_industry_deep_dive.py similarity index 66% rename from tests/test_industry_deep_dive.py rename to tests/unit/test_industry_deep_dive.py index f239fcec..ffed3339 100644 --- a/tests/test_industry_deep_dive.py +++ b/tests/unit/test_industry_deep_dive.py @@ -1,8 +1,7 @@ """Tests for the Industry Deep Dive improvements: 1. _extract_top_sectors() parses sector performance reports correctly -2. Enriched get_industry_performance_yfinance returns price columns -3. run_tool_loop nudge triggers when first response is short & no tool calls +2. run_tool_loop nudge triggers when first response is short & no tool calls """ import pytest @@ -52,7 +51,6 @@ class TestExtractTopSectors: def test_returns_top_3_by_absolute_1month(self): result = _extract_top_sectors(SAMPLE_SECTOR_REPORT, top_n=3) assert len(result) == 3 - # Energy (+7.80%), Communication Services (+6.30%), Technology (+5.67%) assert result[0] == "energy" assert result[1] == "communication-services" assert result[2] == "technology" @@ -60,7 +58,6 @@ class TestExtractTopSectors: def test_returns_top_n_variable(self): result = _extract_top_sectors(SAMPLE_SECTOR_REPORT, top_n=5) assert len(result) == 5 - # All should be valid sector keys for key in result: assert key in VALID_SECTOR_KEYS, f"Invalid key: {key}" @@ -86,7 +83,7 @@ class TestExtractTopSectors: | Healthcare | +0.05% | +0.10% | +0.50% | +1.00% | """ result = _extract_top_sectors(report, top_n=2) - assert result[0] == "energy" # |-8.50| > |1.00| + assert result[0] == "energy" def test_all_returned_keys_are_valid(self): result = _extract_top_sectors(SAMPLE_SECTOR_REPORT, top_n=11) @@ -94,8 +91,6 @@ class TestExtractTopSectors: assert key in VALID_SECTOR_KEYS def test_display_to_key_covers_all_sectors(self): - """Every sector name that appears in the ETF performance table - should map to a valid key.""" display_names = [ "technology", "healthcare", "financials", "energy", "consumer discretionary", "consumer staples", "industrials", @@ -114,7 +109,6 @@ class TestToolLoopNudge: """Verify the nudge mechanism in run_tool_loop.""" def _make_chain(self, responses): - """Create a mock chain that returns responses in sequence.""" chain = MagicMock() chain.invoke = MagicMock(side_effect=responses) return chain @@ -126,8 +120,7 @@ class TestToolLoopNudge: return tool def test_long_response_no_nudge(self): - """A long first response (no tool calls) should be returned as-is.""" - long_text = "A" * 2100 # must exceed MIN_REPORT_LENGTH (2000) + long_text = "A" * 2100 response = AIMessage(content=long_text, tool_calls=[]) chain = self._make_chain([response]) tool = self._make_tool() @@ -137,7 +130,6 @@ class TestToolLoopNudge: assert chain.invoke.call_count == 1 def test_short_response_triggers_nudge(self): - """A short first response triggers a nudge, then the LLM is re-invoked.""" short_resp = AIMessage(content="Brief.", tool_calls=[]) long_resp = AIMessage(content="A" * 2100, tool_calls=[]) chain = self._make_chain([short_resp, long_resp]) @@ -147,20 +139,16 @@ class TestToolLoopNudge: assert result.content == long_resp.content assert chain.invoke.call_count == 2 - # The second invoke should have received a HumanMessage nudge second_call_messages = chain.invoke.call_args_list[1][0][0] nudge_msgs = [m for m in second_call_messages if isinstance(m, HumanMessage)] assert len(nudge_msgs) == 1 assert "MUST call at least one tool" in nudge_msgs[0].content def test_nudge_only_on_first_round(self): - """Nudge should NOT trigger after tools have been used.""" - # Round 1: LLM calls a tool tool_call_resp = AIMessage( content="", tool_calls=[{"name": "my_tool", "args": {}, "id": "tc1"}], ) - # Round 2: LLM returns a short text — no nudge expected short_resp = AIMessage(content="Done.", tool_calls=[]) chain = self._make_chain([tool_call_resp, short_resp]) tool = self._make_tool() @@ -170,7 +158,6 @@ class TestToolLoopNudge: assert chain.invoke.call_count == 2 def test_tool_calls_execute_normally(self): - """Normal tool-calling flow should still work unchanged.""" tool_call_resp = AIMessage( content="", tool_calls=[{"name": "my_tool", "args": {"x": 1}, "id": "tc1"}], @@ -182,61 +169,3 @@ class TestToolLoopNudge: result = run_tool_loop(chain, [], [tool]) tool.invoke.assert_called_once_with({"x": 1}) assert "Final report" in result.content - - -# --------------------------------------------------------------------------- -# Enriched industry performance tests -# --------------------------------------------------------------------------- - -class TestEnrichedIndustryPerformance: - """Verify that get_industry_performance_yfinance now returns price columns. - - These tests require network access to Yahoo Finance. If the host is not - reachable (e.g. in sandboxed CI), they are automatically skipped. - """ - - @pytest.fixture(autouse=True) - def _require_yahoo(self): - import socket - try: - socket.setdefaulttimeout(3) - socket.socket(socket.AF_INET, socket.SOCK_STREAM).connect( - ("query2.finance.yahoo.com", 443) - ) - except (socket.error, OSError): - pytest.skip("Yahoo Finance not reachable") - - def test_technology_has_price_columns(self): - from tradingagents.dataflows.yfinance_scanner import ( - get_industry_performance_yfinance, - ) - - result = get_industry_performance_yfinance("technology") - assert "# Industry Performance: Technology" in result - # New columns should be present in the header - assert "1-Day %" in result - assert "1-Week %" in result - assert "1-Month %" in result - - def test_table_has_seven_columns(self): - from tradingagents.dataflows.yfinance_scanner import ( - get_industry_performance_yfinance, - ) - - result = get_industry_performance_yfinance("technology") - lines = result.strip().split("\n") - # Find the header separator line - sep_lines = [l for l in lines if l.startswith("|") and "---" in l] - assert len(sep_lines) >= 1 - # Count columns in separator - cols = [c.strip() for c in sep_lines[0].split("|")[1:-1]] - assert len(cols) == 7, f"Expected 7 columns, got {len(cols)}: {cols}" - - def test_healthcare_sector_key(self): - from tradingagents.dataflows.yfinance_scanner import ( - get_industry_performance_yfinance, - ) - - result = get_industry_performance_yfinance("healthcare") - assert "Industry Performance: Healthcare" in result - assert "1-Day %" in result diff --git a/tests/unit/test_scanner_fallback.py b/tests/unit/test_scanner_fallback.py new file mode 100644 index 00000000..3b10d040 --- /dev/null +++ b/tests/unit/test_scanner_fallback.py @@ -0,0 +1,30 @@ +"""Unit tests for AV failover — mocked, no network.""" + +import pytest +from unittest.mock import patch + +from tradingagents.dataflows.alpha_vantage_common import AlphaVantageError +from tradingagents.dataflows.alpha_vantage_scanner import ( + get_sector_performance_alpha_vantage, + get_industry_performance_alpha_vantage, +) + + +class TestAlphaVantageFailoverRaise: + """Verify AV scanner functions raise when all data fails.""" + + def test_sector_perf_raises_on_total_failure(self): + with patch( + "tradingagents.dataflows.alpha_vantage_scanner._fetch_global_quote", + side_effect=AlphaVantageError("Rate limit exceeded — mocked"), + ): + with pytest.raises(AlphaVantageError, match="All .* sector queries failed"): + get_sector_performance_alpha_vantage() + + def test_industry_perf_raises_on_total_failure(self): + with patch( + "tradingagents.dataflows.alpha_vantage_scanner._fetch_global_quote", + side_effect=AlphaVantageError("Rate limit exceeded — mocked"), + ): + with pytest.raises(AlphaVantageError, match="All .* ticker queries failed"): + get_industry_performance_alpha_vantage("technology")