tests/test_external_vla_baseline.py

from __future__ import annotations

import json
import subprocess
import sys
from pathlib import Path

from dovla_cil.eval.external_vla_baseline import (
    ExternalVLABaselineSpec,
    assess_external_vla_baseline,
    build_external_vla_plan,
    redact_command,
    run_external_vla_entrypoint,
    write_external_vla_plan,
)


def test_external_vla_plan_is_secret_free_and_not_ready_without_adapter(tmp_path: Path) -> None:
    dataset = tmp_path / "dataset"
    dataset.mkdir()
    spec = ExternalVLABaselineSpec(
        model_family="smolvla",
        checkpoint_path=str(tmp_path / "missing-checkpoint"),
        dataset_dir=str(dataset),
        out_dir=str(tmp_path / "run"),
    )

    status = assess_external_vla_baseline(spec)
    plan = build_external_vla_plan(spec)
    payload = json.dumps(plan)

    assert not status.ready
    assert "adapter_entrypoint" in status.missing
    assert "hf download lerobot/smolvla_base" in plan["commands"]["download"]
    assert "TOKEN" not in payload
    assert "OPENCLAUDE_API_KEY" not in payload


def test_external_vla_redacts_token_like_command_fragments() -> None:
    command = "HF_TOKEN=abc hf download repo --token xyz --api-key nope"

    redacted = redact_command(command)

    assert "abc" not in redacted
    assert "xyz" not in redacted
    assert "nope" not in redacted
    assert "<redacted>" in redacted


def test_external_vla_dry_run_cli_writes_plan(tmp_path: Path) -> None:
    dataset = tmp_path / "dataset"
    dataset.mkdir()
    out = tmp_path / "out"

    result = subprocess.run(
        [
            sys.executable,
            "scripts/run_external_vla_baseline.py",
            "--model-family",
            "smolvla",
            "--checkpoint",
            str(tmp_path / "missing-checkpoint"),
            "--dataset",
            str(dataset),
            "--out",
            str(out),
            "--dry-run",
        ],
        check=True,
        capture_output=True,
        text=True,
    )

    assert "external VLA plan" in result.stdout
    assert (out / "external_vla_baseline_plan.json").exists()


def test_external_vla_entrypoint_contract(tmp_path: Path, monkeypatch) -> None:
    module_path = tmp_path / "fake_external_adapter.py"
    module_path.write_text(
        "def run(spec, plan):\n"
        "    return {'model_family': spec['model_family'], 'success_rate': 0.25}\n",
        encoding="utf-8",
    )
    monkeypatch.syspath_prepend(str(tmp_path))
    dataset = tmp_path / "dataset"
    checkpoint = tmp_path / "checkpoint"
    dataset.mkdir()
    checkpoint.mkdir()
    spec = ExternalVLABaselineSpec(
        model_family="smolvla",
        checkpoint_path=str(checkpoint),
        dataset_dir=str(dataset),
        out_dir=str(tmp_path / "run"),
        adapter_entrypoint="fake_external_adapter:run",
    )

    plan_path = write_external_vla_plan(spec)
    result = run_external_vla_entrypoint("fake_external_adapter:run", spec)

    assert plan_path.exists()
    assert result == {"model_family": "smolvla", "success_rate": 0.25}


def test_external_vla_cli_passes_secret_free_adapter_config(tmp_path: Path) -> None:
    dataset = tmp_path / "dataset"
    checkpoint = tmp_path / "checkpoint"
    out = tmp_path / "out"
    dataset.mkdir()
    checkpoint.mkdir()
    adapter_config = tmp_path / "adapter.json"
    adapter_config.write_text('{"steps": 2, "state_dim": 32}', encoding="utf-8")

    subprocess.run(
        [
            sys.executable,
            "scripts/run_external_vla_baseline.py",
            "--checkpoint",
            str(checkpoint),
            "--dataset",
            str(dataset),
            "--out",
            str(out),
            "--adapter-config",
            str(adapter_config),
            "--dry-run",
        ],
        check=True,
        capture_output=True,
        text=True,
    )

    plan = json.loads((out / "external_vla_baseline_plan.json").read_text(encoding="utf-8"))
    assert plan["spec"]["metadata"] == {"steps": 2, "state_dim": 32}