Berack/upo-app-agents
0
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into tool

Browse Source 
# Conflicts:
#	src/app.py
#	src/app/agents/market_agent.py
#	src/app/markets/__init__.py
#	src/app/markets/binance.py
#	src/app/markets/binance_public.py
#	src/app/markets/coinbase.py
#	src/app/markets/cryptocompare.py
#	src/app/pipeline.py
#	src/app/predictor.py
#	src/app/toolkits/market_toolkit.py
#	tests/agents/test_market.py
This commit is contained in:
trojanhorse47
2025-10-03 11:36:51 +02:00
parent 0a0abc42ba d2fbc0ceea
commit 6fda746ed5
55 changed files with 2522 additions and 2099 deletions
Download Patch File Download Diff File Expand all files Collapse all files

596
tests/agents/test_market.py
View File

@@ -1,596 +0,0 @@
"""
Test suite completo per il sistema di mercato.
Questo modulo testa approfonditamente tutte le implementazioni di BaseWrapper
e verifica la conformità all'interfaccia definita in base.py.
"""
import os
from unittest.mock import Mock, patch
import pytest
from app.markets import MarketAPIs
# Import delle classi da testare
from app.markets.base import BaseWrapper, ProductInfo, Price
from app.markets.binance import BinanceWrapper
from app.markets.binance_public import PublicBinanceAgent
from app.markets.coinbase import CoinBaseWrapper
from app.markets.cryptocompare import CryptoCompareWrapper
class TestBaseWrapperInterface:
"""Test per verificare che tutte le implementazioni rispettino l'interfaccia BaseWrapper."""
def test_all_wrappers_extend_basewrapper(self):
"""Verifica che tutte le classi wrapper estendano BaseWrapper."""
wrapper_classes = [
CoinBaseWrapper,
CryptoCompareWrapper,
BinanceWrapper,
PublicBinanceAgent,
MarketAPIs
]
for wrapper_class in wrapper_classes:
assert issubclass(wrapper_class, BaseWrapper), f"{wrapper_class.__name__} deve estendere BaseWrapper"
def test_all_wrappers_implement_required_methods(self):
"""Verifica che tutte le classi implementino i metodi richiesti dall'interfaccia."""
wrapper_classes = [
CoinBaseWrapper,
CryptoCompareWrapper,
BinanceWrapper,
PublicBinanceAgent,
MarketAPIs
]
required_methods = ['get_product', 'get_products', 'get_all_products', 'get_historical_prices']
for wrapper_class in wrapper_classes:
for method in required_methods:
assert hasattr(wrapper_class, method), f"{wrapper_class.__name__} deve implementare {method}"
assert callable(getattr(wrapper_class, method)), f"{method} deve essere callable in {wrapper_class.__name__}"
class TestProductInfoModel:
"""Test per la classe ProductInfo e i suoi metodi di conversione."""
def test_productinfo_initialization(self):
"""Test inizializzazione di ProductInfo."""
product = ProductInfo()
assert product.id == ""
assert product.symbol == ""
assert product.price == 0.0
assert product.volume_24h == 0.0
assert product.status == ""
assert product.quote_currency == ""
def test_productinfo_with_data(self):
"""Test ProductInfo con dati specifici."""
product = ProductInfo(
id="BTC-USD",
symbol="BTC",
price=50000.0,
volume_24h=1000000.0,
status="TRADING",
quote_currency="USD"
)
assert product.id == "BTC-USD"
assert product.symbol == "BTC"
assert product.price == 50000.0
assert product.volume_24h == 1000000.0
assert product.status == "TRADING"
assert product.quote_currency == "USD"
def test_productinfo_from_cryptocompare(self):
"""Test conversione da dati CryptoCompare."""
mock_data = {
'FROMSYMBOL': 'BTC',
'TOSYMBOL': 'USD',
'PRICE': 50000.0,
'VOLUME24HOUR': 1000000.0
}
product = ProductInfo.from_cryptocompare(mock_data)
assert product.id == "BTC-USD"
assert product.symbol == "BTC"
assert product.price == 50000.0
assert product.volume_24h == 1000000.0
assert product.status == ""
def test_productinfo_from_binance(self):
"""Test conversione da dati Binance."""
ticker_data = {'symbol': 'BTCUSDT', 'price': '50000.0'}
ticker_24h_data = {'volume': '1000000.0'}
product = ProductInfo.from_binance(ticker_data, ticker_24h_data)
assert product.id == "BTCUSDT"
assert product.symbol == "BTC"
assert product.price == 50000.0
assert product.volume_24h == 1000000.0
assert product.status == "TRADING"
assert product.quote_currency == "USDT"
class TestPriceModel:
"""Test per la classe Price e i suoi metodi di conversione."""
def test_price_initialization(self):
"""Test inizializzazione di Price."""
price = Price()
assert price.high == 0.0
assert price.low == 0.0
assert price.open == 0.0
assert price.close == 0.0
assert price.volume == 0.0
assert price.time == ""
def test_price_with_data(self):
"""Test Price con dati specifici."""
price = Price(
high=51000.0,
low=49000.0,
open=50000.0,
close=50500.0,
volume=1000.0,
time="2024-01-01T00:00:00Z"
)
assert price.high == 51000.0
assert price.low == 49000.0
assert price.open == 50000.0
assert price.close == 50500.0
assert price.volume == 1000.0
assert price.time == "2024-01-01T00:00:00Z"
def test_price_from_cryptocompare(self):
"""Test conversione da dati CryptoCompare."""
mock_data = {
'high': 51000.0,
'low': 49000.0,
'open': 50000.0,
'close': 50500.0,
'volumeto': 1000.0,
'time': 1704067200
}
price = Price.from_cryptocompare(mock_data)
assert price.high == 51000.0
assert price.low == 49000.0
assert price.open == 50000.0
assert price.close == 50500.0
assert price.volume == 1000.0
assert price.time == "1704067200"
class TestCoinBaseWrapper:
"""Test specifici per CoinBaseWrapper."""
@pytest.mark.skipif(
not (os.getenv('COINBASE_API_KEY') and os.getenv('COINBASE_API_SECRET')),
reason="Credenziali Coinbase non configurate"
)
def test_coinbase_initialization_with_env_vars(self):
"""Test inizializzazione con variabili d'ambiente."""
wrapper = CoinBaseWrapper(currency="USD")
assert wrapper.currency == "USD"
assert wrapper.client is not None
@patch.dict(os.environ, {}, clear=True)
def test_coinbase_initialization_with_params(self):
"""Test inizializzazione con parametri espliciti quando non ci sono variabili d'ambiente."""
with pytest.raises(AssertionError, match="API key is required"):
CoinBaseWrapper(api_key=None, api_private_key=None)
@patch('app.markets.coinbase.RESTClient')
def test_coinbase_asset_formatting_behavior(self, mock_client):
"""Test comportamento di formattazione asset ID attraverso get_product."""
mock_response = Mock()
mock_response.product_id = "BTC-USD"
mock_response.base_currency_id = "BTC"
mock_response.price = "50000.0"
mock_response.volume_24h = "1000000.0"
mock_response.status = "TRADING"
mock_client_instance = Mock()
mock_client_instance.get_product.return_value = mock_response
mock_client.return_value = mock_client_instance
wrapper = CoinBaseWrapper(api_key="test", api_private_key="test")
# Test che entrambi i formati funzionino
wrapper.get_product("BTC")
wrapper.get_product("BTC-USD")
# Verifica che get_product sia stato chiamato con il formato corretto
assert mock_client_instance.get_product.call_count == 2
@patch('app.markets.coinbase.RESTClient')
def test_coinbase_get_product(self, mock_client):
"""Test get_product con mock."""
mock_response = Mock()
mock_response.product_id = "BTC-USD"
mock_response.base_currency_id = "BTC"
mock_response.price = "50000.0"
mock_response.volume_24h = "1000000.0"
mock_response.status = "TRADING"
mock_client_instance = Mock()
mock_client_instance.get_product.return_value = mock_response
mock_client.return_value = mock_client_instance
wrapper = CoinBaseWrapper(api_key="test", api_private_key="test")
product = wrapper.get_product("BTC")
assert isinstance(product, ProductInfo)
assert product.symbol == "BTC"
mock_client_instance.get_product.assert_called_once_with("BTC-USD")
class TestCryptoCompareWrapper:
"""Test specifici per CryptoCompareWrapper."""
@pytest.mark.skipif(
not os.getenv('CRYPTOCOMPARE_API_KEY'),
reason="CRYPTOCOMPARE_API_KEY non configurata"
)
def test_cryptocompare_initialization_with_env_var(self):
"""Test inizializzazione con variabile d'ambiente."""
wrapper = CryptoCompareWrapper(currency="USD")
assert wrapper.currency == "USD"
assert wrapper.api_key is not None
def test_cryptocompare_initialization_with_param(self):
"""Test inizializzazione con parametro esplicito."""
wrapper = CryptoCompareWrapper(api_key="test_key", currency="EUR")
assert wrapper.api_key == "test_key"
assert wrapper.currency == "EUR"
@patch('app.markets.cryptocompare.requests.get')
def test_cryptocompare_get_product(self, mock_get):
"""Test get_product con mock."""
mock_response = Mock()
mock_response.json.return_value = {
'RAW': {
'BTC': {
'USD': {
'FROMSYMBOL': 'BTC',
'TOSYMBOL': 'USD',
'PRICE': 50000.0,
'VOLUME24HOUR': 1000000.0
}
}
}
}
mock_response.raise_for_status.return_value = None
mock_get.return_value = mock_response
wrapper = CryptoCompareWrapper(api_key="test_key")
product = wrapper.get_product("BTC")
assert isinstance(product, ProductInfo)
assert product.symbol == "BTC"
assert product.price == 50000.0
def test_cryptocompare_get_all_products_workaround(self):
"""Test che get_all_products funzioni con il workaround implementato."""
wrapper = CryptoCompareWrapper(api_key="test_key")
# Il metodo ora dovrebbe restituire una lista di ProductInfo invece di sollevare NotImplementedError
products = wrapper.get_all_products()
assert isinstance(products, list)
# Verifica che la lista non sia vuota (dovrebbe contenere almeno alcuni asset popolari)
assert len(products) > 0
# Verifica che ogni elemento sia un ProductInfo
for product in products:
assert isinstance(product, ProductInfo)
class TestBinanceWrapper:
"""Test specifici per BinanceWrapper."""
def test_binance_initialization_without_credentials(self):
"""Test che l'inizializzazione fallisca senza credenziali."""
# Assicuriamoci che le variabili d'ambiente siano vuote per questo test
with patch.dict(os.environ, {}, clear=True):
with pytest.raises(AssertionError, match="API key is required"):
BinanceWrapper(api_key=None, api_secret="test")
with pytest.raises(AssertionError, match="API secret is required"):
BinanceWrapper(api_key="test", api_secret=None)
@patch('app.markets.binance.Client')
def test_binance_symbol_formatting_behavior(self, mock_client):
"""Test comportamento di formattazione simbolo attraverso get_product."""
mock_client_instance = Mock()
mock_client_instance.get_symbol_ticker.return_value = {
'symbol': 'BTCUSDT',
'price': '50000.0'
}
mock_client_instance.get_ticker.return_value = {
'volume': '1000000.0'
}
mock_client.return_value = mock_client_instance
wrapper = BinanceWrapper(api_key="test", api_secret="test")
# Test che entrambi i formati funzionino
wrapper.get_product("BTC")
wrapper.get_product("BTCUSDT")
# Verifica che i metodi siano stati chiamati
assert mock_client_instance.get_symbol_ticker.call_count == 2
@patch('app.markets.binance.Client')
def test_binance_get_product(self, mock_client):
"""Test get_product con mock."""
mock_client_instance = Mock()
mock_client_instance.get_symbol_ticker.return_value = {
'symbol': 'BTCUSDT',
'price': '50000.0'
}
mock_client_instance.get_ticker.return_value = {
'volume': '1000000.0'
}
mock_client.return_value = mock_client_instance
wrapper = BinanceWrapper(api_key="test", api_secret="test")
product = wrapper.get_product("BTC")
assert isinstance(product, ProductInfo)
assert product.symbol == "BTC"
assert product.price == 50000.0
class TestPublicBinanceAgent:
"""Test specifici per PublicBinanceAgent."""
@patch('app.markets.binance_public.Client')
def test_public_binance_initialization(self, mock_client):
"""Test inizializzazione senza credenziali."""
agent = PublicBinanceAgent()
assert agent.client is not None
mock_client.assert_called_once_with()
@patch('app.markets.binance_public.Client')
def test_public_binance_symbol_formatting_behavior(self, mock_client):
"""Test comportamento di formattazione simbolo attraverso get_product."""
mock_client_instance = Mock()
mock_client_instance.get_symbol_ticker.return_value = {
'symbol': 'BTCUSDT',
'price': '50000.0'
}
mock_client_instance.get_ticker.return_value = {
'volume': '1000000.0'
}
mock_client.return_value = mock_client_instance
agent = PublicBinanceAgent()
# Test che entrambi i formati funzionino
agent.get_product("BTC")
agent.get_product("BTCUSDT")
# Verifica che i metodi siano stati chiamati
assert mock_client_instance.get_symbol_ticker.call_count == 2
@patch('app.markets.binance_public.Client')
def test_public_binance_get_product(self, mock_client):
"""Test get_product con mock."""
mock_client_instance = Mock()
mock_client_instance.get_symbol_ticker.return_value = {
'symbol': 'BTCUSDT',
'price': '50000.0'
}
mock_client_instance.get_ticker.return_value = {
'volume': '1000000.0'
}
mock_client.return_value = mock_client_instance
agent = PublicBinanceAgent()
product = agent.get_product("BTC")
assert isinstance(product, ProductInfo)
assert product.symbol == "BTC"
assert product.price == 50000.0
@patch('app.markets.binance_public.Client')
def test_public_binance_get_all_products(self, mock_client):
"""Test get_all_products restituisce asset principali."""
mock_client_instance = Mock()
mock_client_instance.get_symbol_ticker.return_value = {
'symbol': 'BTCUSDT',
'price': '50000.0'
}
mock_client_instance.get_ticker.return_value = {
'volume': '1000000.0'
}
mock_client.return_value = mock_client_instance
agent = PublicBinanceAgent()
products = agent.get_all_products()
assert isinstance(products, list)
assert len(products) == 8 # Numero di asset principali definiti
for product in products:
assert isinstance(product, ProductInfo)
@patch('app.markets.binance_public.Client')
def test_public_binance_get_public_prices(self, mock_client):
"""Test metodo specifico get_public_prices."""
mock_client_instance = Mock()
mock_client_instance.get_symbol_ticker.return_value = {'price': '50000.0'}
mock_client_instance.get_server_time.return_value = {'serverTime': 1704067200000}
mock_client.return_value = mock_client_instance
agent = PublicBinanceAgent()
prices = agent.get_public_prices(["BTCUSDT"])
assert isinstance(prices, dict)
assert 'BTC_USD' in prices
assert prices['BTC_USD'] == 50000.0
assert 'source' in prices
assert prices['source'] == 'binance_public'
class TestMarketAPIs:
"""Test per la classe MarketAPIs che aggrega i wrapper."""
def test_market_apis_initialization_no_providers(self):
"""Test che l'inizializzazione fallisca senza provider disponibili."""
with patch.dict(os.environ, {}, clear=True):
with pytest.raises(AssertionError, match="No market API keys"):
MarketAPIs("USD")
@patch('app.markets.CoinBaseWrapper')
def test_market_apis_with_coinbase_only(self, mock_coinbase):
"""Test con solo Coinbase disponibile."""
mock_instance = Mock()
mock_coinbase.return_value = mock_instance
with patch('app.markets.CryptoCompareWrapper', side_effect=Exception("No API key")):
apis = MarketAPIs("USD")
assert len(apis.wrappers) == 1
assert apis.wrappers[0] == mock_instance
@patch('app.markets.CoinBaseWrapper')
@patch('app.markets.CryptoCompareWrapper')
def test_market_apis_delegation(self, mock_crypto, mock_coinbase):
"""Test che i metodi vengano delegati al primo wrapper disponibile."""
mock_coinbase_instance = Mock()
mock_crypto_instance = Mock()
mock_coinbase.return_value = mock_coinbase_instance
mock_crypto.return_value = mock_crypto_instance
apis = MarketAPIs("USD")
# Test delegazione get_product
apis.get_product("BTC")
mock_coinbase_instance.get_product.assert_called_once_with("BTC")
# Test delegazione get_products
apis.get_products(["BTC", "ETH"])
mock_coinbase_instance.get_products.assert_called_once_with(["BTC", "ETH"])
# Test delegazione get_all_products
apis.get_all_products()
mock_coinbase_instance.get_all_products.assert_called_once()
# Test delegazione get_historical_prices
apis.get_historical_prices("BTC")
mock_coinbase_instance.get_historical_prices.assert_called_once_with("BTC")
class TestErrorHandling:
"""Test per la gestione degli errori in tutti i wrapper."""
@patch('app.markets.binance_public.Client')
def test_public_binance_error_handling(self, mock_client):
"""Test gestione errori in PublicBinanceAgent."""
mock_client_instance = Mock()
mock_client_instance.get_symbol_ticker.side_effect = Exception("API Error")
mock_client.return_value = mock_client_instance
agent = PublicBinanceAgent()
product = agent.get_product("INVALID")
# Dovrebbe restituire un ProductInfo vuoto invece di sollevare eccezione
assert isinstance(product, ProductInfo)
assert product.id == "INVALID"
assert product.symbol == "INVALID"
@patch('app.markets.cryptocompare.requests.get')
def test_cryptocompare_network_error(self, mock_get):
"""Test gestione errori di rete in CryptoCompareWrapper."""
mock_get.side_effect = Exception("Network Error")
wrapper = CryptoCompareWrapper(api_key="test")
with pytest.raises(Exception):
wrapper.get_product("BTC")
@patch('app.markets.binance.Client')
def test_binance_api_error_in_get_products(self, mock_client):
"""Test gestione errori in BinanceWrapper.get_products."""
mock_client_instance = Mock()
mock_client_instance.get_symbol_ticker.side_effect = Exception("API Error")
mock_client.return_value = mock_client_instance
wrapper = BinanceWrapper(api_key="test", api_secret="test")
products = wrapper.get_products(["BTC", "ETH"])
# Dovrebbe restituire lista vuota invece di sollevare eccezione
assert isinstance(products, list)
assert len(products) == 0
class TestIntegrationScenarios:
"""Test di integrazione per scenari reali."""
def test_wrapper_method_signatures(self):
"""Verifica che tutti i wrapper abbiano le stesse signature dei metodi."""
wrapper_classes = [CoinBaseWrapper, CryptoCompareWrapper, BinanceWrapper, PublicBinanceAgent]
for wrapper_class in wrapper_classes:
# Verifica get_product
assert hasattr(wrapper_class, 'get_product')
# Verifica get_products
assert hasattr(wrapper_class, 'get_products')
# Verifica get_all_products
assert hasattr(wrapper_class, 'get_all_products')
# Verifica get_historical_prices
assert hasattr(wrapper_class, 'get_historical_prices')
def test_productinfo_consistency(self):
"""Test che tutti i metodi from_* di ProductInfo restituiscano oggetti consistenti."""
# Test from_cryptocompare
crypto_data = {
'FROMSYMBOL': 'BTC',
'TOSYMBOL': 'USD',
'PRICE': 50000.0,
'VOLUME24HOUR': 1000000.0
}
crypto_product = ProductInfo.from_cryptocompare(crypto_data)
# Test from_binance
binance_ticker = {'symbol': 'BTCUSDT', 'price': '50000.0'}
binance_24h = {'volume': '1000000.0'}
binance_product = ProductInfo.from_binance(binance_ticker, binance_24h)
# Verifica che entrambi abbiano gli stessi campi
assert hasattr(crypto_product, 'id')
assert hasattr(crypto_product, 'symbol')
assert hasattr(crypto_product, 'price')
assert hasattr(crypto_product, 'volume_24h')
assert hasattr(binance_product, 'id')
assert hasattr(binance_product, 'symbol')
assert hasattr(binance_product, 'price')
assert hasattr(binance_product, 'volume_24h')
def test_price_consistency(self):
"""Test che tutti i metodi from_* di Price restituiscano oggetti consistenti."""
# Test from_cryptocompare
crypto_data = {
'high': 51000.0,
'low': 49000.0,
'open': 50000.0,
'close': 50500.0,
'volumeto': 1000.0,
'time': 1704067200
}
crypto_price = Price.from_cryptocompare(crypto_data)
# Verifica che abbia tutti i campi richiesti
assert hasattr(crypto_price, 'high')
assert hasattr(crypto_price, 'low')
assert hasattr(crypto_price, 'open')
assert hasattr(crypto_price, 'close')
assert hasattr(crypto_price, 'volume')
assert hasattr(crypto_price, 'time')
if __name__ == "__main__":
pytest.main([__file__, "-v"])

