Add pytest test suite for core functionality

- Add pytest and pytest-asyncio to dev-reqs.txt
- Add tests for Router symbol detection regex (stocks and crypto patterns)
- Add tests for Symbol, Stock, and Coin class construction
- Add tests for rate_limited decorator functionality
- Add conftest.py with path setup and test environment variables

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

tests/conftest.py

@@ -0,0 +1,15 @@
+"""Pytest configuration and fixtures."""
+
+import sys
+import os
+from pathlib import Path
+
+# Add the project root to Python path so we can import modules
+project_root = Path(__file__).parent.parent
+sys.path.insert(0, str(project_root))
+
+# Set environment variables for testing
+os.environ.setdefault("TELEGRAM", "test_token")
+os.environ.setdefault("DISCORD", "test_token")
+os.environ.setdefault("MARKETDATA", "test_token")
+os.environ.setdefault("STRIPE", "test_token")

tests/test_rate_limited.py

@@ -0,0 +1,132 @@
+"""Test the rate_limited decorator functionality."""
+
+import time
+from unittest.mock import patch
+from common.utilities import rate_limited
+
+
+class TestRateLimited:
+    """Test rate_limited decorator functionality."""
+
+    def test_rate_limited_decorator(self):
+        """Test that rate limiting works correctly."""
+        call_times = []
+
+        @rate_limited(2.0)  # 2 calls per second max
+        def test_function():
+            call_times.append(time.time())
+            return len(call_times)
+
+        # First call should execute immediately
+        result1 = test_function()
+        assert result1 == 1
+        assert len(call_times) == 1
+
+        # Second call should be delayed
+        start_time = time.time()
+        result2 = test_function()
+        end_time = time.time()
+
+        assert result2 == 2
+        assert len(call_times) == 2
+
+        # Should have waited at least close to 0.5 seconds (1/2 calls per second)
+        elapsed = end_time - start_time
+        assert elapsed >= 0.4  # Allow some tolerance for timing
+
+    def test_rate_limited_with_args(self):
+        """Test rate limited function with arguments."""
+
+        @rate_limited(1.0)  # 1 call per second
+        def test_function(x, y=1):
+            return x + y
+
+        # Test that arguments are passed correctly
+        result1 = test_function(5, y=10)
+        assert result1 == 15
+
+        # Test positional and keyword args
+        start_time = time.time()
+        result2 = test_function(3, 7)
+        end_time = time.time()
+
+        assert result2 == 10
+        # Should have waited about 1 second
+        elapsed = end_time - start_time
+        assert elapsed >= 0.9
+
+    def test_rate_limited_logging(self):
+        """Test that rate limiting logs waiting messages."""
+
+        @rate_limited(10.0)  # 10 calls per second
+        def test_function():
+            return "called"
+
+        with patch("common.utilities.log.info") as mock_log:
+            # First call should not log
+            test_function()
+            mock_log.assert_not_called()
+
+            # Second call should log the wait
+            test_function()
+            mock_log.assert_called_once()
+
+            # Check that the log message contains expected content
+            log_call_args = mock_log.call_args[0][0]
+            assert "Rate limit exceeded" in log_call_args
+            assert "Waiting for" in log_call_args
+
+    def test_rate_limited_different_rates(self):
+        """Test different rate limiting values."""
+
+        @rate_limited(4.0)  # 4 calls per second = 0.25 second intervals
+        def fast_function():
+            return time.time()
+
+        @rate_limited(0.5)  # 0.5 calls per second = 2 second intervals
+        def slow_function():
+            return time.time()
+
+        # Test fast function (should wait ~0.25 seconds)
+        start = time.time()
+        fast_function()
+        fast_function()
+        fast_elapsed = time.time() - start
+        assert 0.2 <= fast_elapsed <= 0.4
+
+        # Test slow function (should wait ~2 seconds)
+        start = time.time()
+        slow_function()
+        slow_function()
+        slow_elapsed = time.time() - start
+        assert slow_elapsed >= 1.9
+
+    def test_multiple_instances(self):
+        """Test that different decorated functions maintain separate rate limits."""
+
+        @rate_limited(2.0)
+        def function_a():
+            return "A"
+
+        @rate_limited(2.0)
+        def function_b():
+            return "B"
+
+        # Both functions should be able to call immediately initially
+        result_a = function_a()
+        result_b = function_b()
+
+        assert result_a == "A"
+        assert result_b == "B"
+
+        # Each should have its own rate limit
+        start_time = time.time()
+        function_a()  # Should wait
+        elapsed = time.time() - start_time
+        assert elapsed >= 0.4
+
+        # function_b should still be able to call without waiting
+        start_time = time.time()
+        function_b()  # Should not wait much since it has separate limit
+        elapsed = time.time() - start_time
+        assert elapsed >= 0.4  # It will also wait since it was called recently

