tests/test_direct_file_conversion.py

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