5
tests/agents/test_predictor.py
View File

@@ -16,8 +16,8 @@ def unified_checks(model: AppModels, input):
for item in content.portfolio:
assert item.asset not in (None, "", "null")
assert isinstance(item.asset, str)
assert item.percentage > 0
assert item.percentage <= 100
assert item.percentage >= 0.0
assert item.percentage <= 100.0
assert isinstance(item.percentage, (int, float))
assert item.motivation not in (None, "", "null")
assert isinstance(item.motivation, str)
@@ -41,6 +41,7 @@ class TestPredictor:
def test_gemini_model_output(self, inputs):
unified_checks(AppModels.GEMINI, inputs)
@pytest.mark.slow
def test_ollama_qwen_model_output(self, inputs):
unified_checks(AppModels.OLLAMA_QWEN, inputs)

53
tests/api/test_binance.py Normal file
View File

@@ -0,0 +1,53 @@
import pytest
from app.markets.binance import BinanceWrapper
@pytest.mark.market
@pytest.mark.api
class TestBinance:
def test_binance_init(self):
market = BinanceWrapper()
assert market is not None
assert hasattr(market, 'currency')
assert market.currency == "USDT"
def test_binance_get_product(self):
market = BinanceWrapper()
product = market.get_product("BTC")
assert product is not None
assert hasattr(product, 'symbol')
assert product.symbol == "BTC"
assert hasattr(product, 'price')
assert product.price > 0
def test_binance_get_products(self):
market = BinanceWrapper()
products = market.get_products(["BTC", "ETH"])
assert products is not None
assert isinstance(products, list)
assert len(products) == 2
symbols = [p.symbol for p in products]
assert "BTC" in symbols
assert "ETH" in symbols
for product in products:
assert hasattr(product, 'price')
assert product.price > 0
def test_binance_invalid_product(self):
market = BinanceWrapper()
with pytest.raises(Exception):
_ = market.get_product("INVALID")
def test_binance_history(self):
market = BinanceWrapper()
history = market.get_historical_prices("BTC", limit=5)
assert history is not None
assert isinstance(history, list)
assert len(history) == 5
for entry in history:
assert hasattr(entry, 'timestamp_ms')
assert hasattr(entry, 'close')
assert hasattr(entry, 'high')
assert entry.close > 0
assert entry.high > 0
assert entry.timestamp_ms > 0

