feat: refine operational control panel pages

central/tests/test_control_panel_pages.py

@@ -0,0 +1,155 @@
+import re
+from datetime import datetime, timedelta, timezone
+
+from fastapi.testclient import TestClient
+from sqlalchemy import select
+
+from central.app.blocked_products import upsert_blocker
+from central.app.events import record_event
+from central.app.main import create_app
+from central.app.models import Agent, AgentSnapshot, Installation, Job
+from central.tests.test_control_plane import seed, setup_db
+
+
+def _panel_client():
+    session_factory = setup_db()
+    agent_id, _ = seed(session_factory)
+    now = datetime(2026, 9, 22, 19, 11, tzinfo=timezone.utc)
+    with session_factory() as session:
+        installation = session.scalar(select(Installation).where(Installation.agent_id == agent_id))
+        agent = session.get(Agent, agent_id)
+        agent.agent_version = "0.9.0"
+        snapshot = session.scalar(select(AgentSnapshot).where(AgentSnapshot.installation_id == installation.id))
+        snapshot.total_products = 50
+        snapshot.fiscally_complete = 49
+        snapshot.coverage_percent = 98
+        job = Job(
+            agent_id=agent_id,
+            kind="REPROCESS_PRODUCTS",
+            status="RUNNING",
+            created_at=now - timedelta(minutes=3),
+            started_at=now - timedelta(minutes=2),
+            payload={"codigos": ["P-001"]},
+        )
+        session.add(job)
+        session.flush()
+        blocker = upsert_blocker(
+            session,
+            installation.tenant_id,
+            {
+                "company": "Acme",
+                "product_code": "P-001",
+                "description": "Produto de teste",
+                "blocker_type": "BLOCKED_MISSING_EVIDENCE",
+                "field": "CFOP",
+                "reason_code": "MISSING_CONTEXT",
+                "input_fingerprint": "page-test",
+                "evidence_summary": {"source": "snapshot"},
+            },
+            installation_id=installation.id,
+        )
+        record_event(
+            session,
+            tenant_id=installation.tenant_id,
+            installation_id=installation.id,
+            agent_id=agent_id,
+            severity="WARNING",
+            event_type="PRODUCT_BLOCKED",
+            message="Contexto fiscal ausente",
+            job_id=job.id,
+            product_code="P-001",
+            blocker_id=blocker.id,
+            occurred_at=now,
+        )
+        session.commit()
+
+    client = TestClient(create_app(database_probe=lambda: True, session_factory=session_factory, bootstrap_token=None))
+    login = client.post("/login", data={"username": "operator", "password": "a-secure-password"}, follow_redirects=False)
+    assert login.status_code == 303
+    return client, agent_id, blocker.id
+
+
+def test_dashboard_renders_operational_queue_labels_and_human_dates():
+    client, _, _ = _panel_client()
+
+    response = client.get("/")
+
+    assert response.status_code == 200
+    for label in ("Agents total", "Online", "Degraded", "Offline", "Failed", "Unknown"):
+        assert label in response.text
+    assert "Produtos bloqueados" in response.text
+    assert "Blockers abertos" in response.text
+    assert "ATENÇÃO NECESSÁRIA" in response.text
+    assert "Ver todos" in response.text
+    assert "Jobs com falha" in response.text
+    assert "Agents desatualizados" in response.text
+    assert not re.search(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}", response.text)
+
+
+def test_agents_render_separate_health_and_fiscal_status_with_server_sort():
+    client, _, _ = _panel_client()
+
+    response = client.get("/agents?sort=worst_coverage&health=ONLINE&search=Acme")
+
+    assert response.status_code == 200
+    for label in ("Cliente", "Nome de exibição", "Hostname", "Saúde técnica", "Status fiscal", "Cobertura fiscal", "Versão"):
+        assert label in response.text
+    assert "Acme ERP" in response.text
+    assert "srv-01" in response.text
+    assert "98,0%" in response.text
+    assert 'name="health"' in response.text
+    assert 'value="Acme"' in response.text
+    assert 'value="worst_coverage"' in response.text
+
+
+def test_jobs_render_client_identity_duration_and_preserve_filters():
+    client, _, _ = _panel_client()
+
+    response = client.get("/jobs?status=RUNNING&search=Acme")
+
+    assert response.status_code == 200
+    for label in ("Cliente", "Nome de exibição", "Hostname", "Job ID", "Tipo", "Status", "Criado", "Finalizado", "Duração", "Erro"):
+        assert label in response.text
+    assert "Acme" in response.text
+    assert "srv-01" in response.text
+    assert "em andamento" in response.text
+    assert 'value="Acme"' in response.text
+    assert '<option value="RUNNING" selected>' in response.text
+    assert "Agent 1" not in response.text
+
+
+def test_blockers_render_visible_filters_summary_priority_and_detail_link():
+    client, _, blocker_id = _panel_client()
+
+    response = client.get("/blockers?client=Acme&installation=srv-01&field=CFOP&blocker_type=BLOCKED_MISSING_EVIDENCE&status=OPEN&age=0h&search=P-001")
+
+    assert response.status_code == 200
+    for label in ("Cliente", "Instalação / Agent / Servidor", "Campo", "Tipo de blocker", "Status", "Idade", "Produto ou busca", "Blockers abertos", "Produtos afetados", "Clientes afetados", "CFOP", "ST", "ICMS", "CSOSN", "NCM", "OTHER", "Prioridade"):
+        assert label in response.text
+    assert f"/blockers/{blocker_id}" in response.text
+    assert "CRITICAL" in response.text or "HIGH" in response.text
+    assert 'value="Acme"' in response.text
+    assert 'value="BLOCKED_MISSING_EVIDENCE"' in response.text
+    assert "window.confirm" not in response.text
+
+
+def test_agent_and_blocker_details_keep_safe_forms_and_human_event_rows():
+    client, agent_id, blocker_id = _panel_client()
+
+    agent_response = client.get(f"/agents/{agent_id}?tab=events&severity=WARNING&event_type=PRODUCT_BLOCKED&page=2")
+    blocker_response = client.get(f"/blockers/{blocker_id}")
+
+    assert agent_response.status_code == 200
+    for label in ("Cliente", "Nome de exibição", "Hostname", "Saúde técnica", "Status fiscal", "Último heartbeat", "Versão", "Empresa", "UF", "CRT", "Eventos recentes", "Severidade", "Tipo de evento"):
+        assert label in agent_response.text
+    assert "Contexto fiscal ausente" in agent_response.text
+    assert not re.search(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}", agent_response.text)
+    assert "severity=WARNING" in agent_response.text
+    assert "event_type=PRODUCT_BLOCKED" in agent_response.text
+
+    assert blocker_response.status_code == 200
+    for label in ("Detalhes do blocker", "Evidência", "Contexto atual", "Salvar contexto e reprocessar", "Chave de contexto", "Valor", "Escopo", "Nota"):
+        assert label in blocker_response.text
+    assert 'name="context"' not in blocker_response.text
+    assert "data-confirm" not in blocker_response.text
+    assert not re.search(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}", blocker_response.text)