tests/test_telegram_commands.py

#!/usr/bin/env python3
"""tests/test_telegram_commands.py — Test funzionale dei command handler Telegram.

Cosa testa
----------
Ogni command handler (19 comandi + 2 callback dispatcher) viene eseguito
davvero, ma tutte le chiamate HTTP verso api.telegram.org sono intercettate.

Nessun messaggio viene inviato su Telegram.
Nessuna env var richiesta.
Nessuna connessione di rete necessaria.

Come funziona
-------------
1. Le dipendenze esterne (fastapi, pydantic, httpx, api.*) vengono stubbate.
2. Le funzioni _tg_reply / _tg_send / _tg_edit / _tg_typing / _tg_react
   vengono rimpiazzate da AsyncMock PRIMA di ogni test, nel namespace del
   modulo importato (import_module caching garantisce coerenza).
3. Il test verifica che ogni handler abbia chiamato almeno una funzione TG.

Uso
---
  cd backend && python -m pytest tests/test_telegram_commands.py -v
  cd backend && python tests/test_telegram_commands.py         # standalone
"""
from __future__ import annotations
import asyncio, importlib, sys, types, unittest
from unittest.mock import AsyncMock, MagicMock, patch

# ─────────────────────────────────────────────────────────────────────────────
# 0. Stub dipendenze esterne (identico a test_telegram_imports.py)
# ─────────────────────────────────────────────────────────────────────────────
def _stub(name: str, **attrs) -> types.ModuleType:
    m = types.ModuleType(name)
    for k, v in attrs.items():
        setattr(m, k, v)
    sys.modules[name] = m
    return m

_fa = _stub("fastapi")
_fa.APIRouter     = type("APIRouter",     (), {
    "__init__": lambda self, **kw: None,
    "post":     lambda self, *a, **kw: (lambda f: f),
    "get":      lambda self, *a, **kw: (lambda f: f),
})
_fa.Request       = type("Request",       (), {})
_fa.HTTPException = type("HTTPException", (Exception,), {})
_fa.BackgroundTasks = type("BackgroundTasks", (), {})
_stub("fastapi.responses", JSONResponse=type("JSONResponse", (), {}))
_stub("fastapi.routing")
_stub("starlette.requests", Request=_fa.Request)
_stub("pydantic",           BaseModel=type("BaseModel", (), {}))

# httpx — il client viene moccato per intercettare tutte le chiamate TG
class _FakeResponse:
    status_code = 200
    def json(self): return {"ok": True, "result": {"message_id": 999}}
    async def aread(self): return b'{"ok":true,"result":{"message_id":999}}'

class _FakeAsyncClient:
    def __init__(self, *a, **kw): pass
    async def __aenter__(self): return self
    async def __aexit__(self, *a): pass
    async def post(self, url, **kw): return _FakeResponse()
    async def get(self,  url, **kw): return _FakeResponse()

_httpx = _stub("httpx", AsyncClient=_FakeAsyncClient, Response=_FakeResponse)
_stub("httpcore")
_stub("anyio")

# Backend interno — stub per evitare import circolari
for _m in [
    "api.agent", "api.providers", "api.state", "api.unified_api",
    "api.structured_log", "api.supabase_client", "api.exec",
    "agents.unified_loop_fallback", "agents.goal_verifier",
    "agents.memory_manager", "agents.audit_semantic_l2",
]:
    _stub(_m)

# ─────────────────────────────────────────────────────────────────────────────
# 1. Import moduli (dopo aver stubbato le dipendenze)
# ─────────────────────────────────────────────────────────────────────────────
def _imp(name: str):
    full = f"api.{name}"
    return sys.modules.get(full) or importlib.import_module(full)

# Import nell'ordine delle dipendenze
_tg_client = _imp("telegram_tg_client")
_keyboards  = _imp("telegram_keyboards")
_mon        = _imp("telegram_cmd_monitoring")
_ai         = _imp("telegram_cmd_ai")
_cb         = _imp("telegram_callbacks")

CHAT_ID = 123456789   # chat_id fittizio — non viene mai usato realmente