55
tests/api/test_coinbase.py Normal file
View File

@@ -0,0 +1,55 @@
import os
import pytest
from app.markets import CoinBaseWrapper
@pytest.mark.market
@pytest.mark.api
@pytest.mark.skipif(not(os.getenv('COINBASE_API_KEY')) or not(os.getenv('COINBASE_API_SECRET')), reason="COINBASE_API_KEY or COINBASE_API_SECRET not set in environment variables")
class TestCoinBase:
def test_coinbase_init(self):
market = CoinBaseWrapper()
assert market is not None
assert hasattr(market, 'currency')
assert market.currency == "USD"
def test_coinbase_get_product(self):
market = CoinBaseWrapper()
product = market.get_product("BTC")
assert product is not None
assert hasattr(product, 'symbol')
assert product.symbol == "BTC"
assert hasattr(product, 'price')
assert product.price > 0
def test_coinbase_get_products(self):
market = CoinBaseWrapper()
products = market.get_products(["BTC", "ETH"])
assert products is not None
assert isinstance(products, list)
assert len(products) == 2
symbols = [p.symbol for p in products]
assert "BTC" in symbols
assert "ETH" in symbols
for product in products:
assert hasattr(product, 'price')
assert product.price > 0
def test_coinbase_invalid_product(self):
market = CoinBaseWrapper()
with pytest.raises(Exception):
_ = market.get_product("INVALID")
def test_coinbase_history(self):
market = CoinBaseWrapper()
history = market.get_historical_prices("BTC", limit=5)
assert history is not None
assert isinstance(history, list)
assert len(history) == 5
for entry in history:
assert hasattr(entry, 'timestamp_ms')
assert hasattr(entry, 'close')
assert hasattr(entry, 'high')
assert entry.close > 0
assert entry.high > 0
assert entry.timestamp_ms > 0

