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

fix type in tests

Browse Source
This commit is contained in:
Giacomo Bertolazzi
2025-10-05 12:44:30 +02:00
parent f7a0660d4c
commit a31acb3a0b
7 changed files with 71 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
tests/agents/test_predictor.py
View File

@@ -3,9 +3,9 @@ from app.agents import AppModels
from app.agents.predictor import PREDICTOR_INSTRUCTIONS, PredictorInput, PredictorOutput, PredictorStyle from app.agents.predictor import PREDICTOR_INSTRUCTIONS, PredictorInput, PredictorOutput, PredictorStyle
from app.base.markets import ProductInfo from app.base.markets import ProductInfo
def unified_checks(model: AppModels, input): def unified_checks(model: AppModels, input: PredictorInput) -> None:
llm = model.get_agent(PREDICTOR_INSTRUCTIONS, output=PredictorOutput) # type: ignore[arg-type] llm = model.get_agent(PREDICTOR_INSTRUCTIONS, output_schema=PredictorOutput) # type: ignore[arg-type]
result = llm.run(input) result = llm.run(input) # type: ignore
content = result.content content = result.content
assert isinstance(content, PredictorOutput) assert isinstance(content, PredictorOutput)
@@ -27,9 +27,8 @@ def unified_checks(model: AppModels, input):
class TestPredictor: class TestPredictor:
@pytest.fixture(scope="class") def inputs(self) -> PredictorInput:
def inputs(self): data: list[ProductInfo] = []
data = []
for symbol, price in [("BTC", 60000.00), ("ETH", 3500.00), ("SOL", 150.00)]: for symbol, price in [("BTC", 60000.00), ("ETH", 3500.00), ("SOL", 150.00)]:
product_info = ProductInfo() product_info = ProductInfo()
product_info.symbol = symbol product_info.symbol = symbol
@@ -38,13 +37,24 @@ class TestPredictor:
return PredictorInput(data=data, style=PredictorStyle.AGGRESSIVE, sentiment="positivo") return PredictorInput(data=data, style=PredictorStyle.AGGRESSIVE, sentiment="positivo")
def test_gemini_model_output(self, inputs): def test_gemini_model_output(self):
inputs = self.inputs()
unified_checks(AppModels.GEMINI, inputs) unified_checks(AppModels.GEMINI, inputs)
def test_ollama_qwen_1b_model_output(self):
inputs = self.inputs()
unified_checks(AppModels.OLLAMA_QWEN_1B, inputs)
def test_ollama_qwen_4b_model_output(self):
inputs = self.inputs()
unified_checks(AppModels.OLLAMA_QWEN_4B, inputs)
@pytest.mark.slow @pytest.mark.slow
def test_ollama_qwen_model_output(self, inputs): def test_ollama_qwen_latest_model_output(self):
inputs = self.inputs()
unified_checks(AppModels.OLLAMA_QWEN, inputs) unified_checks(AppModels.OLLAMA_QWEN, inputs)
@pytest.mark.slow @pytest.mark.slow
def test_ollama_gpt_oss_model_output(self, inputs): def test_ollama_gpt_oss_model_output(self):
inputs = self.inputs()
unified_checks(AppModels.OLLAMA_GPT, inputs) unified_checks(AppModels.OLLAMA_GPT, inputs)

3
tests/api/test_reddit.py
View File

@@ -1,6 +1,5 @@
import os import os
import pytest import pytest
from praw import Reddit
from app.social.reddit import MAX_COMMENTS, RedditWrapper from app.social.reddit import MAX_COMMENTS, RedditWrapper
@pytest.mark.social @pytest.mark.social
@@ -10,7 +9,7 @@ class TestRedditWrapper:
def test_initialization(self): def test_initialization(self):
wrapper = RedditWrapper() wrapper = RedditWrapper()
assert wrapper is not None assert wrapper is not None
assert isinstance(wrapper.tool, Reddit) assert wrapper.tool is not None
def test_get_top_crypto_posts(self): def test_get_top_crypto_posts(self):
wrapper = RedditWrapper() wrapper = RedditWrapper()

2
tests/conftest.py
View File