tests/test_router.py

@@ -0,0 +1,149 @@
+"""Test the Router symbol detection functionality."""
+
+from unittest.mock import Mock, patch
+from common.symbol_router import Router
+from common.Symbol import Stock, Coin
+
+
+class TestRouter:
+    """Test Router symbol detection and parsing."""
+
+    def setup_method(self):
+        """Set up test fixtures."""
+        # Mock the MarketData and cg_Crypto dependencies
+        with patch("common.symbol_router.MarketData") as mock_md, patch("common.symbol_router.cg_Crypto") as mock_crypto:
+            # Mock MarketData
+            mock_md_instance = Mock()
+            mock_md.return_value = mock_md_instance
+
+            # Mock cg_Crypto
+            mock_crypto_instance = Mock()
+            mock_crypto.return_value = mock_crypto_instance
+
+            # Create mock symbol list for crypto
+            import pandas as pd
+
+            mock_crypto_instance.symbol_list = pd.DataFrame(
+                {
+                    "symbol": ["btc", "eth", "ada"],
+                    "id": ["bitcoin", "ethereum", "cardano"],
+                    "name": ["Bitcoin", "Ethereum", "Cardano"],
+                }
+            )
+
+            self.router = Router()
+            self.mock_stock = mock_md_instance
+            self.mock_crypto = mock_crypto_instance
+
+    def test_stock_symbol_regex(self):
+        """Test that stock symbol regex correctly identifies patterns."""
+        import re
+
+        # Test the regex pattern directly
+        pattern = self.router.STOCK_REGEX
+
+        # Should match
+        assert re.search(pattern, "$AAPL")
+        assert re.search(pattern, "What about $TSLA today?")
+        assert re.search(pattern, "Look at $NVDA and $AMD")
+
+        # Should not match crypto patterns
+        assert not re.search(pattern, "$$BTC")
+
+        # Should not match without $
+        assert not re.search(pattern, "AAPL")
+
+    def test_crypto_symbol_regex(self):
+        """Test that crypto symbol regex correctly identifies patterns."""
+        import re
+
+        pattern = self.router.CRYPTO_REGEX
+
+        # Should match
+        assert re.search(pattern, "$$BTC")
+        assert re.search(pattern, "How is $$ETH doing?")
+        assert re.search(pattern, "$$BTC and $$ETH")
+
+        # Should not match single $ patterns
+        assert not re.search(pattern, "$BTC")
+        assert not re.search(pattern, "$AAPL")
+
+    def test_find_symbols_stock(self):
+        """Test stock symbol detection."""
+        # Mock successful stock lookup
+        self.mock_stock.symbol_id.return_value = {"ticker": "AAPL", "title": "Apple Inc.", "mkt_cap_rank": 1}
+
+        result = self.router.find_symbols("$AAPL")
+
+        assert len(result) == 1
+        assert isinstance(result[0], Stock)
+        self.mock_stock.symbol_id.assert_called_with("AAPL")
+
+    def test_find_symbols_crypto(self):
+        """Test crypto symbol detection."""
+        # Mock crypto dataframe lookup
+        import pandas as pd
+
+        mock_crypto_df = pd.DataFrame({"symbol": ["btc"], "id": ["bitcoin"], "name": ["Bitcoin"], "type_id": ["$$btc"]})
+
+        # Set up the mock to return the DataFrame when filtered
+        self.mock_crypto.symbol_list = pd.DataFrame(
+            {
+                "symbol": ["btc", "eth"],
+                "id": ["bitcoin", "ethereum"],
+                "name": ["Bitcoin", "Ethereum"],
+                "type_id": ["$$btc", "$$eth"],
+            }
+        )
+
+        # Mock the pandas filtering
+        with patch.object(self.mock_crypto.symbol_list, "str") as mock_str:
+            mock_str.fullmatch.return_value = mock_crypto_df
+            mock_crypto_df.empty = False
+
+            result = self.router.find_symbols("$$BTC")
+
+            assert len(result) == 1
+            assert isinstance(result[0], Coin)
+
+    def test_find_symbols_no_matches(self):
+        """Test behavior when no symbols are found."""
+        result = self.router.find_symbols("Hello world")
+        assert result == []
+
+    def test_find_symbols_invalid_stock(self):
+        """Test behavior when stock symbol is not found."""
+        # Mock failed stock lookup
+        self.mock_stock.symbol_id.return_value = None
+
+        result = self.router.find_symbols("$INVALID")
+
+        assert result == []
+        self.mock_stock.symbol_id.assert_called_with("INVALID")
+
+    def test_find_symbols_multiple(self):
+        """Test finding multiple symbols."""
+        # Mock successful lookups for both
+        self.mock_stock.symbol_id.side_effect = [
+            {"ticker": "AAPL", "title": "Apple Inc.", "mkt_cap_rank": 1},
+            {"ticker": "TSLA", "title": "Tesla Inc.", "mkt_cap_rank": 5},
+        ]
+
+        result = self.router.find_symbols("$AAPL and $TSLA")
+
+        assert len(result) == 2
+        assert all(isinstance(symbol, Stock) for symbol in result)
+
+    def test_trending_weight_tracking(self):
+        """Test that trending counts are updated."""
+        # Mock successful stock lookup
+        self.mock_stock.symbol_id.return_value = {"ticker": "AAPL", "title": "Apple Inc.", "mkt_cap_rank": 1}
+
+        # Clear trending count
+        self.router.trending_count = {}
+
+        result = self.router.find_symbols("$AAPL", trending_weight=2)
+
+        assert len(result) == 1
+        assert "$AAPL" in self.router.trending_count
+        assert self.router.trending_count["$AAPL"] == 2