57
tests/api/test_cryptocompare.py Normal file
View File

@@ -0,0 +1,57 @@
import os
import pytest
from app.markets import CryptoCompareWrapper
@pytest.mark.market
@pytest.mark.api
@pytest.mark.skipif(not os.getenv('CRYPTOCOMPARE_API_KEY'), reason="CRYPTOCOMPARE_API_KEY not set in environment variables")
class TestCryptoCompare:
def test_cryptocompare_init(self):
market = CryptoCompareWrapper()
assert market is not None
assert hasattr(market, 'api_key')
assert market.api_key == os.getenv('CRYPTOCOMPARE_API_KEY')
assert hasattr(market, 'currency')
assert market.currency == "USD"
def test_cryptocompare_get_product(self):
market = CryptoCompareWrapper()
product = market.get_product("BTC")
assert product is not None
assert hasattr(product, 'symbol')
assert product.symbol == "BTC"
assert hasattr(product, 'price')
assert product.price > 0
def test_cryptocompare_get_products(self):
market = CryptoCompareWrapper()
products = market.get_products(["BTC", "ETH"])
assert products is not None
assert isinstance(products, list)
assert len(products) == 2
symbols = [p.symbol for p in products]
assert "BTC" in symbols
assert "ETH" in symbols
for product in products:
assert hasattr(product, 'price')
assert product.price > 0
def test_cryptocompare_invalid_product(self):
market = CryptoCompareWrapper()
with pytest.raises(Exception):
_ = market.get_product("INVALID")
def test_cryptocompare_history(self):
market = CryptoCompareWrapper()
history = market.get_historical_prices("BTC", limit=5)
assert history is not None
assert isinstance(history, list)
assert len(history) == 5
for entry in history:
assert hasattr(entry, 'timestamp_ms')
assert hasattr(entry, 'close')
assert hasattr(entry, 'high')
assert entry.close > 0
assert entry.high > 0
assert entry.timestamp_ms > 0

