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"