mirror of
https://gitlab.com/simple-stock-bots/simple-stock-bot.git
synced 2026-06-03 21:00:26 +00:00
Add comprehensive test suite with 174 tests
Test categories: - Unit tests for Symbol/Stock/Coin class construction, repr, str, hash - Router regex parsing tests (stock/crypto patterns, edge cases) - Router.find_symbols with mocked MarketData and cg_Crypto - Router.trending_decay math tests - Router.price_reply dispatching tests - MarketData.price_reply formatting (up/down/flat/error/None cases) - MarketData.intra_reply and chart_reply DataFrame construction - cg_Crypto.price_reply formatting tests - cg_Crypto retry/backoff behavior on 429s - rate_limited decorator timing and instance independence - Telegram bot command handler tests (mocked updates) - Integration tests for full message flows - Edge case tests (empty messages, API errors, malformed responses) - Unicode and long input handling tests Uses pytest with fixtures, conftest.py, and parametrize where appropriate. All external API calls are mocked with unittest.mock. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,333 @@
|
||||
"""Integration tests for full message flow."""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
import pandas as pd
|
||||
import pytest
|
||||
|
||||
from common.symbol_router import Router
|
||||
from common.Symbol import Stock, Coin
|
||||
|
||||
|
||||
class TestStockMessageFlow:
|
||||
"""Integration tests for stock symbol message flow."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_router(self, mock_symbol_list, mock_crypto_symbol_list):
|
||||
"""Create Router with all dependencies mocked."""
|
||||
with patch.object(Router, "__init__", lambda self: None):
|
||||
router = Router()
|
||||
router.stock = MagicMock()
|
||||
router.crypto = MagicMock()
|
||||
router.stock.symbol_id = lambda s: mock_symbol_list.get(s.upper())
|
||||
router.stock.symbol_list = mock_symbol_list
|
||||
router.crypto.symbol_list = mock_crypto_symbol_list
|
||||
router.trending_count = {}
|
||||
return router
|
||||
|
||||
def test_full_stock_price_flow(self, mock_router):
|
||||
"""Test: Message with $AAPL -> Router finds it -> MarketData returns price."""
|
||||
# Setup mock response
|
||||
mock_router.stock.price_reply = MagicMock(
|
||||
return_value="The current price of Apple Inc. is $150.25 and is currently up 1.5% for the day."
|
||||
)
|
||||
|
||||
# Simulate message flow
|
||||
message = "What do you think about $AAPL?"
|
||||
symbols = mock_router.find_symbols(message)
|
||||
|
||||
assert len(symbols) == 1
|
||||
assert isinstance(symbols[0], Stock)
|
||||
assert symbols[0].symbol == "AAPL"
|
||||
|
||||
# Get price reply
|
||||
replies = mock_router.price_reply(symbols)
|
||||
|
||||
assert len(replies) == 1
|
||||
assert "Apple Inc." in replies[0]
|
||||
assert "150.25" in replies[0]
|
||||
assert "1.5%" in replies[0]
|
||||
|
||||
def test_multiple_stocks_flow(self, mock_router):
|
||||
"""Test message with multiple stock symbols."""
|
||||
mock_router.stock.price_reply = MagicMock(
|
||||
side_effect=[
|
||||
"Apple Inc. price reply",
|
||||
"Tesla, Inc. price reply",
|
||||
]
|
||||
)
|
||||
|
||||
message = "Comparing $AAPL vs $TSLA"
|
||||
symbols = mock_router.find_symbols(message)
|
||||
|
||||
assert len(symbols) == 2
|
||||
replies = mock_router.price_reply(symbols)
|
||||
assert len(replies) == 2
|
||||
|
||||
|
||||
class TestCryptoMessageFlow:
|
||||
"""Integration tests for crypto symbol message flow."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_router(self, mock_symbol_list, mock_crypto_symbol_list):
|
||||
"""Create Router with all dependencies mocked."""
|
||||
with patch.object(Router, "__init__", lambda self: None):
|
||||
router = Router()
|
||||
router.stock = MagicMock()
|
||||
router.crypto = MagicMock()
|
||||
router.stock.symbol_id = lambda s: mock_symbol_list.get(s.upper())
|
||||
router.crypto.symbol_list = mock_crypto_symbol_list
|
||||
router.trending_count = {}
|
||||
return router
|
||||
|
||||
def test_full_crypto_price_flow(self, mock_router):
|
||||
"""Test: Message with $$BTC -> Router finds it -> CoinGecko returns price."""
|
||||
mock_router.crypto.price_reply = MagicMock(
|
||||
return_value="The current price of Bitcoin is $45,000, the coin is currently up 2.5% for today"
|
||||
)
|
||||
|
||||
message = "How is $$BTC doing today?"
|
||||
symbols = mock_router.find_symbols(message)
|
||||
|
||||
assert len(symbols) == 1
|
||||
assert isinstance(symbols[0], Coin)
|
||||
assert symbols[0].id == "bitcoin"
|
||||
|
||||
replies = mock_router.price_reply(symbols)
|
||||
|
||||
assert len(replies) == 1
|
||||
assert "Bitcoin" in replies[0]
|
||||
assert "45,000" in replies[0]
|
||||
|
||||
def test_multiple_cryptos_flow(self, mock_router):
|
||||
"""Test message with multiple crypto symbols."""
|
||||
mock_router.crypto.price_reply = MagicMock(
|
||||
side_effect=[
|
||||
"Bitcoin price reply",
|
||||
"Ethereum price reply",
|
||||
]
|
||||
)
|
||||
|
||||
message = "$$BTC and $$ETH looking good"
|
||||
symbols = mock_router.find_symbols(message)
|
||||
|
||||
assert len(symbols) == 2
|
||||
replies = mock_router.price_reply(symbols)
|
||||
assert len(replies) == 2
|
||||
|
||||
|
||||
class TestMixedMessageFlow:
|
||||
"""Integration tests for messages with both stocks and crypto."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_router(self, mock_symbol_list, mock_crypto_symbol_list):
|
||||
"""Create Router with all dependencies mocked."""
|
||||
with patch.object(Router, "__init__", lambda self: None):
|
||||
router = Router()
|
||||
router.stock = MagicMock()
|
||||
router.crypto = MagicMock()
|
||||
router.stock.symbol_id = lambda s: mock_symbol_list.get(s.upper())
|
||||
router.crypto.symbol_list = mock_crypto_symbol_list
|
||||
router.trending_count = {}
|
||||
return router
|
||||
|
||||
def test_mixed_symbols_resolved(self, mock_router):
|
||||
"""Test message with both stock and crypto symbols."""
|
||||
mock_router.stock.price_reply = MagicMock(return_value="Apple price")
|
||||
mock_router.crypto.price_reply = MagicMock(return_value="Bitcoin price")
|
||||
|
||||
message = "I own $AAPL and $$BTC"
|
||||
symbols = mock_router.find_symbols(message)
|
||||
|
||||
assert len(symbols) == 2
|
||||
|
||||
# Check we have one of each type
|
||||
types = {type(s).__name__ for s in symbols}
|
||||
assert types == {"Stock", "Coin"}
|
||||
|
||||
replies = mock_router.price_reply(symbols)
|
||||
assert len(replies) == 2
|
||||
|
||||
|
||||
class TestChartCommandFlow:
|
||||
"""Integration tests for chart commands."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_router(self, mock_symbol_list, mock_crypto_symbol_list):
|
||||
"""Create Router with all dependencies mocked."""
|
||||
with patch.object(Router, "__init__", lambda self: None):
|
||||
router = Router()
|
||||
router.stock = MagicMock()
|
||||
router.crypto = MagicMock()
|
||||
router.stock.symbol_id = lambda s: mock_symbol_list.get(s.upper())
|
||||
router.crypto.symbol_list = mock_crypto_symbol_list
|
||||
router.trending_count = {}
|
||||
return router
|
||||
|
||||
def test_chart_command_stock(self, mock_router):
|
||||
"""Test /chart command with stock symbol."""
|
||||
expected_df = pd.DataFrame(
|
||||
{
|
||||
"Open": [100, 101, 102],
|
||||
"High": [101, 102, 103],
|
||||
"Low": [99, 100, 101],
|
||||
"Close": [100.5, 101.5, 102.5],
|
||||
}
|
||||
)
|
||||
mock_router.stock.chart_reply = MagicMock(return_value=expected_df)
|
||||
|
||||
message = "/chart $AAPL"
|
||||
symbols = mock_router.find_symbols(message)
|
||||
|
||||
assert len(symbols) == 1
|
||||
df = mock_router.chart_reply(symbols[0])
|
||||
|
||||
assert not df.empty
|
||||
assert "Open" in df.columns
|
||||
|
||||
def test_chart_command_crypto(self, mock_router):
|
||||
"""Test /chart command with crypto symbol."""
|
||||
expected_df = pd.DataFrame(
|
||||
{
|
||||
"Open": [29000, 29200, 29800],
|
||||
"High": [29500, 30000, 30500],
|
||||
"Low": [28500, 29000, 29500],
|
||||
"Close": [29200, 29800, 30200],
|
||||
}
|
||||
)
|
||||
mock_router.crypto.chart_reply = MagicMock(return_value=expected_df)
|
||||
|
||||
message = "/chart $$BTC"
|
||||
symbols = mock_router.find_symbols(message)
|
||||
|
||||
assert len(symbols) == 1
|
||||
df = mock_router.chart_reply(symbols[0])
|
||||
|
||||
assert not df.empty
|
||||
|
||||
|
||||
class TestTrendingCommandFlow:
|
||||
"""Integration tests for trending command."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_router(self, mock_symbol_list, mock_crypto_symbol_list):
|
||||
"""Create Router with all dependencies mocked."""
|
||||
with patch.object(Router, "__init__", lambda self: None):
|
||||
router = Router()
|
||||
router.stock = MagicMock()
|
||||
router.crypto = MagicMock()
|
||||
router.stock.symbol_id = lambda s: mock_symbol_list.get(s.upper())
|
||||
router.crypto.symbol_list = mock_crypto_symbol_list
|
||||
router.trending_count = {"$AAPL": 10.0, "$TSLA": 5.0}
|
||||
return router
|
||||
|
||||
def test_trending_with_bot_tracking(self, mock_router):
|
||||
"""Test trending includes bot-tracked symbols."""
|
||||
mock_router.crypto.trending = MagicMock(
|
||||
return_value=[
|
||||
"`$$BTC`: Bitcoin, 2.5%",
|
||||
"`$$ETH`: Ethereum, 1.2%",
|
||||
]
|
||||
)
|
||||
mock_router.stock.spark_reply = MagicMock(return_value="`$AAPL`: 1.5%")
|
||||
mock_router.find_symbols = MagicMock(return_value=[MagicMock()])
|
||||
|
||||
# Check trending_count is populated
|
||||
assert "$AAPL" in mock_router.trending_count
|
||||
assert mock_router.trending_count["$AAPL"] == 10.0
|
||||
|
||||
|
||||
class TestOptionsDetectionFlow:
|
||||
"""Integration tests for options detection."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_router(self, mock_symbol_list, mock_crypto_symbol_list):
|
||||
"""Create Router with all dependencies mocked."""
|
||||
with patch.object(Router, "__init__", lambda self: None):
|
||||
router = Router()
|
||||
router.stock = MagicMock()
|
||||
router.crypto = MagicMock()
|
||||
router.stock.symbol_id = lambda s: mock_symbol_list.get(s.upper())
|
||||
router.crypto.symbol_list = mock_crypto_symbol_list
|
||||
router.trending_count = {}
|
||||
return router
|
||||
|
||||
def test_options_call_detected(self, mock_router):
|
||||
"""Test that 'call' keyword triggers options lookup."""
|
||||
mock_router.stock.options_reply = MagicMock(
|
||||
return_value={
|
||||
"Option Symbol": "AAPL230120C150",
|
||||
"Underlying": "$AAPL",
|
||||
"Expiration": "in 30 days",
|
||||
"side": "call",
|
||||
"strike": 150,
|
||||
}
|
||||
)
|
||||
|
||||
message = "$AAPL call 150"
|
||||
symbols = mock_router.find_symbols(message)
|
||||
|
||||
# Check 'call' is in message
|
||||
assert "call" in message.lower()
|
||||
assert len(symbols) == 1
|
||||
|
||||
def test_options_put_detected(self, mock_router):
|
||||
"""Test that 'put' keyword triggers options lookup."""
|
||||
message = "$TSLA put 200"
|
||||
|
||||
# Check 'put' is in message
|
||||
assert "put" in message.lower()
|
||||
|
||||
symbols = mock_router.find_symbols(message)
|
||||
assert len(symbols) == 1
|
||||
|
||||
|
||||
class TestEdgeCaseFlows:
|
||||
"""Integration tests for edge cases."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_router(self, mock_symbol_list, mock_crypto_symbol_list):
|
||||
"""Create Router with all dependencies mocked."""
|
||||
with patch.object(Router, "__init__", lambda self: None):
|
||||
router = Router()
|
||||
router.stock = MagicMock()
|
||||
router.crypto = MagicMock()
|
||||
router.stock.symbol_id = lambda s: mock_symbol_list.get(s.upper())
|
||||
router.crypto.symbol_list = mock_crypto_symbol_list
|
||||
router.trending_count = {}
|
||||
return router
|
||||
|
||||
def test_empty_message(self, mock_router):
|
||||
"""Test empty message returns no symbols."""
|
||||
symbols = mock_router.find_symbols("")
|
||||
assert len(symbols) == 0
|
||||
|
||||
def test_no_symbols_in_message(self, mock_router):
|
||||
"""Test message with no symbols."""
|
||||
symbols = mock_router.find_symbols("Just a regular message")
|
||||
assert len(symbols) == 0
|
||||
|
||||
def test_invalid_symbols_filtered(self, mock_router):
|
||||
"""Test invalid symbols are filtered out."""
|
||||
symbols = mock_router.find_symbols("$INVALID and $$NOTACOIN")
|
||||
assert len(symbols) == 0
|
||||
|
||||
def test_unicode_in_message(self, mock_router):
|
||||
"""Test unicode characters in message."""
|
||||
symbols = mock_router.find_symbols("$AAPL and $$BTC")
|
||||
assert len(symbols) == 2
|
||||
|
||||
def test_very_long_message(self, mock_router):
|
||||
"""Test very long message."""
|
||||
long_text = "blah " * 1000 + "$AAPL" + " blah " * 1000
|
||||
symbols = mock_router.find_symbols(long_text)
|
||||
assert len(symbols) == 1
|
||||
|
||||
def test_api_error_handling(self, mock_router):
|
||||
"""Test handling of API errors."""
|
||||
mock_router.stock.price_reply = MagicMock(return_value="Error getting quote")
|
||||
|
||||
symbols = mock_router.find_symbols("$AAPL")
|
||||
replies = mock_router.price_reply(symbols)
|
||||
|
||||
assert len(replies) == 1
|
||||
# Should return error message, not crash
|
||||
Reference in New Issue
Block a user