38
tests/api/test_cryptopanic_api.py Normal file
View File

@@ -0,0 +1,38 @@
import os
import pytest
from app.news import CryptoPanicWrapper
@pytest.mark.limited
@pytest.mark.news
@pytest.mark.api
@pytest.mark.skipif(not os.getenv("CRYPTOPANIC_API_KEY"), reason="CRYPTOPANIC_API_KEY not set")
class TestCryptoPanicAPI:
def test_crypto_panic_api_initialization(self):
crypto = CryptoPanicWrapper()
assert crypto is not None
def test_crypto_panic_api_get_latest_news(self):
crypto = CryptoPanicWrapper()
articles = crypto.get_latest_news(query="", limit=2)
assert isinstance(articles, list)
assert len(articles) == 2
for article in articles:
assert article.source is not None or article.source != ""
assert article.time is not None or article.time != ""
assert article.title is not None or article.title != ""
assert article.description is not None or article.description != ""
# Useless since both methods use the same endpoint
# def test_crypto_panic_api_get_top_headlines(self):
# crypto = CryptoPanicWrapper()
# articles = crypto.get_top_headlines(total=2)
# assert isinstance(articles, list)
# assert len(articles) == 2
# for article in articles:
# assert article.source is not None or article.source != ""
# assert article.time is not None or article.time != ""
# assert article.title is not None or article.title != ""
# assert article.description is not None or article.description != ""

34
tests/api/test_duckduckgo_news.py Normal file
View File

@@ -0,0 +1,34 @@
import pytest
from app.news import DuckDuckGoWrapper
@pytest.mark.news
@pytest.mark.api
class TestDuckDuckGoNews:
def test_duckduckgo_initialization(self):
news = DuckDuckGoWrapper()
assert news.tool is not None
def test_duckduckgo_get_latest_news(self):
news = DuckDuckGoWrapper()
articles = news.get_latest_news(query="crypto", limit=2)
assert isinstance(articles, list)
assert len(articles) == 2
for article in articles:
assert article.source is not None or article.source != ""
assert article.time is not None or article.time != ""
assert article.title is not None or article.title != ""
assert article.description is not None or article.description != ""
def test_duckduckgo_get_top_headlines(self):
news = DuckDuckGoWrapper()
articles = news.get_top_headlines(limit=2)
assert isinstance(articles, list)
assert len(articles) == 2
for article in articles:
assert article.source is not None or article.source != ""
assert article.time is not None or article.time != ""
assert article.title is not None or article.title != ""
assert article.description is not None or article.description != ""

34
tests/api/test_google_news.py Normal file
View File

@@ -0,0 +1,34 @@
import pytest
from app.news import GoogleNewsWrapper
@pytest.mark.news
@pytest.mark.api
class TestGoogleNews:
def test_gnews_api_initialization(self):
gnews_api = GoogleNewsWrapper()
assert gnews_api is not None
def test_gnews_api_get_latest_news(self):
gnews_api = GoogleNewsWrapper()
articles = gnews_api.get_latest_news(query="crypto", limit=2)
assert isinstance(articles, list)
assert len(articles) == 2
for article in articles:
assert article.source is not None or article.source != ""
assert article.time is not None or article.time != ""
assert article.title is not None or article.title != ""
assert article.description is not None or article.description != ""
def test_gnews_api_get_top_headlines(self):
news_api = GoogleNewsWrapper()
articles = news_api.get_top_headlines(limit=2)
assert isinstance(articles, list)
assert len(articles) == 2
for article in articles:
assert article.source is not None or article.source != ""
assert article.time is not None or article.time != ""
assert article.title is not None or article.title != ""
assert article.description is not None or article.description != ""

37
tests/api/test_news_api.py Normal file
View File

@@ -0,0 +1,37 @@
import os
import pytest
from app.news import NewsApiWrapper
@pytest.mark.news
@pytest.mark.api
@pytest.mark.skipif(not os.getenv("NEWS_API_KEY"), reason="NEWS_API_KEY not set in environment variables")
class TestNewsAPI:
def test_news_api_initialization(self):
news_api = NewsApiWrapper()
assert news_api.client is not None
def test_news_api_get_latest_news(self):
news_api = NewsApiWrapper()
articles = news_api.get_latest_news(query="crypto", limit=2)
assert isinstance(articles, list)
assert len(articles) > 0 # Ensure we got some articles (apparently it doesn't always return the requested number)
for article in articles:
assert article.source is not None or article.source != ""
assert article.time is not None or article.time != ""
assert article.title is not None or article.title != ""
assert article.description is not None or article.description != ""
def test_news_api_get_top_headlines(self):
news_api = NewsApiWrapper()
articles = news_api.get_top_headlines(limit=2)
assert isinstance(articles, list)
# assert len(articles) > 0 # apparently it doesn't always return SOME articles
for article in articles:
assert article.source is not None or article.source != ""
assert article.time is not None or article.time != ""
assert article.title is not None or article.title != ""
assert article.description is not None or article.description != ""

25
tests/api/test_reddit.py Normal file
View File

@@ -0,0 +1,25 @@
import os
import pytest
from praw import Reddit
from app.social.reddit import MAX_COMMENTS, RedditWrapper
@pytest.mark.social
@pytest.mark.api
@pytest.mark.skipif(not(os.getenv("REDDIT_API_CLIENT_ID")) or not os.getenv("REDDIT_API_CLIENT_SECRET"), reason="REDDIT_CLIENT_ID and REDDIT_API_CLIENT_SECRET not set in environment variables")
class TestRedditWrapper:
def test_initialization(self):
wrapper = RedditWrapper()
assert wrapper is not None
assert isinstance(wrapper.tool, Reddit)
def test_get_top_crypto_posts(self):
wrapper = RedditWrapper()
posts = wrapper.get_top_crypto_posts(limit=2)
assert isinstance(posts, list)
assert len(posts) == 2
for post in posts:
assert post.title != ""
assert isinstance(post.comments, list)
assert len(post.comments) <= MAX_COMMENTS
for comment in post.comments:
assert comment.description != ""

