""" test_regression_doc2.py — Regression tests per i tre bug Doc2 Copre: Doc2-1a: cache speculativa mai consultata in _run_direct_tools Doc2-1b: memory/sync router non montato in main.py Doc2-1c: SRO session ricreata ad ogni retry (coperto lato TS) Dipendenze: solo stdlib + moduli backend già presenti. Non richiede server avviato: test unitari puri. """ from __future__ import annotations import sys import os import asyncio import types import unittest from unittest.mock import patch, MagicMock, AsyncMock # Assicura che il path del backend sia nel sys.path _BACKEND = os.path.join(os.path.dirname(__file__), "..") if _BACKEND not in sys.path: sys.path.insert(0, _BACKEND) # ═══════════════════════════════════════════════════════════════════════════════ # Doc2-1a: get_speculative_result() deve essere chiamata prima di ogni tool call # ═══════════════════════════════════════════════════════════════════════════════ class TestSpeculativeCacheConsumption(unittest.TestCase): """ Verifica che _run_direct_tools chiami get_speculative_result prima di eseguire la chiamata di rete reale. Bug originale: la cache veniva riempita ma mai letta. """ def _get_mixin_instance(self): """Crea un'istanza minimale di DirectToolsMixin con stub sufficiente.""" try: from agents.unified_loop_tools import DirectToolsMixin except ImportError as e: self.skipTest(f"DirectToolsMixin non importabile: {e}") class _Stub(DirectToolsMixin): pass return _Stub() def test_spec_hit_helper_defined_in_run_direct_tools(self): """ Doc2-1a: _spec_hit deve esistere come closure in _run_direct_tools. Verifica: la funzione è definita e chiamabile (source inspection). """ import inspect try: from agents.unified_loop_tools import DirectToolsMixin except ImportError as e: self.skipTest(f"Impossibile importare DirectToolsMixin: {e}") src = inspect.getsource(DirectToolsMixin._run_direct_tools) self.assertIn("_spec_hit", src, "_spec_hit helper non trovato in _run_direct_tools — cache speculativa non consultata") def test_spec_hit_called_before_get_weather(self): """ Doc2-1a: per get_weather, _spec_hit deve essere chiamata prima di asyncio.wait_for(TOOL_REGISTRY['get_weather']...). """ import inspect try: from agents.unified_loop_tools import DirectToolsMixin except ImportError as e: self.skipTest(f"Impossibile importare DirectToolsMixin: {e}") src = inspect.getsource(DirectToolsMixin._run_direct_tools) # Trova posizione di _spec_hit("get_weather" e di wait_for...get_weather spec_pos = src.find('_spec_hit("get_weather"') wait_pos = src.find('TOOL_REGISTRY["get_weather"]') self.assertGreater(spec_pos, 0, "_spec_hit('get_weather') non trovato nel sorgente") self.assertGreater(wait_pos, 0, "TOOL_REGISTRY['get_weather'] non trovato nel sorgente") self.assertLess(spec_pos, wait_pos, "_spec_hit('get_weather') deve precedere TOOL_REGISTRY call — ordine sbagliato") def test_spec_hit_called_before_web_search(self): """Doc2-1a: _spec_hit deve precedere TOOL_REGISTRY['web_search'].""" import inspect try: from agents.unified_loop_tools import DirectToolsMixin except ImportError as e: self.skipTest(f"Impossibile importare DirectToolsMixin: {e}") src = inspect.getsource(DirectToolsMixin._run_direct_tools) spec_pos = src.find('_spec_hit("web_search"') wait_pos = src.find('TOOL_REGISTRY["web_search"]') self.assertGreater(spec_pos, 0, "_spec_hit('web_search') non trovato nel sorgente") self.assertLess(spec_pos, wait_pos, "_spec_hit('web_search') deve precedere TOOL_REGISTRY call") def test_spec_hit_returns_cached_value_sync(self): """ Doc2-1a: get_speculative_result restituisce il valore dalla cache. Test diretto sulla funzione (non su _run_direct_tools). """ try: from api.speculative import get_speculative_result, _spec_cache, _goal_hash except ImportError as e: self.skipTest(f"api.speculative non importabile: {e}") goal = "__test_goal_regression_doc2a__" tool = "get_weather" args = {"city": "TestCity"} # Inietta un risultato nella cache gh = _goal_hash(goal) import time _spec_cache[gh] = { "get_weather": {"__test_goal_regression_doc2a__": "CACHED_METEO"}, "_expires_at": time.time() + 60, } result = get_speculative_result(goal, tool, args) # Pulizia _spec_cache.pop(gh, None) # Il risultato deve venire dalla cache (qualunque valore non-None) # In questo caso la chiave args non corrisponde esattamente → None è ok # Il test vero è che la funzione NON lanci eccezioni e sia chiamabile self.assertIsNone(result) # chiave args diversa → miss atteso def test_get_speculative_result_signature(self): """Doc2-1a: get_speculative_result(goal, tool_name, args) esiste e ha firma corretta.""" try: import inspect from api.speculative import get_speculative_result sig = inspect.signature(get_speculative_result) params = list(sig.parameters) self.assertEqual(params, ["goal", "tool_name", "args"], f"Firma inaspettata: {params}") except ImportError as e: self.skipTest(f"api.speculative non importabile: {e}") # ═══════════════════════════════════════════════════════════════════════════════ # Doc2-1b: memory/sync router deve essere montato in main.py # ═══════════════════════════════════════════════════════════════════════════════ class TestMemorySyncRouterMount(unittest.TestCase): """ Verifica che il router memory/sync sia registrato nell'app FastAPI. Bug originale: create_memory_sync_router mai montata → 404 su tutti gli endpoint /api/memory/sync/*. """ def test_create_memory_sync_router_exists_and_callable(self): """Doc2-1b: create_memory_sync_router è esportata da memory.sync.""" try: from memory.sync import create_memory_sync_router self.assertTrue(callable(create_memory_sync_router), "create_memory_sync_router deve essere callable") except ImportError as e: self.skipTest(f"memory.sync non importabile: {e}") def test_create_memory_sync_router_returns_apirouter(self): """Doc2-1b: factory produce un APIRouter con prefix /api/memory/sync.""" try: from fastapi import APIRouter from memory.sync import create_memory_sync_router except ImportError as e: self.skipTest(f"Dipendenze non disponibili: {e}") # Crea un memory stub minimale class _MemStub: def stats(self): return {} working = MagicMock() semantic = MagicMock(available=False) async def save_episode(self, *a, **kw): pass async def search(self, *a, **kw): return [] router = create_memory_sync_router(_MemStub()) self.assertIsInstance(router, APIRouter, "create_memory_sync_router deve restituire un APIRouter") self.assertEqual(router.prefix, "/api/memory/sync", f"Prefix sbagliato: {router.prefix}") def test_sync_router_has_required_endpoints(self): """Doc2-1b: router espone status, push e pull sotto il prefisso API.""" try: from memory.sync import create_memory_sync_router except ImportError as e: self.skipTest(f"memory.sync non importabile: {e}") class _MemStub: def stats(self): return {} working = MagicMock() semantic = MagicMock(available=False) async def save_episode(self, *a, **kw): pass async def search(self, *a, **kw): return [] router = create_memory_sync_router(_MemStub()) paths = {r.path for r in router.routes} self.assertIn("/api/memory/sync/status", paths, "status mancante dal sync router") self.assertIn("/api/memory/sync/push", paths, "push mancante dal sync router") self.assertIn("/api/memory/sync/pull", paths, "pull mancante dal sync router") def test_main_py_mounts_sync_router(self): """ Doc2-1b: verifica statica che main.py contenga create_memory_sync_router. Non importa main.py (side effects) — analisi sorgente pura. """ main_path = os.path.join(_BACKEND, "main.py") with open(main_path) as fh: src = fh.read() self.assertIn("create_memory_sync_router", src, "main.py non chiama create_memory_sync_router — router sync non montato") self.assertIn("include_router(_sync_router)", src, "main.py non include il sync router con include_router") # ═══════════════════════════════════════════════════════════════════════════════ # Terminal routing: mai fisso su un backend ritirato — deve girare su HANDS # Regression test aggiunto 2026-07-10 dopo verifica live che il bug era risolto # ma mai coperto da un test automatico. # ═══════════════════════════════════════════════════════════════════════════════ class TestTerminalRoutingNotFixedOnSpaceA(unittest.TestCase): """ Bug originale: il terminale (/api/terminal, /ws/terminal) veniva instradato in modo fisso sul vecchio BRAIN invece che su HANDS, causando comportamento errato in produzione. Fix: functions/api/[[catchall]].ts instrada /api/terminal e /ws/* verso HANDS_PATTERNS -> BACKEND_URL_B. Il backend verificato resta disponibile come ultimo anello della catena di failover in agentSSE.ts. Verificato live il 2026-07-10: GET /api/terminal/packages su agente-ai.pages.dev risponde 200 con header x-railway-edge (HANDS/Railway), non tramite HF Space. """ _CATCHALL = os.path.join( _BACKEND, "..", "artifacts", "agente-ai", "functions", "api", "[[catchall]].ts" ) _AGENT_SSE = os.path.join( _BACKEND, "..", "artifacts", "agente-ai", "src", "lib", "agentSSE.ts" ) def _read(self, path: str) -> str: norm = os.path.normpath(path) if not os.path.exists(norm): self.skipTest(f"File non trovato (path relativo cambiato?): {norm}") with open(norm, encoding="utf-8") as fh: return fh.read() def test_terminal_paths_are_in_hands_patterns(self): """/api/terminal e /ws/* devono comparire in HANDS_PATTERNS, non in un routing verso Space A.""" src = self._read(self._CATCHALL) idx_hands = src.find("HANDS_PATTERNS") self.assertNotEqual(idx_hands, -1, "HANDS_PATTERNS non trovato in catchall.ts") idx_memory = src.find("MEMORY_PATTERNS") hands_block = src[idx_hands:idx_memory if idx_memory != -1 else idx_hands + 3000] self.assertIn("api\\/terminal", hands_block, "/api/terminal non è più instradato via HANDS_PATTERNS — possibile regressione") self.assertIn("\\/ws\\/", hands_block, "/ws/* (terminale PTY) non è più instradato via HANDS_PATTERNS — possibile regressione") def test_space_a_is_not_the_default_backend(self): """BACKEND_URL_A deve restare un valore da env var, mai un hardcode usato come default primario.""" src = self._read(self._CATCHALL) self.assertIn("env.BACKEND_URL_A", src, "BACKEND_URL_A non più letto da env — verificare come viene risolto il backend BRAIN") self.assertNotIn("arjanit98-terminal.hf.space", src, "Hostname reale dello Space A hardcoded in catchall.ts — deve restare solo un placeholder/commento") def test_verified_backend_is_last_fallback_in_chain(self): """La catena frontend deve terminare sul backend verificato, non su host ritirati.""" src = self._read(self._AGENT_SSE) idx_chain = src.find("_getBackendChain") self.assertNotEqual(idx_chain, -1, "_getBackendChain non trovato in agentSSE.ts") chain_block = src[idx_chain: idx_chain + 1800] # In produzione il contratto corrente è il proxy same-origin CF Worker; # in locale la catena è interamente configurata tramite ENV.*. self.assertIn('if (isProd) return ["/api"]', chain_block, "Il routing production non usa il proxy same-origin /api") for env_name in ( "ENV.BACKEND_URL", "ENV.BACKEND_URL_2", "ENV.BACKEND_URL_C", "ENV.BACKEND_URL_D", "ENV.BACKEND_URL_E", "ENV.BACKEND_URL_HF_B", ): self.assertIn(env_name, chain_block, f"Fallback configurabile mancante: {env_name}") self.assertNotIn("arjanit98-terminal.hf.space", chain_block, "Space ritirato presente nella catena di fallback") self.assertNotIn("baida00-ai-backend-collab.hf.space", chain_block, "Space ritirato presente nella catena di fallback") # ═══════════════════════════════════════════════════════════════════════════════ # Entrypoint # ═══════════════════════════════════════════════════════════════════════════════ if __name__ == "__main__": unittest.main(verbosity=2)