69 lines
2.7 KiB
Python
69 lines
2.7 KiB
Python
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def read(path: str) -> str:
|
|
return (ROOT / path).read_text(encoding="utf-8")
|
|
|
|
|
|
def test_v480_outbox_stale_recovery_service_and_worker_are_present():
|
|
service = read("app/integration_outbox_service.py")
|
|
worker = read("scripts/process_outbox.py")
|
|
script = read("scripts/recover_stale_outbox.py")
|
|
assert "def recover_stale_processing_outbox" in service
|
|
assert "OUTBOX_STALE_PROCESSING_MINUTES" in service
|
|
assert "OUTBOX_STALE_RECOVERY_MODE" in service
|
|
assert "manual_only" in service
|
|
assert "retry_pending" in service
|
|
assert "mark_failed" in service
|
|
assert "recover_stale_processing_outbox" in worker
|
|
assert "OUTBOX_RECOVER_STALE_BEFORE_PROCESS" in worker
|
|
assert "recover_stale_processing_outbox" in script
|
|
|
|
|
|
def test_v480_operator_audit_service_and_sensitive_actions_are_audited():
|
|
audit = read("app/operator_audit_service.py")
|
|
tasks = read("app/task_service.py")
|
|
outbox_page = read("app/admin_ui/pages/outbox.py")
|
|
assert "def record_operator_action" in audit
|
|
assert "event_type = 'operator_action'" in audit
|
|
assert "task_completed" in tasks
|
|
assert "task_skipped" in tasks
|
|
assert "task_reclassified" in tasks
|
|
assert "outbox_reprocess_requested" in outbox_page
|
|
assert "outbox_marked_failed" in outbox_page
|
|
assert "outbox_ignored" in outbox_page
|
|
|
|
|
|
def test_v480_sensitive_actions_have_htmx_confirmation():
|
|
tasks_page = read("app/admin_ui/pages/tasks.py")
|
|
outbox_page = read("app/admin_ui/pages/outbox.py")
|
|
assert "hx-confirm=\"Confirmar conclusão desta tarefa?\"" in tasks_page
|
|
assert "hx-confirm=\"Ignorar esta tarefa?" in tasks_page
|
|
assert "hx-confirm=\"Guardar reclassificação manual desta tarefa?\"" in tasks_page
|
|
assert "hx-confirm=\"Reprocessar este item da outbox?\"" in outbox_page
|
|
assert "hx-confirm=\"Ignorar este item da outbox?\"" in outbox_page
|
|
assert "hx-confirm=\"Marcar este item como failed?\"" in outbox_page
|
|
|
|
|
|
def test_v480_operational_health_and_operations_filters_are_expanded():
|
|
ops = read("app/admin_ui/view_models/operations.py")
|
|
service = read("app/operations_service.py")
|
|
system = read("app/admin_ui/pages/system.py")
|
|
assert "Ambíguas" in ops
|
|
assert "Atrasadas" in ops
|
|
assert "is_overdue" in ops
|
|
assert "operator_actions_24h" in service
|
|
assert "outbox_stale" in service
|
|
assert "Outbox stale" in system
|
|
assert "Ações operador 24h" in system
|
|
|
|
|
|
def test_v480_documentation_exists():
|
|
doc = read("docs/CLIENTFLOW_V480_OPERATIONAL_AUTOMATION_AUDIT_RECOVERY.md")
|
|
assert "Operational Automation, Audit & Recovery" in doc
|
|
assert "OUTBOX_STALE_PROCESSING_MINUTES" in doc
|
|
assert "operator_action" in doc
|
|
assert "hx-confirm" in doc
|