Spaces:
Running
Running
File size: 5,833 Bytes
c8365f5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 | 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_inline_csv_conversion_writes_source_and_semantically_equivalent_json():
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 = """Esegui solo nel workspace VFS locale questa conversione deterministica.
Crea capability_catalog.csv con contenuto esatto: id,name,active
1,alpha,true
2,beta,false
. Poi crea capability_catalog.json con lo stesso catalogo come array JSON valido di due oggetti.
Non usare rete, shell, servizi esterni o provider aggiuntivi."""
output, called, succeeded, errors = asyncio.run(
Harness()._run_direct_tools(goal, on_step=on_step)
)
assert output.startswith("[DIRECT_TERMINAL]\nE2E_CONVERSION_OK")
assert "verificati 2 record" in output
assert called == 1
assert succeeded == 1
assert errors == 0
assert writes["capability_catalog.csv"] == "id,name,active\n1,alpha,true\n2,beta,false\n"
assert writes["capability_catalog.json"] == (
'[\n {\n "id": 1,\n "name": "alpha",\n "active": "true"\n },\n'
' {\n "id": 2,\n "name": "beta",\n "active": "false"\n }\n]\n'
)
assert {event.get("path") for event in events if event.get("action") == "file_written"} == {
"capability_catalog.csv", "capability_catalog.json"
}
def test_textual_checkout_plan_does_not_trigger_direct_image_generation():
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(
"Crea un breve piano per verificare un errore intermittente nel checkout e indica il primo dato da raccogliere.",
on_step=on_step,
)
)
assert "E2E_IMAGE_OK" not in output
assert not any(event.get("action") == "file_written" for event in events)
assert not any(path.startswith("generated-image-") for path in writes)
assert (called, succeeded, errors) == (0, 0, 0)
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
|