# ─────────────────────────────────────────────────────────────────────────────
# 2. Helper — patch di tutte le _tg_* nel modulo target
# ─────────────────────────────────────────────────────────────────────────────
TG_FUNS = ["_tg_reply", "_tg_send", "_tg_edit", "_tg_typing",
           "_tg_react", "_tg_photo", "_tg_answer_callback"]

class TGCapture:
    """Context manager: patcha le funzioni TG in 'mod', registra le chiamate."""
    def __init__(self, *mods):
        self.mods   = mods
        self.mocks  = {}   # "mod.fun" → AsyncMock
        self._patches = []

    def __enter__(self):
        for mod in self.mods:
            for fun in TG_FUNS:
                if hasattr(mod, fun):
                    m = AsyncMock(return_value=999)
                    p = patch.object(mod, fun, m)
                    p.start()
                    self._patches.append(p)
                    self.mocks[f"{mod.__name__}.{fun}"] = m
        return self

    def __exit__(self, *a):
        for p in self._patches:
            p.stop()

    def called_any(self) -> bool:
        return any(m.called for m in self.mocks.values())

    def calls(self) -> list[str]:
        return [k.split(".")[-1] for k,m in self.mocks.items() if m.called]

def run(coro):
    return asyncio.get_event_loop().run_until_complete(coro)

