refactor(tests): consolidate live-API tests into integration/, move mocked tests to unit/
- 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 <noreply@anthropic.com>
This commit is contained in:
Ahmet Guzererler
parent
8c9183cf10
commit
57c9b3ecaa
|
|
@ -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"
|
||||
)
|
||||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -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!")
|
||||
|
|
@ -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"])
|
||||
|
|
@ -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"])
|
||||
|
|
@ -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
|
||||
|
|
@ -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!")
|
||||
|
|
@ -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"])
|
||||
|
|
@ -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
|
||||
|
|
@ -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")
|
||||
Loading…
Reference in New Issue