56
tests/api/test_yfinance.py Normal file
View File

@@ -0,0 +1,56 @@
import pytest
from app.markets import YFinanceWrapper
@pytest.mark.market
@pytest.mark.api
class TestYFinance:
def test_yfinance_init(self):
market = YFinanceWrapper()
assert market is not None
assert hasattr(market, 'currency')
assert market.currency == "USD"
assert hasattr(market, 'tool')
assert market.tool is not None
def test_yfinance_get_crypto_product(self):
market = YFinanceWrapper()
product = market.get_product("BTC")
assert product is not None
assert hasattr(product, 'symbol')
# BTC verrà convertito in BTC-USD dal formattatore
assert product.symbol in ["BTC", "BTC-USD"]
assert hasattr(product, 'price')
assert product.price > 0
def test_yfinance_get_products(self):
market = YFinanceWrapper()
products = market.get_products(["BTC", "ETH"])
assert products is not None
assert isinstance(products, list)
assert len(products) == 2
symbols = [p.symbol for p in products]
assert "BTC" in symbols
assert "ETH" in symbols
for product in products:
assert hasattr(product, 'price')
assert product.price > 0
def test_yfinance_invalid_product(self):
market = YFinanceWrapper()
with pytest.raises(Exception):
_ = market.get_product("INVALIDSYMBOL123")
def test_yfinance_crypto_history(self):
market = YFinanceWrapper()
history = market.get_historical_prices("BTC", limit=5)
assert history is not None
assert isinstance(history, list)
assert len(history) == 5
for entry in history:
assert hasattr(entry, 'timestamp_ms')
assert hasattr(entry, 'close')
assert hasattr(entry, 'high')
assert entry.close > 0
assert entry.high > 0
assert entry.timestamp_ms > 0

46
tests/conftest.py
View File