@@ -33,7 +33,7 @@ def pytest_configure(config:pytest.Config):
line = f"{marker[0]}: {marker[1]}" line = f"{marker[0]}: {marker[1]}"
config.addinivalue_line("markers", line) config.addinivalue_line("markers", line)
def pytest_collection_modifyitems(config, items): def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None:
"""Modifica automaticamente degli item di test rimovendoli""" """Modifica automaticamente degli item di test rimovendoli"""
# Rimuovo i test "limited" e "slow" se non richiesti esplicitamente # Rimuovo i test "limited" e "slow" se non richiesti esplicitamente
mark_to_remove = ['limited', 'slow'] mark_to_remove = ['limited', 'slow']

12
tests/tools/test_market_tool.py
View File

@@ -7,15 +7,15 @@ from app.markets import MarketAPIsTool
@pytest.mark.api @pytest.mark.api
class TestMarketAPIsTool: class TestMarketAPIsTool:
def test_wrapper_initialization(self): def test_wrapper_initialization(self):
market_wrapper = MarketAPIsTool("USD") market_wrapper = MarketAPIsTool("EUR")
assert market_wrapper is not None assert market_wrapper is not None
assert hasattr(market_wrapper, 'get_product') assert hasattr(market_wrapper, 'get_product')
assert hasattr(market_wrapper, 'get_products') assert hasattr(market_wrapper, 'get_products')
assert hasattr(market_wrapper, 'get_historical_prices') assert hasattr(market_wrapper, 'get_historical_prices')
def test_wrapper_capabilities(self): def test_wrapper_capabilities(self):
market_wrapper = MarketAPIsTool("USD") market_wrapper = MarketAPIsTool("EUR")
capabilities = [] capabilities: list[str] = []
if hasattr(market_wrapper, 'get_product'): if hasattr(market_wrapper, 'get_product'):
capabilities.append('single_product') capabilities.append('single_product')
if hasattr(market_wrapper, 'get_products'): if hasattr(market_wrapper, 'get_products'):
@@ -25,7 +25,7 @@ class TestMarketAPIsTool:
assert len(capabilities) > 0 assert len(capabilities) > 0
def test_market_data_retrieval(self): def test_market_data_retrieval(self):
market_wrapper = MarketAPIsTool("USD") market_wrapper = MarketAPIsTool("EUR")
btc_product = market_wrapper.get_product("BTC") btc_product = market_wrapper.get_product("BTC")
assert btc_product is not None assert btc_product is not None
assert hasattr(btc_product, 'symbol') assert hasattr(btc_product, 'symbol')
@@ -34,8 +34,8 @@ class TestMarketAPIsTool:
def test_error_handling(self): def test_error_handling(self):
try: try:
market_wrapper = MarketAPIsTool("USD") market_wrapper = MarketAPIsTool("EUR")
fake_product = market_wrapper.get_product("NONEXISTENT_CRYPTO_SYMBOL_12345") fake_product = market_wrapper.get_product("NONEXISTENT_CRYPTO_SYMBOL_12345")
assert fake_product is None or fake_product.price == 0 assert fake_product is None or fake_product.price == 0
except Exception as e: except Exception as _:
pass pass

4
tests/tools/test_news_tool.py
View File

@@ -33,7 +33,7 @@ class TestNewsAPITool:
result = tool.wrapper_handler.try_call_all(lambda w: w.get_top_headlines(limit=2)) result = tool.wrapper_handler.try_call_all(lambda w: w.get_top_headlines(limit=2))
assert isinstance(result, dict) assert isinstance(result, dict)
assert len(result.keys()) > 0 assert len(result.keys()) > 0
for provider, articles in result.items(): for _provider, articles in result.items():
for article in articles: for article in articles:
assert article.title is not None assert article.title is not None
assert article.source is not None assert article.source is not None
@@ -43,7 +43,7 @@ class TestNewsAPITool:
result = tool.wrapper_handler.try_call_all(lambda w: w.get_latest_news(query="crypto", limit=2)) result = tool.wrapper_handler.try_call_all(lambda w: w.get_latest_news(query="crypto", limit=2))
assert isinstance(result, dict) assert isinstance(result, dict)
assert len(result.keys()) > 0 assert len(result.keys()) > 0
for provider, articles in result.items(): for _provider, articles in result.items():
for article in articles: for article in articles:
assert article.title is not None assert article.title is not None
assert article.source is not None assert article.source is not None

28
tests/utils/test_market_aggregator.py
View File

