From 0846b9d1b076c60c57ad2a30a3071480a680bccf Mon Sep 17 00:00:00 2001 From: gti Date: Tue, 22 Sep 2026 16:59:05 -0300 Subject: [PATCH] feat: add operational panel presentation helpers --- central/app/config.py | 2 + central/app/human_auth.py | 4 + central/app/panel_presentation.py | 101 +++++++++++++++++++++++ central/tests/test_control_plane.py | 16 +++- central/tests/test_panel_presentation.py | 51 ++++++++++++ docker-compose.yml | 1 + 6 files changed, 174 insertions(+), 1 deletion(-) create mode 100644 central/app/panel_presentation.py create mode 100644 central/tests/test_panel_presentation.py diff --git a/central/app/config.py b/central/app/config.py index cd27f4c..57214a0 100644 --- a/central/app/config.py +++ b/central/app/config.py @@ -36,6 +36,8 @@ class Settings: web_cookie_name: str = os.getenv("WEB_COOKIE_NAME", "gti_admin_session") web_max_event_message: int = int(os.getenv("WEB_MAX_EVENT_MESSAGE", "2000")) web_max_event_context_bytes: int = int(os.getenv("WEB_MAX_EVENT_CONTEXT_BYTES", "12000")) + ui_timezone: str = os.getenv("UI_TIMEZONE", "America/Bahia") + allow_weak_dev_password: bool = os.getenv("ALLOW_WEAK_DEV_PASSWORD", "0") == "1" ncm_dataset_url: str = os.getenv("NCM_DATASET_URL", "https://portalunico.siscomex.gov.br/classif/api/publico/nomenclatura/download/json") cest_dataset_url: str = os.getenv("CEST_DATASET_URL", "https://www.confaz.fazenda.gov.br/legislacao/convenios/2018/CV142_18") dataset_storage_path: str = os.getenv("DATASET_STORAGE_PATH", "./data/datasets") diff --git a/central/app/human_auth.py b/central/app/human_auth.py index 329b0ac..176fc48 100644 --- a/central/app/human_auth.py +++ b/central/app/human_auth.py @@ -39,6 +39,10 @@ def create_admin(session, username: str, password: str): username = username.strip().lower() if len(username) < 3: raise ValueError("username must have 3 characters") + environment = os.getenv("APP_ENV") or settings.environment + weak_passwords_allowed = os.getenv("ALLOW_WEAK_DEV_PASSWORD", "1" if settings.allow_weak_dev_password else "0") == "1" + if len(password) < 12 and (environment.strip().lower() == "production" or not weak_passwords_allowed): + raise ValueError("password must have 12 characters") if session.scalar(select(AdminUser).where(AdminUser.username == username)): raise ValueError("username already exists") admin = AdminUser(username=username, password_hash=hash_password(password)) diff --git a/central/app/panel_presentation.py b/central/app/panel_presentation.py new file mode 100644 index 0000000..67cc5d4 --- /dev/null +++ b/central/app/panel_presentation.py @@ -0,0 +1,101 @@ +"""Human-readable values used by the operational control panel.""" + +from __future__ import annotations + +from datetime import datetime, timezone +import os +from zoneinfo import ZoneInfo + +from .config import settings + + +def _as_utc(value: datetime | None) -> datetime | None: + if value is None: + return None + if value.tzinfo is None: + value = value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +def _ui_timezone() -> ZoneInfo: + return ZoneInfo(os.getenv("UI_TIMEZONE", settings.ui_timezone)) + + +def format_ui_datetime(value: datetime | None) -> str: + """Format a stored timestamp in the configured UI timezone.""" + value = _as_utc(value) + if value is None: + return "-" + return value.astimezone(_ui_timezone()).strftime("%d/%m/%Y %H:%M") + + +def format_age(value: datetime | None, now: datetime | None = None) -> str: + value = _as_utc(value) + if value is None: + return "-" + now = _as_utc(now or datetime.now(timezone.utc)) + seconds = int((now - value).total_seconds()) + if seconds <= 0: + return "agora" + minutes = seconds // 60 + if minutes < 1: + return "agora" + if minutes < 60: + return f"há {minutes} min" + return f"há {minutes // 60} h" + + +def format_duration(started_at: datetime | None, finished_at: datetime | None, now: datetime | None = None) -> str: + del now + started_at = _as_utc(started_at) + if started_at is None: + return "-" + if finished_at is None: + return "em andamento" + elapsed = max(0, int((_as_utc(finished_at) - started_at).total_seconds())) + hours, remainder = divmod(elapsed, 3600) + minutes, seconds = divmod(remainder, 60) + parts = [] + if hours: + parts.append(f"{hours} h") + if minutes: + parts.append(f"{minutes} min") + if seconds or not parts: + parts.append(f"{seconds} s") + return " ".join(parts) + + +def _blocker_value(blocker, name: str) -> str: + if isinstance(blocker, dict): + return str(blocker.get(name) or "").upper() + return str(getattr(blocker, name, "") or "").upper() + + +def operational_priority(blocker) -> str: + blocker_type = _blocker_value(blocker, "blocker_type") + field = _blocker_value(blocker, "field") + reason = _blocker_value(blocker, "reason_code") + operational_text = f"{blocker_type} {reason}" + if blocker_type in {"TECHNICAL_FAILURE", "CONNECTION_ERROR", "AGENT_OFFLINE", "JOB_FAILED"} or any(token in operational_text for token in ("TECHNICAL", "CONNECTION", "TIMEOUT")): + return "CRITICAL" + if blocker_type in {"BLOCKED_MISSING_EVIDENCE", "MISSING_EVIDENCE"}: + return "HIGH" + if field in {"CFOP", "ST", "ICMS", "CSOSN", "NCM"} or reason in {"INSUFFICIENT_EVIDENCE", "NO_EVIDENCE"}: + return "NORMAL" + return "LOW" + + +_HEALTH_LABELS = { + "ONLINE": "Online", + "DEGRADED": "Degradado", + "OFFLINE": "Offline", + "FAILED": "Falha", + "UNKNOWN": "Desconhecido", +} + + +def health_label(value) -> str: + if value is None or value == "": + return "-" + text = str(value) + return _HEALTH_LABELS.get(text.upper(), text) diff --git a/central/tests/test_control_plane.py b/central/tests/test_control_plane.py index 40232ce..7b115fb 100644 --- a/central/tests/test_control_plane.py +++ b/central/tests/test_control_plane.py @@ -1,4 +1,5 @@ from datetime import datetime, timedelta, timezone +import pytest from fastapi.testclient import TestClient from sqlalchemy import create_engine @@ -44,14 +45,27 @@ def test_password_and_event_redaction_and_retention(): assert session.get(AgentEvent, old.id) is None and session.get(AgentEvent, keep.id) is not None -def test_local_development_admin_accepts_operator_supplied_short_password(): +def test_short_local_password_requires_explicit_setting(monkeypatch): session_factory = setup_db() with session_factory() as session: + monkeypatch.delenv("ALLOW_WEAK_DEV_PASSWORD", raising=False) + with pytest.raises(ValueError): + create_admin(session, "admin", "admin") + monkeypatch.setenv("ALLOW_WEAK_DEV_PASSWORD", "1") admin = create_admin(session, "admin", "admin") assert admin.username == "admin" assert verify_password("admin", admin.password_hash) +def test_production_admin_keeps_strong_password_minimum(monkeypatch): + session_factory = setup_db() + with session_factory() as session: + monkeypatch.setenv("APP_ENV", "production") + monkeypatch.setenv("ALLOW_WEAK_DEV_PASSWORD", "1") + with pytest.raises(ValueError): + create_admin(session, "admin", "admin") + + def test_health_model_separates_safe_blocked_from_technical_failure(): now = datetime.now(timezone.utc) recent = SimpleNamespace(last_seen_at=now, failed_products=0, last_job_status="COMPLETED") diff --git a/central/tests/test_panel_presentation.py b/central/tests/test_panel_presentation.py new file mode 100644 index 0000000..a6011f2 --- /dev/null +++ b/central/tests/test_panel_presentation.py @@ -0,0 +1,51 @@ +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace + +from central.app.panel_presentation import ( + format_age, + format_duration, + format_ui_datetime, + health_label, + operational_priority, +) + + +def test_ui_times_use_configured_timezone_and_human_age(monkeypatch): + monkeypatch.setenv("UI_TIMEZONE", "America/Bahia") + value = datetime(2026, 9, 22, 19, 11, tzinfo=timezone.utc) + + assert format_ui_datetime(value) == "22/09/2026 16:11" + assert format_age(value, datetime(2026, 9, 22, 19, 19, tzinfo=timezone.utc)) == "há 8 min" + + +def test_ui_datetime_treats_naive_values_as_utc(): + value = datetime(2026, 9, 22, 19, 11) + + assert format_ui_datetime(value) == "22/09/2026 16:11" + + +def test_age_handles_empty_and_future_values(): + now = datetime(2026, 9, 22, 19, 19, tzinfo=timezone.utc) + + assert format_age(None, now) == "-" + assert format_age(now + timedelta(minutes=1), now) == "agora" + assert format_age(now - timedelta(hours=2), now) == "há 2 h" + + +def test_duration_and_priority_are_deterministic(): + start = datetime(2026, 9, 22, 19, 0, tzinfo=timezone.utc) + finish = start + timedelta(seconds=194) + + assert format_duration(None, finish) == "-" + assert format_duration(start, None) == "em andamento" + assert format_duration(start, finish) == "3 min 14 s" + assert operational_priority(SimpleNamespace(blocker_type="TECHNICAL_FAILURE", field="", reason_code="X")) == "CRITICAL" + + +def test_health_label_maps_known_states_and_preserves_unknown_states(): + assert health_label("ONLINE") == "Online" + assert health_label("DEGRADED") == "Degradado" + assert health_label("OFFLINE") == "Offline" + assert health_label("FAILED") == "Falha" + assert health_label("UNKNOWN") == "Desconhecido" + assert health_label("NEW_STATE") == "NEW_STATE" diff --git a/docker-compose.yml b/docker-compose.yml index 93869dc..8126a56 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -29,6 +29,7 @@ services: # created with central/scripts/register_agent.py, never via this variable. AGENT_DEV_TOKEN: ${AGENT_DEV_TOKEN:-} APP_ENV: ${APP_ENV:-development} + ALLOW_WEAK_DEV_PASSWORD: "1" LOG_LEVEL: ${LOG_LEVEL:-INFO} BASE_URL: ${BASE_URL:-http://localhost:8000} AGENT_ONLINE_THRESHOLD_SECONDS: ${AGENT_ONLINE_THRESHOLD_SECONDS:-120}