688 lines
28 KiB
Python
688 lines
28 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import shlex
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
ORC_DIR = REPO_ROOT / "skills" / "orc"
|
|
SCRIPT = ORC_DIR / "scripts" / "resolve_profile.py"
|
|
CONFIG = ORC_DIR / "config.yaml"
|
|
|
|
spec = importlib.util.spec_from_file_location("orc_resolve_profile", SCRIPT)
|
|
assert spec is not None and spec.loader is not None
|
|
orc_profiles = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(orc_profiles)
|
|
|
|
|
|
class OrcSkillTests(unittest.TestCase):
|
|
def test_orc_is_explicit_and_routes_to_independent_skills(self) -> None:
|
|
skill = (ORC_DIR / "SKILL.md").read_text(encoding="utf-8")
|
|
routing = (ORC_DIR / "references" / "routing.md").read_text(encoding="utf-8")
|
|
metadata = (ORC_DIR / "agents" / "openai.yaml").read_text(encoding="utf-8")
|
|
|
|
self.assertIn("allow_implicit_invocation: false", metadata)
|
|
self.assertIn("low", skill)
|
|
self.assertIn("mid", skill)
|
|
self.assertIn("high", skill)
|
|
self.assertIn("`codex` 和 `cursor-agent`", skill)
|
|
self.assertIn("<orc-skill-dir>/config.yaml", skill)
|
|
self.assertIn("不得在项目中创建 `docs/orc/config.yaml`", skill)
|
|
self.assertIn("薄路由器", skill)
|
|
self.assertIn("不做领域判断", skill)
|
|
self.assertIn("当前 shell", skill)
|
|
self.assertNotIn("用户未指定档位时采用以下判断", skill)
|
|
self.assertNotIn("若该档位不足以安全完成", skill)
|
|
for child in ("$ack", "$manage-release", "$deb-publisher", "$publish-docker-image"):
|
|
self.assertIn(child, routing)
|
|
|
|
def test_children_do_not_reference_orc(self) -> None:
|
|
for child in ("ack", "manage-release", "deb-publisher", "publish-docker-image"):
|
|
for path in (REPO_ROOT / "skills" / child).rglob("*"):
|
|
if not path.is_file() or "__pycache__" in path.parts:
|
|
continue
|
|
content = path.read_text(encoding="utf-8", errors="ignore")
|
|
self.assertNotIn("$orc", content, str(path))
|
|
self.assertNotRegex(
|
|
content,
|
|
r"(?<![A-Za-z0-9_-])/orc(?![A-Za-z0-9_-])",
|
|
str(path),
|
|
)
|
|
self.assertNotRegex(content, r"\bORC\b", str(path))
|
|
self.assertNotIn("skills/orc", content, str(path))
|
|
|
|
def test_routing_pins_normal_pr_and_unsupported_release_gate(self) -> None:
|
|
routing = (ORC_DIR / "references" / "routing.md").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
|
|
self.assertIn("普通非发布 PR/MR", routing)
|
|
self.assertIn("最多到", routing)
|
|
self.assertIn("review_ready", routing)
|
|
self.assertIn("当前 task 粒度无法安全表达", routing)
|
|
self.assertRegex(
|
|
routing,
|
|
r"不判断版本号、\s*实现方案、发布风险或产物策略",
|
|
)
|
|
|
|
def test_shared_config_and_cli_validate(self) -> None:
|
|
config = orc_profiles.load_config(CONFIG)
|
|
self.assertEqual(config["version"], 2)
|
|
self.assertEqual(config["cliPolicy"], "current-host")
|
|
self.assertEqual(set(config["profiles"]), {"codex", "cursor-agent"})
|
|
for cli in ("codex", "cursor-agent"):
|
|
self.assertEqual(set(config["profiles"][cli]), {"low", "mid", "high"})
|
|
completed = subprocess.run(
|
|
[sys.executable, str(SCRIPT), "validate"],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
self.assertEqual(completed.returncode, 0, completed.stderr)
|
|
self.assertIn('"ok": true', completed.stdout)
|
|
|
|
project_config_argument = subprocess.run(
|
|
[sys.executable, str(SCRIPT), "validate", str(CONFIG)],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
self.assertEqual(project_config_argument.returncode, 2)
|
|
|
|
clean_python = subprocess.run(
|
|
[sys.executable, "-I", "-S", str(SCRIPT), "validate"],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
self.assertEqual(clean_python.returncode, 0, clean_python.stderr)
|
|
|
|
def test_level_precedence_is_stage_then_global_then_config(self) -> None:
|
|
config = orc_profiles.load_config(CONFIG)
|
|
stage = orc_profiles.resolve_profile(
|
|
config,
|
|
stage="docker",
|
|
host_cli="codex",
|
|
global_level="mid",
|
|
stage_level="high",
|
|
)
|
|
global_choice = orc_profiles.resolve_profile(
|
|
config, stage="docker", host_cli="codex", global_level="mid"
|
|
)
|
|
configured = orc_profiles.resolve_profile(
|
|
config, stage="docker", host_cli="codex"
|
|
)
|
|
fallback_config = copy.deepcopy(config)
|
|
del fallback_config["stageDefaults"]["docker"]
|
|
fallback = orc_profiles.resolve_profile(
|
|
fallback_config, stage="docker", host_cli="codex"
|
|
)
|
|
|
|
self.assertEqual((stage["level"], stage["selectionSource"]), ("high", "request.stage"))
|
|
self.assertEqual((global_choice["level"], global_choice["selectionSource"]), ("mid", "request.global"))
|
|
self.assertEqual((configured["level"], configured["selectionSource"]), ("low", "config.stageDefaults.docker"))
|
|
self.assertEqual((fallback["level"], fallback["selectionSource"]), ("mid", "config.defaultLevel"))
|
|
self.assertEqual(configured["cli"], "codex")
|
|
self.assertEqual(configured["cliSelectionSource"], "runtime.host")
|
|
|
|
with self.assertRaisesRegex(orc_profiles.ConfigError, "host CLI is required"):
|
|
orc_profiles.resolve_profile(config, stage="docker")
|
|
|
|
def test_resolver_builds_fixed_worker_args_without_free_command_fields(self) -> None:
|
|
config = orc_profiles.load_config(CONFIG)
|
|
plan = orc_profiles.resolve_profile(
|
|
config, stage="release", host_cli="codex"
|
|
)
|
|
|
|
self.assertEqual(plan["workerArgs"][:2], ["--model", "gpt-5.6-terra"])
|
|
self.assertIn("--strict-config", plan["workerArgs"])
|
|
approval_index = plan["workerArgs"].index("--ask-for-approval")
|
|
self.assertEqual(plan["workerArgs"][approval_index + 1], "on-request")
|
|
self.assertIn('approvals_reviewer="auto_review"', plan["workerArgs"])
|
|
self.assertNotIn(
|
|
"sandbox_workspace_write.network_access=true",
|
|
plan["workerArgs"],
|
|
)
|
|
self.assertNotIn("danger-full-access", " ".join(plan["workerArgs"]))
|
|
self.assertNotIn("env", plan["profile"])
|
|
self.assertNotIn("command", plan["profile"])
|
|
|
|
cursor = orc_profiles.resolve_profile(
|
|
config,
|
|
stage="release",
|
|
host_cli="cursor-agent",
|
|
stage_level="low",
|
|
)
|
|
self.assertEqual(cursor["cli"], "cursor-agent")
|
|
self.assertEqual(cursor["cliSelectionSource"], "runtime.host")
|
|
self.assertEqual(cursor["modelAuth"], "cursor-login")
|
|
self.assertEqual(
|
|
cursor["workerArgs"],
|
|
[
|
|
"--model",
|
|
"auto",
|
|
"--auto-review",
|
|
"--sandbox",
|
|
"enabled",
|
|
],
|
|
)
|
|
with self.assertRaisesRegex(orc_profiles.ConfigError, "not valid for cursor-agent"):
|
|
orc_profiles.resolve_profile(
|
|
config,
|
|
stage="release",
|
|
host_cli="cursor-agent",
|
|
model_auth="openai",
|
|
)
|
|
with self.assertRaisesRegex(orc_profiles.ConfigError, "not valid for codex"):
|
|
orc_profiles.resolve_profile(
|
|
config,
|
|
stage="release",
|
|
host_cli="codex",
|
|
model_auth="cursor-api-key",
|
|
)
|
|
|
|
def test_worker_environment_selects_one_model_and_remote_auth(self) -> None:
|
|
ambient = {
|
|
"OPENAI_API_KEY": "openai-secret",
|
|
"AZURE_OPENAI_API_KEY": "azure-secret",
|
|
"GITHUB_TOKEN": "github-secret",
|
|
"GITLAB_TOKEN": "gitlab-secret",
|
|
"GITEA_TOKEN": "gitea-secret",
|
|
"FORGEJO_TOKEN": "forgejo-secret",
|
|
"DEB_TOKEN": "deb-secret",
|
|
"SSH_AUTH_SOCK": "/tmp/agent.sock",
|
|
"HTTPS_PROXY": "https://user:secret@proxy.example",
|
|
}
|
|
with mock.patch.dict(os.environ, ambient, clear=True):
|
|
environment = orc_profiles.worker_environment(
|
|
"release",
|
|
model_auth="openai",
|
|
remote_auth="forgejo-token",
|
|
)
|
|
|
|
self.assertEqual(environment["OPENAI_API_KEY"], "openai-secret")
|
|
self.assertEqual(environment["FORGEJO_TOKEN"], "forgejo-secret")
|
|
for rejected in (
|
|
"AZURE_OPENAI_API_KEY",
|
|
"GITHUB_TOKEN",
|
|
"GITLAB_TOKEN",
|
|
"GITEA_TOKEN",
|
|
"DEB_TOKEN",
|
|
"SSH_AUTH_SOCK",
|
|
"HTTPS_PROXY",
|
|
):
|
|
self.assertNotIn(rejected, environment)
|
|
|
|
cursor_ambient = {
|
|
"CURSOR_API_KEY": "cursor-secret",
|
|
"OPENAI_API_KEY": "openai-secret",
|
|
}
|
|
with mock.patch.dict(os.environ, cursor_ambient, clear=True):
|
|
cursor_environment = orc_profiles.worker_environment(
|
|
"release",
|
|
model_auth="cursor-api-key",
|
|
remote_auth="none",
|
|
)
|
|
self.assertEqual(cursor_environment["CURSOR_API_KEY"], "cursor-secret")
|
|
self.assertNotIn("OPENAI_API_KEY", cursor_environment)
|
|
|
|
with self.assertRaisesRegex(orc_profiles.ConfigError, "deb remote auth"):
|
|
orc_profiles.worker_environment(
|
|
"deb",
|
|
model_auth="codex-login",
|
|
remote_auth="forgejo-token",
|
|
)
|
|
|
|
with mock.patch.dict(os.environ, {}, clear=True):
|
|
with self.assertRaisesRegex(
|
|
orc_profiles.ConfigError,
|
|
"selected authentication variable is unavailable",
|
|
):
|
|
orc_profiles.worker_environment(
|
|
"release",
|
|
model_auth="openai",
|
|
remote_auth="none",
|
|
)
|
|
|
|
with mock.patch.dict(os.environ, {"OPENAI_API_KEY": ""}, clear=True):
|
|
with self.assertRaisesRegex(
|
|
orc_profiles.ConfigError,
|
|
"selected authentication variable is unavailable",
|
|
):
|
|
orc_profiles.worker_environment(
|
|
"release",
|
|
model_auth="openai",
|
|
remote_auth="none",
|
|
)
|
|
|
|
def test_release_remote_requires_one_identical_fetch_and_push_url(self) -> None:
|
|
git = shutil.which("git")
|
|
self.assertIsNotNone(git)
|
|
assert git is not None
|
|
git_path = Path(git).resolve()
|
|
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary).resolve()
|
|
subprocess.run(
|
|
[str(git_path), "init", "-q", str(root)],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
remote = "https://git.yumee.top/laily/musicpilot.git"
|
|
subprocess.run(
|
|
[str(git_path), "-C", str(root), "remote", "add", "origin", remote],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
with mock.patch.object(
|
|
orc_profiles,
|
|
"resolve_trusted_executable",
|
|
return_value=git_path,
|
|
):
|
|
facts = orc_profiles.resolve_release_remote(root)
|
|
|
|
self.assertEqual(facts["fetchUrl"], remote)
|
|
self.assertEqual(facts["pushUrl"], remote)
|
|
self.assertEqual(facts["host"], "git.yumee.top")
|
|
self.assertEqual(facts["networkHosts"], ["git.yumee.top"])
|
|
|
|
subprocess.run(
|
|
[
|
|
str(git_path),
|
|
"-C",
|
|
str(root),
|
|
"config",
|
|
"remote.origin.pushurl",
|
|
"https://git.example.invalid/other/repo.git",
|
|
],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
with mock.patch.object(
|
|
orc_profiles,
|
|
"resolve_trusted_executable",
|
|
return_value=git_path,
|
|
):
|
|
with self.assertRaisesRegex(
|
|
orc_profiles.ConfigError,
|
|
"fetch and push URLs must match",
|
|
):
|
|
orc_profiles.resolve_release_remote(root)
|
|
|
|
subprocess.run(
|
|
[
|
|
str(git_path),
|
|
"-C",
|
|
str(root),
|
|
"config",
|
|
"--add",
|
|
"remote.origin.pushurl",
|
|
"https://git.example.invalid/second/repo.git",
|
|
],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
with mock.patch.object(
|
|
orc_profiles,
|
|
"resolve_trusted_executable",
|
|
return_value=git_path,
|
|
):
|
|
with self.assertRaisesRegex(
|
|
orc_profiles.ConfigError,
|
|
"exactly one push URL",
|
|
):
|
|
orc_profiles.resolve_release_remote(root)
|
|
|
|
with self.assertRaisesRegex(orc_profiles.ConfigError, "safe remote URL"):
|
|
orc_profiles._remote_url_facts(
|
|
"https://git.example.invalid:notaport/repo.git",
|
|
"test remote",
|
|
)
|
|
scp = orc_profiles._remote_url_facts(
|
|
"git@git.yumee.top:laily/musicpilot.git",
|
|
"scp remote",
|
|
)
|
|
ssh = orc_profiles._remote_url_facts(
|
|
"ssh://git@git.yumee.top/laily/musicpilot.git",
|
|
"ssh remote",
|
|
)
|
|
self.assertEqual(scp["canonical"], ssh["canonical"])
|
|
with self.assertRaisesRegex(orc_profiles.ConfigError, "local host"):
|
|
orc_profiles._remote_url_facts(
|
|
"https://localhost/laily/musicpilot.git",
|
|
"local remote",
|
|
)
|
|
|
|
with mock.patch.object(
|
|
orc_profiles,
|
|
"_origin_urls",
|
|
side_effect=[
|
|
["https://github.com/example/project.git"],
|
|
["https://github.com/example/project.git"],
|
|
],
|
|
), mock.patch.object(
|
|
orc_profiles,
|
|
"resolve_trusted_executable",
|
|
return_value=git_path,
|
|
):
|
|
github = orc_profiles.resolve_release_remote(Path("/tmp/project"))
|
|
self.assertEqual(
|
|
github["networkHosts"],
|
|
["github.com", "api.github.com", "uploads.github.com"],
|
|
)
|
|
|
|
def test_ssh_auth_requires_a_trusted_user_socket(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
socket_path = root / "agent.sock"
|
|
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as agent:
|
|
agent.bind(str(socket_path))
|
|
socket_path.chmod(0o600)
|
|
with mock.patch.dict(
|
|
os.environ,
|
|
{"SSH_AUTH_SOCK": str(socket_path)},
|
|
clear=True,
|
|
):
|
|
environment = orc_profiles.worker_environment(
|
|
"release",
|
|
model_auth="codex-login",
|
|
remote_auth="ssh-agent",
|
|
)
|
|
self.assertEqual(environment["SSH_AUTH_SOCK"], str(socket_path))
|
|
|
|
regular_file = root / "not-a-socket"
|
|
regular_file.write_text("not a socket", encoding="utf-8")
|
|
regular_file.chmod(0o600)
|
|
with mock.patch.dict(
|
|
os.environ,
|
|
{"SSH_AUTH_SOCK": str(regular_file)},
|
|
clear=True,
|
|
):
|
|
with self.assertRaisesRegex(
|
|
orc_profiles.ConfigError,
|
|
"not a trusted user socket",
|
|
):
|
|
orc_profiles.worker_environment(
|
|
"release",
|
|
model_auth="codex-login",
|
|
remote_auth="ssh-agent",
|
|
)
|
|
|
|
def test_trusted_orca_ignores_ambient_path(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
trusted = root / "trusted"
|
|
hostile = root / "hostile"
|
|
trusted.mkdir()
|
|
hostile.mkdir()
|
|
trusted_orca = trusted / "orca"
|
|
hostile_orca = hostile / "orca"
|
|
for executable in (trusted_orca, hostile_orca):
|
|
executable.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
|
executable.chmod(0o700)
|
|
|
|
with (
|
|
mock.patch.object(
|
|
orc_profiles,
|
|
"trusted_path_entries",
|
|
return_value=[trusted],
|
|
),
|
|
mock.patch.dict(os.environ, {"PATH": str(hostile)}, clear=True),
|
|
):
|
|
resolved = orc_profiles.resolve_trusted_executable("orca")
|
|
|
|
self.assertEqual(resolved, trusted_orca)
|
|
|
|
def test_config_rejects_unsafe_or_ambiguous_profiles(self) -> None:
|
|
base = orc_profiles.load_config(CONFIG)
|
|
|
|
free_command = copy.deepcopy(base)
|
|
free_command["profiles"]["codex"]["low"]["command"] = "codex --dangerously-bypass"
|
|
with self.assertRaisesRegex(orc_profiles.ConfigError, "unknown fields"):
|
|
orc_profiles.validate_config(free_command)
|
|
|
|
full_access = copy.deepcopy(base)
|
|
full_access["profiles"]["codex"]["high"]["permissionMode"] = "danger-full-access"
|
|
with self.assertRaisesRegex(orc_profiles.ConfigError, "workspace-write in ORC v2"):
|
|
orc_profiles.validate_config(full_access)
|
|
|
|
read_only = copy.deepcopy(base)
|
|
read_only["profiles"]["codex"]["low"]["permissionMode"] = "read-only"
|
|
with self.assertRaisesRegex(orc_profiles.ConfigError, "workspace-write in ORC v2"):
|
|
orc_profiles.validate_config(read_only)
|
|
|
|
missing_level = copy.deepcopy(base)
|
|
del missing_level["profiles"]["cursor-agent"]["mid"]
|
|
with self.assertRaisesRegex(orc_profiles.ConfigError, "missing fields: mid"):
|
|
orc_profiles.validate_config(missing_level)
|
|
|
|
option_model = copy.deepcopy(base)
|
|
option_model["profiles"]["codex"]["low"]["model"] = "--model"
|
|
with self.assertRaisesRegex(orc_profiles.ConfigError, "safe exact model ID"):
|
|
orc_profiles.validate_config(option_model)
|
|
|
|
list_policy = copy.deepcopy(base)
|
|
list_policy["profiles"]["codex"]["low"]["approvalPolicy"] = ["never"]
|
|
with self.assertRaisesRegex(orc_profiles.ConfigError, "approvalPolicy"):
|
|
orc_profiles.validate_config(list_policy)
|
|
|
|
cursor_effort = copy.deepcopy(base)
|
|
cursor_effort["profiles"]["cursor-agent"]["low"]["reasoningEffort"] = "low"
|
|
with self.assertRaisesRegex(orc_profiles.ConfigError, "Cursor requires null"):
|
|
orc_profiles.validate_config(cursor_effort)
|
|
|
|
mislabeled_cursor = copy.deepcopy(base)
|
|
mislabeled_cursor["profiles"]["cursor-agent"]["low"]["model"] = (
|
|
"gpt-5.6-sol-high"
|
|
)
|
|
with self.assertRaisesRegex(orc_profiles.ConfigError, "encode the low"):
|
|
orc_profiles.validate_config(mislabeled_cursor)
|
|
|
|
project_specific = copy.deepcopy(base)
|
|
project_specific["worktreePolicy"] = ["."]
|
|
with self.assertRaisesRegex(
|
|
orc_profiles.ConfigError,
|
|
"registered-same-repository",
|
|
):
|
|
orc_profiles.validate_config(project_specific)
|
|
|
|
default_cli = copy.deepcopy(base)
|
|
default_cli["cliPolicy"] = "default-codex"
|
|
with self.assertRaisesRegex(orc_profiles.ConfigError, "current-host"):
|
|
orc_profiles.validate_config(default_cli)
|
|
|
|
def test_config_reader_rejects_symlinks_and_redacts_parser_input(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
real = root / "real.yaml"
|
|
real.write_text(CONFIG.read_text(encoding="utf-8"), encoding="utf-8")
|
|
linked = root / "linked.yaml"
|
|
linked.symlink_to(real)
|
|
with self.assertRaisesRegex(orc_profiles.ConfigError, "safely read"):
|
|
orc_profiles.load_config(linked)
|
|
|
|
malformed = root / "malformed.yaml"
|
|
secret_marker = "PRIVATE_MATERIAL_MUST_NOT_APPEAR"
|
|
malformed.write_text(f"{secret_marker}: [unterminated\n", encoding="utf-8")
|
|
with self.assertRaises(orc_profiles.ConfigError) as error:
|
|
orc_profiles.load_config(malformed)
|
|
self.assertNotIn(secret_marker, str(error.exception))
|
|
|
|
def test_bound_plan_enforces_registered_same_repository_worktree(self) -> None:
|
|
git = shutil.which("git")
|
|
self.assertIsNotNone(git)
|
|
assert git is not None
|
|
git_path = Path(git).resolve()
|
|
|
|
with tempfile.TemporaryDirectory(prefix="orc shell $(id) ' ") as temporary:
|
|
temporary_root = Path(temporary).resolve()
|
|
root = temporary_root / "project"
|
|
root.mkdir()
|
|
subprocess.run(
|
|
[str(git_path), "init", "-q", str(root)],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
subprocess.run(
|
|
[
|
|
str(git_path),
|
|
"-C",
|
|
str(root),
|
|
"remote",
|
|
"add",
|
|
"origin",
|
|
"https://git.yumee.top/laily/project.git",
|
|
],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
fake_codex = root / "trusted-codex" / "codex"
|
|
fake_codex.parent.mkdir()
|
|
fake_codex.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
|
fake_codex.chmod(0o700)
|
|
fake_cursor = root / "trusted-cursor" / "cursor-agent"
|
|
fake_cursor.parent.mkdir()
|
|
fake_cursor.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
|
fake_cursor.chmod(0o700)
|
|
fake_orca = root / "trusted-orca" / "orca"
|
|
fake_orca.parent.mkdir()
|
|
fake_orca.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
|
fake_orca.chmod(0o700)
|
|
|
|
def executable(name: str) -> Path:
|
|
return {
|
|
"git": git_path,
|
|
"codex": fake_codex,
|
|
"cursor-agent": fake_cursor,
|
|
"orca": fake_orca,
|
|
}[name]
|
|
|
|
def executable_facts(path: Path) -> dict[str, object]:
|
|
return {
|
|
"path": str(path),
|
|
"device": 1,
|
|
"inode": 2,
|
|
"size": 3,
|
|
"mtimeNs": 4,
|
|
"version": f"{path.name}-test 1",
|
|
}
|
|
|
|
with (
|
|
mock.patch.object(
|
|
orc_profiles,
|
|
"resolve_trusted_executable",
|
|
side_effect=executable,
|
|
),
|
|
mock.patch.object(
|
|
orc_profiles,
|
|
"resolve_trusted_python",
|
|
return_value=Path(sys.executable).resolve(),
|
|
),
|
|
mock.patch.object(
|
|
orc_profiles,
|
|
"_executable_facts",
|
|
side_effect=executable_facts,
|
|
),
|
|
):
|
|
plan = orc_profiles.build_launch_plan(
|
|
project_root=root,
|
|
worktree=root,
|
|
stage="code",
|
|
host_cli="codex",
|
|
stage_level="high",
|
|
)
|
|
|
|
cursor_plan = orc_profiles.build_launch_plan(
|
|
project_root=root,
|
|
worktree=root,
|
|
stage="release",
|
|
host_cli="cursor-agent",
|
|
stage_level="low",
|
|
)
|
|
|
|
release_plan = orc_profiles.build_launch_plan(
|
|
project_root=root,
|
|
worktree=root,
|
|
stage="release",
|
|
host_cli="codex",
|
|
stage_level="low",
|
|
)
|
|
|
|
self.assertEqual(plan["argv"][0], str(fake_codex))
|
|
self.assertEqual(plan["config"]["path"], str(CONFIG))
|
|
self.assertEqual(cursor_plan["argv"][0], str(fake_cursor))
|
|
self.assertEqual(cursor_plan["executable"]["path"], str(fake_cursor))
|
|
self.assertEqual(cursor_plan["argv"][1:3], ["--model", "auto"])
|
|
self.assertIn("--auto-review", cursor_plan["argv"])
|
|
self.assertEqual(cursor_plan["modelAuth"], "cursor-login")
|
|
self.assertEqual(cursor_plan["releaseRemote"]["host"], "git.yumee.top")
|
|
self.assertIn("--host-cli", cursor_plan["launcherArgv"])
|
|
release_args = release_plan["workerArgs"]
|
|
self.assertIn("sandbox_workspace_write.network_access=true", release_args)
|
|
self.assertIn("features.network_proxy.enabled=true", release_args)
|
|
self.assertIn("features.network_proxy.allow_upstream_proxy=false", release_args)
|
|
self.assertIn("features.network_proxy.unix_sockets={}", release_args)
|
|
self.assertIn(
|
|
'features.network_proxy.domains={ "git.yumee.top" = "allow" }',
|
|
release_args,
|
|
)
|
|
release_approval = release_args.index("--ask-for-approval")
|
|
self.assertEqual(release_args[release_approval + 1], "on-request")
|
|
self.assertIn('approvals_reviewer="auto_review"', release_args)
|
|
self.assertEqual(plan["terminalCreateArgv"][0], str(fake_orca))
|
|
self.assertEqual(plan["worktree"], str(root))
|
|
self.assertEqual(plan["selectionSource"], "request.stage")
|
|
self.assertTrue(plan["launchFingerprint"].startswith("sha256:"))
|
|
self.assertIn("_launch", plan["launcherArgv"])
|
|
self.assertEqual(plan["launcherArgv"][1:3], ["-I", "-S"])
|
|
self.assertIn("python", plan)
|
|
self.assertEqual(plan["worktreeSelector"], f"path:{root}")
|
|
self.assertNotIn("PRIVATE", json.dumps(plan))
|
|
self.assertEqual(shlex.split(plan["terminalCommand"]), plan["launcherArgv"])
|
|
self.assertEqual(
|
|
shlex.split(plan["terminalCreateShellCommand"]),
|
|
plan["terminalCreateArgv"],
|
|
)
|
|
command_index = plan["terminalCreateArgv"].index("--command")
|
|
self.assertEqual(
|
|
plan["terminalCreateArgv"][command_index + 1],
|
|
plan["terminalCommand"],
|
|
)
|
|
|
|
outside = temporary_root / "outside"
|
|
outside.mkdir()
|
|
subprocess.run(
|
|
[str(git_path), "init", "-q", str(outside)],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
with mock.patch.object(
|
|
orc_profiles,
|
|
"resolve_trusted_executable",
|
|
side_effect=executable,
|
|
):
|
|
with self.assertRaisesRegex(
|
|
orc_profiles.ConfigError,
|
|
"not registered in the project repository",
|
|
):
|
|
orc_profiles.validate_worktree(
|
|
orc_profiles.load_config(CONFIG),
|
|
project_root_value=root,
|
|
worktree_value=outside,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|