@@ -41,8 +41,8 @@ class TestMarketDataAggregator:
assert info.symbol == "BTC" 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) 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.price == pytest.approx(avg_weighted_price, rel=1e-3) # type: ignore
assert info.volume_24h == pytest.approx(1000.0, rel=1e-3) assert info.volume_24h == pytest.approx(1000.0, rel=1e-3) # type: ignore
assert info.quote_currency == "USD" assert info.quote_currency == "USD"
def test_aggregate_product_info_multiple_symbols(self): def test_aggregate_product_info_multiple_symbols(self):
@@ -65,18 +65,18 @@ class TestMarketDataAggregator:
assert btc_info is not None assert btc_info is not None
avg_weighted_price_btc = (50000.0 * 1000.0 + 50100.0 * 1100.0) / (1000.0 + 1100.0) 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.price == pytest.approx(avg_weighted_price_btc, rel=1e-3) # type: ignore
assert btc_info.volume_24h == pytest.approx(1050.0, rel=1e-3) assert btc_info.volume_24h == pytest.approx(1050.0, rel=1e-3) # type: ignore
assert btc_info.quote_currency == "USD" assert btc_info.quote_currency == "USD"
assert eth_info is not None assert eth_info is not None
avg_weighted_price_eth = (4000.0 * 2000.0 + 4050.0 * 2100.0) / (2000.0 + 2100.0) 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.price == pytest.approx(avg_weighted_price_eth, rel=1e-3) # type: ignore
assert eth_info.volume_24h == pytest.approx(2050.0, rel=1e-3) assert eth_info.volume_24h == pytest.approx(2050.0, rel=1e-3) # type: ignore
assert eth_info.quote_currency == "USD" assert eth_info.quote_currency == "USD"
def test_aggregate_product_info_with_no_data(self): def test_aggregate_product_info_with_no_data(self):
products = { products: dict[str, list[ProductInfo]] = {
"Provider1": [], "Provider1": [],
"Provider2": [], "Provider2": [],
} }
@@ -84,7 +84,7 @@ class TestMarketDataAggregator:
assert len(aggregated) == 0 assert len(aggregated) == 0
def test_aggregate_product_info_with_partial_data(self): def test_aggregate_product_info_with_partial_data(self):
products = { products: dict[str, list[ProductInfo]] = {
"Provider1": [self.__product("BTC", 50000.0, 1000.0, "USD")], "Provider1": [self.__product("BTC", 50000.0, 1000.0, "USD")],
"Provider2": [], "Provider2": [],
} }
@@ -92,8 +92,8 @@ class TestMarketDataAggregator:
assert len(aggregated) == 1 assert len(aggregated) == 1
info = aggregated[0] info = aggregated[0]
assert info.symbol == "BTC" assert info.symbol == "BTC"
assert info.price == pytest.approx(50000.0, rel=1e-3) assert info.price == pytest.approx(50000.0, rel=1e-3) # type: ignore
assert info.volume_24h == pytest.approx(1000.0, rel=1e-3) assert info.volume_24h == pytest.approx(1000.0, rel=1e-3) # type: ignore
assert info.quote_currency == "USD" assert info.quote_currency == "USD"
def test_aggregate_history_prices(self): def test_aggregate_history_prices(self):
@@ -113,8 +113,8 @@ class TestMarketDataAggregator:
aggregated = aggregate_history_prices(prices) aggregated = aggregate_history_prices(prices)
assert len(aggregated) == 2 assert len(aggregated) == 2
assert aggregated[0].timestamp_ms == 1685577600000 assert aggregated[0].timestamp_ms == 1685577600000
assert aggregated[0].high == pytest.approx(50050.0, rel=1e-3) assert aggregated[0].high == pytest.approx(50050.0, rel=1e-3) # type: ignore
assert aggregated[0].low == pytest.approx(49550.0, rel=1e-3) assert aggregated[0].low == pytest.approx(49550.0, rel=1e-3) # type: ignore
assert aggregated[1].timestamp_ms == 1685581200000 assert aggregated[1].timestamp_ms == 1685581200000
assert aggregated[1].high == pytest.approx(50250.0, rel=1e-3) assert aggregated[1].high == pytest.approx(50250.0, rel=1e-3) # type: ignore
assert aggregated[1].low == pytest.approx(49850.0, rel=1e-3) assert aggregated[1].low == pytest.approx(49850.0, rel=1e-3) # type: ignore

56
tests/utils/test_wrapper_handler.py
View File