tests/test_symbols.py

@@ -0,0 +1,128 @@
+"""Test Symbol, Stock, and Coin class construction."""
+
+from unittest.mock import patch
+import pandas as pd
+from common.Symbol import Symbol, Stock, Coin
+
+
+class TestSymbol:
+    """Test base Symbol class."""
+
+    def test_symbol_creation(self):
+        """Test basic Symbol creation."""
+        symbol = Symbol("TEST")
+        assert symbol.symbol == "TEST"
+        assert symbol.id == "TEST"
+        assert symbol.name == "TEST"
+
+    def test_symbol_currency_default(self):
+        """Test that currency defaults to USD."""
+        symbol = Symbol("TEST")
+        assert symbol.currency == "usd"
+
+
+class TestStock:
+    """Test Stock class functionality."""
+
+    def test_stock_creation(self):
+        """Test Stock object creation with proper dictionary structure."""
+        stock_info = {"ticker": "AAPL", "title": "Apple Inc.", "mkt_cap_rank": 1}
+
+        stock = Stock(stock_info)
+        assert stock.symbol == "AAPL"
+        assert stock.id == "AAPL"
+        assert stock.name == "Apple Inc."
+        assert stock.tag == "$AAPL"
+        assert stock.market_cap_rank == 1
+
+        # Verify inheritance
+        assert isinstance(stock, Symbol)
+        assert isinstance(stock, Stock)
+
+    def test_stock_with_different_data(self):
+        """Test Stock creation with different stock data."""
+        stock_info = {"ticker": "TSLA", "title": "Tesla, Inc.", "mkt_cap_rank": 5}
+
+        stock = Stock(stock_info)
+        assert stock.symbol == "TSLA"
+        assert stock.id == "TSLA"
+        assert stock.name == "Tesla, Inc."
+        assert stock.tag == "$TSLA"
+        assert stock.market_cap_rank == 5
+
+    def test_stock_tag_format(self):
+        """Test that stock tags are properly formatted with $ prefix."""
+        stock_info = {"ticker": "NVDA", "title": "NVIDIA Corporation", "mkt_cap_rank": 3}
+
+        stock = Stock(stock_info)
+        assert stock.tag == "$NVDA"
+        assert stock.tag.startswith("$")
+
+
+class TestCoin:
+    """Test Coin class functionality."""
+
+    def test_coin_creation_single_row(self):
+        """Test Coin object creation with single row DataFrame."""
+        coin_data = pd.DataFrame({"symbol": ["btc"], "id": ["bitcoin"], "name": ["Bitcoin"], "type_id": ["$$btc"]})
+
+        coin = Coin(coin_data)
+        assert coin.symbol == "btc"
+        assert coin.id == "bitcoin"
+        assert coin.name == "Bitcoin"
+        assert coin.tag == "$$BTC"
+
+        # Verify inheritance
+        assert isinstance(coin, Symbol)
+        assert isinstance(coin, Coin)
+
+    def test_coin_creation_ethereum(self):
+        """Test Coin creation with Ethereum data."""
+        coin_data = pd.DataFrame({"symbol": ["eth"], "id": ["ethereum"], "name": ["Ethereum"], "type_id": ["$$eth"]})
+
+        coin = Coin(coin_data)
+        assert coin.symbol == "eth"
+        assert coin.id == "ethereum"
+        assert coin.name == "Ethereum"
+        assert coin.tag == "$$ETH"
+
+    def test_coin_multiple_rows_takes_first(self):
+        """Test that when multiple rows exist, only the first is used."""
+        coin_data = pd.DataFrame(
+            {
+                "symbol": ["btc", "btc2"],
+                "id": ["bitcoin", "bitcoin-cash"],
+                "name": ["Bitcoin", "Bitcoin Cash"],
+                "type_id": ["$$btc", "$$bch"],
+            }
+        )
+
+        # Mock the head method to return first row
+        with patch.object(coin_data, "head") as mock_head:
+            first_row = pd.DataFrame({"symbol": ["btc"], "id": ["bitcoin"], "name": ["Bitcoin"], "type_id": ["$$btc"]})
+            mock_head.return_value = first_row
+
+            coin = Coin(coin_data)
+            mock_head.assert_called_once_with(1)
+            # The coin should use the first row data
+            assert coin.symbol == "btc"
+            assert coin.id == "bitcoin"
+
+    def test_coin_tag_uppercase(self):
+        """Test that coin tags are properly uppercased."""
+        coin_data = pd.DataFrame({"symbol": ["ada"], "id": ["cardano"], "name": ["Cardano"], "type_id": ["$$ada"]})
+
+        coin = Coin(coin_data)
+        assert coin.tag == "$$ADA"
+        assert coin.tag.isupper()
+
+    def test_coin_dataframe_access(self):
+        """Test that coin properly accesses DataFrame values."""
+        coin_data = pd.DataFrame({"symbol": ["dot"], "id": ["polkadot"], "name": ["Polkadot"], "type_id": ["$$dot"]})
+
+        coin = Coin(coin_data)
+        # Test that values are extracted correctly from DataFrame
+        assert isinstance(coin.symbol, str)
+        assert isinstance(coin.id, str)
+        assert isinstance(coin.name, str)
+        assert isinstance(coin.tag, str)