Spaces:
Running
Running
| import asyncio | |
| import sys | |
| import types | |
| def _install_tool_stubs(writes, image_url="https://example.test/generated.png"): | |
| tools_pkg = types.ModuleType("tools") | |
| registry = types.ModuleType("tools.registry") | |
| async def write_file(path, content): | |
| writes[path] = content | |
| # Il registry reale usa il contratto `ok` per le operazioni filesystem. | |
| return {"ok": True, "path": path} | |
| async def generate_image(prompt, width=512, height=512, prefer_internal=True): | |
| # Il percorso interattivo deve ottenere un URL renderizzabile subito, | |
| # senza attendere il cold-start del generatore interno. | |
| assert prefer_internal is False | |
| return {"url": image_url, "prompt": prompt, "width": width, "height": height} | |
| registry.TOOL_REGISTRY = { | |
| "write_file": {"_fn": write_file}, | |
| "generate_image": {"_fn": generate_image}, | |
| } | |
| sys.modules["tools"] = tools_pkg | |
| sys.modules["tools.registry"] = registry | |
| api_pkg = types.ModuleType("api") | |
| speculative = types.ModuleType("api.speculative") | |
| speculative.get_speculative_result = lambda *_args, **_kwargs: None | |
| sys.modules["api"] = api_pkg | |
| sys.modules["api.speculative"] = speculative | |
| def test_direct_csv_conversion_writes_vfs_content_and_returns_terminal_output(): | |
| writes = {} | |
| _install_tool_stubs(writes) | |
| from agents.unified_loop_tools import DirectToolsMixin | |
| class Harness(DirectToolsMixin): | |
| def _max_tokens_for_goal(self, _goal): | |
| return 4096 | |
| events = [] | |
| async def on_step(event): | |
| events.append(event) | |
| goal = """Converti e2e_metrics.csv in un file chiamato e2e_metrics.json. | |
| --- **File allegati:** | |
| ### 📎 e2e_metrics.csv (excel, 66B) | |
| ``` | |
| mese,richieste,successi | |
| 2026-01,12,11 | |
| 2026-02,15,14 | |
| ``` | |
| """ | |
| output, called, succeeded, errors = asyncio.run( | |
| Harness()._run_direct_tools(goal, on_step=on_step) | |
| ) | |
| assert output.startswith("[DIRECT_TERMINAL]\nE2E_CONVERSION_OK") | |
| assert called == 1 | |
| assert succeeded == 1 | |
| assert errors == 0 | |
| assert '"richieste": 15' in writes["e2e_metrics.json"] | |
| assert any(event.get("action") == "file_written" for event in events) | |
| def test_direct_image_returns_renderable_terminal_markdown_without_llm(): | |
| writes = {} | |
| _install_tool_stubs(writes) | |
| from agents.unified_loop_tools import DirectToolsMixin | |
| class Harness(DirectToolsMixin): | |
| def _max_tokens_for_goal(self, _goal): | |
| return 4096 | |
| events = [] | |
| async def on_step(event): | |
| events.append(event) | |
| output, called, succeeded, errors = asyncio.run( | |
| Harness()._run_direct_tools( | |
| "Genera un’immagine quadrata di un aeroplanino arancione su fondo blu notte.", | |
| on_step=on_step, | |
| ) | |
| ) | |
| assert output.startswith("[DIRECT_TERMINAL]\n![Immagine generata](https://image.pollinations.ai/prompt/") | |
| assert "E2E_IMAGE_OK" in output | |
| assert "aeroplanino%20arancione" in output | |
| assert "File VFS: `generated-image-" in output | |
| image_event = next(event for event in events if event.get("action") == "file_written") | |
| assert image_event["path"].startswith("generated-image-") | |
| assert image_event["path"].endswith(".jpg") | |
| assert image_event["source_url"].startswith("https://image.pollinations.ai/prompt/") | |
| assert image_event["mime_type"] == "image/jpeg" | |
| assert called == 1 | |
| assert succeeded == 1 | |
| assert errors == 0 | |