TieMeasureFlow/src/vision/tests/test_runner.py

"""The runner: graph plus image in, numbers out. No HTTP, no database."""
import pytest

# The vision tree requires Python >=3.13 (vs-task/vs-measure/vs-pm2d declare it
# upstream, see pyproject.toml's `vision` extra), while backend and frontend
# stay on 3.11. This must run before `import numpy` - numpy itself is only
# installed alongside the vision extra - so that `uv run pytest` on the
# default 3.11 environment reports this module as skipped, with a reason,
# instead of erroring out of the whole collection.
pytest.importorskip(
    "visionsuite.task",
    reason=(
        "the vision tree requires Python >=3.13 (vs-task/vs-measure/vs-pm2d "
        "declare it upstream); run it with "
        "`uv run --python 3.13 --extra vision --extra dev pytest src/vision/tests`"
    ),
)

import numpy as np

from src.vision import runner
from src.vision.runner import RunOutcome, engine_version, run_graph


def a_disc(radius: int = 60, size: int = 400) -> np.ndarray:
    """A light disc on a dark ground: a fake part, but a measurable one."""
    image = np.zeros((size, size), dtype=np.uint8)
    yy, xx = np.ogrid[:size, :size]
    centre = size // 2
    image[(yy - centre) ** 2 + (xx - centre) ** 2 <= radius ** 2] = 255
    return image


def a_circle_graph() -> dict:
    """A two-tool graph, built with vs-task's own API."""
    from visionsuite.task import Task
    from visionsuite.task.tools import DetectCircleTool, DiameterTool

    task = Task(name="prova")
    found = task.add(DetectCircleTool(name="disco"))
    found.center_row, found.center_col, found.radius = 200.0, 200.0, 60.0
    found.scan_size = 30.0

    quota = task.add(DiameterTool(name="diametro"))
    quota.link("circle", found, "circle")
    return task.to_dict()


def test_the_graph_produces_named_outputs():
    outcome = run_graph(a_disc(), a_circle_graph())

    assert isinstance(outcome, RunOutcome)
    assert outcome.failures == []
    # Keys are "tool id dot output name" - the same shape a subtask stores in
    # vision_output.
    assert any(key.endswith(".diameter") for key in outcome.outputs)
    diameter = next(v for k, v in outcome.outputs.items() if k.endswith(".diameter"))
    assert diameter == pytest.approx(120.0, abs=4.0)


def test_a_broken_tool_does_not_sink_the_others():
    """vs-task does not stop the run on an error: the runner reports it instead."""
    graph = a_circle_graph()
    graph["tools"].append(
        {"id": "ghost", "kind": "DetectCircle", "name": "assente",
         "properties": {"center_row": 5.0, "center_col": 5.0, "radius": 900.0}}
    )

    outcome = run_graph(a_disc(), graph)

    assert [f.tool_id for f in outcome.failures] == ["ghost"]
    assert any(key.endswith(".diameter") for key in outcome.outputs)


class _FakeResult:
    """Stands in for a vs-task tool result: only what run_graph touches."""

    def __init__(self, tool_id, tool_name, ok, outputs=None, error=None):
        self.tool_id = tool_id
        self.tool_name = tool_name
        self.ok = ok
        self.outputs = outputs or {}
        self.error = error


def test_a_nan_output_is_dropped_not_measured(monkeypatch):
    """vs-measure emits NaN on an empty slice (caliper.py:112 per the
    revisione) - a NaN must not become a quote's value (#16)."""
    def fake_run(self, context=None):
        return [_FakeResult("t1", "diametro", True, outputs={"diameter": float("nan")})]
    monkeypatch.setattr(runner.Task, "run", fake_run)

    outcome = run_graph(a_disc(), a_circle_graph())

    assert outcome.outputs == {}
    assert outcome.invalid_outputs == {"t1.diameter": "nan"}


def test_an_infinite_output_is_dropped_not_measured(monkeypatch):
    def fake_run(self, context=None):
        return [_FakeResult("t1", "diametro", True, outputs={"diameter": float("inf")})]
    monkeypatch.setattr(runner.Task, "run", fake_run)

    outcome = run_graph(a_disc(), a_circle_graph())

    assert outcome.outputs == {}
    assert outcome.invalid_outputs == {"t1.diameter": "infinite"}


def test_a_bool_output_is_not_treated_as_a_measurable_number(monkeypatch):
    """bool is a subclass of int - a flag output must not become a quote."""
    def fake_run(self, context=None):
        return [_FakeResult("t1", "flag", True, outputs={"present": True})]
    monkeypatch.setattr(runner.Task, "run", fake_run)

    outcome = run_graph(a_disc(), a_circle_graph())

    assert outcome.outputs == {}
    assert outcome.invalid_outputs == {}


def test_the_engine_says_which_version_it_is():
    version = engine_version()
    assert len(version) == 40  # a full git SHA


def test_engine_version_honours_the_environment_variable(monkeypatch):
    """A build-stamped VISION_ENGINE_VERSION wins over git, if set (Ruling R8:
    the container has no usable .git for the vendored submodule, so the
    worker image stamps the commit at build time instead of discovering it)."""
    stamped = "a" * 40
    monkeypatch.setenv("VISION_ENGINE_VERSION", stamped)
    runner._engine_version_from_git.cache_clear()

    try:
        assert engine_version() == stamped
    finally:
        runner._engine_version_from_git.cache_clear()


def test_engine_version_raises_rather_than_inventing_one(monkeypatch):
    """No env var and no resolvable git HEAD: refuse loudly. An engine that
    cannot say what it is has no business producing measurements - it must
    never fall back to a guess, an empty string, or "unknown"."""
    monkeypatch.delenv("VISION_ENGINE_VERSION", raising=False)
    runner._engine_version_from_git.cache_clear()

    def fake_check_output(*args, **kwargs):
        raise FileNotFoundError("git: command not found")

    monkeypatch.setattr(runner.subprocess, "check_output", fake_check_output)

    try:
        with pytest.raises(RuntimeError, match="VISION_ENGINE_VERSION"):
            engine_version()
    finally:
        runner._engine_version_from_git.cache_clear()