From 9080dbc0c06946fd7415a507abc17b216a9d43d3 Mon Sep 17 00:00:00 2001 From: Youssef Aitousarrah Date: Sun, 23 Nov 2025 17:14:43 -0800 Subject: [PATCH] Add pytest tests for multi-vendor routing logic Tests verify: - Single vendor stops after first success - Multi-vendor stops after all primaries (even if they fail) - Fallback vendors are not attempted when primaries are configured - Tool-level config overrides category-level config Tests use pytest with fixtures and mocked vendors, can run without API keys in CI/CD. Run with: pytest tests/test_multi_vendor_routing.py -v --- test_multi_vendor_routing.py | 181 ++++++++++++++++++++++++++++ tests/__init__.py | 0 tests/test_multi_vendor_routing.py | 183 +++++++++++++++++++++++++++++ 3 files changed, 364 insertions(+) create mode 100644 test_multi_vendor_routing.py create mode 100644 tests/__init__.py create mode 100644 tests/test_multi_vendor_routing.py diff --git a/test_multi_vendor_routing.py b/test_multi_vendor_routing.py new file mode 100644 index 00000000..00581c0b --- /dev/null +++ b/test_multi_vendor_routing.py @@ -0,0 +1,181 @@ +""" +Behavioral tests for multi-vendor routing logic. + +These tests verify the stopping behavior by analyzing the debug output +from the routing system. +""" + +import os +import sys +import io +from contextlib import redirect_stdout +from dotenv import load_dotenv + +# Load environment variables +load_dotenv() + +# Add current directory to path +sys.path.append(os.getcwd()) + +from tradingagents.dataflows.config import set_config, get_config + + +def capture_routing_behavior(vendor_config, method, *args): + """Capture the debug output from route_to_vendor to analyze behavior.""" + from tradingagents.dataflows.interface import route_to_vendor + + # Set configuration + config = get_config() + config["data_vendors"]["news_data"] = vendor_config + set_config(config) + + # Capture stdout + f = io.StringIO() + try: + with redirect_stdout(f): + result = route_to_vendor(method, *args) + output = f.getvalue() + return output, result, None + except Exception as e: + output = f.getvalue() + return output, None, e + + +def test_single_vendor_stops_after_success(): + """Test that single vendor config stops after first success.""" + print("\n" + "="*70) + print("TEST 1: Single Vendor - Should stop after first success") + print("="*70) + + output, result, error = capture_routing_behavior( + "alpha_vantage", + "get_news", + "NVDA", + "2024-11-20", + "2024-11-21" + ) + + # Check that it stopped after primary vendor + assert "Stopping after successful vendor 'alpha_vantage' (single-vendor config)" in output, \ + "Should stop after single vendor succeeds" + + # Check that fallback vendors were not attempted + assert "Attempting FALLBACK vendor" not in output, \ + "Should not attempt fallback vendors when primary succeeds" + + # Check vendor attempt count + assert "completed with 1 result(s) from 1 vendor attempt(s)" in output, \ + "Should only attempt 1 vendor" + + print("✅ PASS: Single vendor stopped after success, no fallbacks attempted") + print(f" Vendor attempts: 1") + + +def test_multi_vendor_stops_after_all_primaries(): + """Test that multi-vendor config stops after all primary vendors.""" + print("\n" + "="*70) + print("TEST 2: Multi-Vendor - Should stop after all primaries") + print("="*70) + + output, result, error = capture_routing_behavior( + "alpha_vantage,google", + "get_news", + "NVDA", + "2024-11-20", + "2024-11-21" + ) + + # Check that both primaries were attempted + assert "Attempting PRIMARY vendor 'alpha_vantage'" in output, \ + "Should attempt first primary vendor" + assert "Attempting PRIMARY vendor 'google'" in output, \ + "Should attempt second primary vendor" + + # Check that it stopped after all primaries + assert "All primary vendors attempted" in output, \ + "Should stop after all primary vendors" + + # Check that fallback vendors were not attempted + assert "Attempting FALLBACK vendor 'openai'" not in output, \ + "Should not attempt fallback vendor (openai)" + assert "Attempting FALLBACK vendor 'local'" not in output, \ + "Should not attempt fallback vendor (local)" + + print("✅ PASS: Multi-vendor stopped after all primaries, no fallbacks attempted") + print(f" Primary vendors: alpha_vantage, google") + + +def test_single_vendor_uses_fallback_on_failure(): + """Test that single vendor uses fallback if primary fails.""" + print("\n" + "="*70) + print("TEST 3: Single Vendor Failure - Should use fallback") + print("="*70) + + # Use a vendor that will likely fail (invalid config) + output, result, error = capture_routing_behavior( + "nonexistent_vendor", + "get_news", + "NVDA", + "2024-11-20", + "2024-11-21" + ) + + # Check that fallback was attempted + assert "Attempting FALLBACK vendor" in output or "Attempting PRIMARY vendor" in output, \ + "Should attempt vendors" + + # Should eventually succeed with a fallback + assert result is not None or error is not None, \ + "Should either succeed with fallback or fail gracefully" + + print("✅ PASS: Fallback mechanism works when primary fails") + + +def test_debug_output_shows_fallback_order(): + """Test that debug output shows the complete fallback order.""" + print("\n" + "="*70) + print("TEST 4: Debug Output - Should show fallback order") + print("="*70) + + output, result, error = capture_routing_behavior( + "alpha_vantage", + "get_news", + "NVDA", + "2024-11-20", + "2024-11-21" + ) + + # Check that fallback order is displayed + assert "Full fallback order:" in output, \ + "Should display full fallback order in debug output" + assert "alpha_vantage" in output, \ + "Should show primary vendor in fallback order" + + print("✅ PASS: Debug output correctly shows fallback order") + + +if __name__ == '__main__': + print("\n" + "="*70) + print("MULTI-VENDOR ROUTING BEHAVIORAL TESTS") + print("="*70) + + try: + test_single_vendor_stops_after_success() + test_multi_vendor_stops_after_all_primaries() + test_single_vendor_uses_fallback_on_failure() + test_debug_output_shows_fallback_order() + + print("\n" + "="*70) + print("ALL TESTS PASSED ✅") + print("="*70 + "\n") + + except AssertionError as e: + print(f"\n❌ TEST FAILED: {e}\n") + import traceback + traceback.print_exc() + sys.exit(1) + except Exception as e: + print(f"\n❌ ERROR: {e}\n") + import traceback + traceback.print_exc() + sys.exit(1) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/test_multi_vendor_routing.py b/tests/test_multi_vendor_routing.py new file mode 100644 index 00000000..d523fbaf --- /dev/null +++ b/tests/test_multi_vendor_routing.py @@ -0,0 +1,183 @@ +""" +Unit tests for multi-vendor routing logic. + +These tests use mocked vendor implementations and can run without API keys, +making them suitable for CI/CD environments. + +Run with: pytest tests/test_multi_vendor_routing.py -v +""" + +import pytest +from unittest.mock import patch, MagicMock +import sys +import os + +# Add parent directory to path +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + + +@pytest.fixture +def mock_vendors(): + """Create mock vendor functions with __name__ attribute.""" + mock_a = MagicMock(return_value="Result from Vendor A") + mock_a.__name__ = "mock_vendor_a" + + mock_b = MagicMock(return_value="Result from Vendor B") + mock_b.__name__ = "mock_vendor_b" + + mock_c = MagicMock(return_value="Result from Vendor C") + mock_c.__name__ = "mock_vendor_c" + + return {'a': mock_a, 'b': mock_b, 'c': mock_c} + + +@pytest.fixture +def mock_routing(mock_vendors): + """Set up mocked routing environment.""" + with patch('tradingagents.dataflows.interface.VENDOR_METHODS', { + 'test_method': { + 'vendor_a': mock_vendors['a'], + 'vendor_b': mock_vendors['b'], + 'vendor_c': mock_vendors['c'], + } + }), \ + patch('tradingagents.dataflows.interface.get_category_for_method', return_value='test_category'), \ + patch('tradingagents.dataflows.interface.get_config') as mock_config: + yield mock_config, mock_vendors + + +def test_single_vendor_stops_after_success(mock_routing): + """Test that single vendor config stops after first successful vendor.""" + mock_config, mock_vendors = mock_routing + + # Configure single vendor + mock_config.return_value = { + 'data_vendors': {'test_category': 'vendor_a'}, + 'tool_vendors': {} + } + + from tradingagents.dataflows.interface import route_to_vendor + + result = route_to_vendor('test_method', 'arg1', 'arg2') + + # Assertions + mock_vendors['a'].assert_called_once_with('arg1', 'arg2') + mock_vendors['b'].assert_not_called() # Should not try fallback + mock_vendors['c'].assert_not_called() # Should not try fallback + assert result == 'Result from Vendor A' + + +def test_multi_vendor_stops_after_all_primaries_success(mock_routing): + """Test that multi-vendor stops after all primaries when they succeed.""" + mock_config, mock_vendors = mock_routing + + # Configure two primary vendors + mock_config.return_value = { + 'data_vendors': {'test_category': 'vendor_a,vendor_b'}, + 'tool_vendors': {} + } + + from tradingagents.dataflows.interface import route_to_vendor + + result = route_to_vendor('test_method', 'arg1') + + # Assertions + mock_vendors['a'].assert_called_once_with('arg1') + mock_vendors['b'].assert_called_once_with('arg1') + mock_vendors['c'].assert_not_called() # Should NOT try fallback + + # Result should contain both + assert 'Result from Vendor A' in result + assert 'Result from Vendor B' in result + + +def test_multi_vendor_stops_after_all_primaries_failure(mock_routing): + """Test that multi-vendor stops after all primaries even when they fail.""" + mock_config, mock_vendors = mock_routing + + # Configure two primary vendors that will fail + mock_vendors['a'].side_effect = Exception("Vendor A failed") + mock_vendors['b'].side_effect = Exception("Vendor B failed") + + mock_config.return_value = { + 'data_vendors': {'test_category': 'vendor_a,vendor_b'}, + 'tool_vendors': {} + } + + from tradingagents.dataflows.interface import route_to_vendor + + # Should raise error after trying all primaries + with pytest.raises(RuntimeError, match="All vendor implementations failed"): + route_to_vendor('test_method', 'arg1') + + # Assertions + mock_vendors['a'].assert_called_once_with('arg1') + mock_vendors['b'].assert_called_once_with('arg1') + mock_vendors['c'].assert_not_called() # Should NOT try fallback + + +def test_multi_vendor_partial_failure_stops_after_primaries(mock_routing): + """Test that multi-vendor stops after all primaries even if one fails.""" + mock_config, mock_vendors = mock_routing + + # First vendor fails, second succeeds + mock_vendors['a'].side_effect = Exception("Vendor A failed") + + mock_config.return_value = { + 'data_vendors': {'test_category': 'vendor_a,vendor_b'}, + 'tool_vendors': {} + } + + from tradingagents.dataflows.interface import route_to_vendor + + result = route_to_vendor('test_method', 'arg1') + + # Assertions + mock_vendors['a'].assert_called_once_with('arg1') + mock_vendors['b'].assert_called_once_with('arg1') + mock_vendors['c'].assert_not_called() # Should NOT try fallback + + assert result == 'Result from Vendor B' + + +def test_single_vendor_uses_fallback_on_failure(mock_routing): + """Test that single vendor uses fallback if primary fails.""" + mock_config, mock_vendors = mock_routing + + # Primary vendor fails + mock_vendors['a'].side_effect = Exception("Vendor A failed") + + mock_config.return_value = { + 'data_vendors': {'test_category': 'vendor_a'}, + 'tool_vendors': {} + } + + from tradingagents.dataflows.interface import route_to_vendor + + result = route_to_vendor('test_method', 'arg1') + + # Assertions + mock_vendors['a'].assert_called_once_with('arg1') + mock_vendors['b'].assert_called_once_with('arg1') # Should try fallback + assert result == 'Result from Vendor B' + + +def test_tool_level_override_takes_precedence(mock_routing): + """Test that tool-level vendor config overrides category-level.""" + mock_config, mock_vendors = mock_routing + + # Category says vendor_a, but tool override says vendor_b + mock_config.return_value = { + 'data_vendors': {'test_category': 'vendor_a'}, + 'tool_vendors': {'test_method': 'vendor_b'} + } + + from tradingagents.dataflows.interface import route_to_vendor + + result = route_to_vendor('test_method', 'arg1') + + # Assertions + mock_vendors['a'].assert_not_called() # Category default ignored + mock_vendors['b'].assert_called_once_with('arg1') # Tool override used + assert result == 'Result from Vendor B' +