Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
148 changes: 148 additions & 0 deletions tests/test_prime_plugin.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,148 @@
from pathlib import Path

import verifiers.cli.plugins.prime as prime_plugin


def _make_workspace(tmp_path: Path) -> tuple[Path, Path]:
workspace = tmp_path / "workspace"
env_dir = workspace / "environments" / "my_env"
env_dir.mkdir(parents=True)
(workspace / "verifiers").mkdir()
(workspace / "pyproject.toml").write_text(
'[project]\nname = "workspace"\nversion = "0.1.0"\n',
encoding="utf-8",
)
return workspace, env_dir


def _touch_python(venv_root: Path) -> Path:
python_bin = prime_plugin._venv_python(venv_root)
python_bin.parent.mkdir(parents=True, exist_ok=True)
python_bin.write_text("", encoding="utf-8")
return python_bin


def test_find_workspace_root_from_nested_environment_dir(tmp_path: Path):
workspace, env_dir = _make_workspace(tmp_path)

assert prime_plugin._find_workspace_root(env_dir) == workspace


def test_resolve_workspace_python_prefers_workspace_venv_over_uv_env(
tmp_path: Path, monkeypatch
):
workspace, env_dir = _make_workspace(tmp_path)
workspace_python = _touch_python(workspace / ".venv")
_touch_python(env_dir / ".venv")

monkeypatch.setattr(prime_plugin, "_python_can_import_module", lambda *_: True)
monkeypatch.setenv("UV_PROJECT_ENVIRONMENT", str(env_dir / ".venv"))
monkeypatch.delenv("VIRTUAL_ENV", raising=False)

assert prime_plugin._resolve_workspace_python(env_dir) == str(workspace_python)


def test_build_module_command_install_adds_workspace_env_path(
tmp_path: Path, monkeypatch
):
workspace, env_dir = _make_workspace(tmp_path)
plugin = prime_plugin.PrimeCLIPlugin()

monkeypatch.setattr(prime_plugin, "_current_cwd", lambda: env_dir)
monkeypatch.setattr(prime_plugin, "_resolve_workspace_python", lambda *_: "python")

command = plugin.build_module_command(plugin.install_module, ["my-env"])

assert command == [
"python",
"-m",
plugin.install_module,
"my-env",
"--path",
str((workspace / "environments").resolve()),
]


def test_build_module_command_eval_rewrites_relative_env_dir_path(
tmp_path: Path, monkeypatch
):
workspace, env_dir = _make_workspace(tmp_path)
plugin = prime_plugin.PrimeCLIPlugin()

monkeypatch.setattr(prime_plugin, "_current_cwd", lambda: env_dir)
monkeypatch.setattr(prime_plugin, "_resolve_workspace_python", lambda *_: "python")

command = plugin.build_module_command(
plugin.eval_module,
["my-env", "--env-dir-path", "./environments"],
)

assert command == [
"python",
"-m",
plugin.eval_module,
"my-env",
"--env-dir-path",
str((workspace / "environments").resolve()),
]


def test_build_module_command_gepa_adds_workspace_env_dir_path(
tmp_path: Path, monkeypatch
):
workspace, env_dir = _make_workspace(tmp_path)
plugin = prime_plugin.PrimeCLIPlugin()

monkeypatch.setattr(prime_plugin, "_current_cwd", lambda: env_dir)
monkeypatch.setattr(prime_plugin, "_resolve_workspace_python", lambda *_: "python")

command = plugin.build_module_command(plugin.gepa_module, ["my-env"])

assert command == [
"python",
"-m",
plugin.gepa_module,
"my-env",
"--env-dir-path",
str((workspace / "environments").resolve()),
]


def test_build_module_command_build_adds_workspace_env_path(
tmp_path: Path, monkeypatch
):
workspace, env_dir = _make_workspace(tmp_path)
plugin = prime_plugin.PrimeCLIPlugin()

monkeypatch.setattr(prime_plugin, "_current_cwd", lambda: env_dir)
monkeypatch.setattr(prime_plugin, "_resolve_workspace_python", lambda *_: "python")

command = plugin.build_module_command(plugin.build_module, ["my-env"])

assert command == [
"python",
"-m",
plugin.build_module,
"my-env",
"--path",
str((workspace / "environments").resolve()),
]


def test_build_module_command_init_adds_workspace_env_path(tmp_path: Path, monkeypatch):
workspace, env_dir = _make_workspace(tmp_path)
plugin = prime_plugin.PrimeCLIPlugin()

monkeypatch.setattr(prime_plugin, "_current_cwd", lambda: env_dir)
monkeypatch.setattr(prime_plugin, "_resolve_workspace_python", lambda *_: "python")

