fix: declare static agent routes before dynamic id route
This commit is contained in:
gti
@@ -117,13 +117,19 @@ def _attention_item(item, problem, timestamp, *, rank, link=None, now=None):
|
||||
|
||||
|
||||
def installation_view(installation, agent, tenant, snapshot=None, *, now=None):
|
||||
has_snapshot = snapshot is not None
|
||||
snapshot = snapshot or AgentSnapshot(installation_id=installation.id)
|
||||
health, reason = compute_agent_health(installation, now=now, supplemental=snapshot)
|
||||
version_drift = snapshot.version_drift
|
||||
if version_drift == "UNKNOWN":
|
||||
if not version_drift or version_drift == "UNKNOWN":
|
||||
version_drift = compute_version_drift(agent.agent_version)
|
||||
fiscal_status = _fiscal_status(snapshot)
|
||||
coverage_partition = _coverage_partition(snapshot)
|
||||
if not has_snapshot:
|
||||
# No heartbeat/snapshot yet (e.g. an enrollment-only E2E row): expose
|
||||
# no counts at all rather than fabricated zeros — the UI shows
|
||||
# "Não informado" instead of None/0.
|
||||
coverage_partition = {"unclassified_products": None, "coverage_overcount_products": 0, "coverage_invalid_counts": False, "coverage_invariant_pass": True}
|
||||
version_drift_label = {"CURRENT": "Atualizada", "OUTDATED": "Desatualizada", "UNKNOWN": "Desconhecida"}.get(version_drift, str(version_drift))
|
||||
agent_version_display = f"{agent.agent_version} · {version_drift_label}" if agent.agent_version else "Desconhecida"
|
||||
return {
|
||||
@@ -153,15 +159,15 @@ def installation_view(installation, agent, tenant, snapshot=None, *, now=None):
|
||||
"last_seen_age": format_age(installation.last_seen_at, now),
|
||||
"started_at": _iso(installation.started_at),
|
||||
"last_status": installation.last_status,
|
||||
"total_products": snapshot.total_products,
|
||||
"fiscally_complete": snapshot.fiscally_complete,
|
||||
"safe_blocked": snapshot.safe_blocked,
|
||||
"failed_products": snapshot.failed_products,
|
||||
"total_products": snapshot.total_products if has_snapshot else None,
|
||||
"fiscally_complete": snapshot.fiscally_complete if has_snapshot else None,
|
||||
"safe_blocked": snapshot.safe_blocked if has_snapshot else None,
|
||||
"failed_products": snapshot.failed_products if has_snapshot else None,
|
||||
**coverage_partition,
|
||||
"coverage_percent": snapshot.coverage_percent,
|
||||
"fiscal_completion_percent": _fiscal_completion(snapshot),
|
||||
"fiscal_status": fiscal_status,
|
||||
"fiscal_status_label": fiscal_status.title(),
|
||||
"coverage_percent": snapshot.coverage_percent if has_snapshot else None,
|
||||
"fiscal_completion_percent": _fiscal_completion(snapshot) if has_snapshot else None,
|
||||
"fiscal_status": fiscal_status if has_snapshot else "UNKNOWN",
|
||||
"fiscal_status_label": fiscal_status.title() if has_snapshot else "Não informado",
|
||||
"blockers": snapshot.blockers or {},
|
||||
"last_job_status": snapshot.last_job_status,
|
||||
"last_job_at": _iso(snapshot.last_job_at),
|
||||
|
||||
+58
-47
@@ -161,53 +161,10 @@ def router(session_factory):
|
||||
return render(request, "agents.html", {"result": list_installations(session, page=page, page_size=50, search=search, health=health, sort=sort), "auth": auth_data, "search": search or "", "health_filter": health or "", "sort": sort, "query_base": query_base(search=search, health=health, sort=sort)})
|
||||
finally: session.close()
|
||||
|
||||
@routes.get("/agents/{agent_id}", response_class=HTMLResponse)
|
||||
def agent_detail(request: Request, agent_id: int, tab: str = "overview", event_page: int = 1, severity: str | None = None, event_type: str | None = None, job_id: int | None = None, product_code: str | None = None, since: str | None = None, until: str | None = None):
|
||||
session, auth_data, redirect = page_auth(request)
|
||||
if redirect: return redirect
|
||||
try:
|
||||
row = session.execute(select(Installation, Agent, Tenant, AgentSnapshot).join(Agent, Agent.id == Installation.agent_id).join(Tenant, Tenant.id == Installation.tenant_id).outerjoin(AgentSnapshot, AgentSnapshot.installation_id == Installation.id).where(Agent.id == agent_id)).first()
|
||||
if row is None: return HTMLResponse("Agent not found", status_code=404)
|
||||
installation, agent, tenant, snapshot = row
|
||||
detail = installation_view(installation, agent, tenant, snapshot)
|
||||
event_page = max(1, event_page)
|
||||
events = list_panel_events(session, installation.id, page=event_page, page_size=25, include_more=True, severity=severity, event_type=event_type, job_id=job_id, product_code=product_code, since=date_filter(since), until=date_filter(until, end_of_day=True))
|
||||
has_more = len(events) > 25
|
||||
events = events[:25]
|
||||
for event in events:
|
||||
event["occurred_display"] = format_ui_datetime(datetime.fromisoformat(event["occurred_at"]) if event["occurred_at"] else None)
|
||||
agent_blockers = list_blockers(session, tenant_id=installation.tenant_id, agent_id=agent_id, page=1, page_size=25)
|
||||
return render(request, "agent_detail.html", {"detail": detail, "tab": tab, "blockers": agent_blockers, "blocker_breakdown": blocker_filter_summary(session, tenant_id=installation.tenant_id, agent_id=agent_id, status="OPEN")["blocker_breakdown"], "jobs": list_jobs(session, agent_id=agent_id, page=1, page_size=25), "events": events, "event_page": max(1, event_page), "has_more": has_more, "severity": severity or "", "event_type": event_type or "", "job_id": job_id or "", "product_code": product_code or "", "since": since or "", "until": until or "", "event_query_base": query_base(tab=tab, severity=severity, event_type=event_type, job_id=job_id, product_code=product_code, since=since, until=until), "auth": auth_data, "csrf_token": request.cookies.get("gti_csrf", "")})
|
||||
finally: session.close()
|
||||
|
||||
@routes.get("/agents/{agent_id}/reprocess-confirmation", response_class=HTMLResponse)
|
||||
def reprocess_confirmation(request: Request, agent_id: int):
|
||||
session, auth_data, redirect = page_auth(request)
|
||||
if redirect: return redirect
|
||||
try:
|
||||
if not session.scalar(select(Agent.id).where(Agent.id == agent_id)):
|
||||
return HTMLResponse("Agent not found", status_code=404)
|
||||
return render(request, "reprocess_confirmation.html", {"agent_id": agent_id, "auth": auth_data})
|
||||
finally: session.close()
|
||||
|
||||
@routes.post("/agents/{agent_id}/reprocess")
|
||||
async def reprocess_agent(request: Request, agent_id: int):
|
||||
session, auth_data, redirect = page_auth(request)
|
||||
if redirect: return redirect
|
||||
try:
|
||||
values = await form_values(request); web_session, admin = auth_data
|
||||
require_csrf(web_session, values.get("csrf_token"))
|
||||
if values.get("confirm") != "yes": return render(request, "reprocess_confirmation.html", {"agent_id": agent_id, "auth": auth_data, "error": "Confirme explicitamente o reprocessamento."}, 400)
|
||||
agent = session.get(Agent, agent_id)
|
||||
installation = session.scalar(select(Installation).where(Installation.agent_id == agent_id))
|
||||
if not agent or not installation: return HTMLResponse("Agent not found", status_code=404)
|
||||
existing = next((candidate for candidate in session.scalars(select(Job).where(Job.agent_id == agent.id, Job.kind == "REPROCESS_ALL", Job.status == "PENDING")).all() if candidate.payload == {}), None)
|
||||
job = existing or create_job(session, agent, "REPROCESS_ALL", {})
|
||||
record_audit(session, admin_user_id=admin.id, tenant_id=installation.tenant_id, action="REPROCESS_ALL", target_type="installation", target_id=installation.installation_id, context={"job_id": job.id})
|
||||
session.commit()
|
||||
return RedirectResponse(f"/agents/{agent_id}?tab=jobs", status_code=303)
|
||||
finally: session.close()
|
||||
|
||||
# --- Static /agents/* routes MUST be declared before /agents/{agent_id}:
|
||||
# Starlette matches in registration order, so a dynamic route declared
|
||||
# first would capture "/agents/new" and "/agents/releases" and fail the
|
||||
# int conversion with a 422 (the production bug this ordering prevents).
|
||||
@routes.get("/agents/releases", response_class=HTMLResponse)
|
||||
def agent_releases(request: Request):
|
||||
session, auth_data, redirect = page_auth(request)
|
||||
@@ -274,6 +231,60 @@ def router(session_factory):
|
||||
return RedirectResponse(f"/agents/pending/{enrollment_uuid}", status_code=303)
|
||||
finally: session.close()
|
||||
|
||||
@routes.get("/agents/{agent_id}", response_class=HTMLResponse)
|
||||
def agent_detail(request: Request, agent_id: str, tab: str = "overview", event_page: int = 1, severity: str | None = None, event_type: str | None = None, job_id: int | None = None, product_code: str | None = None, since: str | None = None, until: str | None = None):
|
||||
# agent_id is parsed here (not typed int in the path) so a non-numeric
|
||||
# slug like /agents/abc gets a controlled HTML 404 instead of a raw
|
||||
# 422 — and can never shadow the static routes above.
|
||||
try:
|
||||
agent_id = int(agent_id)
|
||||
except ValueError:
|
||||
return HTMLResponse("Agent não encontrado (identificador inválido)", status_code=404)
|
||||
session, auth_data, redirect = page_auth(request)
|
||||
if redirect: return redirect
|
||||
try:
|
||||
row = session.execute(select(Installation, Agent, Tenant, AgentSnapshot).join(Agent, Agent.id == Installation.agent_id).join(Tenant, Tenant.id == Installation.tenant_id).outerjoin(AgentSnapshot, AgentSnapshot.installation_id == Installation.id).where(Agent.id == agent_id)).first()
|
||||
if row is None: return HTMLResponse("Agent not found", status_code=404)
|
||||
installation, agent, tenant, snapshot = row
|
||||
detail = installation_view(installation, agent, tenant, snapshot)
|
||||
event_page = max(1, event_page)
|
||||
events = list_panel_events(session, installation.id, page=event_page, page_size=25, include_more=True, severity=severity, event_type=event_type, job_id=job_id, product_code=product_code, since=date_filter(since), until=date_filter(until, end_of_day=True))
|
||||
has_more = len(events) > 25
|
||||
events = events[:25]
|
||||
for event in events:
|
||||
event["occurred_display"] = format_ui_datetime(datetime.fromisoformat(event["occurred_at"]) if event["occurred_at"] else None)
|
||||
agent_blockers = list_blockers(session, tenant_id=installation.tenant_id, agent_id=agent_id, page=1, page_size=25)
|
||||
return render(request, "agent_detail.html", {"detail": detail, "tab": tab, "blockers": agent_blockers, "blocker_breakdown": blocker_filter_summary(session, tenant_id=installation.tenant_id, agent_id=agent_id, status="OPEN")["blocker_breakdown"], "jobs": list_jobs(session, agent_id=agent_id, page=1, page_size=25), "events": events, "event_page": max(1, event_page), "has_more": has_more, "severity": severity or "", "event_type": event_type or "", "job_id": job_id or "", "product_code": product_code or "", "since": since or "", "until": until or "", "event_query_base": query_base(tab=tab, severity=severity, event_type=event_type, job_id=job_id, product_code=product_code, since=since, until=until), "auth": auth_data, "csrf_token": request.cookies.get("gti_csrf", "")})
|
||||
finally: session.close()
|
||||
|
||||
@routes.get("/agents/{agent_id}/reprocess-confirmation", response_class=HTMLResponse)
|
||||
def reprocess_confirmation(request: Request, agent_id: int):
|
||||
session, auth_data, redirect = page_auth(request)
|
||||
if redirect: return redirect
|
||||
try:
|
||||
if not session.scalar(select(Agent.id).where(Agent.id == agent_id)):
|
||||
return HTMLResponse("Agent not found", status_code=404)
|
||||
return render(request, "reprocess_confirmation.html", {"agent_id": agent_id, "auth": auth_data})
|
||||
finally: session.close()
|
||||
|
||||
@routes.post("/agents/{agent_id}/reprocess")
|
||||
async def reprocess_agent(request: Request, agent_id: int):
|
||||
session, auth_data, redirect = page_auth(request)
|
||||
if redirect: return redirect
|
||||
try:
|
||||
values = await form_values(request); web_session, admin = auth_data
|
||||
require_csrf(web_session, values.get("csrf_token"))
|
||||
if values.get("confirm") != "yes": return render(request, "reprocess_confirmation.html", {"agent_id": agent_id, "auth": auth_data, "error": "Confirme explicitamente o reprocessamento."}, 400)
|
||||
agent = session.get(Agent, agent_id)
|
||||
installation = session.scalar(select(Installation).where(Installation.agent_id == agent_id))
|
||||
if not agent or not installation: return HTMLResponse("Agent not found", status_code=404)
|
||||
existing = next((candidate for candidate in session.scalars(select(Job).where(Job.agent_id == agent.id, Job.kind == "REPROCESS_ALL", Job.status == "PENDING")).all() if candidate.payload == {}), None)
|
||||
job = existing or create_job(session, agent, "REPROCESS_ALL", {})
|
||||
record_audit(session, admin_user_id=admin.id, tenant_id=installation.tenant_id, action="REPROCESS_ALL", target_type="installation", target_id=installation.installation_id, context={"job_id": job.id})
|
||||
session.commit()
|
||||
return RedirectResponse(f"/agents/{agent_id}?tab=jobs", status_code=303)
|
||||
finally: session.close()
|
||||
|
||||
def _pending_or_404(session, enrollment_uuid):
|
||||
return session.scalar(select(AgentEnrollment).where(AgentEnrollment.enrollment_uuid == enrollment_uuid))
|
||||
|
||||
|
||||
Reference in New Issue
Block a user