252 lines
8.7 KiB
Python
252 lines
8.7 KiB
Python
"""Integration tests: model surface, ai_log_job translation fan-out,
|
|
route-level localized fetch, settings PATCH validation."""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
|
|
def _build_session_factory(tmp_path):
|
|
"""Per-test sqlite engine + factory. Mirrors test_referral_conversion.py."""
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app import db as db_mod
|
|
from app.db import Base
|
|
import app.models # noqa: F401 — registers models on Base.metadata
|
|
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/loc.db")
|
|
factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
db_mod._engine = engine
|
|
db_mod._session_factory = factory
|
|
|
|
async def _setup():
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
return engine, factory, _setup
|
|
|
|
|
|
def test_user_has_lang_column_with_default_en():
|
|
from sqlalchemy import inspect
|
|
from app.models import User
|
|
|
|
cols = {c.name: c for c in inspect(User).columns}
|
|
assert "lang" in cols
|
|
assert cols["lang"].nullable is False
|
|
# SQLAlchemy default may be a callable or a literal — check both.
|
|
default = cols["lang"].default
|
|
assert default is not None
|
|
if hasattr(default, "arg"):
|
|
assert default.arg == "en"
|
|
|
|
|
|
def test_strategic_log_translation_model_columns():
|
|
from sqlalchemy import inspect
|
|
from app.models import StrategicLogTranslation
|
|
|
|
cols = {c.name: c for c in inspect(StrategicLogTranslation).columns}
|
|
assert "log_id" in cols
|
|
assert "lang" in cols
|
|
assert "content_md" in cols
|
|
assert "generated_at" in cols
|
|
assert "llm_model" in cols
|
|
assert "llm_cost_usd" in cols
|
|
assert cols["log_id"].nullable is False
|
|
assert cols["lang"].nullable is False
|
|
assert cols["content_md"].nullable is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_log_translation_fanout_no_active_non_en_users(tmp_path, monkeypatch):
|
|
"""When no users have an active non-en lang, the fan-out makes no
|
|
translation calls and no rows are inserted."""
|
|
from unittest.mock import AsyncMock
|
|
from sqlalchemy import select
|
|
|
|
from app.db import utcnow
|
|
from app.models import StrategicLog, StrategicLogTranslation, User
|
|
from app.jobs import ai_log_job
|
|
|
|
_, factory, setup = _build_session_factory(tmp_path)
|
|
await setup()
|
|
|
|
fake_translate = AsyncMock()
|
|
monkeypatch.setattr(ai_log_job, "translate", fake_translate)
|
|
|
|
# Seed an English user (no non-en users).
|
|
async with factory() as session:
|
|
session.add(User(id=1, email="en@x", tier="paid", lang="en"))
|
|
slog = StrategicLog(
|
|
generated_at=utcnow(), content="# Open\n\nDown 0.4%.",
|
|
model="test-model",
|
|
tone="INTERMEDIATE", analysis="NORMAL",
|
|
)
|
|
session.add(slog)
|
|
await session.commit()
|
|
log_id = slog.id
|
|
|
|
async with factory() as session:
|
|
await ai_log_job.translate_log_for_active_languages(session, log_id)
|
|
|
|
fake_translate.assert_not_awaited()
|
|
async with factory() as session:
|
|
rows = (await session.execute(select(StrategicLogTranslation))).scalars().all()
|
|
assert rows == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_log_translation_fanout_italian_user(tmp_path, monkeypatch):
|
|
"""One user at lang=it triggers one translation; the row lands with
|
|
the right lang and log_id."""
|
|
from sqlalchemy import select
|
|
|
|
from app.db import utcnow
|
|
from app.models import StrategicLog, StrategicLogTranslation, User
|
|
from app.services.openrouter import LogResult
|
|
from app.jobs import ai_log_job
|
|
|
|
_, factory, setup = _build_session_factory(tmp_path)
|
|
await setup()
|
|
|
|
async def _fake_translate(client, text, target_lang):
|
|
assert target_lang == "it"
|
|
return "# Apertura\n\nIn calo 0,4%.", LogResult(
|
|
content="# Apertura\n\nIn calo 0,4%.",
|
|
model="deepseek/deepseek-v4-flash",
|
|
prompt_tokens=300, completion_tokens=80, cost_usd=0.00002,
|
|
)
|
|
monkeypatch.setattr(ai_log_job, "translate", _fake_translate)
|
|
|
|
async with factory() as session:
|
|
session.add(User(id=2, email="it@x", tier="paid", lang="it"))
|
|
slog = StrategicLog(
|
|
generated_at=utcnow(), content="# Open\n\nDown 0.4%.",
|
|
model="test-model",
|
|
tone="INTERMEDIATE", analysis="NORMAL",
|
|
)
|
|
session.add(slog)
|
|
await session.commit()
|
|
log_id = slog.id
|
|
|
|
async with factory() as session:
|
|
await ai_log_job.translate_log_for_active_languages(session, log_id)
|
|
|
|
async with factory() as session:
|
|
rows = (await session.execute(select(StrategicLogTranslation))).scalars().all()
|
|
assert len(rows) == 1
|
|
row = rows[0]
|
|
assert row.log_id == log_id
|
|
assert row.lang == "it"
|
|
assert row.content_md.startswith("# Apertura")
|
|
assert row.llm_model == "deepseek/deepseek-v4-flash"
|
|
assert row.llm_cost_usd == pytest.approx(0.00002)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_log_translation_fanout_per_language_failure_isolated(tmp_path, monkeypatch):
|
|
"""If one language's translation fails, the others (if any) still land
|
|
and the job does not raise."""
|
|
from sqlalchemy import select
|
|
|
|
from app.db import utcnow
|
|
from app.models import StrategicLog, StrategicLogTranslation, User
|
|
from app.jobs import ai_log_job
|
|
|
|
_, factory, setup = _build_session_factory(tmp_path)
|
|
await setup()
|
|
|
|
async def _fake_translate(client, text, target_lang):
|
|
raise RuntimeError("upstream down")
|
|
monkeypatch.setattr(ai_log_job, "translate", _fake_translate)
|
|
|
|
async with factory() as session:
|
|
session.add(User(id=3, email="it@x", tier="paid", lang="it"))
|
|
slog = StrategicLog(
|
|
generated_at=utcnow(), content="# Open",
|
|
model="test-model",
|
|
tone="INTERMEDIATE", analysis="NORMAL",
|
|
)
|
|
session.add(slog)
|
|
await session.commit()
|
|
log_id = slog.id
|
|
|
|
# Must NOT raise.
|
|
async with factory() as session:
|
|
await ai_log_job.translate_log_for_active_languages(session, log_id)
|
|
|
|
async with factory() as session:
|
|
rows = (await session.execute(select(StrategicLogTranslation))).scalars().all()
|
|
assert rows == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_analyse_threads_lang_into_system_prompt(tmp_path, monkeypatch):
|
|
"""When lang='it', the system prompt sent to call_llm contains
|
|
'Respond in Italian.' — the LLM does the rest."""
|
|
from app.services import portfolio_analysis as pa
|
|
from app.services.openrouter import LogResult
|
|
|
|
captured = {}
|
|
|
|
async def _fake_call_llm(client, messages, **kw):
|
|
captured["messages"] = messages
|
|
return LogResult(
|
|
content="Analisi del portafoglio in italiano.",
|
|
model="m", prompt_tokens=400, completion_tokens=100, cost_usd=0.0001,
|
|
)
|
|
monkeypatch.setattr(pa, "call_llm", _fake_call_llm)
|
|
|
|
_, factory, setup = _build_session_factory(tmp_path)
|
|
await setup()
|
|
|
|
payload = {
|
|
"positions": [{"yahoo_ticker": "AAPL", "qty": 10, "avg_cost": 150.0,
|
|
"currency": "USD", "name": "Apple Inc"}],
|
|
"prices": {"AAPL": {"p": 172.4, "c": "USD"}},
|
|
"fx": {"USD": 1.0},
|
|
"base_currency": "USD",
|
|
"tone": "INTERMEDIATE",
|
|
"analysis": "NORMAL",
|
|
"lang": "it",
|
|
}
|
|
req = pa.parse_request(payload)
|
|
assert req.lang == "it"
|
|
|
|
async with factory() as session:
|
|
await pa.analyse(session, req)
|
|
system = next(m["content"] for m in captured["messages"] if m["role"] == "system")
|
|
assert "Respond in Italian" in system
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_analyse_no_clause_when_lang_is_en(tmp_path, monkeypatch):
|
|
from app.services import portfolio_analysis as pa
|
|
from app.services.openrouter import LogResult
|
|
|
|
captured = {}
|
|
|
|
async def _fake_call_llm(client, messages, **kw):
|
|
captured["messages"] = messages
|
|
return LogResult(
|
|
content="Portfolio analysis in English.",
|
|
model="m", prompt_tokens=400, completion_tokens=100, cost_usd=0.0001,
|
|
)
|
|
monkeypatch.setattr(pa, "call_llm", _fake_call_llm)
|
|
|
|
_, factory, setup = _build_session_factory(tmp_path)
|
|
await setup()
|
|
|
|
payload = {
|
|
"positions": [{"yahoo_ticker": "AAPL", "qty": 10, "avg_cost": 150.0,
|
|
"currency": "USD", "name": "Apple Inc"}],
|
|
"prices": {"AAPL": {"p": 172.4, "c": "USD"}},
|
|
"fx": {"USD": 1.0},
|
|
"base_currency": "USD",
|
|
"tone": "INTERMEDIATE",
|
|
"analysis": "NORMAL",
|
|
"lang": "en",
|
|
}
|
|
req = pa.parse_request(payload)
|
|
async with factory() as session:
|
|
await pa.analyse(session, req)
|
|
system = next(m["content"] for m in captured["messages"] if m["role"] == "system")
|
|
assert "Respond in" not in system
|