# ─────────────────────────────────────────────────────────────────────────────
# 3. Test suite
# ─────────────────────────────────────────────────────────────────────────────
class TestTelegramCommands(unittest.TestCase):

    # ── Monitoring ─────────────────────────────────────────────────────────
    def test_cmd_help(self):
        with TGCapture(_mon) as cap:
            run(_mon._cmd_help(CHAT_ID))
        self.assertTrue(cap.called_any(), f"_cmd_help non ha chiamato nulla. Mocks: {cap.calls()}")

    def test_cmd_logs(self):
        with TGCapture(_mon) as cap:
            run(_mon._cmd_logs(CHAT_ID))
        self.assertTrue(cap.called_any(), f"_cmd_logs: {cap.calls()}")

    def test_cmd_logs_level(self):
        """_cmd_logs accetta livello custom."""
        with TGCapture(_mon) as cap:
            run(_mon._cmd_logs(CHAT_ID, level="ERROR"))
        self.assertTrue(cap.called_any())

    def test_cmd_status(self):
        with TGCapture(_mon) as cap:
            run(_mon._cmd_status(CHAT_ID))
        self.assertTrue(cap.called_any(), f"_cmd_status: {cap.calls()}")

    def test_cmd_commit_summary(self):
        with TGCapture(_mon) as cap:
            run(_mon._cmd_commit_summary(CHAT_ID))
        self.assertTrue(cap.called_any(), f"_cmd_commit_summary: {cap.calls()}")

    def test_cmd_check(self):
        with TGCapture(_mon) as cap:
            run(_mon._cmd_check(CHAT_ID))
        self.assertTrue(cap.called_any(), f"_cmd_check: {cap.calls()}")

    def test_cmd_tasks(self):
        with TGCapture(_mon) as cap:
            run(_mon._cmd_tasks(CHAT_ID))
        self.assertTrue(cap.called_any(), f"_cmd_tasks: {cap.calls()}")

    # ── AI commands ────────────────────────────────────────────────────────
    def test_cmd_do_empty_goal(self):
        """_cmd_do con goal vuoto deve rispondere con la keyboard quick-pick."""
        with TGCapture(_ai) as cap:
            run(_ai._cmd_do(CHAT_ID, goal=""))
        self.assertTrue(cap.called_any(), f"_cmd_do(goal=''): {cap.calls()}")

    def test_cmd_do_with_goal(self):
        """_cmd_do con goal reale avvia lo streaming — la prima chiamata TG è _tg_send."""
        with TGCapture(_ai) as cap:
            run(_ai._cmd_do(CHAT_ID, goal="analizza i log"))
        self.assertTrue(cap.called_any(), f"_cmd_do(goal): {cap.calls()}")

    def test_cmd_do_saves_last_goal(self):
        """_cmd_do salva il goal in _LAST_GOAL per il tasto Rifai."""
        with TGCapture(_ai):
            run(_ai._cmd_do(CHAT_ID, goal="test retry goal"))
        from api.telegram_keyboards import _LAST_GOAL
        self.assertEqual(_LAST_GOAL.get(CHAT_ID), "test retry goal")

    def test_cmd_autofix(self):
        with TGCapture(_ai) as cap:
            run(_ai._cmd_autofix(CHAT_ID))
        self.assertTrue(cap.called_any(), f"_cmd_autofix: {cap.calls()}")

    def test_cmd_autofix_with_hint(self):
        with TGCapture(_ai) as cap:
            run(_ai._cmd_autofix(CHAT_ID, hint="TypeError in providers.py"))
        self.assertTrue(cap.called_any())

    def test_cmd_nota(self):
        with TGCapture(_ai) as cap:
            run(_ai._cmd_nota(CHAT_ID, text="ricordati di aggiornare il token"))
        self.assertTrue(cap.called_any(), f"_cmd_nota: {cap.calls()}")

    def test_cmd_cerca(self):
        with TGCapture(_ai) as cap:
            run(_ai._cmd_cerca(CHAT_ID, query="FastAPI async streaming"))
        self.assertTrue(cap.called_any(), f"_cmd_cerca: {cap.calls()}")

    def test_cmd_meteo(self):
        with TGCapture(_ai) as cap:
            run(_ai._cmd_meteo(CHAT_ID, city="Milano"))
        self.assertTrue(cap.called_any(), f"_cmd_meteo: {cap.calls()}")

    def test_cmd_riepilogo(self):
        with TGCapture(_ai) as cap:
            run(_ai._cmd_riepilogo(CHAT_ID))
        self.assertTrue(cap.called_any(), f"_cmd_riepilogo: {cap.calls()}")

    def test_cmd_scan_now(self):
        with TGCapture(_ai) as cap:
            run(_ai._cmd_scan_now(CHAT_ID))
        self.assertTrue(cap.called_any(), f"_cmd_scan_now: {cap.calls()}")

    def test_cmd_coord(self):
        with TGCapture(_ai) as cap:
            run(_ai._cmd_coord(CHAT_ID))
        self.assertTrue(cap.called_any(), f"_cmd_coord: {cap.calls()}")

    def test_cmd_git(self):
        with TGCapture(_ai) as cap:
            run(_ai._cmd_git(CHAT_ID))
        self.assertTrue(cap.called_any(), f"_cmd_git: {cap.calls()}")

    def test_cmd_git_custom_n(self):
        with TGCapture(_ai) as cap:
            run(_ai._cmd_git(CHAT_ID, n=10))
        self.assertTrue(cap.called_any())

    def test_cmd_telemetry(self):
        with TGCapture(_ai) as cap:
            run(_ai._cmd_telemetry(CHAT_ID))
        self.assertTrue(cap.called_any(), f"_cmd_telemetry: {cap.calls()}")

    def test_cmd_score(self):
        with TGCapture(_ai) as cap:
            run(_ai._cmd_score(CHAT_ID))
        self.assertTrue(cap.called_any(), f"_cmd_score: {cap.calls()}")

    def test_cmd_bench(self):
        with TGCapture(_ai) as cap:
            run(_ai._cmd_bench(CHAT_ID))
        self.assertTrue(cap.called_any(), f"_cmd_bench: {cap.calls()}")

    def test_cmd_improve(self):
        with TGCapture(_ai) as cap:
            run(_ai._cmd_improve(CHAT_ID))
        self.assertTrue(cap.called_any(), f"_cmd_improve: {cap.calls()}")

    # ── Callback dispatcher ────────────────────────────────────────────────
    def _make_cb(self, data: str) -> dict:
        return {
            "id": "cb_001",
            "from": {"id": CHAT_ID, "first_name": "Test"},
            "message": {"message_id": 42, "chat": {"id": CHAT_ID}},
            "chat_instance": "ci",
            "data": data,
        }

    def test_handle_callback_help(self):
        with TGCapture(_mon, _ai, _cb) as cap:
            run(_cb._handle_callback(self._make_cb("tgw_help"), token="fake"))
        self.assertTrue(cap.called_any(), "callback tgw_help: nessuna chiamata TG")

    def test_handle_callback_status(self):
        with TGCapture(_mon, _ai, _cb) as cap:
            run(_cb._handle_callback(self._make_cb("tgw_status"), token="fake"))
        self.assertTrue(cap.called_any(), "callback tgw_status")

    def test_handle_callback_tasks(self):
        with TGCapture(_mon, _ai, _cb) as cap:
            run(_cb._handle_callback(self._make_cb("tgw_tasks"), token="fake"))
        self.assertTrue(cap.called_any(), "callback tgw_tasks")

    def test_handle_callback_do(self):
        with TGCapture(_mon, _ai, _cb) as cap:
            run(_cb._handle_callback(self._make_cb("agent"), token="fake"))
        self.assertTrue(cap.called_any(), "callback agent (→ _cmd_do)")

    def test_handle_callback_autofix(self):
        with TGCapture(_mon, _ai, _cb) as cap:
            run(_cb._handle_callback(self._make_cb("tgw_autofix"), token="fake"))
        self.assertTrue(cap.called_any(), "callback tgw_autofix")

    def test_handle_callback_bench(self):
        with TGCapture(_mon, _ai, _cb) as cap:
            run(_cb._handle_callback(self._make_cb("tgw_bench"), token="fake"))
        self.assertTrue(cap.called_any(), "callback tgw_bench")

    def test_handle_callback_score(self):
        with TGCapture(_mon, _ai, _cb) as cap:
            run(_cb._handle_callback(self._make_cb("tgw_score"), token="fake"))
        self.assertTrue(cap.called_any(), "callback tgw_score")

    def test_handle_callback_improve(self):
        with TGCapture(_mon, _ai, _cb) as cap:
            run(_cb._handle_callback(self._make_cb("tgw_improve"), token="fake"))
        self.assertTrue(cap.called_any(), "callback tgw_improve")

    def test_handle_callback_retry(self):
        """tgw_retry rilegge _LAST_GOAL — deve funzionare anche se vuoto."""
        with TGCapture(_mon, _ai, _cb) as cap:
            run(_cb._handle_callback(self._make_cb("tgw_retry"), token="fake"))
        self.assertTrue(cap.called_any(), "callback tgw_retry")

    def test_handle_callback_qp_bug(self):
        """quick-pick qp_bug lancia _cmd_do con goal preimpostato."""
        with TGCapture(_mon, _ai, _cb) as cap:
            run(_cb._handle_callback(self._make_cb("qp_bug"), token="fake"))
        self.assertTrue(cap.called_any(), "callback qp_bug")

    def test_handle_inline_empty(self):
        """Inline query vuota → risponde con lista comandi."""
        iq = {"id": "iq_001", "from": {"id": CHAT_ID}, "query": ""}
        with TGCapture(_cb) as cap:
            run(_cb._handle_inline(iq, token="fake"))
        # La risposta va a answerInlineQuery — httpx è stubbato, non fallisce

    def test_handle_inline_with_query(self):
        iq = {"id": "iq_002", "from": {"id": CHAT_ID}, "query": "analizza bug login"}
        with TGCapture(_cb):
            run(_cb._handle_inline(iq, token="fake"))

    # ── Guardrail: nessun messaggio reale inviato ──────────────────────────
    def test_no_real_http_calls(self):
        """Verifica che httpx.AsyncClient.post non chiami api.telegram.org reale."""
        real_calls: list[str] = []
        original_post = _FakeAsyncClient.post
        async def spy_post(self, url, **kw):
            real_calls.append(url)
            return _FakeResponse()
        _FakeAsyncClient.post = spy_post
        try:
            with TGCapture(_mon):
                run(_mon._cmd_help(CHAT_ID))
            # Se arriva qui senza errori di rete → ok (le _tg_* erano moccate,
            # quindi httpx.post non è stato chiamato direttamente)
            for url in real_calls:
                self.assertNotIn("api.telegram.org", url,
                    f"Chiamata reale a api.telegram.org intercettata: {url}")
        finally:
            _FakeAsyncClient.post = original_post


if __name__ == "__main__":
    import pathlib
    backend = pathlib.Path(__file__).parent.parent
    if str(backend) not in sys.path:
        sys.path.insert(0, str(backend))
    runner = unittest.TextTestRunner(verbosity=2)
    result = runner.run(unittest.TestLoader().loadTestsFromTestCase(TestTelegramCommands))
    sys.exit(0 if result.wasSuccessful() else 1)