"""Offline scheduler fixtures: no actual allocation, inference or cluster contact.""" from __future__ import annotations import importlib.util import json from pathlib import Path import subprocess import tarfile from types import SimpleNamespace import pytest SCRIPT = Path(__file__).resolve().parents[1] / "scripts/run-openenv-after-nemo.py" SPEC = importlib.util.spec_from_file_location("openenv_queue", SCRIPT) queue = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(queue) COMMAND = "70717071-7071-4071-8071-707170717071" PLAN_SHA = "a" * 64 @pytest.fixture(autouse=True) def explicit_fixture_predecessor(monkeypatch): monkeypatch.setattr(queue, "NEMO_ID", "80818081-8081-4081-8081-808180818081") def test_scheduler_filters_secrets_and_captures_native_output(monkeypatch): def run(args, **kwargs): assert args == ["det", "shell", "list", "--json"] assert kwargs["capture_output"] and kwargs["timeout"] == 30 return SimpleNamespace(stdout=json.dumps([{ "id": queue.NEMO_ID, "state": "STATE_QUEUED", "privateKey": "SECRET", "container": {"token": "SECRET"}}])) monkeypatch.setattr(queue.subprocess, "run", run) rows = queue.scheduler("shell") assert rows[0]["id"] == queue.NEMO_ID assert "SECRET" not in json.dumps(rows) @pytest.mark.parametrize("state,ack,expected", [ ("QUEUED", True, False), ("RUNNING", True, False), ("TERMINATING", True, False), ("TERMINATED", True, True), (None, True, True), ("TERMINATED", False, False), ]) def test_release_requires_matching_receipt_and_scheduler(tmp_path, state, ack, expected): plan = {"nemo_origin": {"path": str(tmp_path / "experiment-plan.json"), "sha256": PLAN_SHA}} status = {"shell_id": queue.NEMO_ID, "workspace_id": 123, "plan_sha256": PLAN_SHA, "event": "allocation_release", "release_acknowledged": ack} (tmp_path / "queue-launcher-status.json").write_text(json.dumps(status)) rows = [{"id": queue.NEMO_ID, "workspaceId": 123, "state": "STATE_" + state}] if state else [] assert queue.predecessor_released(plan, rows) is expected status["plan_sha256"] = "b" * 64 (tmp_path / "queue-launcher-status.json").write_text(json.dumps(status)) assert not queue.predecessor_released(plan, rows) def test_create_is_one_detached_bounded_native_command(monkeypatch, tmp_path): calls = [] def run(args, **kwargs): calls.append(args) assert kwargs["capture_output"] and kwargs["timeout"] == 90 return SimpleNamespace(stdout="Created command " + COMMAND) monkeypatch.setattr(queue.subprocess, "run", run) assert queue.create_command(tmp_path / "plan.json", PLAN_SHA, "fixture") == COMMAND assert len(calls) == 1 args = calls[0] assert args[:3] == ["det", "command", "run"] assert "--detach" in args and "description=fixture" in args assert args[args.index("--workspace-name") + 1] == "JAIF" assert args[args.index("/usr/bin/timeout") + 1:][:2] == ["--signal=KILL", "28800s"] assert "--worker" in args and args[-1] == PLAN_SHA def test_reconciliation_rejects_duplicate_and_wrong_pool(): row = {"id": COMMAND, "description": "fixture", "resourcePool": "42_Compute"} assert queue.find_created([row], "fixture") == COMMAND assert queue.find_created([row], "different") is None with pytest.raises(ValueError, match="Ambiguous"): queue.find_created([row, row], "fixture") with pytest.raises(ValueError, match="pool"): queue.find_created([row | {"resourcePool": "other"}], "fixture") @pytest.fixture def controller(tmp_path, monkeypatch): root = tmp_path / "openenv" root.mkdir() plan = {"root": str(root), "nemo_origin": {}} calls = SimpleNamespace(created=0, killed=[], waits=0, slept=0) monkeypatch.setattr(queue, "verify", lambda *args: plan) monkeypatch.setattr(queue, "digest", lambda *args: "b" * 64) monkeypatch.setattr(queue, "predecessor_released", lambda *args: True) def create(*args): calls.created += 1 return COMMAND def kill(identifier): calls.killed.append(identifier) return True monkeypatch.setattr(queue, "create_command", create) monkeypatch.setattr(queue, "kill_owned", kill) monkeypatch.setattr(queue.time, "sleep", lambda seconds: setattr(calls, "slept", calls.slept + seconds)) return root, calls def row(state="STATE_QUEUED"): return {"id": COMMAND, "description": "hessian-openenv-" + PLAN_SHA[:16], "resourcePool": "42_Compute", "state": state} def test_waits_for_nemo_then_one_create_and_native_completion(controller, monkeypatch): root, calls = controller states = iter([[], [row()], [row("STATE_COMPLETED")]]) monkeypatch.setattr(queue, "scheduler", lambda kind: [] if kind == "shell" else next(states)) release = iter([False, False, True]) monkeypatch.setattr(queue, "predecessor_released", lambda *args: next(release)) (root / "experiment-status.json").write_text(json.dumps({"status": "completed", "plan_sha256": PLAN_SHA})) # Delay writing worker completion until after the transient initial list gap. completed = (root / "experiment-status.json").read_text() (root / "experiment-status.json").unlink() original_sleep = queue.time.sleep def sleep(seconds): original_sleep(seconds) if calls.slept >= 4: (root / "experiment-status.json").write_text(completed) monkeypatch.setattr(queue.time, "sleep", sleep) assert queue.launch(root / "plan.json", PLAN_SHA, poll=1) == 0 assert calls.created == 1 and calls.slept >= 4 and not calls.killed def test_interrupt_before_release_creates_and_kills_nothing(controller, monkeypatch): root, calls = controller monkeypatch.setattr(queue, "scheduler", lambda kind: []) monkeypatch.setattr(queue, "predecessor_released", lambda *args: False) monkeypatch.setattr(queue.time, "sleep", lambda *args: (_ for _ in ()).throw(KeyboardInterrupt())) with pytest.raises(KeyboardInterrupt): queue.launch(root / "plan.json", PLAN_SHA) assert calls.created == 0 and calls.killed == [] def test_resume_reconciles_uncertain_create_without_resubmission(controller, monkeypatch): root, calls = controller (root / "command-launcher-status.json").write_text(json.dumps({ "plan_sha256": PLAN_SHA, "launcher_sha256": "b" * 64, "description": row()["description"], "command_id": None, "creation_started": True, "active_since_epoch": None})) monkeypatch.setattr(queue, "scheduler", lambda kind: [] if kind == "shell" else [row("STATE_ERROR")]) assert queue.launch(root / "plan.json", PLAN_SHA) == 1 assert calls.created == 0 def test_uncertain_missing_creation_fails_without_duplicate(controller, monkeypatch): root, calls = controller (root / "command-launcher-status.json").write_text(json.dumps({ "plan_sha256": PLAN_SHA, "launcher_sha256": "b" * 64, "description": row()["description"], "command_id": None, "creation_started": True, "active_since_epoch": None})) monkeypatch.setattr(queue, "scheduler", lambda kind: []) with pytest.raises(RuntimeError, match="refusing a duplicate"): queue.launch(root / "plan.json", PLAN_SHA) assert calls.created == 0 and calls.killed == [] def test_status_is_read_after_exclusive_lock(controller, monkeypatch): root, calls = controller def acquire(*args): # Another controller completed a submission before this lock was acquired. (root / "command-launcher-status.json").write_text(json.dumps({ "plan_sha256": PLAN_SHA, "launcher_sha256": "b" * 64, "description": row()["description"], "command_id": COMMAND, "creation_started": True, "active_since_epoch": None})) monkeypatch.setattr(queue.fcntl, "flock", acquire) monkeypatch.setattr(queue, "scheduler", lambda kind: [row("STATE_ERROR")]) assert queue.launch(root / "plan.json", PLAN_SHA) == 1 assert calls.created == 0 def test_pull_time_consumes_budget_and_only_owned_job_is_stopped(controller, monkeypatch): root, calls = controller monkeypatch.setattr(queue, "scheduler", lambda kind: [] if kind == "shell" else [row("STATE_PULLING")]) ticks = iter([0, 1, 1, 28801]) monkeypatch.setattr(queue.time, "time", lambda: next(ticks)) with pytest.raises(TimeoutError): queue.launch(root / "plan.json", PLAN_SHA) assert calls.created == 1 and calls.killed == [COMMAND] status = json.loads((root / "command-launcher-status.json").read_text()) assert status["error_type"] == "TimeoutError" and status["owned_command_stop_acknowledged"] def test_missing_job_gets_visibility_grace_then_owned_cleanup(controller, monkeypatch): root, calls = controller monkeypatch.setattr(queue, "scheduler", lambda kind: []) ticks = iter([0, 1, 1, 122]) monkeypatch.setattr(queue.time, "time", lambda: next(ticks)) with pytest.raises(RuntimeError, match="absent"): queue.launch(root / "plan.json", PLAN_SHA) assert calls.slept == 15 and calls.killed == [COMMAND] def test_foreign_completed_evidence_is_rejected(controller, monkeypatch): root, calls = controller (root / "experiment-status.json").write_text(json.dumps({ "status": "completed", "plan_sha256": "b" * 64})) monkeypatch.setattr(queue, "scheduler", lambda kind: [] if kind == "shell" else [row("STATE_COMPLETED")]) with pytest.raises(ValueError, match="another frozen plan"): queue.launch(root / "plan.json", PLAN_SHA) assert calls.created == 1 and calls.killed == [COMMAND] def test_stop_retries_only_the_recorded_uuid(monkeypatch): calls = [] def run(args, **kwargs): calls.append(args) if len(calls) == 1: raise subprocess.TimeoutExpired(args, 30) return SimpleNamespace(returncode=0) monkeypatch.setattr(queue.subprocess, "run", run) monkeypatch.setattr(queue.time, "sleep", lambda *args: None) assert queue.kill_owned(COMMAND) assert calls == [["det", "command", "kill", COMMAND]] * 2 @pytest.mark.parametrize("name,kind", [("../escape", tarfile.REGTYPE), ("/escape", tarfile.REGTYPE), ("symlink", tarfile.SYMTYPE)]) def test_staging_rejects_links_and_path_escape(name, kind): member = tarfile.TarInfo(name) member.type = kind with pytest.raises(ValueError, match="Unsafe"): queue.validate_member(member) def test_unset_predecessor_cannot_verify_inputs_or_contact_cluster(monkeypatch): monkeypatch.setattr(queue, "NEMO_ID", None) def forbidden(*args, **kwargs): raise AssertionError("No plan/cluster operations without an explicit predecessor") monkeypatch.setattr(queue, "verify", forbidden) monkeypatch.setattr(queue, "scheduler", forbidden) with pytest.raises(ValueError, match="Explicit JAIF predecessor"): queue.launch(Path("unused.json"), PLAN_SHA)