🚩 gremlins: Windows-only gate in where, init and launches
This commit is contained in:
@@ -119,6 +119,14 @@ USER_SETTINGS_PATH: str | None = None
|
|||||||
PHASE2_COMMANDS = () # commands that tolerate unknown arguments (none: all are implemented)
|
PHASE2_COMMANDS = () # commands that tolerate unknown arguments (none: all are implemented)
|
||||||
# N3: a gremlin may only run where, list, usage and wait (`tick` launches pings).
|
# N3: a gremlin may only run where, list, usage and wait (`tick` launches pings).
|
||||||
CHILD_REFUSED = ("train", "spawn", "continue", "rebase", "kill", "expire", "init", "tick")
|
CHILD_REFUSED = ("train", "spawn", "continue", "rebase", "kill", "expire", "init", "tick")
|
||||||
|
# The platform gate (digest §4, amendment A1): on an unsupported platform these launch nothing.
|
||||||
|
# `init` checks the gate itself, as its first step (B1); `where` reports it.
|
||||||
|
GATED_COMMANDS = ("train", "spawn", "continue", "rebase", "tick")
|
||||||
|
# Porting aid only (C12, documented in the platform handoff, not in SKILL.md): `=1` opens the
|
||||||
|
# gate for that call on an unsupported platform.
|
||||||
|
UNSUPPORTED_OK_ENV = "SPIRE_GREMLIN_UNSUPPORTED_OK"
|
||||||
|
GATE_OVERRIDE_WARNING = ("platform gate overridden (%s=1); this platform is untested, use it only "
|
||||||
|
"to port and test spire:gremlins" % UNSUPPORTED_OK_ENV)
|
||||||
|
|
||||||
# Warmth and keep-warm (design 5, block 2.1). Module-level so tests can shorten them.
|
# Warmth and keep-warm (design 5, block 2.1). Module-level so tests can shorten them.
|
||||||
TICK_SUP_S = 60.0 # a supervisor runs one tick pass this often
|
TICK_SUP_S = 60.0 # a supervisor runs one tick pass this often
|
||||||
@@ -726,6 +734,38 @@ PLATFORM = select_platform()
|
|||||||
# --- platform seam: end ---------------------------------------------------------------
|
# --- platform seam: end ---------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------------------
|
||||||
|
# Platform gate (digest §4): fail closed where the helper is untested
|
||||||
|
# --------------------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def gate_overridden() -> bool:
|
||||||
|
"""A12: the override counts only where the gate is closed (on a supported platform the env
|
||||||
|
var changes nothing)."""
|
||||||
|
return os.environ.get(UNSUPPORTED_OK_ENV) == "1" and not PLATFORM.supported
|
||||||
|
|
||||||
|
|
||||||
|
def platform_report() -> dict:
|
||||||
|
"""`where`'s `platform` block: `PLATFORM.describe()` plus `override`."""
|
||||||
|
return {**PLATFORM.describe(), "override": gate_overridden()}
|
||||||
|
|
||||||
|
|
||||||
|
def platform_gate() -> str | None:
|
||||||
|
"""None when the platform is supported or the gate is overridden; else the reason."""
|
||||||
|
if PLATFORM.supported or gate_overridden():
|
||||||
|
return None
|
||||||
|
d = PLATFORM.describe()
|
||||||
|
return ("spire:gremlins is not tested on %s%s yet; use fresh agents for every block "
|
||||||
|
"(see SKILL.md, Platforms)." % (d.get("sys_platform"), ", WSL" if d.get("wsl") else ""))
|
||||||
|
|
||||||
|
|
||||||
|
def require_platform() -> None:
|
||||||
|
"""Refuse (exit 2) on a closed gate: `init` and the launching commands (A1)."""
|
||||||
|
reason = platform_gate()
|
||||||
|
if reason is not None:
|
||||||
|
raise GremlinsError("platform not supported", reason=reason, platform=platform_report())
|
||||||
|
|
||||||
|
|
||||||
# --------------------------------------------------------------------------------------
|
# --------------------------------------------------------------------------------------
|
||||||
# Paths
|
# Paths
|
||||||
# --------------------------------------------------------------------------------------
|
# --------------------------------------------------------------------------------------
|
||||||
@@ -2216,14 +2256,28 @@ def build_parser() -> _Parser:
|
|||||||
|
|
||||||
|
|
||||||
def cmd_where(args) -> tuple[dict, int]:
|
def cmd_where(args) -> tuple[dict, int]:
|
||||||
|
"""The preflight report. On a closed platform gate: `ok: false` with the gate reason first,
|
||||||
|
and every other part still filled (the porting hosts need it)."""
|
||||||
reasons = []
|
reasons = []
|
||||||
|
gate = platform_gate()
|
||||||
|
if gate is not None:
|
||||||
|
reasons.append(gate)
|
||||||
py_ok = sys.version_info >= (3, 10)
|
py_ok = sys.version_info >= (3, 10)
|
||||||
if not py_ok:
|
if not py_ok:
|
||||||
reasons.append("python < 3.10")
|
reasons.append("python < 3.10")
|
||||||
py = {"version": platform.python_version(), "executable": sys.executable,
|
py = {"version": platform.python_version(), "executable": sys.executable,
|
||||||
"base_executable": getattr(sys, "_base_executable", sys.executable),
|
"base_executable": getattr(sys, "_base_executable", sys.executable),
|
||||||
"supervisor_interpreter": supervisor_interpreter(), "ok": py_ok}
|
"supervisor_interpreter": supervisor_interpreter(), "ok": py_ok}
|
||||||
|
try:
|
||||||
path, how = resolve_binary(args.binary)
|
path, how = resolve_binary(args.binary)
|
||||||
|
except GremlinsError as e: # A11: a path form this platform refuses
|
||||||
|
path = None
|
||||||
|
how = ("--binary" if args.binary else
|
||||||
|
"CLAUDE_CODE_EXECPATH" if os.environ.get("CLAUDE_CODE_EXECPATH") else "PATH")
|
||||||
|
binary = {"path": path, "how": how, "error": e.payload["error"],
|
||||||
|
"given": e.payload.get("path", args.binary)}
|
||||||
|
reasons.append("%s: %s" % (e.payload["error"], binary["given"]))
|
||||||
|
else:
|
||||||
binary = {"path": path, "how": how}
|
binary = {"path": path, "how": how}
|
||||||
if path is None:
|
if path is None:
|
||||||
reasons.append("no claude binary (no --binary, no $CLAUDE_CODE_EXECPATH, none on PATH)")
|
reasons.append("no claude binary (no --binary, no $CLAUDE_CODE_EXECPATH, none on PATH)")
|
||||||
@@ -2239,6 +2293,8 @@ def cmd_where(args) -> tuple[dict, int]:
|
|||||||
warnings.append("ANTHROPIC_BASE_URL is %r, not the default %s; it is passed to every "
|
warnings.append("ANTHROPIC_BASE_URL is %r, not the default %s; it is passed to every "
|
||||||
"base and gremlin, so they authenticate and bill through that endpoint"
|
"base and gremlin, so they authenticate and bill through that endpoint"
|
||||||
% (base_url, DEFAULT_BASE_URL))
|
% (base_url, DEFAULT_BASE_URL))
|
||||||
|
if gate_overridden():
|
||||||
|
warnings.append(GATE_OVERRIDE_WARNING)
|
||||||
out = {
|
out = {
|
||||||
"script": script_path(),
|
"script": script_path(),
|
||||||
"python": py,
|
"python": py,
|
||||||
@@ -2246,7 +2302,7 @@ def cmd_where(args) -> tuple[dict, int]:
|
|||||||
"binary": binary,
|
"binary": binary,
|
||||||
"git": _git_report(),
|
"git": _git_report(),
|
||||||
"scrub": scrub_report(),
|
"scrub": scrub_report(),
|
||||||
"platform": PLATFORM.describe(),
|
"platform": platform_report(),
|
||||||
"warnings": warnings,
|
"warnings": warnings,
|
||||||
"ok": not reasons,
|
"ok": not reasons,
|
||||||
}
|
}
|
||||||
@@ -2417,6 +2473,7 @@ def _diff_fields(a, b, prefix="") -> list[str]:
|
|||||||
|
|
||||||
|
|
||||||
def cmd_init(args) -> tuple[dict, int]:
|
def cmd_init(args) -> tuple[dict, int]:
|
||||||
|
require_platform() # B1: the first step, before any path is read or written
|
||||||
state = state_dir_for(args.run)
|
state = state_dir_for(args.run)
|
||||||
snap = read_registry(state) if os.path.isdir(state) else None
|
snap = read_registry(state) if os.path.isdir(state) else None
|
||||||
old_cfg = (snap or {}).get("config")
|
old_cfg = (snap or {}).get("config")
|
||||||
@@ -3250,7 +3307,10 @@ def tick_pass(state: str) -> dict:
|
|||||||
"""One pass of the ping rule over every base of the run. The decision and the `ping_in_flight`
|
"""One pass of the ping rule over every base of the run. The decision and the `ping_in_flight`
|
||||||
mark are made under the registry lock (C3d), so concurrent tickers launch one ping per base.
|
mark are made under the registry lock (C3d), so concurrent tickers launch one ping per base.
|
||||||
The binary check runs before the lock for the bases that look eligible. Pings go through the
|
The binary check runs before the lock for the bases that look eligible. Pings go through the
|
||||||
same hidden spawn as any job. Returns {"pinged": [job...], "bases": [decision...]}."""
|
same hidden spawn as any job. Returns {"pinged": [job...], "bases": [decision...]}.
|
||||||
|
On a closed platform gate (A1) it returns at once: no ping, nothing read or written."""
|
||||||
|
if platform_gate() is not None:
|
||||||
|
return {"pinged": [], "bases": []}
|
||||||
reg = read_registry(state)
|
reg = read_registry(state)
|
||||||
if not reg or not reg.get("config"):
|
if not reg or not reg.get("config"):
|
||||||
return {"pinged": [], "bases": []}
|
return {"pinged": [], "bases": []}
|
||||||
@@ -3977,6 +4037,8 @@ def run(argv) -> tuple[dict, int]:
|
|||||||
if args.command in CHILD_REFUSED and in_child():
|
if args.command in CHILD_REFUSED and in_child():
|
||||||
raise GremlinsError("`%s` is refused inside a gremlin (%s is set); a gremlin may only run where, "
|
raise GremlinsError("`%s` is refused inside a gremlin (%s is set); a gremlin may only run where, "
|
||||||
"list, usage and wait" % (args.command, CHILD_MARKER), command=args.command)
|
"list, usage and wait" % (args.command, CHILD_MARKER), command=args.command)
|
||||||
|
if args.command in GATED_COMMANDS:
|
||||||
|
require_platform() # A1: an existing registry launches nothing on a closed gate
|
||||||
return HANDLERS[args.command](args)
|
return HANDLERS[args.command](args)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -71,6 +71,12 @@ def child_env():
|
|||||||
return env
|
return env
|
||||||
|
|
||||||
|
|
||||||
|
def gate_warning(out: dict) -> list[str]:
|
||||||
|
"""The warning `where` adds when the platform gate is overridden (C13 on a closed platform;
|
||||||
|
never on Windows), so that the exact `warnings` assertions hold on every host."""
|
||||||
|
return [gremlins.GATE_OVERRIDE_WARNING] if out["platform"]["override"] else []
|
||||||
|
|
||||||
|
|
||||||
class Base(unittest.TestCase):
|
class Base(unittest.TestCase):
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
self.tmp = tempfile.mkdtemp(prefix="gremlins-test-")
|
self.tmp = tempfile.mkdtemp(prefix="gremlins-test-")
|
||||||
@@ -79,6 +85,9 @@ class Base(unittest.TestCase):
|
|||||||
env_patch.start()
|
env_patch.start()
|
||||||
self.addCleanup(env_patch.stop)
|
self.addCleanup(env_patch.stop)
|
||||||
os.environ.pop(gremlins.CHILD_MARKER, None)
|
os.environ.pop(gremlins.CHILD_MARKER, None)
|
||||||
|
# C13: the suite runs with the platform gate overridden, so it exercises the POSIX code on
|
||||||
|
# the porting hosts; on Windows (supported) the variable changes nothing. GateTests drop it.
|
||||||
|
os.environ[gremlins.UNSUPPORTED_OK_ENV] = "1"
|
||||||
|
|
||||||
def assert_one_json_line(self, raw: str):
|
def assert_one_json_line(self, raw: str):
|
||||||
self.assertTrue(raw.endswith("\n"), repr(raw[-40:]))
|
self.assertTrue(raw.endswith("\n"), repr(raw[-40:]))
|
||||||
@@ -1086,15 +1095,51 @@ class CliTests(Base):
|
|||||||
code, out, raw = call(["where", "--binary", stub])
|
code, out, raw = call(["where", "--binary", stub])
|
||||||
self.assertEqual(code, 0, out)
|
self.assertEqual(code, 0, out)
|
||||||
self.assert_one_json_line(raw)
|
self.assert_one_json_line(raw)
|
||||||
self.assertEqual(out["warnings"], [])
|
self.assertEqual(out["warnings"], gate_warning(out))
|
||||||
os.environ["ANTHROPIC_BASE_URL"] = "https://llm-gateway.example.internal/anthropic"
|
os.environ["ANTHROPIC_BASE_URL"] = "https://llm-gateway.example.internal/anthropic"
|
||||||
code, out, raw = call(["where", "--binary", stub])
|
code, out, raw = call(["where", "--binary", stub])
|
||||||
self.assertEqual(code, 0, out)
|
self.assertEqual(code, 0, out)
|
||||||
self.assert_one_json_line(raw)
|
self.assert_one_json_line(raw)
|
||||||
self.assertTrue(out["ok"])
|
self.assertTrue(out["ok"])
|
||||||
self.assertEqual(len(out["warnings"]), 1, out["warnings"])
|
self.assertEqual(len(out["warnings"]), 1 + len(gate_warning(out)), out["warnings"])
|
||||||
self.assertIn("ANTHROPIC_BASE_URL", out["warnings"][0])
|
self.assertIn("ANTHROPIC_BASE_URL", out["warnings"][0])
|
||||||
self.assertIn("llm-gateway.example.internal", out["warnings"][0])
|
self.assertIn("llm-gateway.example.internal", out["warnings"][0])
|
||||||
|
self.assertEqual(out["warnings"][1:], gate_warning(out))
|
||||||
|
|
||||||
|
def test_where_platform(self):
|
||||||
|
"""Digest §3.6/§4, A12: `where.platform` is structured; `override` is false on a supported
|
||||||
|
platform even though the suite sets the override env (C13)."""
|
||||||
|
stub = make_stub_binary(self.tmp)
|
||||||
|
self.assertEqual(os.environ.get(gremlins.UNSUPPORTED_OK_ENV), "1")
|
||||||
|
code, out, raw = call(["where", "--binary", stub])
|
||||||
|
self.assertEqual(code, 0, out)
|
||||||
|
self.assert_one_json_line(raw)
|
||||||
|
self.assertNotIn("spawn", out)
|
||||||
|
pf = out["platform"]
|
||||||
|
self.assertEqual(set(pf), {"name", "sys_platform", "wsl", "supported", "detach", "override"})
|
||||||
|
self.assertEqual(pf["sys_platform"], sys.platform)
|
||||||
|
self.assertEqual(pf["override"], not gremlins.PLATFORM.supported)
|
||||||
|
if WIN_PLATFORM:
|
||||||
|
self.assertEqual(pf, {"name": "windows", "sys_platform": sys.platform, "wsl": False,
|
||||||
|
"supported": True, "detach": "cim-hidden", "override": False})
|
||||||
|
self.assertEqual(out["warnings"], [])
|
||||||
|
|
||||||
|
def test_where_refused_binary_path(self):
|
||||||
|
"""A11: a `--binary` path form this platform refuses lands in `binary.error` and the
|
||||||
|
reasons (exit 2); the rest of the report is still filled."""
|
||||||
|
bad = "/mnt/c/x/claude.exe" if WIN_PLATFORM else "C:\\x\\claude.exe"
|
||||||
|
code, out, raw = call(["where", "--binary", bad])
|
||||||
|
self.assertEqual(code, 2, out)
|
||||||
|
self.assert_one_json_line(raw)
|
||||||
|
self.assertFalse(out["ok"])
|
||||||
|
self.assertIsNone(out["binary"]["path"])
|
||||||
|
self.assertEqual(out["binary"]["how"], "--binary")
|
||||||
|
self.assertEqual(out["binary"]["given"], bad)
|
||||||
|
self.assertIn("path form", out["binary"]["error"])
|
||||||
|
self.assertIn(out["binary"]["error"], out["reason"])
|
||||||
|
self.assertEqual(out["error"], out["reason"])
|
||||||
|
for k in ("script", "python", "uv", "git", "scrub", "platform", "warnings"):
|
||||||
|
self.assertIn(k, out)
|
||||||
|
|
||||||
def test_child_refuses_kill_expire_init(self):
|
def test_child_refuses_kill_expire_init(self):
|
||||||
"""M4: a gremlin may not run kill, expire or init."""
|
"""M4: a gremlin may not run kill, expire or init."""
|
||||||
@@ -1142,6 +1187,175 @@ class CliTests(Base):
|
|||||||
self.assertIn("error", out)
|
self.assertIn("error", out)
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------------------
|
||||||
|
def closed_platform():
|
||||||
|
"""B1: a fake closed platform. It subclasses the host's own platform class (on Windows a
|
||||||
|
`WindowsPlatform`), so path handling stays the host's and `--binary`, `--run` and
|
||||||
|
`script_path()` work; it reports itself as WSL Linux and unsupported."""
|
||||||
|
host = type(gremlins.PLATFORM)
|
||||||
|
|
||||||
|
class ClosedPlatform(host):
|
||||||
|
supported = False
|
||||||
|
|
||||||
|
def is_wsl(self):
|
||||||
|
return True
|
||||||
|
|
||||||
|
def describe(self):
|
||||||
|
return {"name": "posix", "sys_platform": "linux", "wsl": True, "supported": False,
|
||||||
|
"detach": "setsid"}
|
||||||
|
return ClosedPlatform()
|
||||||
|
|
||||||
|
|
||||||
|
class GateTests(Base):
|
||||||
|
"""Digest §4 with amendments B1, A1, A7, A12: the Windows-only gate."""
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
super().setUp()
|
||||||
|
os.environ.pop(gremlins.UNSUPPORTED_OK_ENV, None) # C13: the gate tests run without it
|
||||||
|
self.cwd = os.path.join(self.tmp, "proj")
|
||||||
|
os.makedirs(self.cwd)
|
||||||
|
os.makedirs(os.path.join(self.tmp, "reports"))
|
||||||
|
self.run_log = os.path.join(self.tmp, "reports", "2026-10-16-20-g-run.md")
|
||||||
|
self.state = gremlins.state_dir_for(self.run_log)
|
||||||
|
p = mock.patch.object(gremlins, "USER_SETTINGS_PATH", os.path.join(self.tmp, "home", "settings.json"))
|
||||||
|
p.start()
|
||||||
|
self.addCleanup(p.stop)
|
||||||
|
self.stub = make_stub_binary(self.tmp)
|
||||||
|
|
||||||
|
def close_gate(self):
|
||||||
|
p = mock.patch.object(gremlins, "PLATFORM", closed_platform())
|
||||||
|
p.start()
|
||||||
|
self.addCleanup(p.stop)
|
||||||
|
|
||||||
|
def init(self):
|
||||||
|
return call(["init", "--run", self.run_log, "--model", "claude-opus-5-5[1m]", "--cwd", self.cwd,
|
||||||
|
"--binary", self.stub])
|
||||||
|
|
||||||
|
def test_supported_flags(self):
|
||||||
|
self.assertIs(gremlins.WindowsPlatform().supported, True)
|
||||||
|
for sp in ("linux", "darwin"):
|
||||||
|
with self.subTest(sys_platform=sp), mock.patch.object(sys, "platform", sp):
|
||||||
|
pp = gremlins.PosixPlatform()
|
||||||
|
self.assertIs(pp.supported, False)
|
||||||
|
self.assertIs(pp.describe()["supported"], False) # A12: read from `supported`
|
||||||
|
self.assertEqual(pp.describe()["sys_platform"], sp)
|
||||||
|
|
||||||
|
def test_is_wsl(self):
|
||||||
|
pp = gremlins.PosixPlatform()
|
||||||
|
os.environ["WSL_DISTRO_NAME"] = "Ubuntu-20.04"
|
||||||
|
self.assertTrue(pp.is_wsl())
|
||||||
|
os.environ.pop("WSL_DISTRO_NAME")
|
||||||
|
with mock.patch("builtins.open", mock.mock_open(read_data=b"5.15.90.1-microsoft-standard-WSL2")):
|
||||||
|
self.assertTrue(pp.is_wsl())
|
||||||
|
with mock.patch("builtins.open", mock.mock_open(read_data=b"6.8.0-45-generic")):
|
||||||
|
self.assertFalse(pp.is_wsl())
|
||||||
|
with mock.patch("builtins.open", side_effect=OSError("no /proc")):
|
||||||
|
self.assertFalse(pp.is_wsl())
|
||||||
|
|
||||||
|
def test_gate_reason(self):
|
||||||
|
"""A7: no version number; names the platform and WSL; A12: read from describe()."""
|
||||||
|
if gremlins.PLATFORM.supported:
|
||||||
|
self.assertIsNone(gremlins.platform_gate())
|
||||||
|
self.close_gate()
|
||||||
|
self.assertEqual(gremlins.platform_gate(),
|
||||||
|
"spire:gremlins is not tested on linux, WSL yet; use fresh agents for every "
|
||||||
|
"block (see SKILL.md, Platforms).")
|
||||||
|
os.environ[gremlins.UNSUPPORTED_OK_ENV] = "1"
|
||||||
|
self.assertIsNone(gremlins.platform_gate())
|
||||||
|
self.assertTrue(gremlins.gate_overridden())
|
||||||
|
|
||||||
|
def test_host_where(self):
|
||||||
|
"""On the real host without the override env: Windows is supported, no override."""
|
||||||
|
code, out, raw = call(["where", "--binary", self.stub])
|
||||||
|
self.assert_one_json_line(raw)
|
||||||
|
if WIN_PLATFORM:
|
||||||
|
self.assertEqual(code, 0, out)
|
||||||
|
self.assertTrue(out["platform"]["supported"])
|
||||||
|
self.assertFalse(out["platform"]["override"])
|
||||||
|
self.assertEqual(out["warnings"], [])
|
||||||
|
|
||||||
|
def test_closed_where(self):
|
||||||
|
self.close_gate()
|
||||||
|
code, out, raw = call(["where", "--binary", self.stub])
|
||||||
|
self.assertEqual(code, 2, out)
|
||||||
|
self.assert_one_json_line(raw)
|
||||||
|
self.assertFalse(out["ok"])
|
||||||
|
self.assertIn("not tested on linux, WSL", out["reason"])
|
||||||
|
self.assertIn("fresh agents", out["reason"])
|
||||||
|
self.assertEqual(out["error"], out["reason"])
|
||||||
|
self.assertEqual(out["platform"], {"name": "posix", "sys_platform": "linux", "wsl": True,
|
||||||
|
"supported": False, "detach": "setsid", "override": False})
|
||||||
|
# everything else is still filled (the porting hosts need it)
|
||||||
|
self.assertEqual(out["binary"]["version"], STUB_VERSION)
|
||||||
|
self.assertNotIn("error", out["binary"])
|
||||||
|
for k in ("script", "python", "uv", "git", "scrub", "warnings"):
|
||||||
|
self.assertIn(k, out)
|
||||||
|
self.assertEqual(out["warnings"], [])
|
||||||
|
|
||||||
|
def test_closed_init(self):
|
||||||
|
self.close_gate()
|
||||||
|
code, out, raw = self.init()
|
||||||
|
self.assertEqual(code, 2, out)
|
||||||
|
self.assert_one_json_line(raw)
|
||||||
|
self.assertEqual(out["error"], "platform not supported")
|
||||||
|
self.assertIn("not tested on linux, WSL", out["reason"])
|
||||||
|
self.assertEqual(out["platform"]["supported"], False)
|
||||||
|
self.assertEqual(out["platform"]["override"], False)
|
||||||
|
self.assertFalse(os.path.exists(self.state))
|
||||||
|
# B1: the gate comes before any path handling (a bad --run is not even looked at)
|
||||||
|
code, out, _ = call(["init", "--run", os.path.join(self.tmp, "not-a-run.txt"), "--model",
|
||||||
|
"claude-opus-5-5[1m]", "--cwd", self.cwd, "--binary", self.stub])
|
||||||
|
self.assertEqual((code, out["error"]), (2, "platform not supported"))
|
||||||
|
|
||||||
|
def test_override(self):
|
||||||
|
self.close_gate()
|
||||||
|
os.environ[gremlins.UNSUPPORTED_OK_ENV] = "1"
|
||||||
|
code, out, raw = call(["where", "--binary", self.stub])
|
||||||
|
self.assertEqual(code, 0, out)
|
||||||
|
self.assert_one_json_line(raw)
|
||||||
|
self.assertTrue(out["ok"])
|
||||||
|
self.assertTrue(out["platform"]["override"])
|
||||||
|
self.assertFalse(out["platform"]["supported"])
|
||||||
|
self.assertEqual(out["warnings"], [gremlins.GATE_OVERRIDE_WARNING])
|
||||||
|
self.assertIn(gremlins.UNSUPPORTED_OK_ENV, out["warnings"][0])
|
||||||
|
code, out, _ = self.init()
|
||||||
|
self.assertEqual(code, 0, out)
|
||||||
|
self.assertTrue(out["created"])
|
||||||
|
self.assertTrue(os.path.isfile(gremlins.registry_path(self.state)))
|
||||||
|
|
||||||
|
def test_closed_gate_on_existing_registry(self):
|
||||||
|
"""A1: a registry made with the gate open; with it closed, nothing launches and no job
|
||||||
|
entry is written; the tick launches no ping (it reads nothing)."""
|
||||||
|
code, out, _ = self.init()
|
||||||
|
self.assertEqual(code, 0, out)
|
||||||
|
before = gremlins.read_registry(self.state)
|
||||||
|
brief = os.path.join(self.tmp, "b.md")
|
||||||
|
with open(brief, "w") as f:
|
||||||
|
f.write("x\n")
|
||||||
|
self.close_gate()
|
||||||
|
for argv in (["spawn", "--base", "B1", "--block", "1.1", "--brief", brief],
|
||||||
|
["train", "--base", "B1", "--brief", brief],
|
||||||
|
["continue", "--job", "j0001", "--brief", brief],
|
||||||
|
["rebase", "--base", "B1", "--as", "B2", "--brief", brief],
|
||||||
|
["tick"]):
|
||||||
|
with self.subTest(cmd=argv[0]):
|
||||||
|
code, out, raw = call([argv[0], "--run", self.run_log, *argv[1:]])
|
||||||
|
self.assertEqual(code, 2, out)
|
||||||
|
self.assert_one_json_line(raw)
|
||||||
|
self.assertEqual(out["error"], "platform not supported")
|
||||||
|
self.assertIn("not tested on linux, WSL", out["reason"])
|
||||||
|
self.assertEqual(out["platform"]["supported"], False)
|
||||||
|
after = gremlins.read_registry(self.state)
|
||||||
|
self.assertEqual(after["jobs"], {})
|
||||||
|
self.assertEqual(after, before)
|
||||||
|
with mock.patch.object(gremlins, "read_registry", side_effect=AssertionError("tick read the registry")):
|
||||||
|
self.assertEqual(gremlins.tick_pass(self.state), {"pinged": [], "bases": []})
|
||||||
|
# list and wait still work on a closed gate
|
||||||
|
self.assertEqual(call(["list", "--run", self.run_log])[0], 0)
|
||||||
|
code, out, _ = call(["wait", "--run", self.run_log, "--all", "--timeout", "1"])
|
||||||
|
self.assertEqual((code, out["pings"]), (0, []))
|
||||||
|
|
||||||
|
|
||||||
# --------------------------------------------------------------------------------------
|
# --------------------------------------------------------------------------------------
|
||||||
@unittest.skipUnless(shutil.which("uv"), "uv not on PATH")
|
@unittest.skipUnless(shutil.which("uv"), "uv not on PATH")
|
||||||
class SubprocessTests(Base):
|
class SubprocessTests(Base):
|
||||||
|
|||||||
Reference in New Issue
Block a user