EvoScientist/tests/test_image_tools_blocking.py

"""Prove the image tool path makes no blocking calls on the event loop.

The langgraph dev server activates blockbuster during runs; any sync file,
sqlite or socket call inside the async tool aborts it. These tests run the
full tool chain (scope resolution -> config load -> adapter -> save) under an
active BlockBuster, once against a scoped conversation and once unscoped.
"""

import json
from types import SimpleNamespace

import pytest
from blockbuster import BlockBuster

import EvoScientist.scope_registry as registry_module
import EvoScientist.workspace_scope as workspace_scope_module
from EvoScientist import paths
from EvoScientist.image_gen import config as image_config
from EvoScientist.image_gen import service
from EvoScientist.tools import image as image_tools
from EvoScientist.workspace_scope import provision_conversation_scope

PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"blockbuster"


class _FakeAdapter:
    def __init__(self, entry, *, client=None, timeout=120.0):
        self.entry = entry

    async def generate(self, **kwargs):
        return [PNG_BYTES]

    async def edit(self, **kwargs):
        return [PNG_BYTES]


@pytest.fixture
def scoped_env(tmp_path, monkeypatch):
    monkeypatch.setattr(paths, "WORKSPACE_ROOT", tmp_path)
    monkeypatch.setenv("EVOSCIENTIST_WORKSPACE_ISOLATION", "optional")
    registry_module._registry_cache.clear()
    record = provision_conversation_scope("thread-a", deployment_id="deployment-a")

    config = tmp_path / "config.yaml"
    config.write_text(
        "image_generation:\n"
        "  models:\n"
        "    - id: gpt-image-2\n"
        "      provider: openai\n"
        "      api_key: sk-x\n",
        encoding="utf-8",
    )
    monkeypatch.setattr(service, "_config_path", lambda: config)
    monkeypatch.setattr(
        service, "ADAPTER_CLASSES", {"openai": _FakeAdapter, "gemini": _FakeAdapter}
    )
    return record


def _blocker():
    return BlockBuster(
        scanned_modules=[
            image_tools,
            service,
            workspace_scope_module,
            registry_module,
            image_config,
        ]
    )


async def test_scoped_generate_makes_no_blocking_calls(scoped_env):
    record = scoped_env
    configurable = {
        "thread_id": "thread-a",
        "workspace_scope_id": record.scope_id,
        "workspace_scope_owner_id": record.primary_owner_id,
        "workspace_scope_revision": record.revision,
        "workspace_deployment_id": "deployment-a",
    }
    blocker = _blocker()
    blocker.activate()
    try:
        result = json.loads(
            await image_tools.generate_image.ainvoke(
                {"prompt": "a cat"}, config={"configurable": configurable}
            )
        )
    finally:
        blocker.deactivate()
    assert result["ok"] is True, result
    saved = record_files_dir(record) / result["paths"][0]
    assert saved.is_file()
    assert saved.read_bytes() == PNG_BYTES


async def test_unscoped_generate_makes_no_blocking_calls(scoped_env, monkeypatch):
    # No scope identifiers in the config: falls back to _active_workspace.
    monkeypatch.setattr(paths, "_active_workspace", paths.WORKSPACE_ROOT)
    blocker = _blocker()
    blocker.activate()
    try:
        result = json.loads(
            await image_tools.generate_image.ainvoke(
                {"prompt": "a cat"}, config={"configurable": {}}
            )
        )
    finally:
        blocker.deactivate()
    assert result["ok"] is True, result
    saved = paths.WORKSPACE_ROOT / result["paths"][0]
    assert saved.is_file()


async def test_scoped_edit_makes_no_blocking_calls(scoped_env):
    record = scoped_env
    files_dir = record_files_dir(record)
    (files_dir / "artifacts").mkdir(parents=True)
    (files_dir / "artifacts" / "src.png").write_bytes(PNG_BYTES)
    configurable = {
        "thread_id": "thread-a",
        "workspace_scope_id": record.scope_id,
        "workspace_scope_owner_id": record.primary_owner_id,
        "workspace_scope_revision": record.revision,
        "workspace_deployment_id": "deployment-a",
    }
    blocker = _blocker()
    blocker.activate()
    try:
        result = json.loads(
            await image_tools.edit_image.ainvoke(
                {"image_path": "artifacts/src.png", "prompt": "make blue"},
                config={"configurable": configurable},
            )
        )
    finally:
        blocker.deactivate()
    assert result["ok"] is True, result
    assert (files_dir / result["paths"][0]).is_file()


def record_files_dir(record):
    return (
        paths.WORKSPACE_ROOT
        / ".evoscientist"
        / "conversations"
        / record.scope_id
        / "files"
    )