Compare commits
8 Commits
hermes-rep
...
master
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d876c7a6fb | ||
|
|
03f5fbf599 | ||
|
|
fe4cead4c6 | ||
|
|
8bc8940d48 | ||
|
|
eebc9b16c4 | ||
|
|
01759e309a | ||
|
|
ee902354c2 | ||
|
|
fe794b8ffe |
@ -217,6 +217,27 @@ def _register_admin_routes(app: FastAPI, require_auth: Callable, deps: Callable[
|
||||
module = deps()
|
||||
return JSONResponse(module.run_testing_triage_diagnosis(module.storage))
|
||||
|
||||
# The scheduler runs this every minute, which is cron's finest granularity
|
||||
# and still up to sixty seconds of silence after a build goes red. That is
|
||||
# the single largest delay between a failure happening and the system
|
||||
# visibly reacting to it. Running the same tick on demand collapses it;
|
||||
# the tick is idempotent - incidents dedupe on job and build number - so
|
||||
# an extra run can only ever be a no-op.
|
||||
@app.post("/api/admin/hermes/autotriage/run")
|
||||
def run_hermes_autotriage_now(ctx: AuthContext = Depends(require_auth)) -> JSONResponse:
|
||||
"""Run one Hermes auto-triage tick immediately."""
|
||||
|
||||
module = deps()
|
||||
module._require_admin(ctx)
|
||||
return JSONResponse(module.run_hermes_autotriage(module.storage))
|
||||
|
||||
@app.post("/api/internal/hermes/autotriage/run")
|
||||
def run_hermes_autotriage_now_internal() -> JSONResponse:
|
||||
"""Run one Hermes auto-triage tick for trusted internal callers."""
|
||||
|
||||
module = deps()
|
||||
return JSONResponse(module.run_hermes_autotriage(module.storage))
|
||||
|
||||
@app.post("/api/admin/access/requests/{username}/approve")
|
||||
async def approve_access_request(
|
||||
username: str,
|
||||
|
||||
@ -1,9 +1,10 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import dataclass, field
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
from ariadne.services import hermes_code_suggestion as code_suggestion_field
|
||||
from ariadne.services import hermes_suggested_remediation as suggestion_field
|
||||
|
||||
|
||||
@ -24,7 +25,10 @@ _TOP_LEVEL_KEYS = {
|
||||
}
|
||||
# Optional because the field only applies when nothing in the allowlist fit, and
|
||||
# because a response written before it existed must keep validating unchanged.
|
||||
_OPTIONAL_TOP_LEVEL_KEYS = {suggestion_field.SUGGESTION_KEY}
|
||||
_OPTIONAL_TOP_LEVEL_KEYS = {
|
||||
suggestion_field.SUGGESTION_KEY,
|
||||
code_suggestion_field.SUGGESTIONS_KEY,
|
||||
}
|
||||
_STRING_FIELDS = ("incident_id", "classification", "first_failed_gate", "reason")
|
||||
_FACT_KEYS = {"statement", "source", "reference"}
|
||||
_ACTION_KEYS = {"type", "id"}
|
||||
@ -79,6 +83,7 @@ class TriageDecision:
|
||||
human_required: bool
|
||||
reason: str
|
||||
suggested_remediation: suggestion_field.SuggestedRemediation | None = None
|
||||
code_suggestions: list[code_suggestion_field.CodeSuggestion] = field(default_factory=list)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
@ -257,6 +262,7 @@ def _validate_payload(payload: dict[str, Any], expected_incident_id: str) -> str
|
||||
or _validate_facts(payload["facts"])
|
||||
or _validate_requested_action(payload["requested_action"])
|
||||
or suggestion_field.validate(payload)
|
||||
or code_suggestion_field.validate(payload)
|
||||
)
|
||||
if type_error:
|
||||
return type_error
|
||||
@ -329,6 +335,7 @@ def _decision_from_payload(payload: dict[str, Any]) -> TriageDecision:
|
||||
human_required=payload["human_required"],
|
||||
reason=payload["reason"],
|
||||
suggested_remediation=suggestion_field.from_payload(payload),
|
||||
code_suggestions=code_suggestion_field.from_payload(payload),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@ -12,6 +12,7 @@ from dataclasses import dataclass
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
from . import hermes_code_suggestion as code_suggestion_field
|
||||
from . import hermes_suggested_remediation as suggestion_field
|
||||
from .hermes_autotriage_metrics import (
|
||||
HERMES_TRIAGE_ACTION_TOTAL,
|
||||
@ -213,6 +214,9 @@ def outcome_phase(outcome: Any) -> dict[str, Any]:
|
||||
"suggested_remediation": suggestion_field.as_detail(
|
||||
getattr(decision, "suggested_remediation", None)
|
||||
),
|
||||
"code_suggestions": code_suggestion_field.as_detail(
|
||||
getattr(decision, "code_suggestions", None)
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
|
||||
@ -7,6 +7,7 @@ from typing import Any
|
||||
import httpx
|
||||
|
||||
from ..utils.logging import get_logger
|
||||
from . import hermes_sonar_client
|
||||
|
||||
|
||||
logger = get_logger(__name__)
|
||||
@ -23,6 +24,11 @@ _REPAIR_BRANCH_PREFIX = "hermes-repair/"
|
||||
_OPEN_PULLS_LIMIT = 50
|
||||
_COMMIT_IDENTITY = {"name": "Hermes Agent", "email": "hermes@bstein.dev"}
|
||||
_COMMIT_OK_STATUSES = {HTTP_OK, HTTP_CREATED}
|
||||
# The Hermes console reopens a finished run from its id at this route.
|
||||
RUN_PATH = "/chat?resume="
|
||||
# Every proposal title is built from this, so the incident an open pull
|
||||
# request belongs to can be read back without storing a second index.
|
||||
PR_TITLE_PREFIX = "fix(hermes): repair "
|
||||
_BRANCH_RETRY_STATUSES = {HTTP_NOT_FOUND, HTTP_UNPROCESSABLE}
|
||||
_BRANCH_UNSAFE = re.compile(r"[^A-Za-z0-9._-]+")
|
||||
|
||||
@ -83,6 +89,45 @@ def find_open_proposal(cfg: dict) -> dict[str, Any]:
|
||||
return _oldest_repair_pull(response, _base_branch(cfg))
|
||||
|
||||
|
||||
def open_proposal_incidents(cfg: dict) -> tuple[list[str], str | None]:
|
||||
"""List the incidents that already have an open repair pull request.
|
||||
|
||||
Inputs: `cfg` as for `fetch_file`. Outputs: (incident_ids, error).
|
||||
|
||||
Read back from the pull request titles rather than from a stored index,
|
||||
because the pull requests are the thing that actually exists - an index
|
||||
could disagree with them, and the whole point of the check is to not
|
||||
propose something a reviewer is already looking at.
|
||||
|
||||
Fails open: on any error the list is empty, so a lookup failure produces a
|
||||
duplicate rather than silently suppressing real work.
|
||||
"""
|
||||
|
||||
base_url = _base_url(cfg)
|
||||
if not base_url:
|
||||
return [], "gitea base url is empty"
|
||||
try:
|
||||
with httpx.Client(timeout=_timeout(cfg)) as client:
|
||||
response = client.get(
|
||||
f"{base_url}/api/v1/repos/{_owner(cfg)}/{_repo(cfg)}/pulls",
|
||||
headers=_headers(cfg),
|
||||
params={"state": "open", "limit": _OPEN_PULLS_LIMIT},
|
||||
)
|
||||
if response.status_code != HTTP_OK:
|
||||
return [], f"open proposal lookup http {response.status_code}"
|
||||
payload = response.json()
|
||||
except Exception as exc:
|
||||
return [], f"open proposal lookup failed: {exc}"
|
||||
if not isinstance(payload, list):
|
||||
return [], "open proposal payload is not a list"
|
||||
found = []
|
||||
for pull in payload:
|
||||
title = str(pull.get("title") or "") if isinstance(pull, dict) else ""
|
||||
if title.startswith(PR_TITLE_PREFIX):
|
||||
found.append(title[len(PR_TITLE_PREFIX) :].strip())
|
||||
return found, None
|
||||
|
||||
|
||||
def push_branch(
|
||||
cfg: dict, incident_id: str, ref: Any, patch: Any, patched_contents: str
|
||||
) -> dict[str, Any]:
|
||||
@ -134,7 +179,7 @@ def open_pull_request( # noqa: PLR0913 - the body needs the full proposal prove
|
||||
payload = {
|
||||
"head": branch,
|
||||
"base": _base_branch(cfg),
|
||||
"title": f"fix(hermes): repair {incident_id}",
|
||||
"title": f"{PR_TITLE_PREFIX}{incident_id}",
|
||||
"body": _pr_body(incident_id, patch, analysis, run_id, cfg),
|
||||
}
|
||||
try:
|
||||
@ -256,6 +301,19 @@ def _pr_result(response: Any) -> dict[str, Any]:
|
||||
return {"pr_number": number, "url": url, "error": None}
|
||||
|
||||
|
||||
def run_url(ui_url: str, run_id: str) -> str:
|
||||
"""Build the deep link that reopens one Hermes run in its console.
|
||||
|
||||
Inputs: the Hermes UI base url and a run id. Outputs: the resume link, or
|
||||
"" when either is missing. Kept here so the pull request and the demo
|
||||
monitor cannot drift into pointing at different pages.
|
||||
"""
|
||||
|
||||
base = str(ui_url or "").rstrip("/")
|
||||
run = str(run_id or "").strip()
|
||||
return f"{base}{RUN_PATH}{run}" if base and run else ""
|
||||
|
||||
|
||||
def _pr_body(incident_id: str, patch: Any, analysis: str, run_id: str, cfg: dict) -> str:
|
||||
"""Render the markdown pull-request body for human review.
|
||||
|
||||
@ -271,16 +329,26 @@ def _pr_body(incident_id: str, patch: Any, analysis: str, run_id: str, cfg: dict
|
||||
"",
|
||||
f"**Incident:** {incident_id}",
|
||||
f"**File:** `{patch.path}`",
|
||||
]
|
||||
finding = hermes_sonar_client.issue_url(incident_id, cfg.get("sonar_ui_url"))
|
||||
if finding:
|
||||
# A sweep proposal exists because of one finding; the reviewer's first
|
||||
# question is what it said, and an issue key is not an answer.
|
||||
lines.append(f"**SonarQube finding:** {finding}")
|
||||
lines += [
|
||||
f"**Analysis:** {analysis}",
|
||||
f"**Rationale:** {patch.rationale}",
|
||||
]
|
||||
if run_id:
|
||||
lines.append(f"**Hermes run:** `{run_id}`")
|
||||
ui_url = str(cfg.get("hermes_ui_url") or "").rstrip("/")
|
||||
if ui_url:
|
||||
lines.append(f"**Hermes run:** [{run_id}]({run_url(ui_url, run_id)})")
|
||||
lines.append(
|
||||
f"The prompt this run was given and every tool call it made are at {ui_url}."
|
||||
"That link opens the run itself: the prompt Hermes was given, the evidence "
|
||||
"bundle it read, the tools it called, and the JSON it returned."
|
||||
)
|
||||
else:
|
||||
lines.append(f"**Hermes run:** `{run_id}`")
|
||||
lines += [
|
||||
"",
|
||||
"Proposed by Hermes; validated and pushed by Ariadne; "
|
||||
|
||||
@ -55,6 +55,8 @@ def build_config(config: Any) -> dict[str, Any]:
|
||||
"repos": _parse_repo_map(getattr(config, "hermes_code_repos", "")),
|
||||
# Shown in the pull request so a reviewer can read the run that wrote it.
|
||||
"hermes_ui_url": str(getattr(config, "hermes_ui_url", "") or ""),
|
||||
# Shown in sweep proposals so the finding itself is one click away.
|
||||
"sonar_ui_url": str(getattr(config, "hermes_sonar_ui_url", "") or ""),
|
||||
"job_prefixes": _parse_list_map(getattr(config, "hermes_code_prefixes", "")),
|
||||
"job_suffixes": _parse_list_map(getattr(config, "hermes_code_suffixes", "")),
|
||||
"job_base_branches": dict(_parse_pairs(getattr(config, "hermes_code_base_branches", ""))),
|
||||
|
||||
157
ariadne/services/hermes_code_suggestion.py
Normal file
157
ariadne/services/hermes_code_suggestion.py
Normal file
@ -0,0 +1,157 @@
|
||||
"""Let an escalated diagnosis carry the fix it could not apply.
|
||||
|
||||
Ariadne opens a pull request when it can: the repository is mapped, the file
|
||||
is inside the write allowlist, and the change is one anchored snippet its
|
||||
validator can check. When any of that fails, the incident escalates to a human
|
||||
with a diagnosis and nothing else - even though the model that wrote the
|
||||
diagnosis frequently knows exactly what the fix is. That knowledge was being
|
||||
discarded at the moment it was most useful, because the cases where no patch
|
||||
is possible are precisely the cases a maintainer has to do by hand.
|
||||
|
||||
So a diagnosis may carry code suggestions: the file, what to change, and the
|
||||
code to change it to. This is not a patch and is deliberately not treated as
|
||||
one. Nothing here is validated against the file, anchored, applied, or pushed;
|
||||
it is rendered into the issue as a starting point for a person who will read
|
||||
it, judge it, and write the real change themselves.
|
||||
|
||||
That distinction is the whole safety story. A patch has to survive the
|
||||
validator because Ariadne acts on it. A suggestion is read by a human who is
|
||||
already going to edit this file, so being wrong costs them a moment's thought
|
||||
rather than a bad commit - which is also why suggestions are allowed to
|
||||
describe changes too large or too diffuse for the patcher to have attempted.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
|
||||
SUGGESTIONS_KEY = "code_suggestions"
|
||||
|
||||
_KEYS = {"path", "explanation", "code"}
|
||||
_MAX_SUGGESTIONS = 3
|
||||
_MAX_EXPLANATION = 600
|
||||
_MAX_CODE = 1600
|
||||
_FENCE = "```"
|
||||
_FENCE_ESCAPE = "```"
|
||||
_TRUNCATED = "\n... (truncated)"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CodeSuggestion:
|
||||
"""Represent one suggested change a person is expected to review.
|
||||
|
||||
Inputs: a validated `code_suggestions[]` entry. Outputs: the file it
|
||||
concerns, why the change is needed, and the suggested code.
|
||||
|
||||
Never applied and never validated against the file. It is advice.
|
||||
"""
|
||||
|
||||
path: str
|
||||
explanation: str
|
||||
code: str
|
||||
|
||||
|
||||
def validate(payload: dict[str, Any]) -> str | None:
|
||||
"""Validate the optional code_suggestions field of a diagnosis.
|
||||
|
||||
Inputs: the whole parsed response payload. Outputs: a specific reject
|
||||
reason, or None when the field is absent, null, or well-formed.
|
||||
|
||||
Bounded at three. A diagnosis that suggests a dozen changes has stopped
|
||||
diagnosing and started rewriting, and an issue that long buries the reason
|
||||
a person was called in the first place.
|
||||
"""
|
||||
|
||||
suggestions = payload.get(SUGGESTIONS_KEY)
|
||||
if suggestions is None:
|
||||
return None
|
||||
if not isinstance(suggestions, list):
|
||||
return "code_suggestions_invalid: must be a list or null"
|
||||
if len(suggestions) > _MAX_SUGGESTIONS:
|
||||
return f"code_suggestions_invalid: at most {_MAX_SUGGESTIONS} suggestions"
|
||||
for index, suggestion in enumerate(suggestions):
|
||||
error = _validate_one(suggestion, index)
|
||||
if error:
|
||||
return error
|
||||
return None
|
||||
|
||||
|
||||
def _validate_one(suggestion: Any, index: int) -> str | None:
|
||||
"""Validate one suggestion entry's shape and fields."""
|
||||
|
||||
if not isinstance(suggestion, dict):
|
||||
return f"code_suggestions_invalid: [{index}] must be an object"
|
||||
if set(suggestion) != _KEYS:
|
||||
return f"code_suggestions_invalid: [{index}] must have exactly path, explanation, code"
|
||||
for key in sorted(_KEYS):
|
||||
if not isinstance(suggestion[key], str) or not suggestion[key].strip():
|
||||
return f"code_suggestions_invalid: [{index}] {key} must be a non-empty string"
|
||||
return None
|
||||
|
||||
|
||||
def from_payload(payload: dict[str, Any]) -> list[CodeSuggestion]:
|
||||
"""Build the typed suggestions from an already-validated payload.
|
||||
|
||||
Inputs: the parsed response payload. Outputs: the suggestions, clipped so
|
||||
a long one cannot crowd the diagnosis out of a bounded issue body.
|
||||
"""
|
||||
|
||||
suggestions = payload.get(SUGGESTIONS_KEY)
|
||||
if not isinstance(suggestions, list):
|
||||
return []
|
||||
built = []
|
||||
for suggestion in suggestions[:_MAX_SUGGESTIONS]:
|
||||
if not isinstance(suggestion, dict):
|
||||
continue
|
||||
built.append(
|
||||
CodeSuggestion(
|
||||
path=str(suggestion.get("path") or "").strip(),
|
||||
explanation=_clip(" ".join(str(suggestion.get("explanation") or "").split()), _MAX_EXPLANATION),
|
||||
code=_clip(str(suggestion.get("code") or "").strip(), _MAX_CODE),
|
||||
)
|
||||
)
|
||||
return built
|
||||
|
||||
|
||||
def as_detail(suggestions: Any) -> list[dict[str, str]]:
|
||||
"""Summarize suggestions for the audit event detail."""
|
||||
|
||||
return [
|
||||
{"path": item.path, "explanation": item.explanation, "code": item.code}
|
||||
for item in (suggestions or [])
|
||||
]
|
||||
|
||||
|
||||
def issue_section(suggestions: Any) -> str:
|
||||
"""Render the suggestions as a markdown section for the incident issue.
|
||||
|
||||
Inputs: the CodeSuggestion list. Outputs: the section, or "" when there
|
||||
are none.
|
||||
|
||||
The heading says "suggested", and the note says nothing was applied,
|
||||
because a code block in an issue reads like a change that was made unless
|
||||
it is told otherwise.
|
||||
"""
|
||||
|
||||
items = list(suggestions or [])
|
||||
if not items:
|
||||
return ""
|
||||
lines = [
|
||||
"## Suggested fix (not applied)",
|
||||
"Hermes could not open a pull request for this, so the change below was "
|
||||
"not written, validated, or pushed anywhere. It is a starting point for "
|
||||
"whoever picks this up, not a reviewed patch.",
|
||||
]
|
||||
for item in items:
|
||||
lines.append("")
|
||||
lines.append(f"**`{item.path}`** — {item.explanation}")
|
||||
lines.append(f"{_FENCE}\n{item.code.replace(_FENCE, _FENCE_ESCAPE)}\n{_FENCE}")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _clip(value: str, limit: int) -> str:
|
||||
"""Clip one field so a long suggestion cannot crowd out the diagnosis."""
|
||||
|
||||
return value if len(value) <= limit else value[:limit].rstrip() + _TRUNCATED
|
||||
@ -15,12 +15,24 @@ from __future__ import annotations
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
from ariadne.services import hermes_code_suggestion as code_suggestion_field
|
||||
from ariadne.services import hermes_incident_evidence_section as evidence_section
|
||||
from ariadne.services import hermes_suggested_remediation as suggestion_field
|
||||
|
||||
|
||||
DEFAULT_MAX_BODY_CHARS = 8000
|
||||
UNDIAGNOSED = "undiagnosed"
|
||||
# A finding is not a build failure, and rendering one through the other's
|
||||
# wording produces an issue that is wrong in every heading: a line number
|
||||
# printed as a build, a "first failed gate" for a build that passed, and a
|
||||
# SonarQube link labelled "failed build".
|
||||
SONAR_PREFIX = "sonar/"
|
||||
|
||||
|
||||
def is_finding(context: dict) -> bool:
|
||||
"""Report whether this issue describes a finding rather than a failure."""
|
||||
|
||||
return str(context.get("incident_id") or "").startswith(SONAR_PREFIX)
|
||||
|
||||
_MAX_TITLE_CHARS = 120
|
||||
_MAX_FACTS = 10
|
||||
@ -37,7 +49,7 @@ _AUDIT_NOTE = (
|
||||
"event types `hermes_autotriage_incident` and `hermes_autotriage_diagnosis`."
|
||||
)
|
||||
_FOOTER_TEMPLATE = (
|
||||
"Filed automatically by Ariadne from a Hermes Agent diagnosis (run `{run_id}`). "
|
||||
"Filed automatically by Ariadne from a Hermes Agent diagnosis ({run}). "
|
||||
"Hermes has no write access to this repository; no files or infrastructure were changed."
|
||||
)
|
||||
# Some escalations never reach a model at all - a build still running has no
|
||||
@ -93,6 +105,9 @@ def issue_title(context: dict) -> str:
|
||||
`[hermes] {job} #{build}: {classification}` clipped to 120 characters.
|
||||
"""
|
||||
|
||||
if is_finding(context):
|
||||
prefix = f"[hermes] {context.get('job')}: "
|
||||
else:
|
||||
prefix = f"[hermes] {context.get('job')} #{context.get('build_number')}: "
|
||||
classification = str(context.get("classification") or UNDIAGNOSED)
|
||||
room = _MAX_TITLE_CHARS - len(prefix)
|
||||
@ -122,6 +137,7 @@ def issue_body(context: dict, max_chars: int = DEFAULT_MAX_BODY_CHARS) -> str:
|
||||
_facts_section(context),
|
||||
evidence_section.evidence_section(context.get("bundle")),
|
||||
_inferences_section(context),
|
||||
code_suggestion_field.issue_section(context.get("code_suggestions")),
|
||||
suggestion_field.issue_section(context.get("suggested_remediation")),
|
||||
_links_section(context),
|
||||
_footer(context),
|
||||
@ -147,6 +163,11 @@ def _summary_line(context: dict) -> str:
|
||||
|
||||
incident_id = context.get("incident_id")
|
||||
classification = context.get("classification")
|
||||
if is_finding(context):
|
||||
return (
|
||||
f"SonarQube reports **{classification}** in `{context.get('finding_path')}`. "
|
||||
"The build is green; this is a standing finding, not a failure."
|
||||
)
|
||||
run_id = str(context.get("run_id") or "")
|
||||
if not run_id or run_id == "unknown":
|
||||
# The headline is the first thing read, so it must not credit a model
|
||||
@ -166,20 +187,27 @@ def _footer(context: dict) -> str:
|
||||
run_id = str(context.get("run_id") or "")
|
||||
if not run_id or run_id == "unknown":
|
||||
return _NO_RUN_FOOTER
|
||||
return _FOOTER_TEMPLATE.format(run_id=run_id)
|
||||
# A bare id is something to copy; a link is a page to open, and that page
|
||||
# is the whole answer to who decided this.
|
||||
run_url = str(context.get("run_url") or "")
|
||||
run = f"run [{run_id}]({run_url})" if run_url else f"run `{run_id}`"
|
||||
return _FOOTER_TEMPLATE.format(run=run)
|
||||
|
||||
|
||||
def _human_section(context: dict) -> str:
|
||||
"""Render the section explaining why the incident needs a person."""
|
||||
|
||||
lines = ["## Why a human is needed", str(context.get("reason") or "no reason recorded")]
|
||||
heading = "## What is wrong" if is_finding(context) else "## Why a human is needed"
|
||||
lines = [heading, str(context.get("reason") or "no reason recorded")]
|
||||
observation = str(context.get("observation") or "").strip()
|
||||
if observation:
|
||||
# Escalations that never reached a model carry their explanation here
|
||||
# rather than in facts, which are only populated from a diagnosis.
|
||||
lines.append(observation)
|
||||
authorize_reason = str(context.get("authorize_reason") or "")
|
||||
if authorize_reason:
|
||||
if authorize_reason and is_finding(context):
|
||||
lines.append(f"Ariadne opened no pull request for it: {authorize_reason}.")
|
||||
elif authorize_reason:
|
||||
lines.append(f"Ariadne did not authorize automated remediation: `{authorize_reason}`.")
|
||||
return "\n\n".join(lines)
|
||||
|
||||
@ -214,6 +242,9 @@ def _links_section(context: dict) -> str:
|
||||
|
||||
build_url = str(context.get("build_url") or "")
|
||||
proposal_url = str(context.get("code_proposal_url") or "")
|
||||
if is_finding(context):
|
||||
lines = ["## Links", f"- SonarQube finding: {build_url}" if build_url else "- SonarQube finding: url unavailable"]
|
||||
else:
|
||||
lines = [
|
||||
"## Links",
|
||||
f"- Failed build: {build_url}" if build_url else "- Failed build: url unavailable",
|
||||
|
||||
@ -22,7 +22,7 @@ from typing import Any
|
||||
import httpx
|
||||
|
||||
from ..utils.logging import get_logger
|
||||
from . import hermes_incident_body as body
|
||||
from . import hermes_code_repair, hermes_incident_body as body
|
||||
|
||||
|
||||
logger = get_logger(__name__)
|
||||
@ -171,7 +171,9 @@ def issue_config(config: Any) -> dict[str, Any]:
|
||||
}
|
||||
|
||||
|
||||
def issue_context(base: dict[str, Any], diagnosis: dict[str, Any]) -> dict[str, Any]:
|
||||
def issue_context(
|
||||
base: dict[str, Any], diagnosis: dict[str, Any], hermes_ui_url: str = ""
|
||||
) -> dict[str, Any]:
|
||||
"""Flatten the incident, evidence bundle, and outcome into body inputs.
|
||||
|
||||
Inputs: the incident identity fields and the diagnosis dict passed to
|
||||
@ -203,6 +205,7 @@ def issue_context(base: dict[str, Any], diagnosis: dict[str, Any]) -> dict[str,
|
||||
"facts": [body.fact_fields(fact) for fact in getattr(decision, "facts", None) or []],
|
||||
"inferences": list(getattr(decision, "inferences", None) or []),
|
||||
"suggested_remediation": getattr(decision, "suggested_remediation", None),
|
||||
"code_suggestions": getattr(decision, "code_suggestions", None) or [],
|
||||
# The raw excerpt, so the issue shows the failure instead of citing
|
||||
# where in Jenkins the failure can be found.
|
||||
"bundle": bundle,
|
||||
@ -211,6 +214,7 @@ def issue_context(base: dict[str, Any], diagnosis: dict[str, Any]) -> dict[str,
|
||||
# console text is the whole explanation and must not be dropped.
|
||||
"observation": _observation(jenkins) if decision is None else "",
|
||||
"run_id": diagnosis.get("run_id"),
|
||||
"run_url": hermes_code_repair.run_url(hermes_ui_url, str(diagnosis.get("run_id") or "")),
|
||||
"code_proposal_url": str(proposal.get("url") or "") if isinstance(proposal, dict) else "",
|
||||
}
|
||||
|
||||
@ -240,7 +244,7 @@ def _file_issue(
|
||||
if not cfg["enabled"] or target is None or _filed_count(tick_state) >= _max_per_tick(cfg):
|
||||
return None
|
||||
repo_cfg = {**cfg, "owner": target[0], "repo": target[1]}
|
||||
context = issue_context(base, diagnosis)
|
||||
context = issue_context(base, diagnosis, str(getattr(config, "hermes_ui_url", "") or ""))
|
||||
existing = find_open_incident_issue(
|
||||
repo_cfg, job, str(context["classification"]), str(context["incident_id"])
|
||||
)
|
||||
|
||||
250
ariadne/services/hermes_sonar_advice.py
Normal file
250
ariadne/services/hermes_sonar_advice.py
Normal file
@ -0,0 +1,250 @@
|
||||
"""File an issue with a suggested fix when a finding cannot become a patch.
|
||||
|
||||
The sweep opens a pull request when everything lines up: the project maps to a
|
||||
repository, the file sits inside the write allowlist, and the change is one
|
||||
anchored snippet the validator can check. Plenty of real findings fail one of
|
||||
those and produce nothing at all - which is the wrong outcome, because a
|
||||
finding nobody can patch automatically is exactly the one a maintainer has to
|
||||
do by hand, and that is when knowing the intended fix is worth most.
|
||||
|
||||
So the fallback is an issue carrying the finding and the code Hermes believes
|
||||
would resolve it. Not a patch: nothing here is anchored, validated against the
|
||||
file, or pushed. It is written for a person who will read it, judge it, and
|
||||
make the change themselves, which is why it may describe work too large or too
|
||||
diffuse for the patcher to have attempted.
|
||||
|
||||
Deduped on the rule, like the pull requests. One rule is one root cause spread
|
||||
over many files, and an issue per instance would bury the repository in near
|
||||
identical tickets - the precise failure this whole path exists to avoid.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
from ..utils.logging import get_logger
|
||||
from . import (
|
||||
hermes_agent_client,
|
||||
hermes_code_repair,
|
||||
hermes_code_suggestion,
|
||||
hermes_incident_issue,
|
||||
hermes_sonar_client,
|
||||
)
|
||||
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
ADVICE_EVENT_TYPE = "hermes_sonar_advice"
|
||||
|
||||
_RUN_COMPLETED = "completed"
|
||||
_MAX_CONTEXT_CHARS = 40000
|
||||
|
||||
_PROMPT = """A static-analysis finding, not a build failure. Nothing is broken and no investigation is needed: the finding and the file are below.
|
||||
You are suggesting a fix for a maintainer to make by hand. Ariadne cannot apply this automatically, so it will be printed in an issue for a person to read.
|
||||
Return ONLY a single JSON object with exactly these keys and no others:
|
||||
{"incident_id": "<must equal __INCIDENT_ID__>", "analysis": "<what is actually wrong, in one or two sentences>", "code_suggestions": [{"path": "<repository-relative file>", "explanation": "<what is wrong there and why this change fixes it>", "code": "<the suggested code>"}], "human_required": <bool>, "reason": "<string>"}
|
||||
Give at most three suggestions and prefer one. Preserve the existing behaviour exactly: fix what the rule objects to without changing what the code does, and do not suppress the rule or add an inline ignore.
|
||||
Nothing you return is applied, validated against the file, or committed anywhere, so suggest the fix you actually believe is right even when it spans several places.
|
||||
|
||||
The finding:
|
||||
__FINDING__
|
||||
|
||||
Current content of __PATH__:
|
||||
__CONTENTS__"""
|
||||
|
||||
|
||||
def advise( # noqa: PLR0913 - one advice attempt needs the whole proposal context
|
||||
storage: Any,
|
||||
config: Any,
|
||||
hermes_cfg: dict,
|
||||
repo_cfg: dict,
|
||||
project: str,
|
||||
issue: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
"""Ask Hermes for a fix and file it as an issue for one finding.
|
||||
|
||||
Inputs: event storage; a settings-like object; the Hermes run config; the
|
||||
resolved per-repo cfg; the SonarQube project key; and one normalized
|
||||
finding. Outputs: {"filed": bool, "reason": str, "url": str|None}.
|
||||
|
||||
Never raises. This runs after a pull request has already been declined, so
|
||||
a failure here must not turn a partial outcome into no outcome at all.
|
||||
"""
|
||||
|
||||
try:
|
||||
return _advise(storage, config, hermes_cfg, repo_cfg, project, issue)
|
||||
except Exception as exc:
|
||||
logger.info(
|
||||
"hermes sonar advice failed",
|
||||
extra={"event": ADVICE_EVENT_TYPE, "status": "error", "detail": str(exc)},
|
||||
)
|
||||
return {"filed": False, "reason": f"advice_failed: {exc}", "url": None}
|
||||
|
||||
|
||||
def _advise( # noqa: PLR0913 - mirrors advise's contract
|
||||
storage: Any,
|
||||
config: Any,
|
||||
hermes_cfg: dict,
|
||||
repo_cfg: dict,
|
||||
project: str,
|
||||
issue: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
"""Run the advice flow, recording one event whatever the outcome."""
|
||||
|
||||
incident_id = f"{hermes_sonar_client.INCIDENT_PREFIX}{project}/{issue.get('rule')}/{issue.get('key')}"
|
||||
rule = str(issue.get("rule") or "unknown_rule")
|
||||
issue_cfg = _issue_config(config, repo_cfg)
|
||||
|
||||
existing = hermes_incident_issue.find_open_incident_issue(issue_cfg, project, rule, incident_id)
|
||||
if existing.get("found"):
|
||||
return _record(storage, incident_id, {"filed": False, "reason": "issue_already_open", "url": existing.get("url")})
|
||||
|
||||
contents, error = hermes_code_repair.fetch_file(repo_cfg, str(issue.get("path") or ""))
|
||||
if contents is None:
|
||||
return _record(storage, incident_id, {"filed": False, "reason": f"file_fetch_failed: {error}", "url": None})
|
||||
|
||||
run = hermes_agent_client.run_triage(hermes_cfg, _prompt(incident_id, issue, contents))
|
||||
if run.status != _RUN_COMPLETED or not run.output:
|
||||
return _record(storage, incident_id, {"filed": False, "reason": f"hermes_run_{run.status}", "url": None})
|
||||
|
||||
suggestions, analysis = parse_advice(run.output, incident_id)
|
||||
if not suggestions:
|
||||
return _record(storage, incident_id, {"filed": False, "reason": "no_suggestions_returned", "url": None})
|
||||
|
||||
created = hermes_incident_issue.create_incident_issue(
|
||||
issue_cfg, _context(project, rule, incident_id, issue, analysis, suggestions, run.run_id, config)
|
||||
)
|
||||
if created.get("error"):
|
||||
return _record(storage, incident_id, {"filed": False, "reason": str(created["error"]), "url": None})
|
||||
return _record(storage, incident_id, {"filed": True, "reason": "issue_filed", "url": created.get("url")})
|
||||
|
||||
|
||||
def parse_advice(raw_output: str, incident_id: str) -> tuple[list[Any], str]:
|
||||
"""Extract the suggestions and analysis from an advice response.
|
||||
|
||||
Inputs: raw model output and the incident it must reference. Outputs:
|
||||
(suggestions, analysis); suggestions is empty when the response is
|
||||
unusable, which the caller treats as nothing to file. Never raises.
|
||||
|
||||
Validated with the same rules the triage schema applies, so an issue can
|
||||
never carry a shape the rest of the system would have rejected.
|
||||
"""
|
||||
|
||||
try:
|
||||
payload = json.loads(_first_object(str(raw_output or "")) or "")
|
||||
except Exception:
|
||||
return [], ""
|
||||
if not isinstance(payload, dict) or payload.get("incident_id") != incident_id:
|
||||
return [], ""
|
||||
if hermes_code_suggestion.validate(payload) is not None:
|
||||
return [], ""
|
||||
analysis = payload.get("analysis")
|
||||
return hermes_code_suggestion.from_payload(payload), analysis if isinstance(analysis, str) else ""
|
||||
|
||||
|
||||
def _first_object(raw: str) -> str | None:
|
||||
"""Return the first balanced top-level JSON object in raw text."""
|
||||
|
||||
depth = 0
|
||||
start = -1
|
||||
in_string = False
|
||||
escaped = False
|
||||
for index, char in enumerate(raw):
|
||||
if in_string:
|
||||
escaped = char == "\\" and not escaped
|
||||
if char == '"' and not escaped:
|
||||
in_string = False
|
||||
continue
|
||||
if char == '"' and depth > 0:
|
||||
in_string = True
|
||||
elif char == "{":
|
||||
if depth == 0:
|
||||
start = index
|
||||
depth += 1
|
||||
elif char == "}" and depth > 0:
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
return raw[start : index + 1]
|
||||
return None
|
||||
|
||||
|
||||
def _prompt(incident_id: str, issue: dict[str, Any], contents: str) -> str:
|
||||
"""Render the advice prompt for one finding and its file."""
|
||||
|
||||
finding = json.dumps(
|
||||
{key: issue.get(key) for key in ("rule", "severity", "type", "effort", "path", "line", "message")},
|
||||
separators=(",", ":"),
|
||||
)
|
||||
return (
|
||||
_PROMPT.replace("__INCIDENT_ID__", incident_id)
|
||||
.replace("__FINDING__", finding)
|
||||
.replace("__PATH__", str(issue.get("path") or ""))
|
||||
.replace("__CONTENTS__", contents[:_MAX_CONTEXT_CHARS])
|
||||
)
|
||||
|
||||
|
||||
def _context( # noqa: PLR0913 - the issue body needs every one of these
|
||||
project: str,
|
||||
rule: str,
|
||||
incident_id: str,
|
||||
issue: dict[str, Any],
|
||||
analysis: str,
|
||||
suggestions: list[Any],
|
||||
run_id: str | None,
|
||||
config: Any,
|
||||
) -> dict[str, Any]:
|
||||
"""Build the issue context for one advised finding."""
|
||||
|
||||
ui_url = str(getattr(config, "hermes_ui_url", "") or "")
|
||||
finding_url = hermes_sonar_client.issue_url(
|
||||
incident_id, str(getattr(config, "hermes_sonar_ui_url", "") or "")
|
||||
)
|
||||
reason = analysis or f"SonarQube reports {rule} in {issue.get('path')}."
|
||||
return {
|
||||
"incident_id": incident_id,
|
||||
"job": project,
|
||||
"build_number": issue.get("line"),
|
||||
"build_url": finding_url,
|
||||
"classification": rule,
|
||||
"confidence": None,
|
||||
"first_failed_gate": "",
|
||||
"finding_path": issue.get("path"),
|
||||
"reason": reason,
|
||||
# Says plainly why nobody patched it, so the issue does not read as a
|
||||
# failure of the automation.
|
||||
"authorize_reason": "no automated patch was possible for this finding",
|
||||
"facts": [
|
||||
{
|
||||
"statement": f"{issue.get('message')} ({issue.get('severity')}, {issue.get('effort')} estimated)",
|
||||
"source": "gitea",
|
||||
"reference": f"{issue.get('path')}:{issue.get('line')}",
|
||||
}
|
||||
],
|
||||
"inferences": [],
|
||||
"code_suggestions": suggestions,
|
||||
"run_id": run_id,
|
||||
"run_url": hermes_code_repair.run_url(ui_url, str(run_id or "")),
|
||||
}
|
||||
|
||||
|
||||
def _issue_config(config: Any, repo_cfg: dict) -> dict[str, Any]:
|
||||
"""Build the Gitea cfg for filing into this finding's repository."""
|
||||
|
||||
return {
|
||||
"gitea_base_url": repo_cfg.get("gitea_base_url") or getattr(config, "hermes_gitea_base_url", ""),
|
||||
"gitea_token": repo_cfg.get("gitea_token") or getattr(config, "hermes_gitea_token", ""),
|
||||
"owner": repo_cfg.get("owner"),
|
||||
"repo": repo_cfg.get("repo"),
|
||||
"timeout_seconds": repo_cfg.get("timeout_seconds") or 15.0,
|
||||
}
|
||||
|
||||
|
||||
def _record(storage: Any, incident_id: str, result: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Record one advice attempt, whatever it decided."""
|
||||
|
||||
try:
|
||||
storage.record_event(ADVICE_EVENT_TYPE, {"incident_id": incident_id, **result})
|
||||
except Exception:
|
||||
pass
|
||||
return result
|
||||
@ -149,6 +149,38 @@ def component_path(component: Any) -> str:
|
||||
return path.strip() if separator else ""
|
||||
|
||||
|
||||
INCIDENT_PREFIX = "sonar/"
|
||||
# The console opens one finding in place, with the rule, the effort estimate
|
||||
# and the offending lines highlighted.
|
||||
_ISSUE_PATH = "/project/issues?resolved=false&id={project}&open={key}"
|
||||
|
||||
|
||||
def issue_url(incident_id: str, ui_url: str) -> str:
|
||||
"""Build the console link for the finding an incident came from.
|
||||
|
||||
Inputs: an incident id shaped `sonar/<project>/<key>`, and the SonarQube
|
||||
base url. Outputs: the deep link, or "" for any incident that did not come
|
||||
from a sweep - a build-driven repair has no finding to point at.
|
||||
|
||||
Parsed from the incident id rather than threaded through the flow: the id
|
||||
already carries both halves, and inventing a second path for them to travel
|
||||
is a second thing that can disagree with the first.
|
||||
"""
|
||||
|
||||
base = str(ui_url or "").rstrip("/")
|
||||
text = str(incident_id or "")
|
||||
if not base or not text.startswith(INCIDENT_PREFIX):
|
||||
return ""
|
||||
# `sonar/<project>/<rule>/<key>`; the rule sits between them so the
|
||||
# project is read from the front and the key from the back.
|
||||
remainder = text[len(INCIDENT_PREFIX) :]
|
||||
project, separator, rest = remainder.partition("/")
|
||||
_, _, key = rest.rpartition("/") if "/" in rest else ("", "", rest)
|
||||
if not separator or not project.strip() or not key.strip():
|
||||
return ""
|
||||
return base + _ISSUE_PATH.format(project=project.strip(), key=key.strip())
|
||||
|
||||
|
||||
def _types(cfg: dict) -> tuple[str, ...]:
|
||||
"""Return the finding types this deployment allows, defaulting to all."""
|
||||
|
||||
|
||||
@ -27,7 +27,7 @@ from typing import Any
|
||||
|
||||
from ..settings import settings
|
||||
from ..utils.logging import get_logger
|
||||
from . import hermes_code_flow, hermes_sonar_client
|
||||
from . import hermes_code_flow, hermes_code_repair, hermes_sonar_advice, hermes_sonar_client
|
||||
|
||||
|
||||
logger = get_logger(__name__)
|
||||
@ -84,7 +84,7 @@ def sweep(storage: Any, config: Any, hermes_cfg: dict) -> dict[str, Any]:
|
||||
"hermes sonar sweep failed",
|
||||
extra={"event": SWEEP_EVENT_TYPE, "status": "error", "detail": str(exc)},
|
||||
)
|
||||
return {"proposed": 0, "skipped": [f"sweep_failed: {exc}"], "projects": 0}
|
||||
return {"proposed": 0, "advised": 0, "skipped": [f"sweep_failed: {exc}"], "projects": 0}
|
||||
|
||||
|
||||
def _sweep(storage: Any, config: Any, hermes_cfg: dict) -> dict[str, Any]:
|
||||
@ -92,11 +92,12 @@ def _sweep(storage: Any, config: Any, hermes_cfg: dict) -> dict[str, Any]:
|
||||
|
||||
projects = _projects(config)
|
||||
if not projects:
|
||||
return {"proposed": 0, "skipped": [_NO_PROJECTS], "projects": 0}
|
||||
return {"proposed": 0, "advised": 0, "skipped": [_NO_PROJECTS], "projects": 0}
|
||||
sonar_cfg = client_config(config)
|
||||
code_cfg = hermes_code_flow.code_config(config)
|
||||
budget = _max_per_sweep(config)
|
||||
proposed = 0
|
||||
advised = 0
|
||||
skipped: list[str] = []
|
||||
for project, job in sorted(projects.items()):
|
||||
if proposed >= budget:
|
||||
@ -106,12 +107,13 @@ def _sweep(storage: Any, config: Any, hermes_cfg: dict) -> dict[str, Any]:
|
||||
storage, project, job, sonar_cfg, code_cfg, hermes_cfg, config
|
||||
)
|
||||
proposed += outcome["proposed"]
|
||||
advised += outcome.get("advised", 0)
|
||||
skipped.extend(outcome["skipped"])
|
||||
logger.info(
|
||||
"hermes sonar sweep finished",
|
||||
extra={"event": SWEEP_EVENT_TYPE, "status": "ok", "detail": f"proposed={proposed}"},
|
||||
)
|
||||
return {"proposed": proposed, "skipped": skipped, "projects": len(projects)}
|
||||
return {"proposed": proposed, "advised": advised, "skipped": skipped, "projects": len(projects)}
|
||||
|
||||
|
||||
def _propose_for_project( # noqa: PLR0913 - one project needs every config the flow does
|
||||
@ -131,9 +133,9 @@ def _propose_for_project( # noqa: PLR0913 - one project needs every config the
|
||||
issues, error = hermes_sonar_client.fetch_issues(sonar_cfg, project)
|
||||
if error:
|
||||
return {"proposed": 0, "skipped": [f"{project}: {error}"]}
|
||||
chosen = select_issue(issues, repo_cfg, config)
|
||||
if chosen is None:
|
||||
return {"proposed": 0, "skipped": [f"{project}: no mechanically fixable finding"]}
|
||||
open_rules = proposed_rules(repo_cfg, project)
|
||||
chosen = select_issue(issues, repo_cfg, config, open_rules)
|
||||
if chosen is not None:
|
||||
result = hermes_code_flow.propose_code_fix(
|
||||
storage,
|
||||
incident_id=incident_id(project, chosen),
|
||||
@ -145,14 +147,100 @@ def _propose_for_project( # noqa: PLR0913 - one project needs every config the
|
||||
)
|
||||
if result.get("status") == "pr_opened":
|
||||
return {"proposed": 1, "skipped": []}
|
||||
return {"proposed": 0, "skipped": [f"{project}: {result.get('reason')}"]}
|
||||
# A declined pull request is not a dead end. The finding is still real
|
||||
# and a person still has to deal with it, so fall through to advice.
|
||||
return _advise(storage, config, hermes_cfg, repo_cfg, project, chosen, str(result.get("reason")))
|
||||
advisable = select_advice_issue(issues, repo_cfg, config, open_rules)
|
||||
if advisable is None:
|
||||
return {"proposed": 0, "skipped": [f"{project}: no new finding to act on"]}
|
||||
return _advise(storage, config, hermes_cfg, repo_cfg, project, advisable, "too large to patch")
|
||||
|
||||
|
||||
def select_issue(issues: list[dict], repo_cfg: dict, config: Any) -> dict[str, Any] | None:
|
||||
def _advise( # noqa: PLR0913 - the advice call needs the whole proposal context
|
||||
storage: Any,
|
||||
config: Any,
|
||||
hermes_cfg: dict,
|
||||
repo_cfg: dict,
|
||||
project: str,
|
||||
issue: dict,
|
||||
why: str,
|
||||
) -> dict[str, Any]:
|
||||
"""File an issue with a suggested fix for a finding no patch covered."""
|
||||
|
||||
if not _advice_enabled(config):
|
||||
return {"proposed": 0, "skipped": [f"{project}: {why} (advice disabled)"]}
|
||||
outcome = hermes_sonar_advice.advise(storage, config, hermes_cfg, repo_cfg, project, issue)
|
||||
if outcome.get("filed"):
|
||||
return {"proposed": 1, "skipped": [], "advised": 1}
|
||||
return {"proposed": 0, "skipped": [f"{project}: {why} -> {outcome.get('reason')}"]}
|
||||
|
||||
|
||||
def select_advice_issue(
|
||||
issues: list[dict], repo_cfg: dict, config: Any, open_rules: set[str] | None = None
|
||||
) -> dict[str, Any] | None:
|
||||
"""Pick the finding most worth explaining to a person.
|
||||
|
||||
Inputs and ordering as `select_issue`, minus the effort ceiling: this path
|
||||
exists precisely for the findings too large to patch, which on this
|
||||
instance is most of the backlog. Still bounded to writable paths, because
|
||||
advising on a file nobody would edit is noise.
|
||||
"""
|
||||
|
||||
seen = open_rules or set()
|
||||
eligible = [
|
||||
issue
|
||||
for issue in issues
|
||||
if str(issue.get("rule") or "") not in seen
|
||||
and _is_writable(str(issue.get("path") or ""), repo_cfg)
|
||||
]
|
||||
if not eligible:
|
||||
return None
|
||||
return sorted(eligible, key=_selection_key)[0]
|
||||
|
||||
|
||||
def _advice_enabled(config: Any) -> bool:
|
||||
"""Report whether findings with no patch may be filed as issues."""
|
||||
|
||||
return bool(getattr(config, "hermes_sonar_advice_enabled", False))
|
||||
|
||||
|
||||
def proposed_rules(repo_cfg: dict, project: str) -> set[str]:
|
||||
"""Return the rules this project already has an open proposal for.
|
||||
|
||||
Inputs: the resolved per-repo cfg and the SonarQube project key. Outputs:
|
||||
the rule ids, read back from the open pull requests' incident ids.
|
||||
|
||||
One rule is usually one root cause spread across many files: S2208 appears
|
||||
in three Ariadne modules, and the cognitive-complexity rule in dozens. One
|
||||
finding at a time with no memory would open a near-identical pull request
|
||||
for every instance, and a reviewer facing thirty of those reads none of
|
||||
them. Until the open one is dealt with, the rest of that rule waits.
|
||||
|
||||
Fails open, like every other duplicate check here: an unreadable list
|
||||
yields an empty set, so the cost of a lookup failure is one extra proposal
|
||||
rather than a whole rule silently going unreported.
|
||||
"""
|
||||
|
||||
incidents, _error = hermes_code_repair.open_proposal_incidents(repo_cfg)
|
||||
prefix = f"{hermes_sonar_client.INCIDENT_PREFIX}{project}/"
|
||||
rules = set()
|
||||
for incident in incidents:
|
||||
if not incident.startswith(prefix):
|
||||
continue
|
||||
rule, separator, _key = incident[len(prefix) :].rpartition("/")
|
||||
if separator and rule:
|
||||
rules.add(rule)
|
||||
return rules
|
||||
|
||||
|
||||
def select_issue(
|
||||
issues: list[dict], repo_cfg: dict, config: Any, open_rules: set[str] | None = None
|
||||
) -> dict[str, Any] | None:
|
||||
"""Pick the single most mechanically fixable finding for one project.
|
||||
|
||||
Inputs: the normalized findings, the resolved per-repo cfg (whose write
|
||||
allowlist decides what is patchable at all), and the settings object.
|
||||
allowlist decides what is patchable at all), the settings object, and the
|
||||
rules that already have an open proposal.
|
||||
Outputs: one finding, or None when none qualify.
|
||||
|
||||
Ordered by effort, then severity, then key. Effort leads because it is the
|
||||
@ -163,10 +251,12 @@ def select_issue(issues: list[dict], repo_cfg: dict, config: Any) -> dict[str, A
|
||||
"""
|
||||
|
||||
ceiling = _max_effort_minutes(config)
|
||||
seen = open_rules or set()
|
||||
eligible = [
|
||||
issue
|
||||
for issue in issues
|
||||
if _is_writable(str(issue.get("path") or ""), repo_cfg)
|
||||
if str(issue.get("rule") or "") not in seen
|
||||
and _is_writable(str(issue.get("path") or ""), repo_cfg)
|
||||
and effort_minutes(issue.get("effort")) is not None
|
||||
and (effort_minutes(issue.get("effort")) or 0) <= ceiling
|
||||
]
|
||||
@ -236,9 +326,14 @@ def bundle_for(project: str, issue: dict[str, Any]) -> dict[str, Any]:
|
||||
|
||||
|
||||
def incident_id(project: str, issue: dict[str, Any]) -> str:
|
||||
"""Name the incident for one finding, stable across sweeps."""
|
||||
"""Name the incident for one finding, stable across sweeps.
|
||||
|
||||
return f"sonar/{project}/{issue.get('key')}"
|
||||
Carries the rule as well as the key so an open proposal announces which
|
||||
root cause is already being reviewed. The alternative was a second index
|
||||
of proposed rules, which could disagree with the pull requests themselves.
|
||||
"""
|
||||
|
||||
return f"sonar/{project}/{issue.get('rule')}/{issue.get('key')}"
|
||||
|
||||
|
||||
def client_config(config: Any) -> dict[str, Any]:
|
||||
|
||||
@ -26,7 +26,7 @@ jenkins.first_failed_stage names the pipeline stage that failed when the build r
|
||||
The jenkins.console_failures array holds excerpts around detected failure markers in chronological order; the earliest region usually contains the first enforced failure, and jenkins.console_tail is the end of the build which often only shows downstream noise.
|
||||
Distinguish facts from inference.
|
||||
Return ONLY a single JSON object with exactly these keys and no others:
|
||||
{"incident_id": "<must equal __INCIDENT_ID__>", "classification": "<short snake_case name for what actually failed>", "confidence": <0..1>, "facts": [{"statement": "...", "source": "jenkins|opensearch|victoriametrics|kubernetes|flux|gitea", "reference": "..."}], "inferences": ["..."], "first_failed_gate": "<string>", "requested_action": <object or null>, "suggested_remediation": <object or null>, "human_required": <bool>, "reason": "<string>"}
|
||||
{"incident_id": "<must equal __INCIDENT_ID__>", "classification": "<short snake_case name for what actually failed>", "confidence": <0..1>, "facts": [{"statement": "...", "source": "jenkins|opensearch|victoriametrics|kubernetes|flux|gitea", "reference": "..."}], "inferences": ["..."], "first_failed_gate": "<string>", "requested_action": <object or null>, "suggested_remediation": <object or null>, "code_suggestions": <array or null>, "human_required": <bool>, "reason": "<string>"}
|
||||
You are diagnosing only; you do not execute anything. Ariadne separately validates and executes any requested action under its own authorization policy, and will refuse anything its own reading of the evidence does not support.
|
||||
Write the reason and the inferences for an engineer who maintains this service and who has no knowledge of how this triage system is configured. Describe the failure and what a fix would involve. Do not discuss classifications, actions, policies, or which of them are permitted; those are Ariadne's concern and are meaningless in the service's issue tracker.
|
||||
Three classifications have a predefined remediation. Use one only when the evidence plainly shows that failure; otherwise leave requested_action null.
|
||||
@ -35,6 +35,7 @@ Use workspace_storage_exhausted with requested_action {"type": "run_ariadne_job"
|
||||
Use jenkins_agent_provisioning_failure with requested_action {"type": "run_ariadne_job", "id": "clear_stuck_agent_pods"} when the evidence shows the build never got an agent (all nodes of a label offline, an agent pod stuck ContainerCreating or Pending, or an error in provisioning) rather than failing once it was running.
|
||||
Otherwise leave requested_action null.
|
||||
When you leave requested_action null because none of the remediations above fits, and you can name a remediation that a maintainer could reasonably automate for this failure, set suggested_remediation to {"action_id": "<snake_case name for the remediation you are proposing>", "summary": "<what it would do, in one or two sentences>", "evidence_required": "<what would have to be present in the evidence before running it is safe>"}; otherwise set it to null. This proposes work for a maintainer to build and does not request anything: nothing you name here can be executed, and it must be null whenever requested_action is set. Propose one only when the same remediation would be correct for any build failing this way, not merely for this build.
|
||||
When the failure is in this repository's own source or tests and you can see what the change should be, set code_suggestions to at most three {"path": "<repository-relative file>", "explanation": "<what is wrong there and why this change fixes it>", "code": "<the suggested code>"}; otherwise set it to null. Write them for a maintainer who will read the change and then make it themselves: nothing you put here is applied, validated against the file, or committed anywhere. Suggest the fix you actually believe is right even when it spans several places or is too large to express as one edit, and prefer correcting the code under test over changing a test's expectations.
|
||||
Set human_required to true when decisive evidence is missing or the failure needs a judgement only a maintainer can make; otherwise set it false and say plainly what you believe is wrong.
|
||||
Do not perform mutations."""
|
||||
|
||||
|
||||
@ -319,11 +319,13 @@ class Settings:
|
||||
hermes_sonar_cron: str
|
||||
hermes_sonar_enabled: bool
|
||||
hermes_sonar_url: str
|
||||
hermes_sonar_ui_url: str
|
||||
hermes_sonar_token: str
|
||||
hermes_sonar_projects: dict[str, str]
|
||||
hermes_sonar_types: list[str]
|
||||
hermes_sonar_severities: list[str]
|
||||
hermes_sonar_max_per_sweep: int
|
||||
hermes_sonar_advice_enabled: bool
|
||||
hermes_sonar_max_effort_minutes: int
|
||||
hermes_sonar_timeout_seconds: float
|
||||
|
||||
|
||||
@ -60,6 +60,7 @@ def _hermes_autotriage_config() -> dict[str, Any]:
|
||||
"ARIADNE_HERMES_SONAR_URL", "http://sonarqube.quality.svc.cluster.local:9000"
|
||||
).rstrip("/"),
|
||||
"hermes_sonar_token": _env("ARIADNE_HERMES_SONAR_TOKEN", ""),
|
||||
"hermes_sonar_ui_url": _env("ARIADNE_HERMES_SONAR_UI_URL", ""),
|
||||
"hermes_sonar_projects": _pair_map(_env("ARIADNE_HERMES_SONAR_PROJECTS", "")),
|
||||
"hermes_sonar_types": [
|
||||
item.strip()
|
||||
@ -72,6 +73,7 @@ def _hermes_autotriage_config() -> dict[str, Any]:
|
||||
if item.strip()
|
||||
],
|
||||
"hermes_sonar_max_per_sweep": _env_int("ARIADNE_HERMES_SONAR_MAX_PER_SWEEP", 1),
|
||||
"hermes_sonar_advice_enabled": _env_bool("ARIADNE_HERMES_SONAR_ADVICE_ENABLED", "false"),
|
||||
"hermes_sonar_max_effort_minutes": _env_int("ARIADNE_HERMES_SONAR_MAX_EFFORT_MINUTES", 20),
|
||||
"hermes_sonar_timeout_seconds": _env_float("ARIADNE_HERMES_SONAR_TIMEOUT_SECONDS", 20.0),
|
||||
"hermes_demo_namespace": _env("ARIADNE_HERMES_DEMO_NAMESPACE", "hermes-triage-demo"),
|
||||
|
||||
@ -37,6 +37,7 @@ def _hermes_cfg() -> dict:
|
||||
def _code_cfg(**overrides) -> dict: # type: ignore[no-untyped-def]
|
||||
base = {
|
||||
"hermes_ui_url": "",
|
||||
"sonar_ui_url": "",
|
||||
"candidate_path": "src/discount.py",
|
||||
"allowed_path_prefixes": ["src/"],
|
||||
"allowed_suffixes": [".py"],
|
||||
|
||||
@ -3,6 +3,8 @@ from __future__ import annotations
|
||||
import base64
|
||||
import json
|
||||
|
||||
import httpx
|
||||
|
||||
from ariadne.services import hermes_code_repair as module
|
||||
from ariadne.services.hermes_code_patch import ProposedPatch
|
||||
|
||||
@ -363,9 +365,11 @@ def test_the_pull_request_names_the_hermes_run(monkeypatch) -> None:
|
||||
)
|
||||
|
||||
body = calls["requests"][0][2]["json"]["body"]
|
||||
assert "**Hermes run:** `run_a5af87af`" in body
|
||||
assert "https://agent.bstein.dev." in body
|
||||
assert "every tool call it made" in body
|
||||
assert (
|
||||
"**Hermes run:** [run_a5af87af]"
|
||||
"(https://agent.bstein.dev/chat?resume=run_a5af87af)"
|
||||
) in body
|
||||
assert "the tools it called" in body
|
||||
|
||||
|
||||
def test_the_pull_request_omits_the_link_when_no_ui_is_configured(monkeypatch) -> None:
|
||||
@ -386,3 +390,90 @@ def test_the_pull_request_stays_readable_without_a_run_id(monkeypatch) -> None:
|
||||
body = calls["requests"][0][2]["json"]["body"]
|
||||
assert "Hermes run" not in body
|
||||
assert "requires human review" in body
|
||||
|
||||
|
||||
def test_the_run_link_reopens_the_run_in_the_hermes_console() -> None:
|
||||
"""One helper, so the pull request and the demo monitor cannot diverge."""
|
||||
|
||||
assert module.run_url("https://agent.bstein.dev/", "run_x") == (
|
||||
"https://agent.bstein.dev/chat?resume=run_x"
|
||||
)
|
||||
assert module.run_url("", "run_x") == ""
|
||||
assert module.run_url("https://agent.bstein.dev", " ") == ""
|
||||
|
||||
|
||||
def test_a_sweep_proposal_links_to_the_finding_that_caused_it(monkeypatch) -> None:
|
||||
"""The reviewer's first question is what the finding said."""
|
||||
|
||||
calls = _install_http(monkeypatch, [FakeResponse(201, {"number": 6, "html_url": "u"})])
|
||||
module.open_pull_request(
|
||||
{**_cfg(), "sonar_ui_url": "https://quality.bstein.dev"},
|
||||
"sonar/ariadne/AZ2y0FYFKy9i4pkIpNlV",
|
||||
"run_x",
|
||||
BRANCH,
|
||||
_patch(),
|
||||
"analysis",
|
||||
)
|
||||
|
||||
body = calls["requests"][0][2]["json"]["body"]
|
||||
assert "**SonarQube finding:** https://quality.bstein.dev/project/issues" in body
|
||||
assert "open=AZ2y0FYFKy9i4pkIpNlV" in body
|
||||
|
||||
|
||||
def test_a_build_driven_proposal_has_no_finding_line(monkeypatch) -> None:
|
||||
calls = _install_http(monkeypatch, [FakeResponse(201, {"number": 6, "html_url": "u"})])
|
||||
module.open_pull_request(
|
||||
{**_cfg(), "sonar_ui_url": "https://quality.bstein.dev"},
|
||||
INCIDENT_ID,
|
||||
"run_x",
|
||||
BRANCH,
|
||||
_patch(),
|
||||
"analysis",
|
||||
)
|
||||
|
||||
assert "SonarQube finding" not in calls["requests"][0][2]["json"]["body"]
|
||||
|
||||
|
||||
def test_open_proposal_incidents_reads_them_from_the_titles(monkeypatch) -> None:
|
||||
"""The pull requests are the thing that exists; an index could disagree."""
|
||||
|
||||
calls = _install_http(
|
||||
monkeypatch,
|
||||
[
|
||||
FakeResponse(
|
||||
200,
|
||||
[
|
||||
{"title": "fix(hermes): repair sonar/ariadne/python:S2208/AZ1"},
|
||||
{"title": "fix(hermes): repair ariadne/408"},
|
||||
{"title": "chore: something a person opened"},
|
||||
{"title": ""},
|
||||
"not-a-dict",
|
||||
],
|
||||
)
|
||||
],
|
||||
)
|
||||
incidents, error = module.open_proposal_incidents(_cfg())
|
||||
|
||||
assert error is None
|
||||
assert incidents == ["sonar/ariadne/python:S2208/AZ1", "ariadne/408"]
|
||||
assert calls["requests"][0][2]["params"]["state"] == "open"
|
||||
|
||||
|
||||
def test_open_proposal_incidents_fails_open(monkeypatch) -> None:
|
||||
"""A duplicate is noise; a suppressed rule is lost work."""
|
||||
|
||||
_install_http(monkeypatch, [FakeResponse(500, None)])
|
||||
assert module.open_proposal_incidents(_cfg()) == ([], "open proposal lookup http 500")
|
||||
|
||||
_install_http(monkeypatch, [FakeResponse(200, {"not": "a list"})])
|
||||
assert module.open_proposal_incidents(_cfg()) == ([], "open proposal payload is not a list")
|
||||
|
||||
_install_http(monkeypatch, [httpx.ConnectError("refused")])
|
||||
incidents, error = module.open_proposal_incidents(_cfg())
|
||||
assert incidents == []
|
||||
assert "open proposal lookup failed" in error
|
||||
|
||||
assert module.open_proposal_incidents({**_cfg(), "gitea_base_url": ""}) == (
|
||||
[],
|
||||
"gitea base url is empty",
|
||||
)
|
||||
|
||||
174
tests/test_hermes_code_suggestion.py
Normal file
174
tests/test_hermes_code_suggestion.py
Normal file
@ -0,0 +1,174 @@
|
||||
"""Tests for code suggestions carried by an escalated diagnosis."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from ariadne.services import hermes_autotriage_decision as decision_module
|
||||
from ariadne.services import hermes_code_suggestion as module
|
||||
from ariadne.services import hermes_incident_body as body
|
||||
|
||||
|
||||
SUGGESTION = {
|
||||
"path": "ariadne/utils/errors.py",
|
||||
"explanation": "safe_error_detail drops the response body, so the assertion never sees it.",
|
||||
"code": 'def safe_error_detail(exc):\n return f"http {exc.response.status_code}: {exc.response.text}"',
|
||||
}
|
||||
|
||||
|
||||
def _payload(**overrides):
|
||||
payload = {
|
||||
"incident_id": "ariadne/408",
|
||||
"classification": "pytest_test_failure",
|
||||
"confidence": 0.9,
|
||||
"facts": [],
|
||||
"inferences": [],
|
||||
"first_failed_gate": "tests",
|
||||
"requested_action": None,
|
||||
"human_required": True,
|
||||
"reason": "a repository test failure",
|
||||
}
|
||||
payload.update(overrides)
|
||||
return payload
|
||||
|
||||
|
||||
def _parse(payload):
|
||||
return decision_module.parse_triage_response(json.dumps(payload), "ariadne/408")
|
||||
|
||||
|
||||
def test_a_response_without_the_field_still_validates() -> None:
|
||||
"""Responses written before the field existed must keep parsing."""
|
||||
|
||||
outcome = _parse(_payload())
|
||||
|
||||
assert outcome.valid
|
||||
assert outcome.decision.code_suggestions == []
|
||||
|
||||
|
||||
def test_suggestions_are_parsed() -> None:
|
||||
outcome = _parse(_payload(code_suggestions=[dict(SUGGESTION)]))
|
||||
|
||||
suggestion = outcome.decision.code_suggestions[0]
|
||||
assert suggestion.path == "ariadne/utils/errors.py"
|
||||
assert "drops the response body" in suggestion.explanation
|
||||
assert "def safe_error_detail" in suggestion.code
|
||||
|
||||
|
||||
def test_a_null_field_is_accepted() -> None:
|
||||
assert _parse(_payload(code_suggestions=None)).decision.code_suggestions == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("suggestions", "expected"),
|
||||
[
|
||||
("nope", "must be a list or null"),
|
||||
([dict(SUGGESTION)] * 4, "at most 3 suggestions"),
|
||||
(["not-an-object"], "[0] must be an object"),
|
||||
([{"path": "a"}], "[0] must have exactly path, explanation, code"),
|
||||
([{**SUGGESTION, "code": " "}], "[0] code must be a non-empty string"),
|
||||
([{**SUGGESTION, "path": 7}], "[0] path must be a non-empty string"),
|
||||
],
|
||||
)
|
||||
def test_a_malformed_field_is_rejected(suggestions, expected) -> None:
|
||||
outcome = _parse(_payload(code_suggestions=suggestions))
|
||||
|
||||
assert not outcome.valid
|
||||
assert expected in outcome.reject_reason
|
||||
|
||||
|
||||
def test_a_long_suggestion_is_clipped() -> None:
|
||||
outcome = _parse(
|
||||
_payload(code_suggestions=[{**SUGGESTION, "code": "x" * 5000, "explanation": "y " * 900}])
|
||||
)
|
||||
|
||||
suggestion = outcome.decision.code_suggestions[0]
|
||||
assert len(suggestion.code) <= 1700
|
||||
assert len(suggestion.explanation) <= 700
|
||||
|
||||
|
||||
def test_a_suggestion_never_becomes_an_action() -> None:
|
||||
"""It is advice in an issue; the gates must not see it as anything else."""
|
||||
|
||||
outcome = _parse(_payload(code_suggestions=[dict(SUGGESTION)], human_required=False))
|
||||
allowed, reason = decision_module.authorize_action(
|
||||
outcome,
|
||||
{
|
||||
"allowed_actions": ["retry_transient_infra"],
|
||||
"action_classifications": {"pytest_test_failure": "retry_transient_infra"},
|
||||
"autoremediation_enabled": True,
|
||||
"min_confidence": 0.5,
|
||||
},
|
||||
prior_action_count=0,
|
||||
build_is_terminal_failure=True,
|
||||
job_allowlisted=True,
|
||||
evidence_has_signature=True,
|
||||
)
|
||||
|
||||
assert not allowed
|
||||
assert reason == "requested_action_missing"
|
||||
|
||||
|
||||
def test_the_issue_section_says_the_change_was_not_applied() -> None:
|
||||
"""A code block in an issue reads as a change that was made."""
|
||||
|
||||
section = module.issue_section(module.from_payload({"code_suggestions": [dict(SUGGESTION)]}))
|
||||
|
||||
assert "## Suggested fix (not applied)" in section
|
||||
assert "not written, validated, or pushed anywhere" in section
|
||||
assert "**`ariadne/utils/errors.py`**" in section
|
||||
assert section.count("```") == 2
|
||||
assert module.issue_section(None) == ""
|
||||
assert module.issue_section([]) == ""
|
||||
|
||||
|
||||
def test_a_fence_inside_a_suggestion_cannot_break_the_block() -> None:
|
||||
section = module.issue_section(
|
||||
module.from_payload({"code_suggestions": [{**SUGGESTION, "code": "a\n```\nb"}]})
|
||||
)
|
||||
|
||||
assert section.count("```") == 2
|
||||
|
||||
|
||||
def test_the_detail_recorded_for_the_audit_trail() -> None:
|
||||
detail = module.as_detail(module.from_payload({"code_suggestions": [dict(SUGGESTION)]}))
|
||||
|
||||
assert detail[0]["path"] == "ariadne/utils/errors.py"
|
||||
assert module.as_detail(None) == []
|
||||
|
||||
|
||||
def test_from_payload_tolerates_junk() -> None:
|
||||
assert module.from_payload({}) == []
|
||||
assert module.from_payload({"code_suggestions": "nope"}) == []
|
||||
assert module.from_payload({"code_suggestions": ["x"]}) == []
|
||||
|
||||
|
||||
def test_the_issue_body_carries_the_suggestion_and_keeps_its_marker() -> None:
|
||||
rendered = body.issue_body(
|
||||
{
|
||||
"incident_id": "ariadne/408",
|
||||
"job": "ariadne",
|
||||
"build_number": 408,
|
||||
"classification": "pytest_test_failure",
|
||||
"reason": "a repository test failure",
|
||||
"run_id": "run-1",
|
||||
"code_suggestions": module.from_payload({"code_suggestions": [dict(SUGGESTION)]}),
|
||||
}
|
||||
)
|
||||
|
||||
assert "## Suggested fix (not applied)" in rendered
|
||||
assert "def safe_error_detail" in rendered
|
||||
assert rendered.index("Suggested fix") < rendered.index("## Links")
|
||||
assert rendered.rstrip().endswith("-->")
|
||||
|
||||
|
||||
def test_the_prompt_asks_for_suggestions_without_promising_to_apply_them() -> None:
|
||||
from ariadne.services import hermes_triage_prompt
|
||||
|
||||
prompt = hermes_triage_prompt.build_prompt("ariadne/408", "ariadne", {})
|
||||
|
||||
assert '"code_suggestions": <array or null>' in prompt
|
||||
assert "at most three" in prompt
|
||||
assert "nothing you put here is applied" in prompt
|
||||
assert "prefer correcting the code under test" in prompt
|
||||
@ -7,6 +7,23 @@ from types import SimpleNamespace
|
||||
from ariadne.services import hermes_incident_body as body
|
||||
from ariadne.services import hermes_incident_issue as module
|
||||
|
||||
|
||||
_BASE = {"incident_id": "ariadne/408", "job": "ariadne", "build_number": 408}
|
||||
|
||||
|
||||
def _diagnosis(**overrides):
|
||||
"""A completed diagnosis, as maybe_file_issue receives it."""
|
||||
|
||||
diagnosis = {
|
||||
"bundle": {"jenkins": {"url": "https://ci.example/job/ariadne/408/"}},
|
||||
"outcome": None,
|
||||
"authorize_reason": "human_required",
|
||||
"run_id": "run_a5af87af",
|
||||
}
|
||||
diagnosis.update(overrides)
|
||||
return diagnosis
|
||||
|
||||
|
||||
def test_body_does_not_claim_a_diagnosis_that_never_happened() -> None:
|
||||
"""A hung build is escalated without any model call.
|
||||
|
||||
@ -99,3 +116,21 @@ def test_explicit_classification_is_used_only_without_a_decision() -> None:
|
||||
"classification": "build_exceeded_time_cap", "run_id": "run_x"},
|
||||
)
|
||||
assert ctx["classification"] == "pytest_test_failure"
|
||||
|
||||
|
||||
def test_the_issue_footer_links_to_the_hermes_run() -> None:
|
||||
"""A bare id is something to copy; a link is a page to open."""
|
||||
|
||||
context = module.issue_context(_BASE, _diagnosis(), "https://agent.bstein.dev")
|
||||
rendered = body.issue_body(context)
|
||||
|
||||
assert context["run_url"].startswith("https://agent.bstein.dev/chat?resume=")
|
||||
assert f"run [{context['run_id']}]({context['run_url']})" in rendered
|
||||
|
||||
|
||||
def test_the_footer_falls_back_to_a_bare_id_without_a_console() -> None:
|
||||
context = module.issue_context(_BASE, _diagnosis())
|
||||
rendered = body.issue_body(context)
|
||||
|
||||
assert context["run_url"] == ""
|
||||
assert f"run `{context['run_id']}`" in rendered
|
||||
|
||||
@ -493,3 +493,4 @@ def test_broken_repo_map_entries_file_nothing(monkeypatch) -> None:
|
||||
is None
|
||||
)
|
||||
assert calls["requests"] == []
|
||||
|
||||
|
||||
289
tests/test_hermes_sonar_advice.py
Normal file
289
tests/test_hermes_sonar_advice.py
Normal file
@ -0,0 +1,289 @@
|
||||
"""Tests for filing a suggested fix when a finding cannot become a patch."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from ariadne.services import hermes_sonar_advice as module
|
||||
|
||||
|
||||
ISSUE = {
|
||||
"key": "AZ1",
|
||||
"rule": "python:S3776",
|
||||
"severity": "CRITICAL",
|
||||
"type": "CODE_SMELL",
|
||||
"effort": "11min",
|
||||
"path": "ariadne/services/thing.py",
|
||||
"line": 42,
|
||||
"message": "Refactor this function to reduce its Cognitive Complexity from 21 to the 15 allowed.",
|
||||
}
|
||||
|
||||
REPO_CFG = {
|
||||
"owner": "bstein",
|
||||
"repo": "ariadne",
|
||||
"gitea_base_url": "https://scm.example",
|
||||
"gitea_token": "t",
|
||||
"base_branch": "master",
|
||||
}
|
||||
|
||||
INCIDENT = "sonar/ariadne/python:S3776/AZ1"
|
||||
|
||||
SUGGESTION = {
|
||||
"path": "ariadne/services/thing.py",
|
||||
"explanation": "Extract the validation branch into its own helper.",
|
||||
"code": "def _validate(row):\n return bool(row)",
|
||||
}
|
||||
|
||||
|
||||
def _config(**overrides):
|
||||
values = {
|
||||
"hermes_ui_url": "https://agent.example",
|
||||
"hermes_sonar_ui_url": "https://quality.example",
|
||||
"hermes_gitea_base_url": "https://scm.example",
|
||||
"hermes_gitea_token": "t",
|
||||
}
|
||||
values.update(overrides)
|
||||
return SimpleNamespace(**values)
|
||||
|
||||
|
||||
class _Storage:
|
||||
def __init__(self):
|
||||
self.events = []
|
||||
|
||||
def record_event(self, event_type, detail):
|
||||
self.events.append((event_type, detail))
|
||||
|
||||
|
||||
def _response(**overrides):
|
||||
payload = {
|
||||
"incident_id": INCIDENT,
|
||||
"analysis": "The function branches five ways over the same row.",
|
||||
"code_suggestions": [dict(SUGGESTION)],
|
||||
"human_required": True,
|
||||
"reason": "needs a maintainer",
|
||||
}
|
||||
payload.update(overrides)
|
||||
return json.dumps(payload)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def wiring(monkeypatch):
|
||||
state = {
|
||||
"existing": {"found": False},
|
||||
"contents": "def thing():\n pass\n",
|
||||
"fetch_error": None,
|
||||
"run": SimpleNamespace(status="completed", output=_response(), run_id="run_x"),
|
||||
"created": {"issue_number": 9, "url": "https://scm.example/issues/9", "error": None},
|
||||
"filed_context": [],
|
||||
}
|
||||
monkeypatch.setattr(
|
||||
module.hermes_incident_issue, "find_open_incident_issue",
|
||||
lambda cfg, job, classification, incident: state["existing"],
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
module.hermes_code_repair, "fetch_file",
|
||||
lambda cfg, path: (state["contents"], state["fetch_error"]),
|
||||
)
|
||||
monkeypatch.setattr(module.hermes_agent_client, "run_triage", lambda cfg, prompt: state["run"])
|
||||
|
||||
def _create(cfg, context):
|
||||
state["filed_context"].append(context)
|
||||
return state["created"]
|
||||
|
||||
monkeypatch.setattr(module.hermes_incident_issue, "create_incident_issue", _create)
|
||||
return state
|
||||
|
||||
|
||||
def test_a_finding_becomes_an_issue_carrying_the_suggested_code(wiring) -> None:
|
||||
storage = _Storage()
|
||||
|
||||
result = module.advise(storage, _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))
|
||||
|
||||
assert result == {"filed": True, "reason": "issue_filed", "url": "https://scm.example/issues/9"}
|
||||
context = wiring["filed_context"][0]
|
||||
assert context["incident_id"] == INCIDENT
|
||||
assert context["classification"] == "python:S3776"
|
||||
assert context["code_suggestions"][0].code.startswith("def _validate")
|
||||
assert context["reason"].startswith("The function branches")
|
||||
|
||||
|
||||
def test_the_issue_links_to_both_the_finding_and_the_run(wiring) -> None:
|
||||
module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))
|
||||
|
||||
context = wiring["filed_context"][0]
|
||||
assert context["build_url"] == (
|
||||
"https://quality.example/project/issues?resolved=false&id=ariadne&open=AZ1"
|
||||
)
|
||||
assert context["run_url"] == "https://agent.example/chat?resume=run_x"
|
||||
|
||||
|
||||
def test_the_issue_says_why_nothing_was_patched(wiring) -> None:
|
||||
"""Otherwise it reads as the automation having failed."""
|
||||
|
||||
module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))
|
||||
|
||||
assert "no automated patch was possible" in wiring["filed_context"][0]["authorize_reason"]
|
||||
|
||||
|
||||
def test_a_rule_already_filed_is_not_filed_again(wiring) -> None:
|
||||
"""One rule is one root cause; an issue per instance buries the repo."""
|
||||
|
||||
wiring["existing"] = {"found": True, "url": "https://scm.example/issues/3"}
|
||||
|
||||
result = module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))
|
||||
|
||||
assert result["filed"] is False
|
||||
assert result["reason"] == "issue_already_open"
|
||||
assert wiring["filed_context"] == []
|
||||
|
||||
|
||||
def test_an_unreadable_file_files_nothing(wiring) -> None:
|
||||
wiring["contents"] = None
|
||||
wiring["fetch_error"] = "http 404"
|
||||
|
||||
result = module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))
|
||||
|
||||
assert result["filed"] is False
|
||||
assert "file_fetch_failed" in result["reason"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("status", "output"),
|
||||
[("failed", ""), ("completed", ""), ("timeout", "{}")],
|
||||
)
|
||||
def test_a_run_that_produced_nothing_files_nothing(wiring, status, output) -> None:
|
||||
wiring["run"] = SimpleNamespace(status=status, output=output, run_id="r")
|
||||
|
||||
assert module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))["filed"] is False
|
||||
|
||||
|
||||
def test_a_response_with_no_suggestions_files_nothing(wiring) -> None:
|
||||
wiring["run"] = SimpleNamespace(
|
||||
status="completed", output=_response(code_suggestions=[]), run_id="r"
|
||||
)
|
||||
|
||||
result = module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))
|
||||
|
||||
assert result["reason"] == "no_suggestions_returned"
|
||||
|
||||
|
||||
def test_a_failed_creation_is_reported(wiring) -> None:
|
||||
wiring["created"] = {"issue_number": None, "url": None, "error": "http 500"}
|
||||
|
||||
assert module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))["filed"] is False
|
||||
|
||||
|
||||
def test_every_attempt_is_recorded(wiring) -> None:
|
||||
storage = _Storage()
|
||||
module.advise(storage, _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))
|
||||
|
||||
assert storage.events[0][0] == module.ADVICE_EVENT_TYPE
|
||||
assert storage.events[0][1]["incident_id"] == INCIDENT
|
||||
|
||||
|
||||
def test_advice_never_raises(monkeypatch) -> None:
|
||||
"""It runs after a pull request was already declined; it must not erase it."""
|
||||
|
||||
monkeypatch.setattr(
|
||||
module.hermes_incident_issue, "find_open_incident_issue",
|
||||
lambda *a, **k: (_ for _ in ()).throw(RuntimeError("boom")),
|
||||
)
|
||||
|
||||
result = module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))
|
||||
|
||||
assert result["filed"] is False
|
||||
assert "advice_failed" in result["reason"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"raw",
|
||||
[
|
||||
"",
|
||||
"not json at all",
|
||||
'{"incident_id": "someone/else", "analysis": "a", "code_suggestions": [], "human_required": true, "reason": "r"}',
|
||||
'{"incident_id": "' + INCIDENT + '", "code_suggestions": "nope"}',
|
||||
],
|
||||
)
|
||||
def test_an_unusable_response_yields_no_suggestions(raw) -> None:
|
||||
assert module.parse_advice(raw, INCIDENT) == ([], "")
|
||||
|
||||
|
||||
def test_a_good_response_is_parsed() -> None:
|
||||
suggestions, analysis = module.parse_advice("noise " + _response() + " trailing", INCIDENT)
|
||||
|
||||
assert len(suggestions) == 1
|
||||
assert suggestions[0].path == "ariadne/services/thing.py"
|
||||
assert analysis.startswith("The function branches")
|
||||
|
||||
|
||||
def test_the_prompt_states_that_nothing_is_applied() -> None:
|
||||
prompt = module._prompt(INCIDENT, dict(ISSUE), "def thing(): pass")
|
||||
|
||||
assert "not a build failure" in prompt
|
||||
assert "Nothing you return is applied" in prompt
|
||||
assert "python:S3776" in prompt
|
||||
assert "def thing(): pass" in prompt
|
||||
|
||||
|
||||
def test_the_prompt_bounds_the_file_it_sends() -> None:
|
||||
prompt = module._prompt(INCIDENT, dict(ISSUE), "x" * 90000)
|
||||
|
||||
assert len(prompt) < 45000
|
||||
|
||||
|
||||
def test_a_finding_issue_never_uses_build_failure_wording() -> None:
|
||||
"""A line number printed as a build number reads as a failure that never happened."""
|
||||
|
||||
from ariadne.services import hermes_incident_body as body
|
||||
from ariadne.services import hermes_code_suggestion as cs
|
||||
|
||||
context = {
|
||||
"incident_id": INCIDENT,
|
||||
"job": "ariadne",
|
||||
"build_number": 42,
|
||||
"classification": "python:S3776",
|
||||
"finding_path": "ariadne/services/thing.py",
|
||||
"reason": "The function branches five ways.",
|
||||
"authorize_reason": "no automated patch was possible for this finding",
|
||||
"build_url": "https://quality.example/project/issues?open=AZ1",
|
||||
"run_id": "run_x",
|
||||
"run_url": "https://agent.example/chat?resume=run_x",
|
||||
"code_suggestions": cs.from_payload({"code_suggestions": [dict(SUGGESTION)]}),
|
||||
}
|
||||
|
||||
assert body.issue_title(context) == "[hermes] ariadne: python:S3776"
|
||||
rendered = body.issue_body(context)
|
||||
assert rendered.startswith("SonarQube reports **python:S3776**")
|
||||
assert "The build is green" in rendered
|
||||
assert "## What is wrong" in rendered
|
||||
assert "- SonarQube finding: https://quality.example" in rendered
|
||||
assert "Failed build" not in rendered
|
||||
assert "first failed gate" not in rendered
|
||||
assert "Why a human is needed" not in rendered
|
||||
|
||||
|
||||
def test_a_build_failure_issue_is_unchanged() -> None:
|
||||
"""The finding wording must not leak into real triage issues."""
|
||||
|
||||
from ariadne.services import hermes_incident_body as body
|
||||
|
||||
context = {
|
||||
"incident_id": "ariadne/408",
|
||||
"job": "ariadne",
|
||||
"build_number": 408,
|
||||
"classification": "pytest_test_failure",
|
||||
"reason": "a repository test failure",
|
||||
"authorize_reason": "human_required",
|
||||
"build_url": "https://ci.example/job/ariadne/408/",
|
||||
"run_id": "run_y",
|
||||
}
|
||||
|
||||
assert body.issue_title(context) == "[hermes] ariadne #408: pytest_test_failure"
|
||||
rendered = body.issue_body(context)
|
||||
assert "## Why a human is needed" in rendered
|
||||
assert "- Failed build: https://ci.example" in rendered
|
||||
assert "did not authorize automated remediation" in rendered
|
||||
assert "SonarQube" not in rendered
|
||||
@ -248,3 +248,27 @@ def test_the_timeout_falls_back_when_unparseable() -> None:
|
||||
)
|
||||
def test_the_component_key_yields_the_repository_path(component, expected) -> None:
|
||||
assert module.component_path(component) == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("incident", "expected"),
|
||||
[
|
||||
(
|
||||
"sonar/ariadne/AZ2y0FYFKy9i4pkIpNlV",
|
||||
"https://quality.bstein.dev/project/issues"
|
||||
"?resolved=false&id=ariadne&open=AZ2y0FYFKy9i4pkIpNlV",
|
||||
),
|
||||
# A build-driven repair has no finding to point at.
|
||||
("ariadne/408", ""),
|
||||
("sonar/ariadne", ""),
|
||||
("sonar//key", ""),
|
||||
("sonar/proj/ ", ""),
|
||||
("", ""),
|
||||
],
|
||||
)
|
||||
def test_the_finding_link_is_derived_from_the_incident_id(incident, expected) -> None:
|
||||
assert module.issue_url(incident, "https://quality.bstein.dev/") == expected
|
||||
|
||||
|
||||
def test_no_finding_link_without_a_configured_console() -> None:
|
||||
assert module.issue_url("sonar/ariadne/AZ1", "") == ""
|
||||
|
||||
@ -42,6 +42,7 @@ def _config(**overrides):
|
||||
"hermes_sonar_max_per_sweep": 1,
|
||||
"hermes_sonar_max_effort_minutes": 20,
|
||||
"hermes_sonar_timeout_seconds": 5,
|
||||
"hermes_sonar_advice_enabled": False,
|
||||
}
|
||||
values.update(overrides)
|
||||
return SimpleNamespace(**values)
|
||||
@ -72,6 +73,10 @@ def wiring(monkeypatch):
|
||||
return calls["result"]
|
||||
|
||||
monkeypatch.setattr(module.hermes_code_flow, "propose_code_fix", _propose)
|
||||
monkeypatch.setattr(
|
||||
module.hermes_code_repair, "open_proposal_incidents",
|
||||
lambda cfg: (calls.get("open_incidents", []), None),
|
||||
)
|
||||
return calls
|
||||
|
||||
|
||||
@ -144,9 +149,9 @@ def test_a_sweep_opens_one_proposal_for_the_chosen_finding(wiring) -> None:
|
||||
|
||||
result = module.sweep(storage, _config(), {})
|
||||
|
||||
assert result == {"proposed": 1, "skipped": [], "projects": 1}
|
||||
assert result == {"proposed": 1, "advised": 0, "skipped": [], "projects": 1}
|
||||
proposal = wiring["proposals"][0]
|
||||
assert proposal["incident_id"] == "sonar/ariadne/AZ-1"
|
||||
assert proposal["incident_id"] == "sonar/ariadne/python:S1172/AZ-1"
|
||||
assert proposal["job"] == "ariadne"
|
||||
assert proposal["build_number"] == "sonar-AZ-1"
|
||||
|
||||
@ -159,7 +164,7 @@ def test_the_bundle_offers_only_the_file_the_finding_names() -> None:
|
||||
assert bundle["sonarqube"] == {"project": "ariadne", "issues": [_issue()]}
|
||||
assert bundle["jenkins"]["console_failures"] == []
|
||||
assert bundle["jenkins"]["console_tail"] == ""
|
||||
assert bundle["incident_id"] == "sonar/ariadne/AZ-1"
|
||||
assert bundle["incident_id"] == "sonar/ariadne/python:S1172/AZ-1"
|
||||
|
||||
|
||||
def test_the_sweep_budget_stops_after_its_quota(wiring) -> None:
|
||||
@ -182,13 +187,13 @@ def test_a_declined_proposal_is_reported_and_does_not_spend_budget(wiring) -> No
|
||||
|
||||
assert result["proposed"] == 0
|
||||
assert len(wiring["proposals"]) == 2
|
||||
assert "a: open_proposal_limit_reached" in result["skipped"]
|
||||
assert any("open_proposal_limit_reached" in item for item in result["skipped"])
|
||||
|
||||
|
||||
def test_no_configured_projects_makes_no_call(wiring) -> None:
|
||||
result = module.sweep(_Storage(), _config(hermes_sonar_projects={}), {})
|
||||
|
||||
assert result == {"proposed": 0, "skipped": ["no_projects_configured"], "projects": 0}
|
||||
assert result == {"proposed": 0, "advised": 0, "skipped": ["no_projects_configured"], "projects": 0}
|
||||
assert wiring["proposals"] == []
|
||||
|
||||
|
||||
@ -211,12 +216,16 @@ def test_a_fetch_error_is_reported_not_raised(wiring) -> None:
|
||||
assert result["skipped"] == ["ariadne: sonar fetch http 503"]
|
||||
|
||||
|
||||
def test_a_project_with_nothing_mechanical_is_reported(wiring) -> None:
|
||||
def test_a_finding_too_large_to_patch_becomes_advice(wiring) -> None:
|
||||
"""The bulk of the backlog is refactors no anchored patch can express."""
|
||||
|
||||
wiring["issues"] = [_issue(effort="8h")]
|
||||
|
||||
result = module.sweep(_Storage(), _config(), {})
|
||||
|
||||
assert result["skipped"] == ["ariadne: no mechanically fixable finding"]
|
||||
assert result["proposed"] == 0
|
||||
assert result["skipped"] == ["ariadne: too large to patch (advice disabled)"]
|
||||
assert wiring["proposals"] == []
|
||||
|
||||
|
||||
def test_a_sweep_never_takes_down_the_scheduler(monkeypatch) -> None:
|
||||
@ -360,3 +369,109 @@ def test_a_bundles_origin_decides_the_framing(bundle, expected) -> None:
|
||||
from ariadne.services import hermes_code_prompt
|
||||
|
||||
assert hermes_code_prompt.is_quality_sweep(bundle) is expected
|
||||
|
||||
|
||||
def test_a_rule_already_under_review_is_not_proposed_again(wiring) -> None:
|
||||
"""One rule is one root cause; thirty near-identical PRs get read as none."""
|
||||
|
||||
wiring["open_incidents"] = ["sonar/ariadne/python:S1172/AZ-OTHER"]
|
||||
wiring["issues"] = [
|
||||
_issue(key="a", rule="python:S1172"),
|
||||
_issue(key="b", rule="python:S2208", effort="10min"),
|
||||
]
|
||||
|
||||
result = module.sweep(_Storage(), _config(), {})
|
||||
|
||||
assert result["proposed"] == 1
|
||||
assert wiring["proposals"][0]["incident_id"] == "sonar/ariadne/python:S2208/b"
|
||||
|
||||
|
||||
def test_every_rule_under_review_means_nothing_new_to_propose(wiring) -> None:
|
||||
wiring["open_incidents"] = ["sonar/ariadne/python:S1172/AZ-OTHER"]
|
||||
wiring["issues"] = [_issue(rule="python:S1172")]
|
||||
|
||||
result = module.sweep(_Storage(), _config(), {})
|
||||
|
||||
assert result["proposed"] == 0
|
||||
assert result["skipped"] == ["ariadne: no new finding to act on"]
|
||||
|
||||
|
||||
def test_another_projects_open_proposal_does_not_block_this_one(wiring) -> None:
|
||||
wiring["open_incidents"] = ["sonar/metis/python:S1172/AZ-X", "ariadne/408"]
|
||||
|
||||
assert module.sweep(_Storage(), _config(), {})["proposed"] == 1
|
||||
|
||||
|
||||
def test_the_rules_under_review_are_read_from_the_open_proposals(monkeypatch) -> None:
|
||||
monkeypatch.setattr(
|
||||
module.hermes_code_repair, "open_proposal_incidents",
|
||||
lambda cfg: (
|
||||
[
|
||||
"sonar/ariadne/python:S2208/AZ1",
|
||||
"sonar/ariadne/python:S3776/AZ2",
|
||||
"sonar/other/python:S1172/AZ3",
|
||||
"ariadne/408",
|
||||
"sonar/ariadne/malformed",
|
||||
],
|
||||
None,
|
||||
),
|
||||
)
|
||||
|
||||
assert module.proposed_rules(REPO_CFG, "ariadne") == {"python:S2208", "python:S3776"}
|
||||
|
||||
|
||||
def test_an_unreadable_proposal_list_costs_a_duplicate_not_a_silent_rule(monkeypatch) -> None:
|
||||
"""Failing open here means one extra PR; failing closed loses a whole rule."""
|
||||
|
||||
monkeypatch.setattr(
|
||||
module.hermes_code_repair, "open_proposal_incidents", lambda cfg: ([], "http 500")
|
||||
)
|
||||
|
||||
assert module.proposed_rules(REPO_CFG, "ariadne") == set()
|
||||
|
||||
|
||||
def test_advice_files_an_issue_when_no_patch_was_possible(monkeypatch, wiring) -> None:
|
||||
"""A finding nobody can patch is exactly the one worth explaining."""
|
||||
|
||||
filed = []
|
||||
monkeypatch.setattr(
|
||||
module.hermes_sonar_advice, "advise",
|
||||
lambda storage, config, hermes_cfg, repo_cfg, project, issue: filed.append(issue)
|
||||
or {"filed": True, "reason": "issue_filed", "url": "https://scm/issues/9"},
|
||||
)
|
||||
wiring["issues"] = [_issue(effort="8h", key="big")]
|
||||
|
||||
result = module.sweep(_Storage(), _config(hermes_sonar_advice_enabled=True), {})
|
||||
|
||||
assert result["advised"] == 1
|
||||
assert filed[0]["key"] == "big"
|
||||
|
||||
|
||||
def test_a_declined_pull_request_falls_through_to_advice(monkeypatch, wiring) -> None:
|
||||
"""A declined proposal is not a dead end; the finding is still real."""
|
||||
|
||||
monkeypatch.setattr(
|
||||
module.hermes_sonar_advice, "advise",
|
||||
lambda *a, **k: {"filed": True, "reason": "issue_filed", "url": "u"},
|
||||
)
|
||||
wiring["result"] = {"status": "human_required", "reason": "patch_rejected: original_ambiguous"}
|
||||
|
||||
result = module.sweep(_Storage(), _config(hermes_sonar_advice_enabled=True), {})
|
||||
|
||||
assert result["advised"] == 1
|
||||
|
||||
|
||||
def test_advice_respects_the_rules_already_under_review(wiring) -> None:
|
||||
wiring["open_incidents"] = ["sonar/ariadne/python:S1172/OTHER"]
|
||||
wiring["issues"] = [_issue(effort="8h", rule="python:S1172")]
|
||||
|
||||
result = module.sweep(_Storage(), _config(hermes_sonar_advice_enabled=True), {})
|
||||
|
||||
assert result["skipped"] == ["ariadne: no new finding to act on"]
|
||||
|
||||
|
||||
def test_advice_stays_inside_the_write_allowlist() -> None:
|
||||
"""Advising on a file nobody would edit is noise, not help."""
|
||||
|
||||
assert module.select_advice_issue([_issue(path="docs/x.md", effort="8h")], REPO_CFG, _config()) is None
|
||||
assert module.select_advice_issue([_issue(effort="8h")], REPO_CFG, _config())["key"] == "AZ-1"
|
||||
|
||||
@ -443,3 +443,39 @@ def test_retry_access_request_reports_update_failure(monkeypatch) -> None:
|
||||
resp = client.post("/api/access/requests/REQ1/retry")
|
||||
|
||||
assert resp.status_code == 502
|
||||
|
||||
def test_hermes_autotriage_can_be_run_on_demand(monkeypatch) -> None:
|
||||
"""The cron tick is once a minute; a demo should not wait for it."""
|
||||
|
||||
ctx = AuthContext(username="bstein", email="", groups=["admin"], claims={})
|
||||
client = _client(monkeypatch, ctx)
|
||||
calls = []
|
||||
monkeypatch.setattr(
|
||||
app_module,
|
||||
"run_hermes_autotriage",
|
||||
lambda storage: calls.append(storage) or {"status": "ok", "jobs": {}},
|
||||
)
|
||||
|
||||
admin_run = client.post(
|
||||
"/api/admin/hermes/autotriage/run",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
internal_run = client.post("/api/internal/hermes/autotriage/run")
|
||||
|
||||
assert admin_run.status_code == 200
|
||||
assert internal_run.status_code == 200
|
||||
assert admin_run.json()["status"] == "ok"
|
||||
assert len(calls) == 2
|
||||
|
||||
|
||||
def test_hermes_autotriage_on_demand_requires_admin(monkeypatch) -> None:
|
||||
ctx = AuthContext(username="nobody", email="", groups=[], claims={})
|
||||
client = _client(monkeypatch, ctx)
|
||||
monkeypatch.setattr(app_module, "run_hermes_autotriage", lambda storage: {"status": "ok"})
|
||||
|
||||
resp = client.post(
|
||||
"/api/admin/hermes/autotriage/run",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
|
||||
assert resp.status_code == 403
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user