@@ -37,7 +37,7 @@ class TestWrapperHandler:
def test_init_failing_with_instances(self): def test_init_failing_with_instances(self):
with pytest.raises(AssertionError) as exc_info: with pytest.raises(AssertionError) as exc_info:
WrapperHandler.build_wrappers([MockWrapper(), MockWrapper2()]) WrapperHandler.build_wrappers([MockWrapper(), MockWrapper2()]) # type: ignore
assert exc_info.type == AssertionError assert exc_info.type == AssertionError
def test_init_not_failing(self): def test_init_not_failing(self):
@@ -49,16 +49,16 @@ class TestWrapperHandler:
assert len(handler.wrappers) == 2 assert len(handler.wrappers) == 2
def test_all_wrappers_fail(self): def test_all_wrappers_fail(self):
wrappers = [FailingWrapper, FailingWrapper] wrappers: list[type[MockWrapper]] = [FailingWrapper, FailingWrapper]
handler: WrapperHandler[MockWrapper] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=2, retry_delay=0) handler = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=2, retry_delay=0)
with pytest.raises(Exception) as exc_info: with pytest.raises(Exception) as exc_info:
handler.try_call(lambda w: w.do_something()) handler.try_call(lambda w: w.do_something())
assert "All wrappers failed" in str(exc_info.value) assert "All wrappers failed" in str(exc_info.value)
def test_success_on_first_try(self): def test_success_on_first_try(self):
wrappers = [MockWrapper, FailingWrapper] wrappers: list[type[MockWrapper]] = [MockWrapper, FailingWrapper]
handler: WrapperHandler[MockWrapper] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=2, retry_delay=0) handler = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=2, retry_delay=0)
result = handler.try_call(lambda w: w.do_something()) result = handler.try_call(lambda w: w.do_something())
assert result == "Success" assert result == "Success"
@@ -66,8 +66,8 @@ class TestWrapperHandler:
assert handler.retry_count == 0 assert handler.retry_count == 0
def test_eventual_success(self): def test_eventual_success(self):
wrappers = [FailingWrapper, MockWrapper] wrappers: list[type[MockWrapper]] = [FailingWrapper, MockWrapper]
handler: WrapperHandler[MockWrapper] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=2, retry_delay=0) handler = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=2, retry_delay=0)
result = handler.try_call(lambda w: w.do_something()) result = handler.try_call(lambda w: w.do_something())
assert result == "Success" assert result == "Success"
@@ -75,8 +75,8 @@ class TestWrapperHandler:
assert handler.retry_count == 0 assert handler.retry_count == 0
def test_partial_failures(self): def test_partial_failures(self):
wrappers = [FailingWrapper, MockWrapper, FailingWrapper] wrappers: list[type[MockWrapper]] = [FailingWrapper, MockWrapper, FailingWrapper]
handler: WrapperHandler[MockWrapper] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0) handler = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0)
result = handler.try_call(lambda w: w.do_something()) result = handler.try_call(lambda w: w.do_something())
assert result == "Success" assert result == "Success"
@@ -96,34 +96,34 @@ class TestWrapperHandler:
assert handler.retry_count == 0 assert handler.retry_count == 0
def test_try_call_all_success(self): def test_try_call_all_success(self):
wrappers = [MockWrapper, MockWrapper2] wrappers: list[type[MockWrapper]] = [MockWrapper, MockWrapper2]
handler: WrapperHandler[MockWrapper] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0) handler = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0)
results = handler.try_call_all(lambda w: w.do_something()) results = handler.try_call_all(lambda w: w.do_something())
assert results == {MockWrapper: "Success", MockWrapper2: "Success 2"} assert results == {MockWrapper.__name__: "Success", MockWrapper2.__name__: "Success 2"}
def test_try_call_all_partial_failures(self): def test_try_call_all_partial_failures(self):
# Only the second wrapper should succeed # Only the second wrapper should succeed
wrappers = [FailingWrapper, MockWrapper, FailingWrapper] wrappers: list[type[MockWrapper]] = [FailingWrapper, MockWrapper, FailingWrapper]
handler: WrapperHandler[MockWrapper] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0) handler = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0)
results = handler.try_call_all(lambda w: w.do_something()) results = handler.try_call_all(lambda w: w.do_something())
assert results == {MockWrapper: "Success"} assert results == {MockWrapper.__name__: "Success"}
# Only the second and fourth wrappers should succeed # Only the second and fourth wrappers should succeed
wrappers = [FailingWrapper, MockWrapper, FailingWrapper, MockWrapper2] wrappers: list[type[MockWrapper]] = [FailingWrapper, MockWrapper, FailingWrapper, MockWrapper2]
handler: WrapperHandler[MockWrapper] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0) handler = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0)
results = handler.try_call_all(lambda w: w.do_something()) results = handler.try_call_all(lambda w: w.do_something())
assert results == {MockWrapper: "Success", MockWrapper2: "Success 2"} assert results == {MockWrapper.__name__: "Success", MockWrapper2.__name__: "Success 2"}
def test_try_call_all_all_fail(self): def test_try_call_all_all_fail(self):
# Test when all wrappers fail # Test when all wrappers fail
handler_all_fail: WrapperHandler[MockWrapper] = WrapperHandler.build_wrappers([FailingWrapper, FailingWrapper], try_per_wrapper=1, retry_delay=0) handler_all_fail = WrapperHandler.build_wrappers([FailingWrapper, FailingWrapper], try_per_wrapper=1, retry_delay=0)
with pytest.raises(Exception) as exc_info: with pytest.raises(Exception) as exc_info:
handler_all_fail.try_call_all(lambda w: w.do_something()) handler_all_fail.try_call_all(lambda w: w.do_something())
assert "All wrappers failed" in str(exc_info.value) assert "All wrappers failed" in str(exc_info.value)
def test_wrappers_with_parameters(self): def test_wrappers_with_parameters(self):
wrappers = [FailingWrapperWithParameters, MockWrapperWithParameters] wrappers: list[type[MockWrapperWithParameters]] = [FailingWrapperWithParameters, MockWrapperWithParameters]
handler: WrapperHandler[MockWrapperWithParameters] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=2, retry_delay=0) handler = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=2, retry_delay=0)
result = handler.try_call(lambda w: w.do_something("test", 42)) result = handler.try_call(lambda w: w.do_something("test", 42))
assert result == "Success test and 42" assert result == "Success test and 42"
@@ -131,22 +131,22 @@ class TestWrapperHandler:
assert handler.retry_count == 0 assert handler.retry_count == 0
def test_wrappers_with_parameters_all_fail(self): def test_wrappers_with_parameters_all_fail(self):
wrappers = [FailingWrapperWithParameters, FailingWrapperWithParameters] wrappers: list[type[MockWrapperWithParameters]] = [FailingWrapperWithParameters, FailingWrapperWithParameters]
handler: WrapperHandler[MockWrapperWithParameters] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0) handler = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0)
with pytest.raises(Exception) as exc_info: with pytest.raises(Exception) as exc_info:
handler.try_call(lambda w: w.do_something("test", 42)) handler.try_call(lambda w: w.do_something("test", 42))
assert "All wrappers failed" in str(exc_info.value) assert "All wrappers failed" in str(exc_info.value)
def test_try_call_all_with_parameters(self): def test_try_call_all_with_parameters(self):
wrappers = [FailingWrapperWithParameters, MockWrapperWithParameters] wrappers: list[type[MockWrapperWithParameters]] = [FailingWrapperWithParameters, MockWrapperWithParameters]
handler: WrapperHandler[MockWrapperWithParameters] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0) handler = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0)
results = handler.try_call_all(lambda w: w.do_something("param", 99)) results = handler.try_call_all(lambda w: w.do_something("param", 99))
assert results == {MockWrapperWithParameters: "Success param and 99"} assert results == {MockWrapperWithParameters.__name__: "Success param and 99"}
def test_try_call_all_with_parameters_all_fail(self): def test_try_call_all_with_parameters_all_fail(self):
wrappers = [FailingWrapperWithParameters, FailingWrapperWithParameters] wrappers: list[type[MockWrapperWithParameters]] = [FailingWrapperWithParameters, FailingWrapperWithParameters]
handler: WrapperHandler[MockWrapperWithParameters] = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0) handler = WrapperHandler.build_wrappers(wrappers, try_per_wrapper=1, retry_delay=0)
with pytest.raises(Exception) as exc_info: with pytest.raises(Exception) as exc_info:
handler.try_call_all(lambda w: w.do_something("param", 99)) handler.try_call_all(lambda w: w.do_something("param", 99))
assert "All wrappers failed" in str(exc_info.value) assert "All wrappers failed" in str(exc_info.value)