diff --git a/testing/tests/test_hermes_cli_dispatch_runtime.py b/testing/tests/test_hermes_cli_dispatch_runtime.py
index 0593efcb..3ffc0b39 100644
--- a/testing/tests/test_hermes_cli_dispatch_runtime.py
+++ b/testing/tests/test_hermes_cli_dispatch_runtime.py
@@ -14,8 +14,6 @@ from testing.tests.test_hermes_cli_support import SCRIPTS, lanes
def test_board_slug_and_connection_failure_paths(monkeypatch):
- assert lanes._owns_local_workspace(SimpleNamespace(workspace_path=" /workspace "))
- assert not lanes._owns_local_workspace(SimpleNamespace(workspace_path=""))
assert lanes._board_slug(SimpleNamespace(slug="cassandra")) == "cassandra"
assert lanes._board_slug(SimpleNamespace(id="fallback")) == "fallback"
failures = []
@@ -103,41 +101,6 @@ def test_claim_ready_handles_zero_limit_empty_boards_and_claim_errors(monkeypatc
assert lanes.claim_ready(set(), 1) == []
-def test_dispatch_skips_unhealthy_ineligible_and_unclaimed_entries(monkeypatch):
- class Connection:
- def __init__(self, board):
- self.board = board
-
- def close(self):
- return None
-
- tasks = [
- SimpleNamespace(id="t_ineligible", status="ready", assignee="cli-auto"),
- SimpleNamespace(id="t_unclaimed", status="ready", assignee="cli-auto"),
- ]
- db = SimpleNamespace(
- list_boards=lambda **_kwargs: [
- {"slug": "offline"},
- {"slug": "cassandra"},
- ],
- scoped_current_board=lambda _board: nullcontext(),
- connect=lambda board: (_ for _ in ()).throw(OSError("offline"))
- if board == "offline"
- else Connection(board),
- recompute_ready=lambda _conn: None,
- list_tasks=lambda _conn: tasks,
- claim_task=lambda *_args, **_kwargs: None,
- )
- monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db))
- assert lanes.claim_ready(
- set(),
- 2,
- eligible=lambda _board, task: task.id == "t_unclaimed",
- ) == []
- monkeypatch.setattr(lanes, "recover_pending_finalizations", lambda: 0)
- lanes.recover_orphans()
-
-
class _Future:
def __init__(self):
self.polls = 0
diff --git a/testing/tests/test_hermes_cli_execution_edges.py b/testing/tests/test_hermes_cli_execution_edges.py
index 42ad6e4c..d1ffbda8 100644
--- a/testing/tests/test_hermes_cli_execution_edges.py
+++ b/testing/tests/test_hermes_cli_execution_edges.py
@@ -16,13 +16,6 @@ class _Connection:
return None
-def test_goal_rejection_history_is_list_bounded():
- execution = __import__("cli_lane_execution")
- history = ["first"]
- assert execution._goal_rejections({"goal_rejections": history}) is history
- assert execution._goal_rejections({"goal_rejections": "invalid"}) == []
-
-
def test_missing_claim_is_a_noop(tmp_path: Path, monkeypatch):
db = SimpleNamespace(
scoped_current_board=lambda _board: nullcontext(),
@@ -71,18 +64,12 @@ def test_transient_callback_storage_errors_do_not_kill_provider(
def provider(*args, **_kwargs):
assert args[6]("still working") is True
- return lanes.ProcessResult(
- 1,
- "failed",
- {"status": "incomplete", "blockers": ["synthetic provider failure"]},
- False,
- )
+ return lanes.ProcessResult(1, "failed", None, False)
monkeypatch.setattr(lanes, "run_provider", provider)
lanes.execute_claim("cassandra", "t_callbacks")
assert blocks and blocks[0]["kind"] == "capability"
- assert blocks[0]["reason"] == "synthetic provider failure"
def test_restart_handoff_survives_missing_prior_log(tmp_path: Path, monkeypatch):
diff --git a/testing/tests/test_hermes_execution_pool_submission.py b/testing/tests/test_hermes_execution_pool_submission.py
index 2ea26692..69df8c53 100644
--- a/testing/tests/test_hermes_execution_pool_submission.py
+++ b/testing/tests/test_hermes_execution_pool_submission.py
@@ -23,7 +23,9 @@ sys.path[:0] = [str(SCRIPTS), str(SCM_SCRIPTS)]
import execution_pool_client as client # noqa: E402
import execution_pool_protocol as protocol # noqa: E402
import execution_pool_scm as scm # noqa: E402
+import receive_pack_scan # noqa: E402
import scm_broker # noqa: E402
+from testing.tests.test_hermes_scm_broker_support import _receive_command # noqa: E402
from testing.tests.test_hermes_execution_pool_mediator import ( # noqa: E402
KEY,
RESULT,
@@ -50,7 +52,7 @@ def test_submission_refs_are_creation_only_and_attempt_scoped():
# Every candidate stays inside a reviewed namespace the broker accepts, and
# none of them is a protected or base ref.
for candidate in first + retry:
- assert scm_broker.FEATURE_REF_RE.fullmatch(f"refs/heads/{candidate}")
+ assert receive_pack_scan.FEATURE_REF_RE.fullmatch(f"refs/heads/{candidate}")
assert candidate not in {"main", "master"}
@@ -65,15 +67,10 @@ def test_submission_refs_drop_oversized_names_and_fail_closed_when_empty():
def test_the_broker_accepts_the_attempt_ref_as_a_creation_and_still_refuses_updates():
zero, head = b"0" * 40, b"1" * 40
ref = b"refs/heads/feature/pool-attempt-2"
- create = b" ".join((zero, head, ref))
- scm_broker._validate_receive_pack(
- b"%04x%s\n0000" % (len(create) + 5, create), "token"
- )
- update = b" ".join((b"2" * 40, head, ref))
- with pytest.raises(scm_broker.PolicyError, match="new namespaced branch"):
- scm_broker._validate_receive_pack(
- b"%04x%s\n0000" % (len(update) + 5, update), "token"
- )
+ scm_broker._validate_receive_pack(_receive_command(zero, head, ref), "token")
+ update = _receive_command(b"2" * 40, head, ref)
+ with pytest.raises(scm_broker.PolicyError, match="feature-branch creation"):
+ scm_broker._validate_receive_pack(update, "token")
def submit_harness(tmp_path, monkeypatch, remote_refs, *, attempt=1):
diff --git a/testing/tests/test_hermes_runtime_access.py b/testing/tests/test_hermes_runtime_access.py
index 42049146..68644623 100644
--- a/testing/tests/test_hermes_runtime_access.py
+++ b/testing/tests/test_hermes_runtime_access.py
@@ -82,82 +82,6 @@ def test_gitea_api_rejects_unavailable_methods(method: str):
)
-def test_jenkins_evidence_ignores_nested_execution_result():
- evidence_reader = _load("jenkins_build_evidence")
- evidence = evidence_reader.parse_build_xml(
- """
-
- abc123
- 10000
- SUCCESS
-
- """,
- job="demo",
- branch="master",
- number=15,
- )
-
- assert evidence.revision == "abc123"
- assert evidence.result is None
- assert evidence.building is True
-
-
-def test_jenkins_evidence_reads_only_top_level_terminal_result():
- evidence_reader = _load("jenkins_build_evidence")
- evidence = evidence_reader.parse_build_xml(
- """
-
- abc123
- 1000250
- FAILURE
- SUCCESS
-
- """,
- job="demo",
- branch=None,
- number=4,
- )
-
- assert evidence.result == "FAILURE"
- assert evidence.building is False
- assert evidence.duration_ms == 250
-
-
-def test_jenkins_evidence_removes_hidden_links_and_ansi_control_sequences():
- evidence_reader = _load("jenkins_build_evidence")
-
- value = (
- "before\n\x1b[8mha:////private-jenkins-payload\x1b[0m[Pipeline] sh\n"
- "\x1b[31m5 passed\x1b[0m\n"
- )
-
- assert evidence_reader.sanitize_log(value) == "before\n[Pipeline] sh\n5 passed\n"
-
-
-def test_jenkins_evidence_falls_back_to_multibranch_suffix(monkeypatch):
- evidence_reader = _load("jenkins_build_evidence")
- calls = []
-
- def fake_find(job, branch, commit, log_lines):
- calls.append((job, branch, commit, log_lines))
- if job == "demo":
- raise ValueError("not a multibranch job")
- return "evidence"
-
- monkeypatch.setattr(evidence_reader, "_find_build_for_job", fake_find)
-
- assert evidence_reader.find_build("demo", "master", "abc", 20) == "evidence"
- assert [call[0] for call in calls] == ["demo", "demo-branches"]
-
-
-@pytest.mark.parametrize("branch", ["../master", "/master", "feature//unsafe"])
-def test_jenkins_evidence_rejects_unsafe_branch_paths(branch: str):
- evidence_reader = _load("jenkins_build_evidence")
-
- with pytest.raises(ValueError):
- evidence_reader._validate_branch(branch)
-
-
def test_agent_runtime_stage_keeps_credentials_in_memory(tmp_path: Path, monkeypatch):
stage = _load("stage_runtime_access")
vault = tmp_path / "vault"
@@ -196,6 +120,7 @@ def test_agent_runtime_stage_keeps_credentials_in_memory(tmp_path: Path, monkeyp
monkeypatch.setattr(stage, "RUNTIME_ROOT", runtime)
monkeypatch.setattr(stage, "PERSISTENT_HOME", home)
monkeypatch.setattr(stage.os, "chown", lambda *_args: None)
+ monkeypatch.setattr(stage.os, "fchown", lambda *_args: None)
stage.stage_agent()
@@ -306,6 +231,7 @@ def test_invalid_runtime_json_is_removed(tmp_path: Path, monkeypatch):
(vault / "credential").write_text("not-json\n", encoding="utf-8")
monkeypatch.setattr(stage, "VAULT_ROOT", vault)
monkeypatch.setattr(stage.os, "chown", lambda *_args: None)
+ monkeypatch.setattr(stage.os, "fchown", lambda *_args: None)
destination = runtime / "credential.json"
try:
diff --git a/testing/tests/test_hermes_runtime_evidence.py b/testing/tests/test_hermes_runtime_evidence.py
new file mode 100644
index 00000000..11ac9bc0
--- /dev/null
+++ b/testing/tests/test_hermes_runtime_evidence.py
@@ -0,0 +1,104 @@
+"""Contracts for Hermes' Jenkins build-evidence reader boundary."""
+
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).parents[2]
+HERMES = ROOT / "services" / "hermes"
+SCRIPTS = HERMES / "scripts"
+SCM_SCRIPTS = HERMES / "scm-common" / "scripts"
+if str(SCRIPTS) not in sys.path:
+ sys.path.insert(0, str(SCRIPTS))
+if str(SCM_SCRIPTS) not in sys.path:
+ sys.path.insert(0, str(SCM_SCRIPTS))
+
+
+def _load(name: str):
+ root = SCM_SCRIPTS if name == "gitea_api" else SCRIPTS
+ spec = importlib.util.spec_from_file_location(name, root / f"{name}.py")
+ assert spec and spec.loader
+ module = importlib.util.module_from_spec(spec)
+ sys.modules[spec.name] = module
+ spec.loader.exec_module(module)
+ return module
+
+
+def test_jenkins_evidence_ignores_nested_execution_result():
+ evidence_reader = _load("jenkins_build_evidence")
+ evidence = evidence_reader.parse_build_xml(
+ """
+
+ abc123
+ 10000
+ SUCCESS
+
+ """,
+ job="demo",
+ branch="master",
+ number=15,
+ )
+
+ assert evidence.revision == "abc123"
+ assert evidence.result is None
+ assert evidence.building is True
+
+
+def test_jenkins_evidence_reads_only_top_level_terminal_result():
+ evidence_reader = _load("jenkins_build_evidence")
+ evidence = evidence_reader.parse_build_xml(
+ """
+
+ abc123
+ 1000250
+ FAILURE
+ SUCCESS
+
+ """,
+ job="demo",
+ branch=None,
+ number=4,
+ )
+
+ assert evidence.result == "FAILURE"
+ assert evidence.building is False
+ assert evidence.duration_ms == 250
+
+
+def test_jenkins_evidence_removes_hidden_links_and_ansi_control_sequences():
+ evidence_reader = _load("jenkins_build_evidence")
+
+ value = (
+ "before\n\x1b[8mha:////private-jenkins-payload\x1b[0m[Pipeline] sh\n"
+ "\x1b[31m5 passed\x1b[0m\n"
+ )
+
+ assert evidence_reader.sanitize_log(value) == "before\n[Pipeline] sh\n5 passed\n"
+
+
+def test_jenkins_evidence_falls_back_to_multibranch_suffix(monkeypatch):
+ evidence_reader = _load("jenkins_build_evidence")
+ calls = []
+
+ def fake_find(job, branch, commit, log_lines):
+ calls.append((job, branch, commit, log_lines))
+ if job == "demo":
+ raise ValueError("not a multibranch job")
+ return "evidence"
+
+ monkeypatch.setattr(evidence_reader, "_find_build_for_job", fake_find)
+
+ assert evidence_reader.find_build("demo", "master", "abc", 20) == "evidence"
+ assert [call[0] for call in calls] == ["demo", "demo-branches"]
+
+
+@pytest.mark.parametrize("branch", ["../master", "/master", "feature//unsafe"])
+def test_jenkins_evidence_rejects_unsafe_branch_paths(branch: str):
+ evidence_reader = _load("jenkins_build_evidence")
+
+ with pytest.raises(ValueError):
+ evidence_reader._validate_branch(branch)
diff --git a/testing/tests/test_hermes_runtime_stage_coverage.py b/testing/tests/test_hermes_runtime_stage_coverage.py
index 3ca8e496..cae8066e 100644
--- a/testing/tests/test_hermes_runtime_stage_coverage.py
+++ b/testing/tests/test_hermes_runtime_stage_coverage.py
@@ -22,6 +22,7 @@ def _stage(tmp_path: Path, monkeypatch):
monkeypatch.setattr(module, "RUNTIME_ROOT", runtime)
monkeypatch.setattr(module, "PERSISTENT_HOME", home)
monkeypatch.setattr(module.os, "chown", lambda *_args: None)
+ monkeypatch.setattr(module.os, "fchown", lambda *_args: None)
return module, vault, runtime, home
diff --git a/testing/tests/test_hermes_scm_broker.py b/testing/tests/test_hermes_scm_broker.py
index 8f1b2ca9..2a0fad44 100644
--- a/testing/tests/test_hermes_scm_broker.py
+++ b/testing/tests/test_hermes_scm_broker.py
@@ -48,7 +48,7 @@ def test_broker_metadata_boundary_reuses_explicit_read_allowlist():
api.authorize_request("GET", path, None)
-def test_receive_pack_allows_only_new_branches_in_reviewed_namespaces():
+def test_receive_pack_allows_only_new_namespaced_feature_branch():
broker = _load("scm_broker")
zero = b"0" * 40
commit = b"1" * 40
@@ -57,18 +57,9 @@ def test_receive_pack_allows_only_new_branches_in_reviewed_namespaces():
_receive_command(zero, commit, b"refs/heads/hermes/focused-fix"),
"runtime-sentinel",
)
- broker._validate_receive_pack(
- _receive_command(zero, commit, b"refs/heads/wt/t_deadbeef"),
- "runtime-sentinel",
- )
- broker._validate_receive_pack(
- _receive_command(zero, commit, b"refs/heads/review/t_deadbeef"),
- "runtime-sentinel",
- )
for old, new, ref in (
(commit, b"2" * 40, b"refs/heads/hermes/focused-fix"),
(commit, zero, b"refs/heads/hermes/focused-fix"),
- (commit, commit, b"refs/heads/hermes/focused-fix"),
(zero, commit, b"refs/heads/main"),
(zero, commit, b"refs/heads/master"),
(zero, commit, b"refs/tags/release"),