hermes: streamline forge evidence tools
This commit is contained in:
parent
c7ac16d206
commit
46f37f32a3
@ -330,6 +330,9 @@ data:
|
||||
is `/opt/coordinator/jenkins_build_evidence.py JOB [--branch BRANCH]
|
||||
[--commit SHA] --wait`; it distinguishes a genuinely terminal build from
|
||||
nested Jenkins execution metadata and returns bounded JSON/log evidence.
|
||||
With a branch, the helper also tries the conventional `JOB-branches`
|
||||
multibranch name. For JSON Gitea mutations, prefer repeatable shell-safe
|
||||
`--field KEY=VALUE` arguments over hand-quoted JSON.
|
||||
After merging, observe the default-branch build to a terminal result and
|
||||
report exact commit, build, and test evidence.
|
||||
|
||||
|
||||
@ -78,15 +78,40 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
data_group.add_argument(
|
||||
"--data-file", type=Path, help="path to a JSON request body"
|
||||
)
|
||||
data_group.add_argument(
|
||||
"--field",
|
||||
action="append",
|
||||
metavar="KEY=VALUE",
|
||||
help=(
|
||||
"repeatable JSON field; bare text remains a string while true, false, "
|
||||
"null, numbers, objects, and arrays are decoded as JSON"
|
||||
),
|
||||
)
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
def parse_fields(values: list[str]) -> dict[str, object]:
|
||||
"""Build a JSON object from shell-safe, repeatable key/value arguments."""
|
||||
result: dict[str, object] = {}
|
||||
for value in values:
|
||||
key, separator, raw = value.partition("=")
|
||||
if not separator or not key or any(char.isspace() for char in key):
|
||||
raise ValueError("each --field must be KEY=VALUE with a non-space key")
|
||||
try:
|
||||
result[key] = json.loads(raw)
|
||||
except json.JSONDecodeError:
|
||||
result[key] = raw
|
||||
return result
|
||||
|
||||
|
||||
def load_data(args: argparse.Namespace) -> object | None:
|
||||
"""Decode the optional JSON body without involving a shell expansion."""
|
||||
if args.data_json is not None:
|
||||
return json.loads(args.data_json)
|
||||
if args.data_file is not None:
|
||||
return json.loads(args.data_file.read_text(encoding="utf-8"))
|
||||
if args.field is not None:
|
||||
return parse_fields(args.field)
|
||||
return None
|
||||
|
||||
|
||||
|
||||
@ -16,6 +16,8 @@ from dataclasses import asdict, dataclass
|
||||
JENKINS_HOME = "/var/jenkins_home/jobs"
|
||||
SAFE_JOB = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
|
||||
SAFE_BRANCH = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]*$")
|
||||
HIDDEN_ANSI = re.compile(r"\x1b\[8m.*?\x1b\[0m", re.DOTALL)
|
||||
ANSI_ESCAPE = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
@ -143,7 +145,12 @@ def parse_build_xml(
|
||||
)
|
||||
|
||||
|
||||
def find_build(
|
||||
def sanitize_log(value: str) -> str:
|
||||
"""Remove Jenkins hidden hyperlinks and terminal control sequences."""
|
||||
return ANSI_ESCAPE.sub("", HIDDEN_ANSI.sub("", value))
|
||||
|
||||
|
||||
def _find_build_for_job(
|
||||
job: str,
|
||||
branch: str | None,
|
||||
commit: str | None,
|
||||
@ -168,11 +175,34 @@ def find_build(
|
||||
continue
|
||||
if commit and not (evidence.revision or "").startswith(commit):
|
||||
continue
|
||||
log_tail = _kubectl("tail", "-n", str(log_lines), f"{build_path}/log")
|
||||
log_tail = sanitize_log(
|
||||
_kubectl("tail", "-n", str(log_lines), f"{build_path}/log")
|
||||
)
|
||||
return BuildEvidence(**{**asdict(evidence), "log_tail": log_tail})
|
||||
return None
|
||||
|
||||
|
||||
def find_build(
|
||||
job: str,
|
||||
branch: str | None,
|
||||
commit: str | None,
|
||||
log_lines: int,
|
||||
) -> BuildEvidence | None:
|
||||
"""Find a build, accepting the common omitted ``-branches`` suffix."""
|
||||
candidates = [job]
|
||||
if branch is not None and not job.endswith("-branches"):
|
||||
candidates.append(f"{job}-branches")
|
||||
last_error: Exception | None = None
|
||||
for candidate in candidates:
|
||||
try:
|
||||
return _find_build_for_job(candidate, branch, commit, log_lines)
|
||||
except (OSError, ValueError, subprocess.SubprocessError) as exc:
|
||||
last_error = exc
|
||||
if last_error is not None:
|
||||
raise last_error
|
||||
return None
|
||||
|
||||
|
||||
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
"""Parse the bounded Jenkins evidence request."""
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
|
||||
@ -61,6 +61,27 @@ def test_gitea_api_rejects_foreign_or_non_api_targets(path: str):
|
||||
gitea_api.api_url("https://scm.bstein.dev", path)
|
||||
|
||||
|
||||
def test_gitea_api_parses_shell_safe_fields_as_json_values():
|
||||
gitea_api = _load("gitea_api")
|
||||
|
||||
assert gitea_api.parse_fields(
|
||||
["Do=manually-merged", "MergeCommitID=abc123", "enabled=true", "count=2"]
|
||||
) == {
|
||||
"Do": "manually-merged",
|
||||
"MergeCommitID": "abc123",
|
||||
"enabled": True,
|
||||
"count": 2,
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("value", ["missing-separator", "=missing-key", "bad key=x"])
|
||||
def test_gitea_api_rejects_invalid_shell_safe_fields(value: str):
|
||||
gitea_api = _load("gitea_api")
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
gitea_api.parse_fields([value])
|
||||
|
||||
|
||||
def test_jenkins_evidence_ignores_nested_execution_result():
|
||||
evidence_reader = _load("jenkins_build_evidence")
|
||||
evidence = evidence_reader.parse_build_xml(
|
||||
@ -102,6 +123,33 @@ def test_jenkins_evidence_reads_only_top_level_terminal_result():
|
||||
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")
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user