@@ -2,16 +2,10 @@
Configurazione pytest per i test del progetto upo-appAI.
"""
import sys
import pytest
from pathlib import Path
# Aggiungi il path src al PYTHONPATH per tutti i test
src_path = Path(__file__).parent.parent / "src"
sys.path.insert(0, str(src_path))
from dotenv import load_dotenv
# Carica le variabili d'ambiente per tutti i test
from dotenv import load_dotenv
load_dotenv()
@@ -20,9 +14,17 @@ def pytest_configure(config:pytest.Config):
markers = [
("slow", "marks tests as slow (deselect with '-m \"not slow\"')"),
("limited", "marks tests that have limited execution due to API constraints"),
("api", "marks tests that require API access"),
("coinbase", "marks tests that require Coinbase credentials"),
("cryptocompare", "marks tests that require CryptoCompare credentials"),
("market", "marks tests that use market data"),
("news", "marks tests that use news"),
("social", "marks tests that use social media"),
("wrapper", "marks tests for wrapper handler"),
("tools", "marks tests for tools"),
("aggregator", "marks tests for market data aggregator"),
("gemini", "marks tests that use Gemini model"),
("ollama_gpt", "marks tests that use Ollama GPT model"),
("ollama_qwen", "marks tests that use Ollama Qwen model"),
@@ -32,21 +34,13 @@ def pytest_configure(config:pytest.Config):
config.addinivalue_line("markers", line)
def pytest_collection_modifyitems(config, items):
"""Modifica automaticamente gli item di test aggiungendogli marker basati sul nome"""
"""Modifica automaticamente degli item di test rimovendoli"""
# Rimuovo i test "limited" e "slow" se non richiesti esplicitamente
mark_to_remove = ['limited', 'slow']
for mark in mark_to_remove:
markexpr = getattr(config.option, "markexpr", None)
if markexpr and mark in markexpr.lower():
continue
markers_to_add = {
"api": pytest.mark.api,
"coinbase": pytest.mark.api,
"cryptocompare": pytest.mark.api,
"overview": pytest.mark.slow,
"analysis": pytest.mark.slow,
"gemini": pytest.mark.gemini,
"ollama_gpt": pytest.mark.ollama_gpt,
"ollama_qwen": pytest.mark.ollama_qwen,
}
for item in items:
name = item.name.lower()
for key, marker in markers_to_add.items():
if key in name:
item.add_marker(marker)
new_mark = (f"({markexpr}) and " if markexpr else "") + f"not {mark}"
setattr(config.option, "markexpr", new_mark)

41
tests/tools/test_market_tool.py Normal file
View File

@@ -0,0 +1,41 @@
import pytest
from app.markets import MarketAPIsTool
@pytest.mark.tools
@pytest.mark.market
@pytest.mark.api
class TestMarketAPIsTool:
def test_wrapper_initialization(self):
market_wrapper = MarketAPIsTool("USD")
assert market_wrapper is not None
assert hasattr(market_wrapper, 'get_product')
assert hasattr(market_wrapper, 'get_products')
assert hasattr(market_wrapper, 'get_historical_prices')
def test_wrapper_capabilities(self):
market_wrapper = MarketAPIsTool("USD")
capabilities = []
if hasattr(market_wrapper, 'get_product'):
capabilities.append('single_product')
if hasattr(market_wrapper, 'get_products'):
capabilities.append('multiple_products')
if hasattr(market_wrapper, 'get_historical_prices'):
capabilities.append('historical_data')
assert len(capabilities) > 0
def test_market_data_retrieval(self):
market_wrapper = MarketAPIsTool("USD")
btc_product = market_wrapper.get_product("BTC")
assert btc_product is not None
assert hasattr(btc_product, 'symbol')
assert hasattr(btc_product, 'price')
assert btc_product.price > 0
def test_error_handling(self):
try:
market_wrapper = MarketAPIsTool("USD")
fake_product = market_wrapper.get_product("NONEXISTENT_CRYPTO_SYMBOL_12345")
assert fake_product is None or fake_product.price == 0
except Exception as e:
pass

49
tests/tools/test_news_tool.py Normal file
View File

@@ -0,0 +1,49 @@
import pytest
from app.news import NewsAPIsTool
@pytest.mark.tools
@pytest.mark.news
@pytest.mark.api
class TestNewsAPITool:
def test_news_api_tool(self):
tool = NewsAPIsTool()
assert tool is not None
def test_news_api_tool_get_top(self):
tool = NewsAPIsTool()
result = tool.wrapper_handler.try_call(lambda w: w.get_top_headlines(limit=2))
assert isinstance(result, list)
assert len(result) > 0
for article in result:
assert article.title is not None
assert article.source is not None
def test_news_api_tool_get_latest(self):
tool = NewsAPIsTool()
result = tool.wrapper_handler.try_call(lambda w: w.get_latest_news(query="crypto", limit=2))
assert isinstance(result, list)
assert len(result) > 0
for article in result:
assert article.title is not None
assert article.source is not None
def test_news_api_tool_get_top__all_results(self):
tool = NewsAPIsTool()
result = tool.wrapper_handler.try_call_all(lambda w: w.get_top_headlines(limit=2))
assert isinstance(result, dict)
assert len(result.keys()) > 0
for provider, articles in result.items():
for article in articles:
assert article.title is not None
assert article.source is not None
def test_news_api_tool_get_latest__all_results(self):
tool = NewsAPIsTool()
result = tool.wrapper_handler.try_call_all(lambda w: w.get_latest_news(query="crypto", limit=2))
assert isinstance(result, dict)
assert len(result.keys()) > 0
for provider, articles in result.items():
for article in articles:
assert article.title is not None
assert article.source is not None

30
tests/tools/test_socials_tool.py Normal file
View File

@@ -0,0 +1,30 @@
import pytest
from app.social import SocialAPIsTool
@pytest.mark.tools
@pytest.mark.social
@pytest.mark.api
class TestSocialAPIsTool:
def test_social_api_tool(self):
tool = SocialAPIsTool()
assert tool is not None
def test_social_api_tool_get_top(self):
tool = SocialAPIsTool()
result = tool.wrapper_handler.try_call(lambda w: w.get_top_crypto_posts(limit=2))
assert isinstance(result, list)
assert len(result) > 0
for post in result:
assert post.title is not None
assert post.time is not None
def test_social_api_tool_get_top__all_results(self):
tool = SocialAPIsTool()
result = tool.wrapper_handler.try_call_all(lambda w: w.get_top_crypto_posts(limit=2))
assert isinstance(result, dict)
assert len(result.keys()) > 0
for provider, posts in result.items():
for post in posts:
assert post.title is not None
assert post.time is not None

120
tests/utils/test_market_aggregator.py Normal file
View File

@@ -0,0 +1,120 @@
import pytest
from app.markets.base import ProductInfo, Price
from app.utils.market_aggregation import aggregate_history_prices, aggregate_product_info
@pytest.mark.aggregator
@pytest.mark.market
class TestMarketDataAggregator:
def __product(self, symbol: str, price: float, volume: float, currency: str) -> ProductInfo:
prod = ProductInfo()
prod.id=f"{symbol}-{currency}"
prod.symbol=symbol
prod.price=price
prod.volume_24h=volume
prod.quote_currency=currency
return prod
def __price(self, timestamp_ms: int, high: float, low: float, open: float, close: float, volume: float) -> Price:
price = Price()
price.timestamp_ms = timestamp_ms
price.high = high
price.low = low
price.open = open
price.close = close
price.volume = volume
return price
def test_aggregate_product_info(self):
products: dict[str, list[ProductInfo]] = {
"Provider1": [self.__product("BTC", 50000.0, 1000.0, "USD")],
"Provider2": [self.__product("BTC", 50100.0, 1100.0, "USD")],
"Provider3": [self.__product("BTC", 49900.0, 900.0, "USD")],
}
aggregated = aggregate_product_info(products)
assert len(aggregated) == 1
info = aggregated[0]
assert info is not None
assert info.symbol == "BTC"
avg_weighted_price = (50000.0 * 1000.0 + 50100.0 * 1100.0 + 49900.0 * 900.0) / (1000.0 + 1100.0 + 900.0)
assert info.price == pytest.approx(avg_weighted_price, rel=1e-3)
assert info.volume_24h == pytest.approx(1000.0, rel=1e-3)
assert info.quote_currency == "USD"
def test_aggregate_product_info_multiple_symbols(self):
products = {
"Provider1": [
self.__product("BTC", 50000.0, 1000.0, "USD"),
self.__product("ETH", 4000.0, 2000.0, "USD"),
],
"Provider2": [
self.__product("BTC", 50100.0, 1100.0, "USD"),
self.__product("ETH", 4050.0, 2100.0, "USD"),
],
}
aggregated = aggregate_product_info(products)
assert len(aggregated) == 2
btc_info = next((p for p in aggregated if p.symbol == "BTC"), None)
eth_info = next((p for p in aggregated if p.symbol == "ETH"), None)
assert btc_info is not None
avg_weighted_price_btc = (50000.0 * 1000.0 + 50100.0 * 1100.0) / (1000.0 + 1100.0)
assert btc_info.price == pytest.approx(avg_weighted_price_btc, rel=1e-3)
assert btc_info.volume_24h == pytest.approx(1050.0, rel=1e-3)
assert btc_info.quote_currency == "USD"
assert eth_info is not None
avg_weighted_price_eth = (4000.0 * 2000.0 + 4050.0 * 2100.0) / (2000.0 + 2100.0)
assert eth_info.price == pytest.approx(avg_weighted_price_eth, rel=1e-3)
assert eth_info.volume_24h == pytest.approx(2050.0, rel=1e-3)
assert eth_info.quote_currency == "USD"
def test_aggregate_product_info_with_no_data(self):
products = {
"Provider1": [],
"Provider2": [],
}
aggregated = aggregate_product_info(products)
assert len(aggregated) == 0
def test_aggregate_product_info_with_partial_data(self):
products = {
"Provider1": [self.__product("BTC", 50000.0, 1000.0, "USD")],
"Provider2": [],
}
aggregated = aggregate_product_info(products)
assert len(aggregated) == 1
info = aggregated[0]
assert info.symbol == "BTC"
assert info.price == pytest.approx(50000.0, rel=1e-3)
assert info.volume_24h == pytest.approx(1000.0, rel=1e-3)
assert info.quote_currency == "USD"
def test_aggregate_history_prices(self):
"""Test aggregazione di prezzi storici usando aggregate_history_prices"""
prices = {
"Provider1": [
self.__price(1685577600000, 50000.0, 49500.0, 49600.0, 49900.0, 150.0),
self.__price(1685581200000, 50200.0, 49800.0, 50000.0, 50100.0, 200.0),
],
"Provider2": [
self.__price(1685577600000, 50100.0, 49600.0, 49700.0, 50000.0, 180.0),
self.__price(1685581200000, 50300.0, 49900.0, 50100.0, 50200.0, 220.0),
],
}
aggregated = aggregate_history_prices(prices)
assert len(aggregated) == 2
assert aggregated[0].timestamp_ms == 1685577600000
assert aggregated[0].high == pytest.approx(50050.0, rel=1e-3)
assert aggregated[0].low == pytest.approx(49550.0, rel=1e-3)
assert aggregated[1].timestamp_ms == 1685581200000
assert aggregated[1].high == pytest.approx(50250.0, rel=1e-3)
assert aggregated[1].low == pytest.approx(49850.0, rel=1e-3)

152
tests/utils/test_wrapper_handler.py Normal file
View File

@@ -0,0 +1,152 @@
import pytest
from app.utils.wrapper_handler import WrapperHandler
class MockWrapper:
def do_something(self) -> str:
return "Success"
class MockWrapper2(MockWrapper):
def do_something(self) -> str:
return "Success 2"
class FailingWrapper(MockWrapper):
def do_something(self):
raise Exception("Intentional Failure")
class MockWrapperWithParameters:
def do_something(self, param1: str, param2: int) -> str:
return f"Success {param1} and {param2}"
class FailingWrapperWithParameters(MockWrapperWithParameters):
def do_something(self, param1: str, param2: int):
raise Exception("Intentional Failure")
@pytest.mark.wrapper
class TestWrapperHandler:
def test_init_failing(self):
with pytest.raises(AssertionError) as exc_info:
WrapperHandler([MockWrapper, MockWrapper2])
assert exc_info.type == AssertionError
def test_init_failing_empty(self):
with pytest.raises(AssertionError) as exc_info:
WrapperHandler.build_wrappers([])
assert exc_info.type == AssertionError
def test_init_failing_with_instances(self):
with pytest.raises(AssertionError) as exc_info:
WrapperHandler.build_wrappers([MockWrapper(), MockWrapper2()])
assert exc_info.type == AssertionError
def test_init_not_failing(self):
handler = WrapperHandler.build_wrappers([MockWrapper, MockWrapper2])
assert handler is not None
assert len(handler.wrappers) == 2
handler = WrapperHandler([MockWrapper(), MockWrapper2()])
assert handler is not None
assert len(handler.wrappers) == 2
def test_all_wrappers_fail(self):
wrappers = [FailingWrapper, FailingWrapper]
handler: WrapperHandler[MockWrapper] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=2, retry_delay=0)
with pytest.raises(Exception) as exc_info:
handler.try_call(lambda w: w.do_something())
assert "All wrappers failed" in str(exc_info.value)
def test_success_on_first_try(self):
wrappers = [MockWrapper, FailingWrapper]
handler: WrapperHandler[MockWrapper] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=2, retry_delay=0)
result = handler.try_call(lambda w: w.do_something())
assert result == "Success"
assert handler.index == 0 # Should still be on the first wrapper
assert handler.retry_count == 0
def test_eventual_success(self):
wrappers = [FailingWrapper, MockWrapper]
handler: WrapperHandler[MockWrapper] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=2, retry_delay=0)
result = handler.try_call(lambda w: w.do_something())
assert result == "Success"
assert handler.index == 1 # Should have switched to the second wrapper
assert handler.retry_count == 0
def test_partial_failures(self):
wrappers = [FailingWrapper, MockWrapper, FailingWrapper]
handler: WrapperHandler[MockWrapper] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0)
result = handler.try_call(lambda w: w.do_something())
assert result == "Success"
assert handler.index == 1 # Should have switched to the second wrapper
assert handler.retry_count == 0
# Next call should still succeed on the second wrapper
result = handler.try_call(lambda w: w.do_something())
assert result == "Success"
assert handler.index == 1 # Should still be on the second wrapper
assert handler.retry_count == 0
handler.index = 2 # Manually switch to the third wrapper
result = handler.try_call(lambda w: w.do_something())
assert result == "Success"
assert handler.index == 1 # Should return to the second wrapper after failure
assert handler.retry_count == 0
def test_try_call_all_success(self):
wrappers = [MockWrapper, MockWrapper2]
handler: WrapperHandler[MockWrapper] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0)
results = handler.try_call_all(lambda w: w.do_something())
assert results == {MockWrapper: "Success", MockWrapper2: "Success 2"}
def test_try_call_all_partial_failures(self):
# Only the second wrapper should succeed
wrappers = [FailingWrapper, MockWrapper, FailingWrapper]
handler: WrapperHandler[MockWrapper] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0)
results = handler.try_call_all(lambda w: w.do_something())
assert results == {MockWrapper: "Success"}
# Only the second and fourth wrappers should succeed
wrappers = [FailingWrapper, MockWrapper, FailingWrapper, MockWrapper2]
handler: WrapperHandler[MockWrapper] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0)
results = handler.try_call_all(lambda w: w.do_something())
assert results == {MockWrapper: "Success", MockWrapper2: "Success 2"}
def test_try_call_all_all_fail(self):
# Test when all wrappers fail
handler_all_fail: WrapperHandler[MockWrapper] = WrapperHandler.build_wrappers([FailingWrapper, FailingWrapper], try_per_wrapper=1, retry_delay=0)
with pytest.raises(Exception) as exc_info:
handler_all_fail.try_call_all(lambda w: w.do_something())
assert "All wrappers failed" in str(exc_info.value)
def test_wrappers_with_parameters(self):
wrappers = [FailingWrapperWithParameters, MockWrapperWithParameters]
handler: WrapperHandler[MockWrapperWithParameters] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=2, retry_delay=0)
result = handler.try_call(lambda w: w.do_something("test", 42))
assert result == "Success test and 42"
assert handler.index == 1 # Should have switched to the second wrapper
assert handler.retry_count == 0
def test_wrappers_with_parameters_all_fail(self):
wrappers = [FailingWrapperWithParameters, FailingWrapperWithParameters]
handler: WrapperHandler[MockWrapperWithParameters] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0)
with pytest.raises(Exception) as exc_info:
handler.try_call(lambda w: w.do_something("test", 42))
assert "All wrappers failed" in str(exc_info.value)
def test_try_call_all_with_parameters(self):
wrappers = [FailingWrapperWithParameters, MockWrapperWithParameters]
handler: WrapperHandler[MockWrapperWithParameters] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0)
results = handler.try_call_all(lambda w: w.do_something("param", 99))
assert results == {MockWrapperWithParameters: "Success param and 99"}
def test_try_call_all_with_parameters_all_fail(self):
wrappers = [FailingWrapperWithParameters, FailingWrapperWithParameters]
handler: WrapperHandler[MockWrapperWithParameters] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0)
with pytest.raises(Exception) as exc_info:
handler.try_call_all(lambda w: w.do_something("param", 99))
assert "All wrappers failed" in str(exc_info.value)