feat: add bounded operational view models
This commit is contained in:
gti
+351
-54
@@ -2,13 +2,19 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from sqlalchemy import func, or_, select
|
||||
from sqlalchemy import exists, func, or_, select
|
||||
from sqlalchemy.orm import aliased
|
||||
|
||||
from .config import settings
|
||||
from .installation_identity import compute_agent_health
|
||||
from .models import Agent, AgentEvent, AgentSnapshot, BlockedProduct, Installation, Job, Tenant
|
||||
from .events import list_events
|
||||
from .installation_identity import compute_agent_health, compute_version_drift
|
||||
from .models import Agent, AgentEvent, AgentSnapshot, AuditEvent, BlockedProduct, Installation, Job, Tenant
|
||||
from .panel_presentation import format_age, format_duration, format_ui_datetime, health_label, operational_priority
|
||||
|
||||
|
||||
HEALTH_STATES = ("ONLINE", "DEGRADED", "OFFLINE", "FAILED", "UNKNOWN")
|
||||
BLOCKER_FIELDS = ("CFOP", "ST", "ICMS", "CSOSN", "NCM", "OTHER")
|
||||
|
||||
|
||||
def _page(page: int, page_size: int):
|
||||
@@ -19,76 +25,367 @@ def _iso(value):
|
||||
return value.isoformat() if value else None
|
||||
|
||||
|
||||
def installation_view(installation, agent, tenant, snapshot=None):
|
||||
snapshot = snapshot or AgentSnapshot(installation_id=installation.id)
|
||||
health, reason = compute_agent_health(installation, supplemental=snapshot)
|
||||
def _fiscal_completion(snapshot) -> float:
|
||||
total = int(getattr(snapshot, "total_products", 0) or 0)
|
||||
complete = int(getattr(snapshot, "fiscally_complete", 0) or 0)
|
||||
if total:
|
||||
return round(max(0, min(total, complete)) * 100 / total, 1)
|
||||
return round(float(getattr(snapshot, "coverage_percent", 0) or 0), 1)
|
||||
|
||||
|
||||
def _fiscal_status(snapshot) -> str:
|
||||
if int(getattr(snapshot, "safe_blocked", 0) or 0):
|
||||
return "BLOCKED"
|
||||
total = int(getattr(snapshot, "total_products", 0) or 0)
|
||||
complete = int(getattr(snapshot, "fiscally_complete", 0) or 0)
|
||||
if total and complete >= total:
|
||||
return "COMPLETE"
|
||||
return "INCOMPLETE"
|
||||
|
||||
|
||||
def _technical_timestamp(item):
|
||||
if item["technical_health"] in {"FAILED", "DEGRADED"}:
|
||||
return item.get("last_job_at") or item.get("last_seen_at")
|
||||
return item.get("last_seen_at")
|
||||
|
||||
|
||||
def _attention_item(item, problem, timestamp, *, rank, link=None, now=None):
|
||||
timestamp_value = datetime.fromisoformat(timestamp) if isinstance(timestamp, str) else timestamp
|
||||
return {
|
||||
"agent_id": agent.id, "installation_id": installation.installation_id, "installation_db_id": installation.id,
|
||||
"client": tenant.name if tenant else None, "tenant_id": installation.tenant_id, "display_name": installation.display_name,
|
||||
"hostname": installation.hostname, "company_name": installation.company_name, "company_code": installation.company_code,
|
||||
"company_cnpj": installation.company_cnpj_masked, "company_uf": installation.company_uf, "company_crt": installation.company_crt,
|
||||
"agent_version": agent.agent_version, "version_drift": snapshot.version_drift, "health": health, "health_reason": reason,
|
||||
"last_seen_at": _iso(installation.last_seen_at), "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, "coverage_percent": snapshot.coverage_percent, "blockers": snapshot.blockers or {},
|
||||
"last_job_status": snapshot.last_job_status, "last_job_at": _iso(snapshot.last_job_at), "last_successful_job_at": _iso(snapshot.last_successful_job_at),
|
||||
"firebird_version": installation.firebird_version, "firebird_ods": installation.firebird_ods, "database_identifier": installation.database_identifier,
|
||||
"capabilities": installation.capabilities or {}, "fiscal_engine_version": installation.fiscal_engine_version, "dataset_versions": installation.dataset_versions or {},
|
||||
"client": item.get("client"),
|
||||
"display_name": item.get("display_name"),
|
||||
"hostname": item.get("hostname"),
|
||||
"problem": problem,
|
||||
"timestamp": timestamp if isinstance(timestamp, str) else _iso(timestamp_value),
|
||||
"timestamp_display": format_ui_datetime(timestamp_value),
|
||||
"age": format_age(timestamp_value, now),
|
||||
"link": link or f"/agents/{item['agent_id']}",
|
||||
"technical_health": item["technical_health"],
|
||||
"rank": rank,
|
||||
}
|
||||
|
||||
|
||||
def list_installations(session, *, page=1, page_size=50, client=None, health=None, version=None, search=None, last_seen_before=None):
|
||||
def installation_view(installation, agent, tenant, snapshot=None, *, now=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":
|
||||
version_drift = compute_version_drift(agent.agent_version)
|
||||
fiscal_status = _fiscal_status(snapshot)
|
||||
return {
|
||||
"agent_id": agent.id,
|
||||
"installation_id": installation.installation_id,
|
||||
"installation_db_id": installation.id,
|
||||
"client": tenant.name if tenant else None,
|
||||
"tenant_id": installation.tenant_id,
|
||||
"display_name": installation.display_name,
|
||||
"hostname": installation.hostname,
|
||||
"company_name": installation.company_name,
|
||||
"company_code": installation.company_code,
|
||||
"company_cnpj": installation.company_cnpj_masked,
|
||||
"company_uf": installation.company_uf,
|
||||
"company_crt": installation.company_crt,
|
||||
"agent_version": agent.agent_version,
|
||||
"version_drift": version_drift,
|
||||
"health": health,
|
||||
"health_reason": reason,
|
||||
"technical_health": health,
|
||||
"health_label": health_label(health),
|
||||
"technical_health_label": health_label(health),
|
||||
"last_seen_at": _iso(installation.last_seen_at),
|
||||
"last_seen_display": format_ui_datetime(installation.last_seen_at),
|
||||
"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,
|
||||
"coverage_percent": snapshot.coverage_percent,
|
||||
"fiscal_completion_percent": _fiscal_completion(snapshot),
|
||||
"fiscal_status": fiscal_status,
|
||||
"fiscal_status_label": fiscal_status.title(),
|
||||
"blockers": snapshot.blockers or {},
|
||||
"last_job_status": snapshot.last_job_status,
|
||||
"last_job_at": _iso(snapshot.last_job_at),
|
||||
"last_successful_job_at": _iso(snapshot.last_successful_job_at),
|
||||
"firebird_version": installation.firebird_version,
|
||||
"firebird_ods": installation.firebird_ods,
|
||||
"database_identifier": installation.database_identifier,
|
||||
"capabilities": installation.capabilities or {},
|
||||
"fiscal_engine_version": installation.fiscal_engine_version,
|
||||
"dataset_versions": installation.dataset_versions or {},
|
||||
"technical_ids": {"agent_id": agent.id, "installation_db_id": installation.id, "tenant_id": installation.tenant_id},
|
||||
"link": f"/agents/{agent.id}",
|
||||
}
|
||||
|
||||
|
||||
def _installation_query_filters(query, *, client=None, version=None, search=None, last_seen_before=None):
|
||||
if client:
|
||||
query = query.where(or_(Tenant.name.ilike(f"%{client}%"), Installation.customer_code.ilike(f"%{client}%")))
|
||||
if version:
|
||||
query = query.where(Agent.agent_version == version)
|
||||
if search:
|
||||
query = query.where(or_(Installation.display_name.ilike(f"%{search}%"), Installation.hostname.ilike(f"%{search}%"), Tenant.name.ilike(f"%{search}%")))
|
||||
if last_seen_before:
|
||||
query = query.where(or_(Installation.last_seen_at.is_(None), Installation.last_seen_at < last_seen_before))
|
||||
return query
|
||||
|
||||
|
||||
def list_installations(session, *, page=1, page_size=50, client=None, health=None, version=None, search=None, last_seen_before=None, sort="display_name", now=None):
|
||||
page, page_size = _page(page, page_size)
|
||||
query = 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)
|
||||
if client: query = query.where(or_(Tenant.name.ilike(f"%{client}%"), Installation.customer_code.ilike(f"%{client}%")))
|
||||
if version: query = query.where(Agent.agent_version == version)
|
||||
if search: query = query.where(or_(Installation.display_name.ilike(f"%{search}%"), Installation.hostname.ilike(f"%{search}%"), Tenant.name.ilike(f"%{search}%")))
|
||||
if last_seen_before: query = query.where(or_(Installation.last_seen_at.is_(None), Installation.last_seen_at < last_seen_before))
|
||||
ordered = query.order_by(Installation.display_name.asc(), Installation.id.asc())
|
||||
query = _installation_query_filters(query, client=client, version=version, search=search, last_seen_before=last_seen_before)
|
||||
if sort in {"worst_coverage", "coverage", "coverage_asc", "fiscal_completion_percent"}:
|
||||
ordered = query.order_by(func.coalesce(AgentSnapshot.coverage_percent, 0).asc(), Installation.display_name.asc(), Installation.id.asc())
|
||||
else:
|
||||
ordered = query.order_by(Installation.display_name.asc(), Installation.id.asc())
|
||||
rows = session.execute(ordered).all() if health else session.execute(ordered.offset((page - 1) * page_size).limit(page_size)).all()
|
||||
items = [installation_view(i, a, t, s) for i, a, t, s in rows]
|
||||
total = None
|
||||
items = [installation_view(i, a, t, s, now=now) for i, a, t, s in rows]
|
||||
if health:
|
||||
items = [item for item in items if item["health"] == health]
|
||||
total = len(items)
|
||||
items = items[(page - 1) * page_size:page * page_size]
|
||||
return {"page": page, "page_size": page_size, "items": items, "total": total if total is not None else len(items), "health_filter_applied": health}
|
||||
else:
|
||||
count_query = select(func.count()).select_from(query.with_only_columns(Installation.id).order_by(None).subquery())
|
||||
total = session.scalar(count_query) or 0
|
||||
return {"page": page, "page_size": page_size, "items": items, "total": total, "health_filter_applied": health}
|
||||
|
||||
|
||||
def dashboard_summary(session):
|
||||
def _blocker_item(row, tenant, installation=None, agent=None, *, now=None):
|
||||
return {
|
||||
"id": row.id,
|
||||
"tenant_id": row.tenant_id,
|
||||
"client": tenant.name if tenant else row.company,
|
||||
"product_code": row.product_code,
|
||||
"description": row.description,
|
||||
"field": row.field,
|
||||
"blocker_type": row.blocker_type,
|
||||
"reason": row.reason_code,
|
||||
"reason_code": row.reason_code,
|
||||
"evidence_summary": row.evidence_summary,
|
||||
"first_seen_at": _iso(row.first_seen_at),
|
||||
"last_seen_at": _iso(row.last_seen_at),
|
||||
"first_seen_display": format_ui_datetime(row.first_seen_at),
|
||||
"last_seen_display": format_ui_datetime(row.last_seen_at),
|
||||
"age": format_age(row.last_seen_at, now),
|
||||
"retry_count": row.retry_count,
|
||||
"resolved_at": _iso(row.resolved_at),
|
||||
"resolution_source": row.resolution_source,
|
||||
"priority": operational_priority(row),
|
||||
"installation_id": installation.installation_id if installation else None,
|
||||
"installation_db_id": installation.id if installation else None,
|
||||
"display_name": installation.display_name if installation else None,
|
||||
"hostname": installation.hostname if installation else None,
|
||||
"agent_id": agent.id if agent else None,
|
||||
"link": f"/blockers/{row.id}",
|
||||
}
|
||||
|
||||
|
||||
def _parse_age(value):
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, (int, float)):
|
||||
return timedelta(days=float(value))
|
||||
text = str(value).strip().lower()
|
||||
try:
|
||||
if text.endswith("h"):
|
||||
return timedelta(hours=float(text[:-1]))
|
||||
if text.endswith("d"):
|
||||
return timedelta(days=float(text[:-1]))
|
||||
return timedelta(days=float(text))
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _blockers_query(session, *, tenant_id=None, agent_id=None, blocker_type=None, field=None, status="OPEN", client=None, installation=None, search=None, age=None):
|
||||
installation_filter = aliased(Installation)
|
||||
representative = select(func.min(Installation.id).label("installation_db_id"), Installation.tenant_id.label("installation_tenant_id")).group_by(Installation.tenant_id).subquery()
|
||||
query = select(BlockedProduct, Tenant, Installation, Agent).join(Tenant, Tenant.id == BlockedProduct.tenant_id).outerjoin(representative, representative.c.installation_tenant_id == BlockedProduct.tenant_id).outerjoin(Installation, Installation.id == representative.c.installation_db_id).outerjoin(Agent, Agent.id == Installation.agent_id)
|
||||
status = (status or "OPEN").upper()
|
||||
if status == "OPEN":
|
||||
query = query.where(BlockedProduct.resolved_at.is_(None))
|
||||
elif status in {"RESOLVED", "CLOSED"}:
|
||||
query = query.where(BlockedProduct.resolved_at.is_not(None))
|
||||
if tenant_id is not None:
|
||||
query = query.where(BlockedProduct.tenant_id == tenant_id)
|
||||
if agent_id is not None:
|
||||
query = query.where(exists(select(installation_filter.id).where(installation_filter.tenant_id == BlockedProduct.tenant_id, installation_filter.agent_id == agent_id)))
|
||||
if blocker_type:
|
||||
query = query.where(BlockedProduct.blocker_type == blocker_type)
|
||||
if field:
|
||||
query = query.where(BlockedProduct.field == field)
|
||||
if client:
|
||||
query = query.where(or_(Tenant.name.ilike(f"%{client}%"), BlockedProduct.company.ilike(f"%{client}%")))
|
||||
if installation:
|
||||
query = query.where(
|
||||
exists(
|
||||
select(installation_filter.id).where(
|
||||
installation_filter.tenant_id == BlockedProduct.tenant_id,
|
||||
or_(
|
||||
installation_filter.installation_id.ilike(f"%{installation}%"),
|
||||
installation_filter.display_name.ilike(f"%{installation}%"),
|
||||
installation_filter.hostname.ilike(f"%{installation}%"),
|
||||
),
|
||||
)
|
||||
)
|
||||
)
|
||||
if search:
|
||||
term = f"%{search}%"
|
||||
query = query.where(or_(BlockedProduct.product_code.ilike(term), BlockedProduct.description.ilike(term), BlockedProduct.field.ilike(term), BlockedProduct.blocker_type.ilike(term), BlockedProduct.reason_code.ilike(term), Tenant.name.ilike(term)))
|
||||
age_delta = _parse_age(age)
|
||||
if age_delta:
|
||||
query = query.where(BlockedProduct.last_seen_at <= datetime.now(timezone.utc) - age_delta)
|
||||
return query
|
||||
|
||||
|
||||
def list_blockers(session, *, tenant_id=None, agent_id=None, page=1, page_size=50, blocker_type=None, field=None, resolved=False, client=None, installation=None, status="OPEN", age=None, search=None):
|
||||
page, page_size = _page(page, page_size)
|
||||
if resolved and status == "OPEN":
|
||||
status = "RESOLVED"
|
||||
query = _blockers_query(session, tenant_id=tenant_id, agent_id=agent_id, blocker_type=blocker_type, field=field, status=status, client=client, installation=installation, search=search, age=age)
|
||||
ordered = query.order_by(BlockedProduct.last_seen_at.desc(), BlockedProduct.id.desc())
|
||||
rows = session.execute(ordered.offset((page - 1) * page_size).limit(page_size)).all()
|
||||
items = [_blocker_item(blocker, tenant, installation_row, agent, now=datetime.now(timezone.utc)) for blocker, tenant, installation_row, agent in rows]
|
||||
count_query = query.with_only_columns(func.count(BlockedProduct.id)).order_by(None)
|
||||
total = session.scalar(count_query) or 0
|
||||
return {"page": page, "page_size": page_size, "items": items, "total": total}
|
||||
|
||||
|
||||
def _job_item(row, agent, installation, tenant, *, now=None):
|
||||
return {
|
||||
"job_id": row.id,
|
||||
"agent_id": row.agent_id,
|
||||
"client": tenant.name if tenant else None,
|
||||
"display_name": installation.display_name if installation else None,
|
||||
"hostname": installation.hostname if installation else None,
|
||||
"kind": row.kind,
|
||||
"status": row.status,
|
||||
"created_at": _iso(row.created_at),
|
||||
"started_at": _iso(row.started_at),
|
||||
"finished_at": _iso(row.finished_at),
|
||||
"created_display": format_ui_datetime(row.created_at),
|
||||
"started_display": format_ui_datetime(row.started_at),
|
||||
"finished_display": format_ui_datetime(row.finished_at),
|
||||
"duration": format_duration(row.started_at, row.finished_at, now),
|
||||
"attempts": row.attempts,
|
||||
"result_summary": row.result_summary,
|
||||
"error": row.error,
|
||||
}
|
||||
|
||||
|
||||
def list_jobs(session, *, agent_id=None, page=1, page_size=50, status=None, kind=None, since=None, until=None, client=None, search=None):
|
||||
page, page_size = _page(page, page_size)
|
||||
query = select(Job, Agent, Installation, Tenant).join(Agent, Agent.id == Job.agent_id).join(Installation, Installation.agent_id == Agent.id).join(Tenant, Tenant.id == Installation.tenant_id)
|
||||
if agent_id is not None:
|
||||
query = query.where(Job.agent_id == agent_id)
|
||||
if status:
|
||||
query = query.where(Job.status == status)
|
||||
if kind:
|
||||
query = query.where(Job.kind == kind)
|
||||
if since:
|
||||
query = query.where(Job.created_at >= since)
|
||||
if until:
|
||||
query = query.where(Job.created_at <= until)
|
||||
if client:
|
||||
query = query.where(or_(Tenant.name.ilike(f"%{client}%"), Installation.customer_code.ilike(f"%{client}%")))
|
||||
if search:
|
||||
term = f"%{search}%"
|
||||
query = query.where(or_(Tenant.name.ilike(term), Installation.display_name.ilike(term), Installation.hostname.ilike(term), Job.kind.ilike(term), Job.status.ilike(term)))
|
||||
ordered = query.order_by(Job.created_at.desc(), Job.id.desc())
|
||||
rows = session.execute(ordered.offset((page - 1) * page_size).limit(page_size)).all()
|
||||
count_query = query.with_only_columns(func.count(Job.id)).order_by(None)
|
||||
total = session.scalar(count_query) or 0
|
||||
return {"page": page, "page_size": page_size, "items": [_job_item(row, agent, installation, tenant) for row, agent, installation, tenant in rows], "total": total}
|
||||
|
||||
|
||||
def dashboard_summary(session, now=None):
|
||||
now = now or datetime.now(timezone.utc)
|
||||
query = 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).order_by(Installation.id.asc())
|
||||
items = [installation_view(i, a, t, s) for i, a, t, s in session.execute(query).all()]
|
||||
counts = {state: sum(item["health"] == state for item in items) for state in ("ONLINE", "DEGRADED", "OFFLINE", "FAILED", "UNKNOWN")}
|
||||
items = [installation_view(i, a, t, s, now=now) for i, a, t, s in session.execute(query).all()]
|
||||
counts = {state: sum(item["technical_health"] == state for item in items) for state in HEALTH_STATES}
|
||||
totals = {key: sum(int(item[key] or 0) for item in items) for key in ("total_products", "fiscally_complete", "safe_blocked", "failed_products")}
|
||||
blocker_counts = {}
|
||||
for item in items:
|
||||
for key, value in (item["blockers"] or {}).items(): blocker_counts[key] = blocker_counts.get(key, 0) + int(value or 0)
|
||||
return {"agents_total": len(items), **{key.lower(): value for key, value in counts.items()}, **totals, "blockers": blocker_counts, "stale_agents": [item for item in items if item["health"] == "OFFLINE"], "outdated_agents": [item for item in items if item["version_drift"] == "OUTDATED"]}
|
||||
for key, value in (item["blockers"] or {}).items():
|
||||
blocker_counts[key] = blocker_counts.get(key, 0) + int(value or 0)
|
||||
|
||||
blocker_rows = session.scalars(select(BlockedProduct).where(BlockedProduct.resolved_at.is_(None))).all()
|
||||
open_blocker_breakdown = {field: 0 for field in BLOCKER_FIELDS}
|
||||
product_pairs = set()
|
||||
for row in blocker_rows:
|
||||
product_pairs.add((row.tenant_id, row.product_code))
|
||||
key = str(row.field or "").upper()
|
||||
open_blocker_breakdown[key if key in open_blocker_breakdown else "OTHER"] += 1
|
||||
|
||||
attention = []
|
||||
for item in items:
|
||||
if item["technical_health"] in {"FAILED", "DEGRADED", "OFFLINE", "UNKNOWN"}:
|
||||
rank = {"FAILED": 0, "DEGRADED": 1, "OFFLINE": 2, "UNKNOWN": 3}[item["technical_health"]]
|
||||
attention.append((_attention_item(item, item["health_reason"], _technical_timestamp(item), rank=rank, now=now), rank, item["agent_id"]))
|
||||
item_by_tenant = {item["tenant_id"]: item for item in items}
|
||||
for row in blocker_rows:
|
||||
item = item_by_tenant.get(row.tenant_id)
|
||||
if item and (row.field or "").upper() in BLOCKER_FIELDS[:-1]:
|
||||
attention.append((_attention_item(item, f"{row.field or row.blocker_type}: {row.reason_code}", row.last_seen_at, rank=4, link=f"/blockers/{row.id}", now=now), 4, row.id))
|
||||
for item in items:
|
||||
if item["version_drift"] == "OUTDATED":
|
||||
attention.append((_attention_item(item, "Versão do agent desatualizada", item.get("last_seen_at"), rank=5, now=now), 5, item["agent_id"]))
|
||||
attention.sort(key=lambda entry: (entry[1], -(entry[0]["timestamp"] is not None), entry[2]))
|
||||
|
||||
failed_jobs = session.scalar(select(func.count(Job.id)).where(Job.status == "FAILED")) or 0
|
||||
return {
|
||||
"agents_total": len(items),
|
||||
**{key.lower(): value for key, value in counts.items()},
|
||||
"health_counts": counts,
|
||||
**totals,
|
||||
"blockers": blocker_counts,
|
||||
"stale_agents": [item for item in items if item["technical_health"] == "OFFLINE"],
|
||||
"outdated_agents": [item for item in items if item["version_drift"] == "OUTDATED"],
|
||||
"safe_blocked_products": len(product_pairs),
|
||||
"open_blockers": len(blocker_rows),
|
||||
"open_blocker_breakdown": open_blocker_breakdown,
|
||||
"attention_items": [entry[0] for entry in attention[:10]],
|
||||
"failed_jobs": failed_jobs,
|
||||
}
|
||||
|
||||
|
||||
def list_jobs(session, *, agent_id=None, page=1, page_size=50, status=None, kind=None, since=None, until=None):
|
||||
page, page_size = _page(page, page_size)
|
||||
query = select(Job).order_by(Job.created_at.desc(), Job.id.desc())
|
||||
if agent_id: query = query.where(Job.agent_id == agent_id)
|
||||
if status: query = query.where(Job.status == status)
|
||||
if kind: query = query.where(Job.kind == kind)
|
||||
if since: query = query.where(Job.created_at >= since)
|
||||
if until: query = query.where(Job.created_at <= until)
|
||||
rows = session.scalars(query.offset((page - 1) * page_size).limit(page_size)).all()
|
||||
return {"page": page, "page_size": page_size, "items": [{"job_id": row.id, "agent_id": row.agent_id, "kind": row.kind, "status": row.status, "created_at": _iso(row.created_at), "started_at": _iso(row.started_at), "finished_at": _iso(row.finished_at), "attempts": row.attempts, "result_summary": row.result_summary, "error": row.error} for row in rows], "total": len(rows)}
|
||||
|
||||
|
||||
def list_blockers(session, *, tenant_id=None, agent_id=None, page=1, page_size=50, blocker_type=None, field=None, resolved=False):
|
||||
page, page_size = _page(page, page_size)
|
||||
query = select(BlockedProduct, Tenant).join(Tenant, Tenant.id == BlockedProduct.tenant_id).where(BlockedProduct.resolved_at.is_not(None) if resolved else BlockedProduct.resolved_at.is_(None)).order_by(BlockedProduct.last_seen_at.desc(), BlockedProduct.id.desc())
|
||||
if tenant_id: query = query.where(BlockedProduct.tenant_id == tenant_id)
|
||||
if blocker_type: query = query.where(BlockedProduct.blocker_type == blocker_type)
|
||||
if field: query = query.where(BlockedProduct.field == field)
|
||||
rows = session.execute(query.offset((page - 1) * page_size).limit(page_size)).all()
|
||||
return {"page": page, "page_size": page_size, "items": [{"id": row.id, "tenant_id": row.tenant_id, "client": tenant.name, "product_code": row.product_code, "description": row.description, "field": row.field, "blocker_type": row.blocker_type, "reason": row.reason_code, "evidence_summary": row.evidence_summary, "first_seen_at": _iso(row.first_seen_at), "last_seen_at": _iso(row.last_seen_at), "retry_count": row.retry_count, "resolved_at": _iso(row.resolved_at)} for row, tenant in rows], "total": len(rows)}
|
||||
def blocker_detail(session, blocker_id):
|
||||
row = session.execute(_blockers_query(session, status="ALL").where(BlockedProduct.id == blocker_id)).first()
|
||||
if not row:
|
||||
return None
|
||||
blocker, tenant, installation, agent = row
|
||||
item = _blocker_item(blocker, tenant, installation, agent)
|
||||
events = session.scalars(select(AgentEvent).where(AgentEvent.blocker_id == blocker_id).order_by(AgentEvent.occurred_at.desc(), AgentEvent.id.desc()).limit(100)).all()
|
||||
event_views = [{"id": event.id, "severity": event.severity, "event_type": event.event_type, "message": event.message, "context": event.structured_context, "job_id": event.job_id, "product_code": event.product_code, "occurred_at": _iso(event.occurred_at)} for event in events]
|
||||
job_ids = {event.job_id for event in events if event.job_id is not None}
|
||||
related_jobs = []
|
||||
if job_ids:
|
||||
job_query = select(Job, Agent, Installation, Tenant).join(Agent, Agent.id == Job.agent_id).join(Installation, Installation.agent_id == Agent.id).join(Tenant, Tenant.id == Installation.tenant_id).where(Job.id.in_(job_ids)).order_by(Job.created_at.desc(), Job.id.desc()).limit(100)
|
||||
related_jobs = [_job_item(job, related_agent, related_installation, related_tenant) for job, related_agent, related_installation, related_tenant in session.execute(job_query).all()]
|
||||
audits = session.scalars(select(AuditEvent).where(AuditEvent.target_type == "blocker", AuditEvent.target_id == str(blocker_id)).order_by(AuditEvent.occurred_at.desc(), AuditEvent.id.desc()).limit(100)).all()
|
||||
audit_views = [{"id": audit.id, "admin_user_id": audit.admin_user_id, "action": audit.action, "target_id": audit.target_id, "context": audit.context, "severity": audit.severity, "occurred_at": _iso(audit.occurred_at)} for audit in audits]
|
||||
server = {
|
||||
"hostname": installation.hostname if installation else None,
|
||||
"company_name": installation.company_name if installation else None,
|
||||
"company_code": installation.company_code if installation else None,
|
||||
"company_uf": installation.company_uf if installation else None,
|
||||
"company_crt": installation.company_crt if installation else None,
|
||||
"firebird_version": installation.firebird_version if installation else None,
|
||||
"firebird_ods": installation.firebird_ods if installation else None,
|
||||
"database_identifier": installation.database_identifier if installation else None,
|
||||
}
|
||||
return {
|
||||
**item,
|
||||
"evidence": blocker.evidence_summary or {},
|
||||
"installation": {"installation_id": item["installation_id"], "installation_db_id": item["installation_db_id"], "display_name": item["display_name"], "hostname": item["hostname"], "agent_id": item["agent_id"]},
|
||||
"server": server,
|
||||
"related_events": event_views,
|
||||
"related_jobs": related_jobs,
|
||||
"resolution_audit": audit_views,
|
||||
}
|
||||
|
||||
|
||||
def list_panel_events(session, installation_db_id, **filters):
|
||||
from .events import list_events
|
||||
rows = list_events(session, installation_db_id, **filters)
|
||||
return [{"id": row.id, "severity": row.severity, "event_type": row.event_type, "message": row.message, "context": row.structured_context, "job_id": row.job_id, "product_code": row.product_code, "occurred_at": _iso(row.occurred_at)} for row in rows]
|
||||
|
||||
@@ -0,0 +1,190 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from central.app.audit import record_audit
|
||||
from central.app.control_plane import dashboard_summary, list_blockers, list_installations, list_jobs
|
||||
from central.app.events import record_event
|
||||
from central.app.models import Agent, AgentSnapshot, AuditEvent, BlockedProduct, Installation, Job, Tenant
|
||||
from central.tests.test_control_plane import setup_db
|
||||
|
||||
|
||||
NOW = datetime(2026, 9, 22, 19, 0, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def _installation(session, index, *, health="ONLINE", coverage=80, blocked=0, version="1.0.0rc1"):
|
||||
tenant = Tenant(name=f"Client {index}")
|
||||
session.add(tenant)
|
||||
session.flush()
|
||||
agent = Agent(tenant_id=tenant.id, name=f"agent-{index}", token_hash=f"token-{index}".ljust(64, "x"), agent_version=version)
|
||||
session.add(agent)
|
||||
session.flush()
|
||||
last_seen = {
|
||||
"ONLINE": NOW - timedelta(seconds=10),
|
||||
"DEGRADED": NOW - timedelta(seconds=10),
|
||||
"OFFLINE": NOW - timedelta(hours=2),
|
||||
"FAILED": NOW - timedelta(seconds=10),
|
||||
"UNKNOWN": None,
|
||||
}[health]
|
||||
installation = Installation(
|
||||
agent_id=agent.id,
|
||||
tenant_id=tenant.id,
|
||||
installation_id=f"00000000-0000-4000-8000-{index:012d}",
|
||||
display_name=f"Installation {index}",
|
||||
hostname=f"server-{index}",
|
||||
last_seen_at=last_seen,
|
||||
last_status="FAILED" if health == "FAILED" else "IDLE",
|
||||
)
|
||||
session.add(installation)
|
||||
session.flush()
|
||||
snapshot = AgentSnapshot(
|
||||
installation_id=installation.id,
|
||||
total_products=100,
|
||||
fiscally_complete=coverage,
|
||||
coverage_percent=coverage,
|
||||
safe_blocked=blocked,
|
||||
failed_products=1 if health == "FAILED" else 0,
|
||||
last_job_status="FAILED" if health == "DEGRADED" else "COMPLETED",
|
||||
agent_health=health,
|
||||
version_drift="OUTDATED" if version != "1.0.0rc1" else "CURRENT",
|
||||
)
|
||||
session.add(snapshot)
|
||||
session.flush()
|
||||
return tenant, agent, installation, snapshot
|
||||
|
||||
|
||||
def _blocker(session, tenant, *, product, field="CFOP", blocker_type="BLOCKED_MISSING_EVIDENCE", index=0):
|
||||
row = BlockedProduct(
|
||||
tenant_id=tenant.id,
|
||||
company=tenant.name,
|
||||
product_code=product,
|
||||
description=f"Product {product}",
|
||||
blocker_type=blocker_type,
|
||||
field=field,
|
||||
reason_code="INSUFFICIENT_EVIDENCE",
|
||||
evidence_summary={"source": "test"},
|
||||
input_fingerprint=f"input-{index}",
|
||||
blocker_fingerprint=f"blocker-{index}",
|
||||
first_seen_at=NOW - timedelta(hours=3),
|
||||
last_seen_at=NOW - timedelta(minutes=20),
|
||||
retry_count=2,
|
||||
)
|
||||
session.add(row)
|
||||
session.flush()
|
||||
return row
|
||||
|
||||
|
||||
def test_dashboard_reconciles_five_health_states_and_counts_distinct_safe_products():
|
||||
session_factory = setup_db()
|
||||
with session_factory() as session:
|
||||
first = _installation(session, 1, health="ONLINE", blocked=2)[0]
|
||||
second = _installation(session, 2, health="DEGRADED")[0]
|
||||
_installation(session, 3, health="OFFLINE")
|
||||
_installation(session, 4, health="FAILED")
|
||||
_installation(session, 5, health="UNKNOWN")
|
||||
_blocker(session, first, product="P1", field="CFOP", index=1)
|
||||
_blocker(session, first, product="P1", field="NCM", index=2)
|
||||
_blocker(session, second, product="P2", field="ICMS", index=3)
|
||||
session.commit()
|
||||
|
||||
summary = dashboard_summary(session, now=NOW)
|
||||
|
||||
assert summary["health_counts"] == {"ONLINE": 1, "DEGRADED": 1, "OFFLINE": 1, "FAILED": 1, "UNKNOWN": 1}
|
||||
assert summary["agents_total"] == sum(summary["health_counts"].values())
|
||||
assert summary["safe_blocked_products"] == 2
|
||||
assert summary["open_blockers"] == 3
|
||||
assert summary["open_blocker_breakdown"] == {"CFOP": 1, "ST": 0, "ICMS": 1, "CSOSN": 0, "NCM": 1, "OTHER": 0}
|
||||
assert summary["online"] == 1 and summary["safe_blocked"] == 2
|
||||
|
||||
|
||||
def test_attention_items_are_ranked_and_capped_at_ten():
|
||||
session_factory = setup_db()
|
||||
with session_factory() as session:
|
||||
tenants = []
|
||||
for index, health in enumerate(["FAILED"] * 2 + ["DEGRADED"] * 2 + ["OFFLINE"] * 2 + ["UNKNOWN"] * 2 + ["ONLINE"] * 4, 1):
|
||||
tenant = _installation(session, index, health=health, version="0.9.0" if index >= 9 else "1.0.0rc1")[0]
|
||||
tenants.append(tenant)
|
||||
for index, tenant in enumerate(tenants[8:10], 20):
|
||||
_blocker(session, tenant, product=f"P{index}", index=index)
|
||||
session.commit()
|
||||
|
||||
items = dashboard_summary(session, now=NOW)["attention_items"]
|
||||
|
||||
assert len(items) <= 10
|
||||
assert [item["technical_health"] for item in items[:8]] == ["FAILED"] * 2 + ["DEGRADED"] * 2 + ["OFFLINE"] * 2 + ["UNKNOWN"] * 2
|
||||
assert all({"client", "display_name", "hostname", "problem", "timestamp", "age", "link", "technical_health"} <= item.keys() for item in items)
|
||||
|
||||
|
||||
def test_installations_expose_separate_fiscal_and_technical_status_and_worst_sort():
|
||||
session_factory = setup_db()
|
||||
with session_factory() as session:
|
||||
_installation(session, 1, coverage=90, blocked=0)
|
||||
_installation(session, 2, coverage=40, blocked=1, version="0.9.0")
|
||||
session.commit()
|
||||
|
||||
result = list_installations(session, sort="worst_coverage", page_size=10, now=NOW)
|
||||
item = result["items"][0]
|
||||
|
||||
assert item["display_name"] == "Installation 2"
|
||||
assert item["fiscal_completion_percent"] == 40
|
||||
assert item["technical_health"] == "ONLINE"
|
||||
assert item["fiscal_status"] == "BLOCKED"
|
||||
assert item["version_drift"] == "OUTDATED"
|
||||
assert item["technical_ids"]["agent_id"] == item["agent_id"]
|
||||
|
||||
|
||||
def test_jobs_join_operational_identity_and_format_duration():
|
||||
session_factory = setup_db()
|
||||
with session_factory() as session:
|
||||
tenant, agent, installation, _ = _installation(session, 1)
|
||||
session.add(Job(agent_id=agent.id, kind="REPROCESS_ALL", status="FAILED", created_at=NOW, started_at=NOW, finished_at=NOW + timedelta(seconds=194), error="failure"))
|
||||
session.commit()
|
||||
|
||||
result = list_jobs(session, client="Client 1", search="server-1", page_size=10)
|
||||
item = result["items"][0]
|
||||
|
||||
assert item["client"] == tenant.name
|
||||
assert item["display_name"] == installation.display_name
|
||||
assert item["hostname"] == installation.hostname
|
||||
assert item["job_id"]
|
||||
assert item["duration"] == "3 min 14 s"
|
||||
assert item["error"] == "failure"
|
||||
|
||||
|
||||
def test_blockers_filter_with_priority_age_and_stable_pagination():
|
||||
session_factory = setup_db()
|
||||
with session_factory() as session:
|
||||
tenant, _, installation, _ = _installation(session, 1)
|
||||
critical = _blocker(session, tenant, product="P1", field="", blocker_type="TECHNICAL_FAILURE", index=1)
|
||||
_blocker(session, tenant, product="P2", field="CFOP", index=2)
|
||||
session.commit()
|
||||
|
||||
result = list_blockers(session, client="Client 1", installation=installation.display_name, status="OPEN", search="P1", page=1, page_size=1)
|
||||
|
||||
assert result["total"] == 1
|
||||
assert result["items"][0]["id"] == critical.id
|
||||
assert result["items"][0]["priority"] == "CRITICAL"
|
||||
assert result["items"][0]["age"].startswith("há ")
|
||||
assert result["items"][0]["installation_id"] == installation.installation_id
|
||||
|
||||
|
||||
def test_blocker_detail_contains_context_events_jobs_and_resolution_audit():
|
||||
session_factory = setup_db()
|
||||
with session_factory() as session:
|
||||
tenant, agent, installation, _ = _installation(session, 1)
|
||||
blocker = _blocker(session, tenant, product="P1", index=1)
|
||||
job = Job(agent_id=agent.id, kind="REPROCESS_ALL", status="FAILED", created_at=NOW, error="failed")
|
||||
session.add(job)
|
||||
session.flush()
|
||||
record_event(session, tenant_id=tenant.id, installation_id=installation.id, agent_id=agent.id, severity="ERROR", event_type="JOB_FAILED", message="job failed", blocker_id=blocker.id, job_id=job.id, occurred_at=NOW)
|
||||
record_audit(session, admin_user_id=None, tenant_id=tenant.id, action="RESOLVE_BLOCKER", target_type="blocker", target_id=str(blocker.id), context={"reason": "reviewed"})
|
||||
session.commit()
|
||||
|
||||
from central.app.control_plane import blocker_detail
|
||||
|
||||
detail = blocker_detail(session, blocker.id)
|
||||
|
||||
assert detail["client"] == tenant.name
|
||||
assert detail["installation"]["installation_id"] == installation.installation_id
|
||||
assert detail["evidence"]["source"] == "test"
|
||||
assert detail["related_events"][0]["job_id"] == job.id
|
||||
assert detail["related_jobs"][0]["job_id"] == job.id
|
||||
assert detail["resolution_audit"][0]["action"] == "RESOLVE_BLOCKER"
|
||||
Reference in New Issue
Block a user