command = plugin.build_module_command(plugin.init_module, ["my-env"])

assert command == [
"python",
"-m",
plugin.init_module,
"my-env",
"--path",
str((workspace / "environments").resolve()),
]
103 changes: 99 additions & 4 deletions verifiers/cli/plugins/prime.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
from typing import Sequence

PRIME_PLUGIN_API_VERSION = 1
WORKSPACE_ENV_DIR = "environments"


def _venv_python(venv_root: Path) -> Path:
Expand Down Expand Up @@ -42,14 +43,21 @@ def _python_can_import_module(

def _resolve_workspace_python(cwd: Path | None = None) -> str:
workspace = (cwd or Path.cwd()).resolve()
workspace_str = str(workspace)
workspace_root = _find_workspace_root(workspace)
workspace_for_probe = workspace_root or workspace
workspace_str = str(workspace_for_probe)
module = "verifiers.cli.commands.eval"

def _usable(candidate: Path) -> bool:
return candidate.exists() and _python_can_import_module(
str(candidate), module, workspace_str
)

if workspace_root is not None:
candidate = _venv_python(workspace_root / ".venv")
if _usable(candidate):
return str(candidate)

uv_project_env = os.environ.get("UV_PROJECT_ENVIRONMENT")
if uv_project_env:
candidate = _venv_python(Path(uv_project_env))
Expand All @@ -71,6 +79,67 @@ def _usable(candidate: Path) -> bool:
return sys.executable


def _find_workspace_root(start: Path) -> Path | None:
for directory in [start, *start.parents]:
if (
(directory / "pyproject.toml").is_file()
and (directory / "verifiers").is_dir()
and (directory / WORKSPACE_ENV_DIR).is_dir()
):
return directory
return None


def _current_cwd() -> Path:
return Path.cwd().resolve()


def _resolve_dir_arg(value: str, cwd: Path) -> str:
path = Path(value)
if path.is_absolute():
return str(path)

for directory in [cwd, *cwd.parents]:
candidate = (directory / path).resolve()
if candidate.exists() and candidate.is_dir():
return str(candidate)

return str((cwd / path).resolve())


def _normalize_or_append_dir_option(
args: Sequence[str] | None,
*,
long_flag: str,
short_flag: str,
fallback_value: str | None,
cwd: Path,
) -> list[str]:
normalized = list(args or [])
found_flag = False
i = 0
while i < len(normalized):
token = normalized[i]
if token in (long_flag, short_flag):
found_flag = True
if i + 1 < len(normalized):
normalized[i + 1] = _resolve_dir_arg(normalized[i + 1], cwd)
i += 1
elif token.startswith(f"{long_flag}="):
found_flag = True
_, raw_value = token.split("=", 1)
normalized[i] = f"{long_flag}={_resolve_dir_arg(raw_value, cwd)}"
elif token.startswith(short_flag) and token != short_flag:
found_flag = True
raw_value = token[len(short_flag) :]
normalized[i] = f"{short_flag}{_resolve_dir_arg(raw_value, cwd)}"
i += 1

if not found_flag and fallback_value is not None:
normalized.extend([long_flag, fallback_value])
return normalized


@dataclass(frozen=True)
class PrimeCLIPlugin:
"""Declarative command surface consumed by prime-cli."""
Expand All @@ -86,9 +155,35 @@ class PrimeCLIPlugin:
def build_module_command(
self, module_name: str, args: Sequence[str] | None = None
) -> list[str]:
command = [_resolve_workspace_python(), "-m", module_name]
if args:
command.extend(args)
cwd = _current_cwd()
workspace_root = _find_workspace_root(cwd)
workspace_env_dir: str | None = None
if workspace_root is not None:
env_dir = workspace_root / WORKSPACE_ENV_DIR
if env_dir.is_dir():
workspace_env_dir = str(env_dir.resolve())

normalized_args = list(args or [])
if module_name in (self.install_module, self.build_module, self.init_module):
normalized_args = _normalize_or_append_dir_option(
normalized_args,
long_flag="--path",
short_flag="-p",
fallback_value=workspace_env_dir,
cwd=cwd,
)
elif module_name in (self.eval_module, self.gepa_module):
normalized_args = _normalize_or_append_dir_option(
normalized_args,
long_flag="--env-dir-path",
short_flag="-p",
fallback_value=workspace_env_dir,
cwd=cwd,
)

command = [_resolve_workspace_python(cwd), "-m", module_name]
if normalized_args:
command.extend(normalized_args)
return command


Expand Down
Loading