Compare commits
2 Commits
5ca21b598f
...
d6e765ce3c
| Author | SHA1 | Date |
|---|---|---|
|
|
d6e765ce3c | 2 weeks ago |
|
|
9a31b0c5ae | 2 weeks ago |
@ -0,0 +1,477 @@
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
|
||||
from admin_app.api.dependencies import get_settings, require_admin_permission
|
||||
from admin_app.api.schemas import (
|
||||
AdminBotFlowReportCatalogResponse,
|
||||
AdminBotFlowReportOverviewResponse,
|
||||
AdminBotFlowReportResponse,
|
||||
AdminConversationTelemetryReportCatalogResponse,
|
||||
AdminConversationTelemetryReportOverviewResponse,
|
||||
AdminConversationTelemetryReportResponse,
|
||||
AdminRentalReportCatalogResponse,
|
||||
AdminRentalReportOverviewResponse,
|
||||
AdminRentalReportResponse,
|
||||
AdminReportDatasetListResponse,
|
||||
AdminReportDatasetResponse,
|
||||
AdminReportOverviewResponse,
|
||||
AdminRevenueReportCatalogResponse,
|
||||
AdminRevenueReportOverviewResponse,
|
||||
AdminRevenueReportResponse,
|
||||
AdminSalesReportCatalogResponse,
|
||||
AdminSalesReportOverviewResponse,
|
||||
AdminSalesReportResponse,
|
||||
)
|
||||
from admin_app.core import AdminSettings, AuthenticatedStaffPrincipal
|
||||
from admin_app.services import ReportService
|
||||
from shared.contracts import AdminPermission
|
||||
|
||||
router = APIRouter(prefix="/reports", tags=["reports"])
|
||||
|
||||
|
||||
def _build_service(settings: AdminSettings) -> ReportService:
|
||||
return ReportService(settings)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/overview",
|
||||
response_model=AdminReportOverviewResponse,
|
||||
)
|
||||
def reports_overview(
|
||||
settings: AdminSettings = Depends(get_settings),
|
||||
_: AuthenticatedStaffPrincipal = Depends(
|
||||
require_admin_permission(AdminPermission.VIEW_REPORTS)
|
||||
),
|
||||
):
|
||||
service = _build_service(settings)
|
||||
payload = service.build_overview_payload()
|
||||
return AdminReportOverviewResponse(
|
||||
service="orquestrador-admin",
|
||||
mode=payload["mode"],
|
||||
metrics=payload["metrics"],
|
||||
materialization=payload["materialization"],
|
||||
report_families=payload["report_families"],
|
||||
next_steps=payload["next_steps"],
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/datasets",
|
||||
response_model=AdminReportDatasetListResponse,
|
||||
)
|
||||
def report_datasets(
|
||||
settings: AdminSettings = Depends(get_settings),
|
||||
_: AuthenticatedStaffPrincipal = Depends(
|
||||
require_admin_permission(AdminPermission.VIEW_REPORTS)
|
||||
),
|
||||
):
|
||||
service = _build_service(settings)
|
||||
payload = service.list_datasets_payload()
|
||||
return AdminReportDatasetListResponse(
|
||||
service="orquestrador-admin",
|
||||
source=payload["source"],
|
||||
materialization=payload["materialization"],
|
||||
datasets=payload["datasets"],
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/datasets/{dataset_key}",
|
||||
response_model=AdminReportDatasetResponse,
|
||||
)
|
||||
def report_dataset_detail(
|
||||
dataset_key: str,
|
||||
settings: AdminSettings = Depends(get_settings),
|
||||
_: AuthenticatedStaffPrincipal = Depends(
|
||||
require_admin_permission(AdminPermission.VIEW_REPORTS)
|
||||
),
|
||||
):
|
||||
service = _build_service(settings)
|
||||
payload = service.get_dataset_payload(dataset_key)
|
||||
if payload is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Dataset operacional nao encontrado para relatorio.",
|
||||
)
|
||||
|
||||
return AdminReportDatasetResponse(
|
||||
service="orquestrador-admin",
|
||||
source=payload["source"],
|
||||
materialization=payload["materialization"],
|
||||
dataset=payload["dataset"],
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/sales/overview",
|
||||
response_model=AdminSalesReportOverviewResponse,
|
||||
)
|
||||
def sales_reports_overview(
|
||||
settings: AdminSettings = Depends(get_settings),
|
||||
_: AuthenticatedStaffPrincipal = Depends(
|
||||
require_admin_permission(AdminPermission.VIEW_REPORTS)
|
||||
),
|
||||
):
|
||||
service = _build_service(settings)
|
||||
payload = service.build_sales_overview_payload()
|
||||
return AdminSalesReportOverviewResponse(
|
||||
service="orquestrador-admin",
|
||||
domain=payload["domain"],
|
||||
mode=payload["mode"],
|
||||
source_dataset_keys=payload["source_dataset_keys"],
|
||||
metrics=payload["metrics"],
|
||||
materialization=payload["materialization"],
|
||||
reports=payload["reports"],
|
||||
next_steps=payload["next_steps"],
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/sales/reports",
|
||||
response_model=AdminSalesReportCatalogResponse,
|
||||
)
|
||||
def sales_reports_catalog(
|
||||
settings: AdminSettings = Depends(get_settings),
|
||||
_: AuthenticatedStaffPrincipal = Depends(
|
||||
require_admin_permission(AdminPermission.VIEW_REPORTS)
|
||||
),
|
||||
):
|
||||
service = _build_service(settings)
|
||||
payload = service.list_sales_reports_payload()
|
||||
return AdminSalesReportCatalogResponse(
|
||||
service="orquestrador-admin",
|
||||
domain=payload["domain"],
|
||||
source=payload["source"],
|
||||
materialization=payload["materialization"],
|
||||
reports=payload["reports"],
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/sales/reports/{report_key}",
|
||||
response_model=AdminSalesReportResponse,
|
||||
)
|
||||
def sales_report_detail(
|
||||
report_key: str,
|
||||
settings: AdminSettings = Depends(get_settings),
|
||||
_: AuthenticatedStaffPrincipal = Depends(
|
||||
require_admin_permission(AdminPermission.VIEW_REPORTS)
|
||||
),
|
||||
):
|
||||
service = _build_service(settings)
|
||||
payload = service.get_sales_report_payload(report_key)
|
||||
if payload is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Relatorio de vendas nao encontrado.",
|
||||
)
|
||||
|
||||
return AdminSalesReportResponse(
|
||||
service="orquestrador-admin",
|
||||
domain=payload["domain"],
|
||||
source=payload["source"],
|
||||
materialization=payload["materialization"],
|
||||
report=payload["report"],
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/arrecadacao/overview",
|
||||
response_model=AdminRevenueReportOverviewResponse,
|
||||
)
|
||||
def revenue_reports_overview(
|
||||
settings: AdminSettings = Depends(get_settings),
|
||||
_: AuthenticatedStaffPrincipal = Depends(
|
||||
require_admin_permission(AdminPermission.VIEW_REPORTS)
|
||||
),
|
||||
):
|
||||
service = _build_service(settings)
|
||||
payload = service.build_revenue_overview_payload()
|
||||
return AdminRevenueReportOverviewResponse(
|
||||
service="orquestrador-admin",
|
||||
area=payload["area"],
|
||||
source_domain=payload["source_domain"],
|
||||
mode=payload["mode"],
|
||||
source_dataset_keys=payload["source_dataset_keys"],
|
||||
metrics=payload["metrics"],
|
||||
materialization=payload["materialization"],
|
||||
reports=payload["reports"],
|
||||
next_steps=payload["next_steps"],
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/arrecadacao/reports",
|
||||
response_model=AdminRevenueReportCatalogResponse,
|
||||
)
|
||||
def revenue_reports_catalog(
|
||||
settings: AdminSettings = Depends(get_settings),
|
||||
_: AuthenticatedStaffPrincipal = Depends(
|
||||
require_admin_permission(AdminPermission.VIEW_REPORTS)
|
||||
),
|
||||
):
|
||||
service = _build_service(settings)
|
||||
payload = service.list_revenue_reports_payload()
|
||||
return AdminRevenueReportCatalogResponse(
|
||||
service="orquestrador-admin",
|
||||
area=payload["area"],
|
||||
source_domain=payload["source_domain"],
|
||||
source=payload["source"],
|
||||
materialization=payload["materialization"],
|
||||
reports=payload["reports"],
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/arrecadacao/reports/{report_key}",
|
||||
response_model=AdminRevenueReportResponse,
|
||||
)
|
||||
def revenue_report_detail(
|
||||
report_key: str,
|
||||
settings: AdminSettings = Depends(get_settings),
|
||||
_: AuthenticatedStaffPrincipal = Depends(
|
||||
require_admin_permission(AdminPermission.VIEW_REPORTS)
|
||||
),
|
||||
):
|
||||
service = _build_service(settings)
|
||||
payload = service.get_revenue_report_payload(report_key)
|
||||
if payload is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Relatorio de arrecadacao nao encontrado.",
|
||||
)
|
||||
|
||||
return AdminRevenueReportResponse(
|
||||
service="orquestrador-admin",
|
||||
area=payload["area"],
|
||||
source_domain=payload["source_domain"],
|
||||
source=payload["source"],
|
||||
materialization=payload["materialization"],
|
||||
report=payload["report"],
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/locacao/overview",
|
||||
response_model=AdminRentalReportOverviewResponse,
|
||||
)
|
||||
def rental_reports_overview(
|
||||
settings: AdminSettings = Depends(get_settings),
|
||||
_: AuthenticatedStaffPrincipal = Depends(
|
||||
require_admin_permission(AdminPermission.VIEW_REPORTS)
|
||||
),
|
||||
):
|
||||
service = _build_service(settings)
|
||||
payload = service.build_rental_overview_payload()
|
||||
return AdminRentalReportOverviewResponse(
|
||||
service="orquestrador-admin",
|
||||
area=payload["area"],
|
||||
source_domain=payload["source_domain"],
|
||||
mode=payload["mode"],
|
||||
source_dataset_keys=payload["source_dataset_keys"],
|
||||
metrics=payload["metrics"],
|
||||
materialization=payload["materialization"],
|
||||
reports=payload["reports"],
|
||||
next_steps=payload["next_steps"],
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/locacao/reports",
|
||||
response_model=AdminRentalReportCatalogResponse,
|
||||
)
|
||||
def rental_reports_catalog(
|
||||
settings: AdminSettings = Depends(get_settings),
|
||||
_: AuthenticatedStaffPrincipal = Depends(
|
||||
require_admin_permission(AdminPermission.VIEW_REPORTS)
|
||||
),
|
||||
):
|
||||
service = _build_service(settings)
|
||||
payload = service.list_rental_reports_payload()
|
||||
return AdminRentalReportCatalogResponse(
|
||||
service="orquestrador-admin",
|
||||
area=payload["area"],
|
||||
source_domain=payload["source_domain"],
|
||||
source=payload["source"],
|
||||
materialization=payload["materialization"],
|
||||
reports=payload["reports"],
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/locacao/reports/{report_key}",
|
||||
response_model=AdminRentalReportResponse,
|
||||
)
|
||||
def rental_report_detail(
|
||||
report_key: str,
|
||||
settings: AdminSettings = Depends(get_settings),
|
||||
_: AuthenticatedStaffPrincipal = Depends(
|
||||
require_admin_permission(AdminPermission.VIEW_REPORTS)
|
||||
),
|
||||
):
|
||||
service = _build_service(settings)
|
||||
payload = service.get_rental_report_payload(report_key)
|
||||
if payload is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Relatorio de locacao nao encontrado.",
|
||||
)
|
||||
|
||||
return AdminRentalReportResponse(
|
||||
service="orquestrador-admin",
|
||||
area=payload["area"],
|
||||
source_domain=payload["source_domain"],
|
||||
source=payload["source"],
|
||||
materialization=payload["materialization"],
|
||||
report=payload["report"],
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/fluxo-bot/overview",
|
||||
response_model=AdminBotFlowReportOverviewResponse,
|
||||
)
|
||||
def bot_flow_reports_overview(
|
||||
settings: AdminSettings = Depends(get_settings),
|
||||
_: AuthenticatedStaffPrincipal = Depends(
|
||||
require_admin_permission(AdminPermission.VIEW_REPORTS)
|
||||
),
|
||||
):
|
||||
service = _build_service(settings)
|
||||
payload = service.build_bot_flow_overview_payload()
|
||||
return AdminBotFlowReportOverviewResponse(
|
||||
service="orquestrador-admin",
|
||||
area=payload["area"],
|
||||
source_domain=payload["source_domain"],
|
||||
mode=payload["mode"],
|
||||
source_dataset_keys=payload["source_dataset_keys"],
|
||||
metrics=payload["metrics"],
|
||||
materialization=payload["materialization"],
|
||||
reports=payload["reports"],
|
||||
next_steps=payload["next_steps"],
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/fluxo-bot/reports",
|
||||
response_model=AdminBotFlowReportCatalogResponse,
|
||||
)
|
||||
def bot_flow_reports_catalog(
|
||||
settings: AdminSettings = Depends(get_settings),
|
||||
_: AuthenticatedStaffPrincipal = Depends(
|
||||
require_admin_permission(AdminPermission.VIEW_REPORTS)
|
||||
),
|
||||
):
|
||||
service = _build_service(settings)
|
||||
payload = service.list_bot_flow_reports_payload()
|
||||
return AdminBotFlowReportCatalogResponse(
|
||||
service="orquestrador-admin",
|
||||
area=payload["area"],
|
||||
source_domain=payload["source_domain"],
|
||||
source=payload["source"],
|
||||
materialization=payload["materialization"],
|
||||
reports=payload["reports"],
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/fluxo-bot/reports/{report_key}",
|
||||
response_model=AdminBotFlowReportResponse,
|
||||
)
|
||||
def bot_flow_report_detail(
|
||||
report_key: str,
|
||||
settings: AdminSettings = Depends(get_settings),
|
||||
_: AuthenticatedStaffPrincipal = Depends(
|
||||
require_admin_permission(AdminPermission.VIEW_REPORTS)
|
||||
),
|
||||
):
|
||||
service = _build_service(settings)
|
||||
payload = service.get_bot_flow_report_payload(report_key)
|
||||
if payload is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Relatorio operacional do fluxo do bot nao encontrado.",
|
||||
)
|
||||
|
||||
return AdminBotFlowReportResponse(
|
||||
service="orquestrador-admin",
|
||||
area=payload["area"],
|
||||
source_domain=payload["source_domain"],
|
||||
source=payload["source"],
|
||||
materialization=payload["materialization"],
|
||||
report=payload["report"],
|
||||
)
|
||||
|
||||
@router.get(
|
||||
"/telemetria-conversacional/overview",
|
||||
response_model=AdminConversationTelemetryReportOverviewResponse,
|
||||
)
|
||||
def conversation_telemetry_reports_overview(
|
||||
settings: AdminSettings = Depends(get_settings),
|
||||
_: AuthenticatedStaffPrincipal = Depends(
|
||||
require_admin_permission(AdminPermission.VIEW_REPORTS)
|
||||
),
|
||||
):
|
||||
service = _build_service(settings)
|
||||
payload = service.build_conversation_telemetry_overview_payload()
|
||||
return AdminConversationTelemetryReportOverviewResponse(
|
||||
service="orquestrador-admin",
|
||||
area=payload["area"],
|
||||
source_domain=payload["source_domain"],
|
||||
mode=payload["mode"],
|
||||
source_dataset_keys=payload["source_dataset_keys"],
|
||||
metrics=payload["metrics"],
|
||||
materialization=payload["materialization"],
|
||||
reports=payload["reports"],
|
||||
next_steps=payload["next_steps"],
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/telemetria-conversacional/reports",
|
||||
response_model=AdminConversationTelemetryReportCatalogResponse,
|
||||
)
|
||||
def conversation_telemetry_reports_catalog(
|
||||
settings: AdminSettings = Depends(get_settings),
|
||||
_: AuthenticatedStaffPrincipal = Depends(
|
||||
require_admin_permission(AdminPermission.VIEW_REPORTS)
|
||||
),
|
||||
):
|
||||
service = _build_service(settings)
|
||||
payload = service.list_conversation_telemetry_reports_payload()
|
||||
return AdminConversationTelemetryReportCatalogResponse(
|
||||
service="orquestrador-admin",
|
||||
area=payload["area"],
|
||||
source_domain=payload["source_domain"],
|
||||
source=payload["source"],
|
||||
materialization=payload["materialization"],
|
||||
reports=payload["reports"],
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/telemetria-conversacional/reports/{report_key}",
|
||||
response_model=AdminConversationTelemetryReportResponse,
|
||||
)
|
||||
def conversation_telemetry_report_detail(
|
||||
report_key: str,
|
||||
settings: AdminSettings = Depends(get_settings),
|
||||
_: AuthenticatedStaffPrincipal = Depends(
|
||||
require_admin_permission(AdminPermission.VIEW_REPORTS)
|
||||
),
|
||||
):
|
||||
service = _build_service(settings)
|
||||
payload = service.get_conversation_telemetry_report_payload(report_key)
|
||||
if payload is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Relatorio de telemetria conversacional nao encontrado.",
|
||||
)
|
||||
|
||||
return AdminConversationTelemetryReportResponse(
|
||||
service="orquestrador-admin",
|
||||
area=payload["area"],
|
||||
source_domain=payload["source_domain"],
|
||||
source=payload["source"],
|
||||
materialization=payload["materialization"],
|
||||
report=payload["report"],
|
||||
)
|
||||
@ -0,0 +1,104 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Iterable
|
||||
|
||||
from shared.contracts import (
|
||||
PRODUCT_OPERATIONAL_DATASETS,
|
||||
SYSTEM_FUNCTIONAL_CONFIGURATIONS,
|
||||
FunctionalConfigurationPropagation,
|
||||
)
|
||||
|
||||
ALLOWED_ADMIN_WRITE_TABLES: tuple[str, ...] = (
|
||||
"admin_audit_logs",
|
||||
"staff_accounts",
|
||||
"staff_sessions",
|
||||
)
|
||||
|
||||
|
||||
class AdminWriteGovernanceViolation(RuntimeError):
|
||||
"""Raised when the admin runtime attempts an ungoverned direct write."""
|
||||
|
||||
|
||||
def ensure_direct_admin_write_allowed(table_name: str) -> None:
|
||||
normalized_table_name = str(table_name or "").strip().lower()
|
||||
if normalized_table_name in ALLOWED_ADMIN_WRITE_TABLES:
|
||||
return
|
||||
|
||||
raise AdminWriteGovernanceViolation(
|
||||
"Escrita direta do admin bloqueada para a tabela "
|
||||
f"'{normalized_table_name or 'desconhecida'}'. "
|
||||
"Use um fluxo governado, versionado e auditavel antes de publicar qualquer efeito no product."
|
||||
)
|
||||
|
||||
|
||||
def enforce_admin_session_write_governance(
|
||||
*,
|
||||
new: Iterable[object] = (),
|
||||
dirty: Iterable[object] = (),
|
||||
deleted: Iterable[object] = (),
|
||||
) -> None:
|
||||
seen_tables: set[str] = set()
|
||||
for instance in (*tuple(new), *tuple(dirty), *tuple(deleted)):
|
||||
table_name = _resolve_table_name(instance)
|
||||
if table_name is None or table_name in seen_tables:
|
||||
continue
|
||||
ensure_direct_admin_write_allowed(table_name)
|
||||
seen_tables.add(table_name)
|
||||
|
||||
|
||||
def build_admin_write_governance_payload() -> dict:
|
||||
governed_configuration_keys = sorted(
|
||||
configuration.config_key
|
||||
for configuration in SYSTEM_FUNCTIONAL_CONFIGURATIONS
|
||||
if configuration.propagation == FunctionalConfigurationPropagation.VERSIONED_PUBLICATION
|
||||
)
|
||||
return {
|
||||
"mode": "admin_internal_tables_only",
|
||||
"allowed_direct_write_tables": list(ALLOWED_ADMIN_WRITE_TABLES),
|
||||
"blocked_operational_dataset_keys": sorted(
|
||||
dataset.dataset_key for dataset in PRODUCT_OPERATIONAL_DATASETS
|
||||
),
|
||||
"blocked_product_source_tables": sorted(
|
||||
{dataset.source_table for dataset in PRODUCT_OPERATIONAL_DATASETS}
|
||||
),
|
||||
"governed_configuration_keys": governed_configuration_keys,
|
||||
"enforcement_points": [
|
||||
"AdminSession.before_flush bloqueia escrita ORM fora do allowlist interno do admin.",
|
||||
"Contratos compartilhados mantem datasets operacionais com write_allowed=false.",
|
||||
"Configuracoes que afetam o runtime do product seguem versioned_publication antes de qualquer efeito operacional.",
|
||||
],
|
||||
"governance_rules": [
|
||||
"O admin nao escreve diretamente nas tabelas operacionais do product.",
|
||||
"Toda alteracao com efeito no product nasce como estado administrativo versionado.",
|
||||
"O product consome apenas configuracao publicada e aprovada.",
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def build_admin_write_governance_source_payload() -> dict:
|
||||
return {
|
||||
"key": "admin_write_governance",
|
||||
"source": "runtime_guard",
|
||||
"mutable": False,
|
||||
"description": (
|
||||
"Guard no AdminSession bloqueia escrita ORM fora das tabelas internas do admin e preserva a governanca versionada antes de qualquer efeito no product."
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def _resolve_table_name(instance: object) -> str | None:
|
||||
table = getattr(instance, "__table__", None)
|
||||
if table is not None:
|
||||
table_name = getattr(table, "name", None)
|
||||
if table_name:
|
||||
return str(table_name).strip().lower()
|
||||
|
||||
class_table_name = getattr(type(instance), "__tablename__", None)
|
||||
if class_table_name:
|
||||
return str(class_table_name).strip().lower()
|
||||
|
||||
instance_table_name = getattr(instance, "__tablename__", None)
|
||||
if instance_table_name:
|
||||
return str(instance_table_name).strip().lower()
|
||||
|
||||
return None
|
||||
@ -0,0 +1,963 @@
|
||||
from admin_app.core.settings import AdminSettings
|
||||
from shared.contracts import (
|
||||
PRODUCT_OPERATIONAL_DATASETS,
|
||||
OperationalDatasetContract,
|
||||
OperationalReadGranularity,
|
||||
get_operational_dataset,
|
||||
)
|
||||
|
||||
_MATERIALIZATION_STATUS = "contract_defined_pending_snapshot_view"
|
||||
_REFRESH_BEHAVIOR = "manual_refresh_triggers_sync_boundary"
|
||||
_REPORT_SOURCE = "shared_contract_catalog"
|
||||
_SALES_DATASET_KEY = "sales_orders"
|
||||
|
||||
_SALES_REPORT_METRICS = {
|
||||
"total_orders": {"key": "total_orders", "label": "Pedidos totais", "aggregation": "count", "description": "Quantidade total de pedidos consolidados no periodo."},
|
||||
"gross_order_value": {"key": "gross_order_value", "label": "Valor bruto negociado", "aggregation": "sum", "description": "Soma do valor negociado dos pedidos incluidos no recorte."},
|
||||
"active_orders": {"key": "active_orders", "label": "Pedidos ativos", "aggregation": "count_where_status_active", "description": "Quantidade de pedidos ainda em fluxo operacional ativo."},
|
||||
"cancelled_orders": {"key": "cancelled_orders", "label": "Pedidos cancelados", "aggregation": "count_where_status_cancelled", "description": "Quantidade de pedidos cancelados no recorte selecionado."},
|
||||
"cancellation_rate": {"key": "cancellation_rate", "label": "Taxa de cancelamento", "aggregation": "ratio", "description": "Relacao entre pedidos cancelados e total de pedidos consolidados."},
|
||||
"average_ticket": {"key": "average_ticket", "label": "Ticket medio", "aggregation": "avg", "description": "Media do valor negociado por pedido dentro do recorte."},
|
||||
}
|
||||
|
||||
_SALES_DIMENSIONS = {
|
||||
"created_at": {"field_name": "created_at", "label": "Periodo de criacao", "description": "Agrupamento temporal da criacao do pedido.", "default_group_by": True},
|
||||
"updated_at": {"field_name": "updated_at", "label": "Periodo de atualizacao", "description": "Agrupamento temporal da ultima atualizacao do pedido.", "default_group_by": True},
|
||||
"data_cancelamento": {"field_name": "data_cancelamento", "label": "Periodo de cancelamento", "description": "Agrupamento temporal do cancelamento registrado.", "default_group_by": True},
|
||||
"status": {"field_name": "status", "label": "Status do pedido", "description": "Separa os pedidos por etapa operacional."},
|
||||
"modelo_veiculo": {"field_name": "modelo_veiculo", "label": "Modelo do veiculo", "description": "Recorte por modelo comercial negociado."},
|
||||
"motivo_cancelamento": {"field_name": "motivo_cancelamento", "label": "Motivo do cancelamento", "description": "Separa cancelamentos pelo motivo operacional registrado."},
|
||||
}
|
||||
|
||||
_SALES_FILTERS = {
|
||||
"created_at": {"field_name": "created_at", "label": "Periodo", "filter_type": "date_range", "description": "Intervalo de criacao do pedido consolidado.", "required": True},
|
||||
"updated_at": {"field_name": "updated_at", "label": "Periodo", "filter_type": "date_range", "description": "Intervalo da ultima atualizacao do pedido.", "required": True},
|
||||
"data_cancelamento": {"field_name": "data_cancelamento", "label": "Periodo", "filter_type": "date_range", "description": "Intervalo em que o cancelamento foi registrado.", "required": True},
|
||||
"status": {"field_name": "status", "label": "Status", "filter_type": "enum", "description": "Restringe o consolidado para um ou mais status operacionais."},
|
||||
"modelo_veiculo": {"field_name": "modelo_veiculo", "label": "Modelo do veiculo", "filter_type": "enum", "description": "Filtra pedidos por modelo comercial reservado."},
|
||||
"motivo_cancelamento": {"field_name": "motivo_cancelamento", "label": "Motivo do cancelamento", "filter_type": "enum", "description": "Restringe o consolidado para um ou mais motivos operacionais."},
|
||||
}
|
||||
|
||||
_SALES_REPORTS = (
|
||||
{"report_key": "orders_volume", "label": "Volume de pedidos", "description": "Acompanha o volume bruto de pedidos por periodo e status operacional.", "default_time_field": "created_at", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("total_orders", "gross_order_value"), "dimension_fields": ("created_at", "status", "modelo_veiculo"), "filter_fields": ("created_at", "status", "modelo_veiculo")},
|
||||
{"report_key": "active_vs_cancelled", "label": "Pedidos ativos e cancelados", "description": "Compara pedidos em andamento com pedidos cancelados para leitura operacional da conversao.", "default_time_field": "updated_at", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("active_orders", "cancelled_orders", "cancellation_rate"), "dimension_fields": ("updated_at", "status", "modelo_veiculo"), "filter_fields": ("updated_at", "status", "modelo_veiculo")},
|
||||
{"report_key": "average_ticket", "label": "Ticket medio", "description": "Consolida a evolucao do valor medio negociado por periodo e por modelo.", "default_time_field": "created_at", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("average_ticket", "gross_order_value", "total_orders"), "dimension_fields": ("created_at", "modelo_veiculo", "status"), "filter_fields": ("created_at", "status", "modelo_veiculo")},
|
||||
{"report_key": "cancellations_by_period", "label": "Cancelamentos por periodo", "description": "Organiza o volume de cancelamentos e seus motivos ao longo do tempo.", "default_time_field": "data_cancelamento", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("cancelled_orders", "cancellation_rate"), "dimension_fields": ("data_cancelamento", "motivo_cancelamento", "modelo_veiculo"), "filter_fields": ("data_cancelamento", "motivo_cancelamento", "modelo_veiculo")},
|
||||
)
|
||||
|
||||
_REVENUE_DATASET_KEY = "rental_payments"
|
||||
|
||||
_REVENUE_REPORT_METRICS = {
|
||||
"total_payments": {"key": "total_payments", "label": "Pagamentos totais", "aggregation": "count", "description": "Quantidade total de pagamentos liquidados no periodo."},
|
||||
"collected_amount": {"key": "collected_amount", "label": "Valor arrecadado", "aggregation": "sum", "description": "Soma do valor liquidado dos pagamentos incluidos no recorte."},
|
||||
"average_payment_amount": {"key": "average_payment_amount", "label": "Valor medio por pagamento", "aggregation": "avg", "description": "Media do valor liquidado por pagamento no recorte selecionado."},
|
||||
"distinct_contracts": {"key": "distinct_contracts", "label": "Contratos conciliados", "aggregation": "count_distinct", "description": "Quantidade de contratos distintos com pagamento consolidado no periodo."},
|
||||
}
|
||||
|
||||
_REVENUE_DIMENSIONS = {
|
||||
"data_pagamento": {"field_name": "data_pagamento", "label": "Periodo do pagamento", "description": "Agrupamento temporal do pagamento liquidado.", "default_group_by": True},
|
||||
"created_at": {"field_name": "created_at", "label": "Periodo de registro", "description": "Agrupamento temporal do registro do pagamento no read model administrativo.", "default_group_by": True},
|
||||
"contrato_numero": {"field_name": "contrato_numero", "label": "Contrato", "description": "Recorte por contrato associado ao pagamento."},
|
||||
"placa": {"field_name": "placa", "label": "Placa", "description": "Recorte por veiculo vinculado ao contrato pago."},
|
||||
"protocolo": {"field_name": "protocolo", "label": "Protocolo", "description": "Rastreio por protocolo publico do pagamento."},
|
||||
}
|
||||
|
||||
_REVENUE_FILTERS = {
|
||||
"data_pagamento": {"field_name": "data_pagamento", "label": "Periodo do pagamento", "filter_type": "date_range", "description": "Intervalo em que o pagamento foi liquidado.", "required": True},
|
||||
"created_at": {"field_name": "created_at", "label": "Periodo de registro", "filter_type": "date_range", "description": "Intervalo em que o pagamento foi registrado no dataset administrativo.", "required": True},
|
||||
"contrato_numero": {"field_name": "contrato_numero", "label": "Contrato", "filter_type": "exact_match", "description": "Filtra pagamentos por contrato associado."},
|
||||
"placa": {"field_name": "placa", "label": "Placa", "filter_type": "exact_match", "description": "Filtra pagamentos pela placa vinculada ao contrato."},
|
||||
"protocolo": {"field_name": "protocolo", "label": "Protocolo", "filter_type": "exact_match", "description": "Filtra o consolidado por protocolo publico do pagamento."},
|
||||
}
|
||||
|
||||
_REVENUE_REPORTS = (
|
||||
{"report_key": "payments_volume", "label": "Volume de pagamentos", "description": "Acompanha a quantidade de pagamentos liquidados por periodo, contrato e veiculo.", "default_time_field": "data_pagamento", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("total_payments", "distinct_contracts"), "dimension_fields": ("data_pagamento", "contrato_numero", "placa"), "filter_fields": ("data_pagamento", "contrato_numero", "placa")},
|
||||
{"report_key": "collected_amount", "label": "Arrecadacao por periodo", "description": "Consolida o valor arrecadado por periodo com apoio de contrato e placa para leitura operacional.", "default_time_field": "data_pagamento", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("collected_amount", "average_payment_amount", "total_payments"), "dimension_fields": ("data_pagamento", "contrato_numero", "placa"), "filter_fields": ("data_pagamento", "contrato_numero", "placa")},
|
||||
{"report_key": "contract_reconciliation", "label": "Pagamentos por contrato", "description": "Organiza pagamentos conciliados por contrato com rastreio por placa e protocolo publico.", "default_time_field": "data_pagamento", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("collected_amount", "total_payments"), "dimension_fields": ("contrato_numero", "placa", "protocolo"), "filter_fields": ("data_pagamento", "contrato_numero", "placa", "protocolo")},
|
||||
)
|
||||
|
||||
_RENTAL_FLEET_DATASET_KEY = "rental_fleet"
|
||||
_RENTAL_CONTRACTS_DATASET_KEY = "rental_contracts"
|
||||
|
||||
_RENTAL_REPORT_METRICS = {
|
||||
"total_fleet_vehicles": {"key": "total_fleet_vehicles", "label": "Veiculos da frota", "aggregation": "count", "description": "Quantidade total de veiculos consolidados na frota administrativa."},
|
||||
"available_fleet_vehicles": {"key": "available_fleet_vehicles", "label": "Veiculos disponiveis", "aggregation": "count_where_status_available", "description": "Quantidade de veiculos em status operacional disponivel para locacao."},
|
||||
"average_daily_rate": {"key": "average_daily_rate", "label": "Diaria media", "aggregation": "avg", "description": "Media do valor de diaria vigente dos veiculos incluidos no recorte."},
|
||||
"total_contracts": {"key": "total_contracts", "label": "Contratos totais", "aggregation": "count", "description": "Quantidade total de contratos consolidados no periodo selecionado."},
|
||||
"active_contracts": {"key": "active_contracts", "label": "Contratos ativos", "aggregation": "count_where_status_active", "description": "Quantidade de contratos ainda em curso no recorte operacional."},
|
||||
"closed_contracts": {"key": "closed_contracts", "label": "Contratos encerrados", "aggregation": "count_where_status_closed", "description": "Quantidade de contratos concluidos ou encerrados no recorte selecionado."},
|
||||
"overdue_contracts": {"key": "overdue_contracts", "label": "Devolucoes em atraso", "aggregation": "count_overdue", "description": "Quantidade de contratos com fim previsto vencido e sem devolucao consolidada."},
|
||||
"occupied_vehicles": {"key": "occupied_vehicles", "label": "Veiculos ocupados", "aggregation": "count_distinct_active_vehicles", "description": "Quantidade de veiculos distintos associados a contratos ativos no periodo."},
|
||||
"projected_revenue": {"key": "projected_revenue", "label": "Receita prevista", "aggregation": "sum", "description": "Soma do valor previsto dos contratos incluidos no recorte."},
|
||||
"final_revenue": {"key": "final_revenue", "label": "Receita final", "aggregation": "sum", "description": "Soma do valor final consolidado dos contratos no recorte selecionado."},
|
||||
"revenue_delta": {"key": "revenue_delta", "label": "Desvio entre previsto e final", "aggregation": "difference", "description": "Diferenca consolidada entre receita prevista e receita final dos contratos."},
|
||||
}
|
||||
|
||||
_RENTAL_DIMENSIONS = {
|
||||
"created_at": {"field_name": "created_at", "label": "Periodo de cadastro", "description": "Agrupamento temporal do cadastro no read model administrativo.", "default_group_by": True},
|
||||
"categoria": {"field_name": "categoria", "label": "Categoria", "description": "Recorte por categoria comercial da locacao."},
|
||||
"status": {"field_name": "status", "label": "Status", "description": "Separa frota ou contratos por status operacional."},
|
||||
"modelo": {"field_name": "modelo", "label": "Modelo", "description": "Recorte por modelo do veiculo de locacao."},
|
||||
"placa": {"field_name": "placa", "label": "Placa", "description": "Rastreio por placa do veiculo locado."},
|
||||
"data_inicio": {"field_name": "data_inicio", "label": "Inicio da locacao", "description": "Agrupamento temporal da abertura do contrato.", "default_group_by": True},
|
||||
"data_fim_prevista": {"field_name": "data_fim_prevista", "label": "Fim previsto", "description": "Agrupamento temporal do fim previsto da locacao.", "default_group_by": True},
|
||||
"data_devolucao": {"field_name": "data_devolucao", "label": "Data de devolucao", "description": "Agrupamento temporal da devolucao consolidada do contrato.", "default_group_by": True},
|
||||
"updated_at": {"field_name": "updated_at", "label": "Ultima atualizacao", "description": "Agrupamento temporal da ultima atualizacao do contrato.", "default_group_by": True},
|
||||
"modelo_veiculo": {"field_name": "modelo_veiculo", "label": "Modelo do veiculo", "description": "Recorte por modelo do veiculo vinculado ao contrato."},
|
||||
"contrato_numero": {"field_name": "contrato_numero", "label": "Contrato", "description": "Rastreio por numero publico do contrato."},
|
||||
}
|
||||
|
||||
_RENTAL_FILTERS = {
|
||||
"created_at": {"field_name": "created_at", "label": "Periodo de cadastro", "filter_type": "date_range", "description": "Intervalo de cadastro no dataset administrativo.", "required": True},
|
||||
"categoria": {"field_name": "categoria", "label": "Categoria", "filter_type": "enum", "description": "Filtra frota ou contratos por categoria comercial."},
|
||||
"status": {"field_name": "status", "label": "Status", "filter_type": "enum", "description": "Restringe o consolidado para um ou mais status operacionais."},
|
||||
"modelo": {"field_name": "modelo", "label": "Modelo", "filter_type": "enum", "description": "Filtra o consolidado por modelo da frota."},
|
||||
"placa": {"field_name": "placa", "label": "Placa", "filter_type": "exact_match", "description": "Filtra o consolidado pela placa do veiculo."},
|
||||
"data_inicio": {"field_name": "data_inicio", "label": "Inicio da locacao", "filter_type": "date_range", "description": "Intervalo de abertura dos contratos de locacao.", "required": True},
|
||||
"data_fim_prevista": {"field_name": "data_fim_prevista", "label": "Fim previsto", "filter_type": "date_range", "description": "Intervalo do fim previsto dos contratos de locacao.", "required": True},
|
||||
"updated_at": {"field_name": "updated_at", "label": "Ultima atualizacao", "filter_type": "date_range", "description": "Intervalo da ultima atualizacao operacional do contrato.", "required": True},
|
||||
"modelo_veiculo": {"field_name": "modelo_veiculo", "label": "Modelo do veiculo", "filter_type": "enum", "description": "Filtra contratos pelo modelo do veiculo locado."},
|
||||
"contrato_numero": {"field_name": "contrato_numero", "label": "Contrato", "filter_type": "exact_match", "description": "Filtra o consolidado por numero publico do contrato."},
|
||||
}
|
||||
|
||||
_RENTAL_REPORTS = (
|
||||
{"report_key": "fleet_availability", "label": "Disponibilidade da frota", "description": "Resume disponibilidade, status e diaria vigente da frota de locacao.", "dataset_key": _RENTAL_FLEET_DATASET_KEY, "default_time_field": "created_at", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("total_fleet_vehicles", "available_fleet_vehicles", "average_daily_rate"), "dimension_fields": ("created_at", "categoria", "status", "modelo"), "filter_fields": ("created_at", "categoria", "status", "modelo", "placa")},
|
||||
{"report_key": "contracts_lifecycle", "label": "Contratos ativos e encerrados", "description": "Organiza o ciclo operacional dos contratos de locacao entre abertos, ativos e encerrados.", "dataset_key": _RENTAL_CONTRACTS_DATASET_KEY, "default_time_field": "data_inicio", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("total_contracts", "active_contracts", "closed_contracts"), "dimension_fields": ("data_inicio", "categoria", "status", "modelo_veiculo"), "filter_fields": ("data_inicio", "categoria", "status", "placa", "contrato_numero")},
|
||||
{"report_key": "overdue_returns", "label": "Devolucoes em atraso", "description": "Acompanha contratos com fim previsto vencido e sem devolucao consolidada.", "dataset_key": _RENTAL_CONTRACTS_DATASET_KEY, "default_time_field": "data_fim_prevista", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("overdue_contracts", "active_contracts"), "dimension_fields": ("data_fim_prevista", "categoria", "status", "placa"), "filter_fields": ("data_fim_prevista", "categoria", "status", "placa", "contrato_numero")},
|
||||
{"report_key": "fleet_occupancy", "label": "Ocupacao da frota", "description": "Consolida o uso da frota por contratos ativos ao longo do tempo e por categoria.", "dataset_key": _RENTAL_CONTRACTS_DATASET_KEY, "default_time_field": "data_inicio", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("occupied_vehicles", "active_contracts", "projected_revenue"), "dimension_fields": ("data_inicio", "categoria", "modelo_veiculo", "status"), "filter_fields": ("data_inicio", "categoria", "status", "modelo_veiculo", "placa")},
|
||||
{"report_key": "projected_vs_final_revenue", "label": "Receita prevista versus final", "description": "Compara o valor previsto na abertura do contrato com o valor final consolidado da locacao.", "dataset_key": _RENTAL_CONTRACTS_DATASET_KEY, "default_time_field": "updated_at", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("projected_revenue", "final_revenue", "revenue_delta"), "dimension_fields": ("updated_at", "categoria", "status", "modelo_veiculo"), "filter_fields": ("updated_at", "categoria", "status", "placa", "contrato_numero")},
|
||||
)
|
||||
|
||||
_BOT_FLOW_DATASET_KEY = "conversation_turns"
|
||||
|
||||
_BOT_FLOW_REPORT_METRICS = {
|
||||
"total_turns": {"key": "total_turns", "label": "Turnos totais", "aggregation": "count", "description": "Quantidade total de turnos processados no recorte operacional."},
|
||||
"completed_turns": {"key": "completed_turns", "label": "Turnos concluidos", "aggregation": "count_where_status_completed", "description": "Quantidade de turnos concluidos pelo fluxo operacional do bot."},
|
||||
"errored_turns": {"key": "errored_turns", "label": "Turnos com falha", "aggregation": "count_where_status_error", "description": "Quantidade de turnos com falha operacional no processamento."},
|
||||
"tool_routed_turns": {"key": "tool_routed_turns", "label": "Turnos com tool", "aggregation": "count_where_tool_called", "description": "Quantidade de turnos que acionaram pelo menos uma tool no fluxo."},
|
||||
"fallback_turns": {"key": "fallback_turns", "label": "Turnos em fallback", "aggregation": "count_where_action_fallback", "description": "Quantidade de turnos encaminhados para fallback funcional do bot."},
|
||||
"handoff_turns": {"key": "handoff_turns", "label": "Turnos em handoff", "aggregation": "count_where_action_handoff", "description": "Quantidade de turnos que escalaram para handoff humano."},
|
||||
}
|
||||
|
||||
_BOT_FLOW_DIMENSIONS = {
|
||||
"started_at": {"field_name": "started_at", "label": "Inicio do turno", "description": "Agrupamento temporal do inicio do processamento do turno.", "default_group_by": True},
|
||||
"completed_at": {"field_name": "completed_at", "label": "Fim do turno", "description": "Agrupamento temporal da finalizacao do turno processado.", "default_group_by": True},
|
||||
"channel": {"field_name": "channel", "label": "Canal", "description": "Recorte por canal operacional do atendimento."},
|
||||
"turn_status": {"field_name": "turn_status", "label": "Status do turno", "description": "Separa o fluxo pelos estados operacionais do turno."},
|
||||
"action": {"field_name": "action", "label": "Acao do fluxo", "description": "Recorte pela acao tomada pelo orquestrador durante o turno."},
|
||||
"tool_name": {"field_name": "tool_name", "label": "Tool acionada", "description": "Rastreio da tool utilizada durante o turno do bot."},
|
||||
"domain": {"field_name": "domain", "label": "Dominio operacional", "description": "Recorte pelo dominio operacional associado ao turno."},
|
||||
"intent": {"field_name": "intent", "label": "Intencao", "description": "Recorte pela intencao classificada para o turno."},
|
||||
}
|
||||
|
||||
_BOT_FLOW_FILTERS = {
|
||||
"started_at": {"field_name": "started_at", "label": "Inicio do turno", "filter_type": "date_range", "description": "Intervalo de inicio do processamento do turno.", "required": True},
|
||||
"completed_at": {"field_name": "completed_at", "label": "Fim do turno", "filter_type": "date_range", "description": "Intervalo de finalizacao do turno processado."},
|
||||
"channel": {"field_name": "channel", "label": "Canal", "filter_type": "enum", "description": "Filtra o fluxo por canal operacional."},
|
||||
"turn_status": {"field_name": "turn_status", "label": "Status do turno", "filter_type": "enum", "description": "Restringe o consolidado para um ou mais status do turno."},
|
||||
"action": {"field_name": "action", "label": "Acao do fluxo", "filter_type": "enum", "description": "Restringe o consolidado para uma ou mais acoes do fluxo do bot."},
|
||||
"tool_name": {"field_name": "tool_name", "label": "Tool acionada", "filter_type": "enum", "description": "Filtra os turnos pela tool utilizada no atendimento."},
|
||||
"domain": {"field_name": "domain", "label": "Dominio operacional", "filter_type": "enum", "description": "Filtra o fluxo pelo dominio operacional associado ao turno."},
|
||||
"intent": {"field_name": "intent", "label": "Intencao", "filter_type": "enum", "description": "Filtra o consolidado pela intencao classificada para o turno."},
|
||||
}
|
||||
|
||||
_BOT_FLOW_REPORTS = (
|
||||
{"report_key": "turn_status_overview", "label": "Status dos turnos", "description": "Acompanha o andamento operacional dos turnos por status, canal e dominio.", "default_time_field": "started_at", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("total_turns", "completed_turns", "errored_turns"), "dimension_fields": ("started_at", "turn_status", "channel", "domain"), "filter_fields": ("started_at", "turn_status", "channel", "domain")},
|
||||
{"report_key": "action_routing_flow", "label": "Roteamento do fluxo", "description": "Organiza as acoes do orquestrador entre resposta, fallback, handoff e outros caminhos operacionais.", "default_time_field": "started_at", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("total_turns", "fallback_turns", "handoff_turns"), "dimension_fields": ("started_at", "action", "channel", "domain"), "filter_fields": ("started_at", "action", "channel", "domain", "intent")},
|
||||
{"report_key": "tool_activation_flow", "label": "Uso operacional de tools", "description": "Mostra quais turnos acionaram tools e como isso se distribui no fluxo do bot.", "default_time_field": "started_at", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("tool_routed_turns", "completed_turns", "errored_turns"), "dimension_fields": ("started_at", "tool_name", "action", "domain"), "filter_fields": ("started_at", "tool_name", "action", "domain", "intent")},
|
||||
{"report_key": "fallback_and_handoff", "label": "Fallback e handoff", "description": "Destaca turnos que saem do fluxo padrao para fallback funcional ou handoff humano.", "default_time_field": "started_at", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("fallback_turns", "handoff_turns", "errored_turns"), "dimension_fields": ("started_at", "action", "channel", "intent"), "filter_fields": ("started_at", "action", "channel", "intent", "domain")},
|
||||
{"report_key": "operational_failures", "label": "Falhas operacionais do fluxo", "description": "Ajuda a triar turnos com falha por status, acao e canal operacional.", "default_time_field": "started_at", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("errored_turns", "total_turns", "tool_routed_turns"), "dimension_fields": ("started_at", "turn_status", "action", "channel"), "filter_fields": ("started_at", "turn_status", "action", "channel", "tool_name")},
|
||||
)
|
||||
|
||||
_CONVERSATION_TELEMETRY_DATASET_KEY = "conversation_turns"
|
||||
|
||||
_CONVERSATION_TELEMETRY_REPORT_METRICS = {
|
||||
"total_turns": {"key": "total_turns", "label": "Turnos totais", "aggregation": "count", "description": "Quantidade total de turnos observados no recorte de telemetria."},
|
||||
"distinct_conversations": {"key": "distinct_conversations", "label": "Conversas distintas", "aggregation": "count_distinct", "description": "Quantidade de conversas distintas observadas no recorte selecionado."},
|
||||
"average_latency_ms": {"key": "average_latency_ms", "label": "Latencia media", "aggregation": "avg", "description": "Media do tempo de processamento do turno em milissegundos."},
|
||||
"p95_latency_ms": {"key": "p95_latency_ms", "label": "Latencia p95", "aggregation": "percentile_p95", "description": "Percentil 95 do tempo de processamento dos turnos observados."},
|
||||
"tool_routed_turns": {"key": "tool_routed_turns", "label": "Turnos com tool", "aggregation": "count_where_tool_called", "description": "Quantidade de turnos que acionaram pelo menos uma tool no atendimento."},
|
||||
"errored_turns": {"key": "errored_turns", "label": "Turnos com falha", "aggregation": "count_where_status_error", "description": "Quantidade de turnos com falha no recorte de telemetria."},
|
||||
}
|
||||
|
||||
_CONVERSATION_TELEMETRY_DIMENSIONS = {
|
||||
"started_at": {"field_name": "started_at", "label": "Inicio do turno", "description": "Agrupamento temporal do inicio do processamento do turno.", "default_group_by": True},
|
||||
"completed_at": {"field_name": "completed_at", "label": "Fim do turno", "description": "Agrupamento temporal da finalizacao do turno.", "default_group_by": True},
|
||||
"channel": {"field_name": "channel", "label": "Canal", "description": "Recorte por canal operacional do atendimento."},
|
||||
"domain": {"field_name": "domain", "label": "Dominio operacional", "description": "Recorte pelo dominio operacional associado ao turno."},
|
||||
"intent": {"field_name": "intent", "label": "Intencao", "description": "Recorte pela intencao classificada para o turno."},
|
||||
"tool_name": {"field_name": "tool_name", "label": "Tool acionada", "description": "Rastreio da tool utilizada durante o turno."},
|
||||
"turn_status": {"field_name": "turn_status", "label": "Status do turno", "description": "Separa a telemetria pelos estados observados do turno."},
|
||||
"action": {"field_name": "action", "label": "Acao do turno", "description": "Recorte pela acao operacional tomada pelo orquestrador."},
|
||||
}
|
||||
|
||||
_CONVERSATION_TELEMETRY_FILTERS = {
|
||||
"started_at": {"field_name": "started_at", "label": "Inicio do turno", "filter_type": "date_range", "description": "Intervalo de inicio do processamento do turno.", "required": True},
|
||||
"completed_at": {"field_name": "completed_at", "label": "Fim do turno", "filter_type": "date_range", "description": "Intervalo de finalizacao do turno processado."},
|
||||
"channel": {"field_name": "channel", "label": "Canal", "filter_type": "enum", "description": "Filtra a telemetria por canal operacional."},
|
||||
"domain": {"field_name": "domain", "label": "Dominio operacional", "filter_type": "enum", "description": "Filtra a telemetria pelo dominio associado ao turno."},
|
||||
"intent": {"field_name": "intent", "label": "Intencao", "filter_type": "enum", "description": "Filtra o recorte pela intencao classificada."},
|
||||
"tool_name": {"field_name": "tool_name", "label": "Tool acionada", "filter_type": "enum", "description": "Filtra os turnos pela tool utilizada durante o atendimento."},
|
||||
"turn_status": {"field_name": "turn_status", "label": "Status do turno", "filter_type": "enum", "description": "Restringe o consolidado para um ou mais status observados."},
|
||||
"action": {"field_name": "action", "label": "Acao do turno", "filter_type": "enum", "description": "Restringe o consolidado para uma ou mais acoes do orquestrador."},
|
||||
}
|
||||
|
||||
_CONVERSATION_TELEMETRY_REPORTS = (
|
||||
{"report_key": "conversation_volume", "label": "Volume de atendimento", "description": "Consolida o volume de turnos e conversas por periodo, canal e dominio.", "default_time_field": "started_at", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("total_turns", "distinct_conversations"), "dimension_fields": ("started_at", "channel", "domain", "intent"), "filter_fields": ("started_at", "channel", "domain", "intent")},
|
||||
{"report_key": "latency_profile", "label": "Perfil de latencia", "description": "Organiza sinais de latencia media e p95 por canal, dominio e intencao.", "default_time_field": "completed_at", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("average_latency_ms", "p95_latency_ms", "total_turns"), "dimension_fields": ("completed_at", "channel", "domain", "intent"), "filter_fields": ("started_at", "completed_at", "channel", "domain", "intent")},
|
||||
{"report_key": "domain_distribution", "label": "Distribuicao por dominio", "description": "Mostra como o atendimento se distribui entre dominios, intencoes e canais.", "default_time_field": "started_at", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("total_turns", "distinct_conversations"), "dimension_fields": ("started_at", "domain", "intent", "channel"), "filter_fields": ("started_at", "domain", "intent", "channel")},
|
||||
{"report_key": "tool_usage_telemetry", "label": "Uso de tools", "description": "Expoe quais tools aparecem com mais frequencia no atendimento e em quais contextos.", "default_time_field": "started_at", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("tool_routed_turns", "total_turns", "errored_turns"), "dimension_fields": ("started_at", "tool_name", "domain", "channel"), "filter_fields": ("started_at", "tool_name", "domain", "channel", "intent")},
|
||||
{"report_key": "turn_health_status", "label": "Saude por status", "description": "Acompanha estados de saude do atendimento por status observado e acao tomada.", "default_time_field": "started_at", "default_granularity": OperationalReadGranularity.AGGREGATE, "metric_keys": ("errored_turns", "total_turns", "average_latency_ms"), "dimension_fields": ("started_at", "turn_status", "action", "channel"), "filter_fields": ("started_at", "turn_status", "action", "channel", "domain")},
|
||||
)
|
||||
|
||||
_REPORT_FAMILIES = (
|
||||
{
|
||||
"key": "sales",
|
||||
"label": "Vendas",
|
||||
"description": "Pedidos, conversao comercial e cancelamentos usados pela operacao interna.",
|
||||
"dataset_keys": ["sales_orders"],
|
||||
},
|
||||
{
|
||||
"key": "arrecadacao",
|
||||
"label": "Arrecadacao",
|
||||
"description": "Recebimentos de locacao e conciliacao operacional do faturamento.",
|
||||
"dataset_keys": ["rental_payments"],
|
||||
},
|
||||
{
|
||||
"key": "operacao",
|
||||
"label": "Operacao",
|
||||
"description": "Estoque, revisoes, frota e contratos que suportam o acompanhamento do dia a dia.",
|
||||
"dataset_keys": [
|
||||
"vehicle_inventory",
|
||||
"review_schedules",
|
||||
"rental_fleet",
|
||||
"rental_contracts",
|
||||
],
|
||||
},
|
||||
{
|
||||
"key": "telemetria_atendimento",
|
||||
"label": "Telemetria de atendimento",
|
||||
"description": "Turnos conversacionais, uso de tools e sinais de eficiencia do bot.",
|
||||
"dataset_keys": ["conversation_turns"],
|
||||
},
|
||||
{
|
||||
"key": "integration_deliveries",
|
||||
"label": "Entregas de integracao",
|
||||
"description": "Rastreio operacional das entregas para provedores e falhas de despacho.",
|
||||
"dataset_keys": ["integration_deliveries"],
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
class ReportService:
|
||||
def __init__(self, settings: AdminSettings):
|
||||
self.settings = settings
|
||||
|
||||
def build_overview_payload(self) -> dict:
|
||||
datasets = PRODUCT_OPERATIONAL_DATASETS
|
||||
near_real_time_count = sum(1 for dataset in datasets if dataset.freshness_target.value == "near_real_time")
|
||||
intra_hour_count = sum(1 for dataset in datasets if dataset.freshness_target.value == "intra_hour")
|
||||
return {
|
||||
"mode": "shared_contract_bootstrap",
|
||||
"metrics": [
|
||||
{
|
||||
"key": "datasets",
|
||||
"label": "Datasets liberados",
|
||||
"value": str(len(datasets)),
|
||||
"description": "Datasets operacionais explicitamente liberados para relatorios administrativos.",
|
||||
},
|
||||
{
|
||||
"key": "domains",
|
||||
"label": "Dominios operacionais",
|
||||
"value": str(len({dataset.domain for dataset in datasets})),
|
||||
"description": "Dominios cobertos pelo catalogo inicial de leitura administrativa.",
|
||||
},
|
||||
{
|
||||
"key": "near_real_time_targets",
|
||||
"label": "Metas near real time",
|
||||
"value": str(near_real_time_count),
|
||||
"description": "Datasets cuja UX espera consolidacao mais frequente sem leitura live do produto.",
|
||||
},
|
||||
{
|
||||
"key": "intra_hour_targets",
|
||||
"label": "Metas intra-hour",
|
||||
"value": str(intra_hour_count),
|
||||
"description": "Datasets de apoio operacional e telemetria servidos por consolidacao eventual intra-horaria.",
|
||||
},
|
||||
],
|
||||
"materialization": self._build_materialization_payload(),
|
||||
"report_families": list(_REPORT_FAMILIES),
|
||||
"next_steps": [
|
||||
"Criar snapshots sanitizados no admin para vendas, arrecadacao e operacao.",
|
||||
"Servir views dedicadas por caso de uso em vez de espelhar o schema operacional do produto.",
|
||||
"Exibir carimbo de atualizacao e watermark quando a camada de sincronizacao entrar em producao.",
|
||||
],
|
||||
}
|
||||
|
||||
def list_datasets_payload(self) -> dict:
|
||||
return {
|
||||
"source": _REPORT_SOURCE,
|
||||
"materialization": self._build_materialization_payload(),
|
||||
"datasets": [
|
||||
self._serialize_dataset_summary(dataset)
|
||||
for dataset in sorted(
|
||||
PRODUCT_OPERATIONAL_DATASETS,
|
||||
key=lambda item: (item.domain.value, item.dataset_key),
|
||||
)
|
||||
],
|
||||
}
|
||||
|
||||
def get_dataset_payload(self, dataset_key: str) -> dict | None:
|
||||
dataset = get_operational_dataset(dataset_key)
|
||||
if dataset is None:
|
||||
return None
|
||||
|
||||
return {
|
||||
"source": _REPORT_SOURCE,
|
||||
"materialization": self._build_materialization_payload(dataset),
|
||||
"dataset": self._serialize_dataset_detail(dataset),
|
||||
}
|
||||
|
||||
def build_sales_overview_payload(self) -> dict:
|
||||
dataset = self._get_sales_dataset()
|
||||
return {
|
||||
"domain": dataset.domain,
|
||||
"mode": "sales_contract_bootstrap",
|
||||
"source_dataset_keys": [dataset.dataset_key],
|
||||
"metrics": [
|
||||
{
|
||||
"key": "source_datasets",
|
||||
"label": "Datasets fonte",
|
||||
"value": "1",
|
||||
"description": "A estrutura inicial de vendas nasce apoiada em um dataset sanitizado de pedidos.",
|
||||
},
|
||||
{
|
||||
"key": "initial_reports",
|
||||
"label": "Relatorios iniciais",
|
||||
"value": str(len(_SALES_REPORTS)),
|
||||
"description": "Casos de uso de vendas previstos para a primeira superficie administrativa do dominio.",
|
||||
},
|
||||
{
|
||||
"key": "allowed_fields",
|
||||
"label": "Campos liberados",
|
||||
"value": str(len(dataset.allowed_fields)),
|
||||
"description": "Campos operacionais expostos para agregacao e filtros de vendas.",
|
||||
},
|
||||
{
|
||||
"key": "blocked_fields",
|
||||
"label": "Campos bloqueados",
|
||||
"value": str(len(dataset.blocked_fields)),
|
||||
"description": "Campos sensiveis que permanecem fora do read model administrativo.",
|
||||
},
|
||||
{
|
||||
"key": "freshness_target",
|
||||
"label": "Meta de frescor",
|
||||
"value": dataset.freshness_target.value,
|
||||
"description": "Objetivo inicial de consolidacao para a UX dos relatorios de vendas.",
|
||||
},
|
||||
],
|
||||
"materialization": self._build_materialization_payload(dataset),
|
||||
"reports": [self._serialize_sales_report_summary(report) for report in _SALES_REPORTS],
|
||||
"next_steps": [
|
||||
"Materializar snapshot sanitizado de sales_orders no banco administrativo.",
|
||||
"Criar dedicated views separadas para volume, ticket medio e cancelamentos.",
|
||||
"Exibir watermark e timestamp da ultima consolidacao quando o ETL incremental entrar em producao.",
|
||||
],
|
||||
}
|
||||
|
||||
def list_sales_reports_payload(self) -> dict:
|
||||
dataset = self._get_sales_dataset()
|
||||
return {
|
||||
"domain": dataset.domain,
|
||||
"source": _REPORT_SOURCE,
|
||||
"materialization": self._build_materialization_payload(dataset),
|
||||
"reports": [self._serialize_sales_report_summary(report) for report in _SALES_REPORTS],
|
||||
}
|
||||
|
||||
def get_sales_report_payload(self, report_key: str) -> dict | None:
|
||||
normalized_report_key = self._normalize_key(report_key)
|
||||
dataset = self._get_sales_dataset()
|
||||
for report in _SALES_REPORTS:
|
||||
if report["report_key"] == normalized_report_key:
|
||||
return {
|
||||
"domain": dataset.domain,
|
||||
"source": _REPORT_SOURCE,
|
||||
"materialization": self._build_materialization_payload(dataset),
|
||||
"report": self._serialize_sales_report_detail(report, dataset),
|
||||
}
|
||||
return None
|
||||
|
||||
def build_revenue_overview_payload(self) -> dict:
|
||||
dataset = self._get_revenue_dataset()
|
||||
return {
|
||||
"area": "arrecadacao",
|
||||
"source_domain": dataset.domain,
|
||||
"mode": "revenue_contract_bootstrap",
|
||||
"source_dataset_keys": [dataset.dataset_key],
|
||||
"metrics": [
|
||||
{
|
||||
"key": "source_datasets",
|
||||
"label": "Datasets fonte",
|
||||
"value": "1",
|
||||
"description": "A estrutura inicial de arrecadacao nasce apoiada em um dataset sanitizado de pagamentos.",
|
||||
},
|
||||
{
|
||||
"key": "initial_reports",
|
||||
"label": "Relatorios iniciais",
|
||||
"value": str(len(_REVENUE_REPORTS)),
|
||||
"description": "Casos de uso iniciais de arrecadacao previstos para a primeira superficie administrativa.",
|
||||
},
|
||||
{
|
||||
"key": "allowed_fields",
|
||||
"label": "Campos liberados",
|
||||
"value": str(len(dataset.allowed_fields)),
|
||||
"description": "Campos operacionais expostos para agregacao e conciliacao de pagamentos.",
|
||||
},
|
||||
{
|
||||
"key": "blocked_fields",
|
||||
"label": "Campos bloqueados",
|
||||
"value": str(len(dataset.blocked_fields)),
|
||||
"description": "Campos sensiveis que permanecem fora do read model administrativo.",
|
||||
},
|
||||
{
|
||||
"key": "freshness_target",
|
||||
"label": "Meta de frescor",
|
||||
"value": dataset.freshness_target.value,
|
||||
"description": "Objetivo inicial de consolidacao para a UX dos relatorios de arrecadacao.",
|
||||
},
|
||||
],
|
||||
"materialization": self._build_materialization_payload(dataset),
|
||||
"reports": [self._serialize_revenue_report_summary(report) for report in _REVENUE_REPORTS],
|
||||
"next_steps": [
|
||||
"Materializar snapshot sanitizado de rental_payments no banco administrativo.",
|
||||
"Criar dedicated views separadas para arrecadacao por periodo e conciliacao por contrato.",
|
||||
"Cruzar contratos e pagamentos em uma etapa futura para abrir inadimplencia operacional sem leitura live do produto.",
|
||||
],
|
||||
}
|
||||
|
||||
def list_revenue_reports_payload(self) -> dict:
|
||||
dataset = self._get_revenue_dataset()
|
||||
return {
|
||||
"area": "arrecadacao",
|
||||
"source_domain": dataset.domain,
|
||||
"source": _REPORT_SOURCE,
|
||||
"materialization": self._build_materialization_payload(dataset),
|
||||
"reports": [self._serialize_revenue_report_summary(report) for report in _REVENUE_REPORTS],
|
||||
}
|
||||
|
||||
def get_revenue_report_payload(self, report_key: str) -> dict | None:
|
||||
normalized_report_key = self._normalize_key(report_key)
|
||||
dataset = self._get_revenue_dataset()
|
||||
for report in _REVENUE_REPORTS:
|
||||
if report["report_key"] == normalized_report_key:
|
||||
return {
|
||||
"area": "arrecadacao",
|
||||
"source_domain": dataset.domain,
|
||||
"source": _REPORT_SOURCE,
|
||||
"materialization": self._build_materialization_payload(dataset),
|
||||
"report": self._serialize_revenue_report_detail(report, dataset),
|
||||
}
|
||||
return None
|
||||
|
||||
def build_rental_overview_payload(self) -> dict:
|
||||
fleet_dataset = self._get_rental_fleet_dataset()
|
||||
contracts_dataset = self._get_rental_contracts_dataset()
|
||||
return {
|
||||
"area": "locacao",
|
||||
"source_domain": contracts_dataset.domain,
|
||||
"mode": "rental_contract_bootstrap",
|
||||
"source_dataset_keys": [fleet_dataset.dataset_key, contracts_dataset.dataset_key],
|
||||
"metrics": [
|
||||
{
|
||||
"key": "source_datasets",
|
||||
"label": "Datasets fonte",
|
||||
"value": "2",
|
||||
"description": "A estrutura inicial de locacao nasce sobre snapshots sanitizados de frota e contratos.",
|
||||
},
|
||||
{
|
||||
"key": "initial_reports",
|
||||
"label": "Relatorios iniciais",
|
||||
"value": str(len(_RENTAL_REPORTS)),
|
||||
"description": "Casos de uso operacionais de locacao previstos para a primeira superficie administrativa.",
|
||||
},
|
||||
{
|
||||
"key": "fleet_allowed_fields",
|
||||
"label": "Campos liberados da frota",
|
||||
"value": str(len(fleet_dataset.allowed_fields)),
|
||||
"description": "Campos expostos para disponibilidade, categoria e diaria vigente da frota.",
|
||||
},
|
||||
{
|
||||
"key": "contracts_allowed_fields",
|
||||
"label": "Campos liberados dos contratos",
|
||||
"value": str(len(contracts_dataset.allowed_fields)),
|
||||
"description": "Campos expostos para ciclo do contrato, ocupacao e devolucao operacional.",
|
||||
},
|
||||
{
|
||||
"key": "freshness_target",
|
||||
"label": "Meta de frescor",
|
||||
"value": contracts_dataset.freshness_target.value,
|
||||
"description": "Objetivo inicial de consolidacao para a UX dos relatorios de locacao.",
|
||||
},
|
||||
],
|
||||
"materialization": self._build_materialization_payload(contracts_dataset),
|
||||
"reports": [self._serialize_rental_report_summary(report) for report in _RENTAL_REPORTS],
|
||||
"next_steps": [
|
||||
"Materializar snapshots sanitizados de rental_fleet e rental_contracts no banco administrativo.",
|
||||
"Criar dedicated views separadas para disponibilidade da frota, contratos em curso e devolucoes em atraso.",
|
||||
"Combinar frota e contratos em uma camada futura de ocupacao sem consultar tabelas live do produto.",
|
||||
],
|
||||
}
|
||||
|
||||
def list_rental_reports_payload(self) -> dict:
|
||||
contracts_dataset = self._get_rental_contracts_dataset()
|
||||
return {
|
||||
"area": "locacao",
|
||||
"source_domain": contracts_dataset.domain,
|
||||
"source": _REPORT_SOURCE,
|
||||
"materialization": self._build_materialization_payload(contracts_dataset),
|
||||
"reports": [self._serialize_rental_report_summary(report) for report in _RENTAL_REPORTS],
|
||||
}
|
||||
|
||||
def get_rental_report_payload(self, report_key: str) -> dict | None:
|
||||
normalized_report_key = self._normalize_key(report_key)
|
||||
for report in _RENTAL_REPORTS:
|
||||
if report["report_key"] == normalized_report_key:
|
||||
dataset = self._get_rental_dataset(report["dataset_key"])
|
||||
return {
|
||||
"area": "locacao",
|
||||
"source_domain": dataset.domain,
|
||||
"source": _REPORT_SOURCE,
|
||||
"materialization": self._build_materialization_payload(dataset),
|
||||
"report": self._serialize_rental_report_detail(report, dataset),
|
||||
}
|
||||
return None
|
||||
|
||||
def build_bot_flow_overview_payload(self) -> dict:
|
||||
dataset = self._get_bot_flow_dataset()
|
||||
return {
|
||||
"area": "fluxo_bot",
|
||||
"source_domain": dataset.domain,
|
||||
"mode": "bot_flow_contract_bootstrap",
|
||||
"source_dataset_keys": [dataset.dataset_key],
|
||||
"metrics": [
|
||||
{
|
||||
"key": "source_datasets",
|
||||
"label": "Datasets fonte",
|
||||
"value": "1",
|
||||
"description": "A estrutura inicial do fluxo do bot nasce apoiada em um dataset sanitizado de turnos operacionais.",
|
||||
},
|
||||
{
|
||||
"key": "initial_reports",
|
||||
"label": "Relatorios iniciais",
|
||||
"value": str(len(_BOT_FLOW_REPORTS)),
|
||||
"description": "Casos de uso de operacao do fluxo do bot previstos para a primeira superficie administrativa.",
|
||||
},
|
||||
{
|
||||
"key": "allowed_fields",
|
||||
"label": "Campos liberados",
|
||||
"value": str(len(dataset.allowed_fields)),
|
||||
"description": "Campos operacionais expostos para triagem de status, acao e uso de tools.",
|
||||
},
|
||||
{
|
||||
"key": "blocked_fields",
|
||||
"label": "Campos bloqueados",
|
||||
"value": str(len(dataset.blocked_fields)),
|
||||
"description": "Campos sensiveis e mensagens livres que permanecem fora da operacao administrativa.",
|
||||
},
|
||||
{
|
||||
"key": "freshness_target",
|
||||
"label": "Meta de frescor",
|
||||
"value": dataset.freshness_target.value,
|
||||
"description": "Objetivo inicial de consolidacao para a UX dos relatorios operacionais do fluxo do bot.",
|
||||
},
|
||||
],
|
||||
"materialization": self._build_materialization_payload(dataset),
|
||||
"reports": [self._serialize_bot_flow_report_summary(report) for report in _BOT_FLOW_REPORTS],
|
||||
"next_steps": [
|
||||
"Materializar snapshot sanitizado de conversation_turns no banco administrativo.",
|
||||
"Criar dedicated views separadas para status do turno, roteamento operacional e falhas do fluxo.",
|
||||
"Reservar latencia e eficiencia detalhada para a etapa seguinte de telemetria conversacional.",
|
||||
],
|
||||
}
|
||||
|
||||
def list_bot_flow_reports_payload(self) -> dict:
|
||||
dataset = self._get_bot_flow_dataset()
|
||||
return {
|
||||
"area": "fluxo_bot",
|
||||
"source_domain": dataset.domain,
|
||||
"source": _REPORT_SOURCE,
|
||||
"materialization": self._build_materialization_payload(dataset),
|
||||
"reports": [self._serialize_bot_flow_report_summary(report) for report in _BOT_FLOW_REPORTS],
|
||||
}
|
||||
|
||||
def get_bot_flow_report_payload(self, report_key: str) -> dict | None:
|
||||
normalized_report_key = self._normalize_key(report_key)
|
||||
dataset = self._get_bot_flow_dataset()
|
||||
for report in _BOT_FLOW_REPORTS:
|
||||
if report["report_key"] == normalized_report_key:
|
||||
return {
|
||||
"area": "fluxo_bot",
|
||||
"source_domain": dataset.domain,
|
||||
"source": _REPORT_SOURCE,
|
||||
"materialization": self._build_materialization_payload(dataset),
|
||||
"report": self._serialize_bot_flow_report_detail(report, dataset),
|
||||
}
|
||||
return None
|
||||
|
||||
def build_conversation_telemetry_overview_payload(self) -> dict:
|
||||
dataset = self._get_conversation_telemetry_dataset()
|
||||
return {
|
||||
"area": "telemetria_conversacional",
|
||||
"source_domain": dataset.domain,
|
||||
"mode": "conversation_telemetry_contract_bootstrap",
|
||||
"source_dataset_keys": [dataset.dataset_key],
|
||||
"metrics": [
|
||||
{
|
||||
"key": "source_datasets",
|
||||
"label": "Datasets fonte",
|
||||
"value": "1",
|
||||
"description": "A estrutura inicial de telemetria conversacional nasce apoiada em um dataset sanitizado de turnos.",
|
||||
},
|
||||
{
|
||||
"key": "initial_reports",
|
||||
"label": "Relatorios iniciais",
|
||||
"value": str(len(_CONVERSATION_TELEMETRY_REPORTS)),
|
||||
"description": "Casos de uso iniciais de observabilidade conversacional previstos para a primeira superficie administrativa.",
|
||||
},
|
||||
{
|
||||
"key": "allowed_fields",
|
||||
"label": "Campos liberados",
|
||||
"value": str(len(dataset.allowed_fields)),
|
||||
"description": "Campos expostos para volume, latencia, distribuicao por dominio e uso de tools.",
|
||||
},
|
||||
{
|
||||
"key": "blocked_fields",
|
||||
"label": "Campos bloqueados",
|
||||
"value": str(len(dataset.blocked_fields)),
|
||||
"description": "Campos sensiveis e texto livre que permanecem fora da telemetria administrativa.",
|
||||
},
|
||||
{
|
||||
"key": "freshness_target",
|
||||
"label": "Meta de frescor",
|
||||
"value": dataset.freshness_target.value,
|
||||
"description": "Objetivo inicial de consolidacao para a UX dos relatorios de telemetria conversacional.",
|
||||
},
|
||||
],
|
||||
"materialization": self._build_materialization_payload(dataset),
|
||||
"reports": [self._serialize_conversation_telemetry_report_summary(report) for report in _CONVERSATION_TELEMETRY_REPORTS],
|
||||
"next_steps": [
|
||||
"Materializar snapshot sanitizado de conversation_turns no banco administrativo.",
|
||||
"Criar dedicated views separadas para volume, latencia e distribuicao por dominio do atendimento.",
|
||||
"Preparar buckets e watermark de consolidacao para comparativos historicos da telemetria.",
|
||||
],
|
||||
}
|
||||
|
||||
def list_conversation_telemetry_reports_payload(self) -> dict:
|
||||
dataset = self._get_conversation_telemetry_dataset()
|
||||
return {
|
||||
"area": "telemetria_conversacional",
|
||||
"source_domain": dataset.domain,
|
||||
"source": _REPORT_SOURCE,
|
||||
"materialization": self._build_materialization_payload(dataset),
|
||||
"reports": [self._serialize_conversation_telemetry_report_summary(report) for report in _CONVERSATION_TELEMETRY_REPORTS],
|
||||
}
|
||||
|
||||
def get_conversation_telemetry_report_payload(self, report_key: str) -> dict | None:
|
||||
normalized_report_key = self._normalize_key(report_key)
|
||||
dataset = self._get_conversation_telemetry_dataset()
|
||||
for report in _CONVERSATION_TELEMETRY_REPORTS:
|
||||
if report["report_key"] == normalized_report_key:
|
||||
return {
|
||||
"area": "telemetria_conversacional",
|
||||
"source_domain": dataset.domain,
|
||||
"source": _REPORT_SOURCE,
|
||||
"materialization": self._build_materialization_payload(dataset),
|
||||
"report": self._serialize_conversation_telemetry_report_detail(report, dataset),
|
||||
}
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _build_materialization_payload(dataset: OperationalDatasetContract | None = None) -> dict:
|
||||
reference_dataset = dataset or PRODUCT_OPERATIONAL_DATASETS[0]
|
||||
return {
|
||||
"report_read_model": reference_dataset.report_read_model,
|
||||
"consistency_model": reference_dataset.consistency_model,
|
||||
"sync_strategy": reference_dataset.sync_strategy,
|
||||
"storage_shape": reference_dataset.storage_shape,
|
||||
"query_surface": reference_dataset.query_surface,
|
||||
"uses_product_replica": reference_dataset.uses_product_replica,
|
||||
"direct_product_query_allowed": reference_dataset.direct_product_query_allowed,
|
||||
"refresh_behavior": _REFRESH_BEHAVIOR,
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _serialize_dataset_summary(dataset) -> dict:
|
||||
return {
|
||||
"dataset_key": dataset.dataset_key,
|
||||
"domain": dataset.domain,
|
||||
"description": dataset.description,
|
||||
"source_table": dataset.source_table,
|
||||
"freshness_target": dataset.freshness_target,
|
||||
"allowed_granularities": list(dataset.allowed_granularities),
|
||||
"allowed_field_count": len(dataset.allowed_fields),
|
||||
"blocked_field_count": len(dataset.blocked_fields),
|
||||
"write_allowed": dataset.write_allowed,
|
||||
"materialization_status": _MATERIALIZATION_STATUS,
|
||||
"last_consolidated_at": None,
|
||||
"source_watermark": None,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _serialize_dataset_detail(cls, dataset) -> dict:
|
||||
return {
|
||||
"dataset_key": dataset.dataset_key,
|
||||
"domain": dataset.domain,
|
||||
"description": dataset.description,
|
||||
"source_table": dataset.source_table,
|
||||
"read_permission": dataset.read_permission,
|
||||
"report_read_model": dataset.report_read_model,
|
||||
"consistency_model": dataset.consistency_model,
|
||||
"sync_strategy": dataset.sync_strategy,
|
||||
"storage_shape": dataset.storage_shape,
|
||||
"query_surface": dataset.query_surface,
|
||||
"uses_product_replica": dataset.uses_product_replica,
|
||||
"direct_product_query_allowed": dataset.direct_product_query_allowed,
|
||||
"freshness_target": dataset.freshness_target,
|
||||
"allowed_granularities": list(dataset.allowed_granularities),
|
||||
"write_allowed": dataset.write_allowed,
|
||||
"materialization_status": _MATERIALIZATION_STATUS,
|
||||
"last_consolidated_at": None,
|
||||
"source_watermark": None,
|
||||
"allowed_fields": [cls._serialize_field(field) for field in dataset.allowed_fields],
|
||||
"blocked_fields": [cls._serialize_field(field) for field in dataset.blocked_fields],
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _serialize_sales_report_summary(cls, report_definition: dict) -> dict:
|
||||
return {
|
||||
"report_key": report_definition["report_key"],
|
||||
"label": report_definition["label"],
|
||||
"description": report_definition["description"],
|
||||
"dataset_key": _SALES_DATASET_KEY,
|
||||
"default_time_field": report_definition["default_time_field"],
|
||||
"default_granularity": report_definition["default_granularity"],
|
||||
"supported_metric_keys": list(report_definition["metric_keys"]),
|
||||
"supported_dimension_fields": list(report_definition["dimension_fields"]),
|
||||
"materialization_status": _MATERIALIZATION_STATUS,
|
||||
"last_consolidated_at": None,
|
||||
"source_watermark": None,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _serialize_sales_report_detail(
|
||||
cls,
|
||||
report_definition: dict,
|
||||
dataset: OperationalDatasetContract,
|
||||
) -> dict:
|
||||
return {
|
||||
"report_key": report_definition["report_key"],
|
||||
"label": report_definition["label"],
|
||||
"description": report_definition["description"],
|
||||
"dataset_key": _SALES_DATASET_KEY,
|
||||
"default_time_field": report_definition["default_time_field"],
|
||||
"default_granularity": report_definition["default_granularity"],
|
||||
"metrics": [dict(_SALES_REPORT_METRICS[key]) for key in report_definition["metric_keys"]],
|
||||
"dimensions": [dict(_SALES_DIMENSIONS[field_name]) for field_name in report_definition["dimension_fields"]],
|
||||
"filters": [dict(_SALES_FILTERS[field_name]) for field_name in report_definition["filter_fields"]],
|
||||
"dataset": cls._serialize_dataset_detail(dataset),
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _serialize_revenue_report_summary(cls, report_definition: dict) -> dict:
|
||||
return {
|
||||
"report_key": report_definition["report_key"],
|
||||
"label": report_definition["label"],
|
||||
"description": report_definition["description"],
|
||||
"dataset_key": _REVENUE_DATASET_KEY,
|
||||
"default_time_field": report_definition["default_time_field"],
|
||||
"default_granularity": report_definition["default_granularity"],
|
||||
"supported_metric_keys": list(report_definition["metric_keys"]),
|
||||
"supported_dimension_fields": list(report_definition["dimension_fields"]),
|
||||
"materialization_status": _MATERIALIZATION_STATUS,
|
||||
"last_consolidated_at": None,
|
||||
"source_watermark": None,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _serialize_revenue_report_detail(
|
||||
cls,
|
||||
report_definition: dict,
|
||||
dataset: OperationalDatasetContract,
|
||||
) -> dict:
|
||||
return {
|
||||
"report_key": report_definition["report_key"],
|
||||
"label": report_definition["label"],
|
||||
"description": report_definition["description"],
|
||||
"dataset_key": _REVENUE_DATASET_KEY,
|
||||
"default_time_field": report_definition["default_time_field"],
|
||||
"default_granularity": report_definition["default_granularity"],
|
||||
"metrics": [dict(_REVENUE_REPORT_METRICS[key]) for key in report_definition["metric_keys"]],
|
||||
"dimensions": [dict(_REVENUE_DIMENSIONS[field_name]) for field_name in report_definition["dimension_fields"]],
|
||||
"filters": [dict(_REVENUE_FILTERS[field_name]) for field_name in report_definition["filter_fields"]],
|
||||
"dataset": cls._serialize_dataset_detail(dataset),
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _serialize_rental_report_summary(cls, report_definition: dict) -> dict:
|
||||
return {
|
||||
"report_key": report_definition["report_key"],
|
||||
"label": report_definition["label"],
|
||||
"description": report_definition["description"],
|
||||
"dataset_key": report_definition["dataset_key"],
|
||||
"default_time_field": report_definition["default_time_field"],
|
||||
"default_granularity": report_definition["default_granularity"],
|
||||
"supported_metric_keys": list(report_definition["metric_keys"]),
|
||||
"supported_dimension_fields": list(report_definition["dimension_fields"]),
|
||||
"materialization_status": _MATERIALIZATION_STATUS,
|
||||
"last_consolidated_at": None,
|
||||
"source_watermark": None,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _serialize_rental_report_detail(
|
||||
cls,
|
||||
report_definition: dict,
|
||||
dataset: OperationalDatasetContract,
|
||||
) -> dict:
|
||||
return {
|
||||
"report_key": report_definition["report_key"],
|
||||
"label": report_definition["label"],
|
||||
"description": report_definition["description"],
|
||||
"dataset_key": report_definition["dataset_key"],
|
||||
"default_time_field": report_definition["default_time_field"],
|
||||
"default_granularity": report_definition["default_granularity"],
|
||||
"metrics": [dict(_RENTAL_REPORT_METRICS[key]) for key in report_definition["metric_keys"]],
|
||||
"dimensions": [dict(_RENTAL_DIMENSIONS[field_name]) for field_name in report_definition["dimension_fields"]],
|
||||
"filters": [dict(_RENTAL_FILTERS[field_name]) for field_name in report_definition["filter_fields"]],
|
||||
"dataset": cls._serialize_dataset_detail(dataset),
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _serialize_bot_flow_report_summary(cls, report_definition: dict) -> dict:
|
||||
return {
|
||||
"report_key": report_definition["report_key"],
|
||||
"label": report_definition["label"],
|
||||
"description": report_definition["description"],
|
||||
"dataset_key": _BOT_FLOW_DATASET_KEY,
|
||||
"default_time_field": report_definition["default_time_field"],
|
||||
"default_granularity": report_definition["default_granularity"],
|
||||
"supported_metric_keys": list(report_definition["metric_keys"]),
|
||||
"supported_dimension_fields": list(report_definition["dimension_fields"]),
|
||||
"materialization_status": _MATERIALIZATION_STATUS,
|
||||
"last_consolidated_at": None,
|
||||
"source_watermark": None,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _serialize_bot_flow_report_detail(
|
||||
cls,
|
||||
report_definition: dict,
|
||||
dataset: OperationalDatasetContract,
|
||||
) -> dict:
|
||||
return {
|
||||
"report_key": report_definition["report_key"],
|
||||
"label": report_definition["label"],
|
||||
"description": report_definition["description"],
|
||||
"dataset_key": _BOT_FLOW_DATASET_KEY,
|
||||
"default_time_field": report_definition["default_time_field"],
|
||||
"default_granularity": report_definition["default_granularity"],
|
||||
"metrics": [dict(_BOT_FLOW_REPORT_METRICS[key]) for key in report_definition["metric_keys"]],
|
||||
"dimensions": [dict(_BOT_FLOW_DIMENSIONS[field_name]) for field_name in report_definition["dimension_fields"]],
|
||||
"filters": [dict(_BOT_FLOW_FILTERS[field_name]) for field_name in report_definition["filter_fields"]],
|
||||
"dataset": cls._serialize_dataset_detail(dataset),
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _serialize_conversation_telemetry_report_summary(cls, report_definition: dict) -> dict:
|
||||
return {
|
||||
"report_key": report_definition["report_key"],
|
||||
"label": report_definition["label"],
|
||||
"description": report_definition["description"],
|
||||
"dataset_key": _CONVERSATION_TELEMETRY_DATASET_KEY,
|
||||
"default_time_field": report_definition["default_time_field"],
|
||||
"default_granularity": report_definition["default_granularity"],
|
||||
"supported_metric_keys": list(report_definition["metric_keys"]),
|
||||
"supported_dimension_fields": list(report_definition["dimension_fields"]),
|
||||
"materialization_status": _MATERIALIZATION_STATUS,
|
||||
"last_consolidated_at": None,
|
||||
"source_watermark": None,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _serialize_conversation_telemetry_report_detail(
|
||||
cls,
|
||||
report_definition: dict,
|
||||
dataset: OperationalDatasetContract,
|
||||
) -> dict:
|
||||
return {
|
||||
"report_key": report_definition["report_key"],
|
||||
"label": report_definition["label"],
|
||||
"description": report_definition["description"],
|
||||
"dataset_key": _CONVERSATION_TELEMETRY_DATASET_KEY,
|
||||
"default_time_field": report_definition["default_time_field"],
|
||||
"default_granularity": report_definition["default_granularity"],
|
||||
"metrics": [dict(_CONVERSATION_TELEMETRY_REPORT_METRICS[key]) for key in report_definition["metric_keys"]],
|
||||
"dimensions": [dict(_CONVERSATION_TELEMETRY_DIMENSIONS[field_name]) for field_name in report_definition["dimension_fields"]],
|
||||
"filters": [dict(_CONVERSATION_TELEMETRY_FILTERS[field_name]) for field_name in report_definition["filter_fields"]],
|
||||
"dataset": cls._serialize_dataset_detail(dataset),
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _serialize_field(field) -> dict:
|
||||
return {
|
||||
"name": field.name,
|
||||
"description": field.description,
|
||||
"sensitivity": field.sensitivity,
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _normalize_key(value: str) -> str:
|
||||
return str(value or "").strip().lower()
|
||||
|
||||
@staticmethod
|
||||
def _get_sales_dataset() -> OperationalDatasetContract:
|
||||
dataset = get_operational_dataset(_SALES_DATASET_KEY)
|
||||
if dataset is None:
|
||||
raise RuntimeError("sales_orders contract is required to build sales reports")
|
||||
return dataset
|
||||
|
||||
@staticmethod
|
||||
def _get_revenue_dataset() -> OperationalDatasetContract:
|
||||
dataset = get_operational_dataset(_REVENUE_DATASET_KEY)
|
||||
if dataset is None:
|
||||
raise RuntimeError("rental_payments contract is required to build revenue reports")
|
||||
return dataset
|
||||
|
||||
@staticmethod
|
||||
def _get_rental_dataset(dataset_key: str) -> OperationalDatasetContract:
|
||||
dataset = get_operational_dataset(dataset_key)
|
||||
if dataset is None:
|
||||
raise RuntimeError(f"{dataset_key} contract is required to build rental reports")
|
||||
return dataset
|
||||
|
||||
@classmethod
|
||||
def _get_rental_fleet_dataset(cls) -> OperationalDatasetContract:
|
||||
return cls._get_rental_dataset(_RENTAL_FLEET_DATASET_KEY)
|
||||
|
||||
@classmethod
|
||||
def _get_rental_contracts_dataset(cls) -> OperationalDatasetContract:
|
||||
return cls._get_rental_dataset(_RENTAL_CONTRACTS_DATASET_KEY)
|
||||
|
||||
@staticmethod
|
||||
def _get_bot_flow_dataset() -> OperationalDatasetContract:
|
||||
dataset = get_operational_dataset(_BOT_FLOW_DATASET_KEY)
|
||||
if dataset is None:
|
||||
raise RuntimeError("conversation_turns contract is required to build bot flow reports")
|
||||
return dataset
|
||||
|
||||
@staticmethod
|
||||
def _get_conversation_telemetry_dataset() -> OperationalDatasetContract:
|
||||
dataset = get_operational_dataset(_CONVERSATION_TELEMETRY_DATASET_KEY)
|
||||
if dataset is None:
|
||||
raise RuntimeError("conversation_turns contract is required to build conversation telemetry reports")
|
||||
return dataset
|
||||
@ -0,0 +1,116 @@
|
||||
import unittest
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from admin_app.api.dependencies import get_current_staff_principal
|
||||
from admin_app.app_factory import create_app
|
||||
from admin_app.core import AdminSettings, AuthenticatedStaffPrincipal
|
||||
from shared.contracts import StaffRole
|
||||
|
||||
|
||||
class AdminBotFlowReportsWebTests(unittest.TestCase):
|
||||
def _build_client_with_role(
|
||||
self,
|
||||
role: StaffRole,
|
||||
settings: AdminSettings | None = None,
|
||||
) -> tuple[TestClient, object]:
|
||||
app = create_app(
|
||||
settings
|
||||
or AdminSettings(
|
||||
admin_auth_token_secret="test-secret",
|
||||
admin_api_prefix="/admin",
|
||||
)
|
||||
)
|
||||
app.dependency_overrides[get_current_staff_principal] = lambda: AuthenticatedStaffPrincipal(
|
||||
id=71,
|
||||
email="colaborador@empresa.com" if role == StaffRole.COLABORADOR else "diretor@empresa.com",
|
||||
display_name="Equipe de Operacao do Bot",
|
||||
role=role,
|
||||
is_active=True,
|
||||
)
|
||||
return TestClient(app), app
|
||||
|
||||
def test_bot_flow_reports_overview_requires_authentication(self):
|
||||
app = create_app(AdminSettings(admin_auth_token_secret="test-secret", admin_api_prefix="/admin"))
|
||||
client = TestClient(app)
|
||||
|
||||
response = client.get("/admin/reports/fluxo-bot/overview")
|
||||
|
||||
self.assertEqual(response.status_code, 401)
|
||||
self.assertEqual(response.json()["detail"], "Autenticacao administrativa obrigatoria.")
|
||||
|
||||
def test_bot_flow_reports_overview_returns_bootstrap_structure_for_colaborador(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get("/admin/reports/fluxo-bot/overview", headers={"Authorization": "Bearer token"})
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()
|
||||
self.assertEqual(payload["service"], "orquestrador-admin")
|
||||
self.assertEqual(payload["area"], "fluxo_bot")
|
||||
self.assertEqual(payload["source_domain"], "conversation")
|
||||
self.assertEqual(payload["mode"], "bot_flow_contract_bootstrap")
|
||||
self.assertEqual(payload["source_dataset_keys"], ["conversation_turns"])
|
||||
self.assertEqual(payload["materialization"]["sync_strategy"], "etl_incremental")
|
||||
self.assertEqual(len(payload["reports"]), 5)
|
||||
self.assertIn("fallback_and_handoff", [item["report_key"] for item in payload["reports"]])
|
||||
self.assertEqual(payload["metrics"][1]["value"], "5")
|
||||
|
||||
def test_bot_flow_reports_catalog_returns_initial_report_definitions(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get("/admin/reports/fluxo-bot/reports", headers={"Authorization": "Bearer token"})
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()
|
||||
self.assertEqual(payload["area"], "fluxo_bot")
|
||||
self.assertEqual(payload["source_domain"], "conversation")
|
||||
self.assertEqual(payload["source"], "shared_contract_catalog")
|
||||
self.assertEqual(len(payload["reports"]), 5)
|
||||
routing = next(item for item in payload["reports"] if item["report_key"] == "action_routing_flow")
|
||||
self.assertEqual(routing["dataset_key"], "conversation_turns")
|
||||
self.assertEqual(routing["default_granularity"], "aggregate")
|
||||
self.assertEqual(routing["materialization_status"], "contract_defined_pending_snapshot_view")
|
||||
self.assertIn("action", routing["supported_dimension_fields"])
|
||||
|
||||
def test_bot_flow_report_detail_returns_metrics_filters_and_dataset_contract(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/reports/fluxo-bot/reports/operational_failures",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()["report"]
|
||||
self.assertEqual(payload["report_key"], "operational_failures")
|
||||
self.assertEqual(payload["dataset_key"], "conversation_turns")
|
||||
self.assertEqual(payload["default_time_field"], "started_at")
|
||||
self.assertIn("errored_turns", [metric["key"] for metric in payload["metrics"]])
|
||||
self.assertIn("action", [item["field_name"] for item in payload["dimensions"]])
|
||||
self.assertIn("tool_name", [item["field_name"] for item in payload["filters"]])
|
||||
self.assertFalse(payload["dataset"]["write_allowed"])
|
||||
self.assertIn("assistant_response", [field["name"] for field in payload["dataset"]["blocked_fields"]])
|
||||
|
||||
def test_bot_flow_report_detail_returns_404_for_unknown_report(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/reports/fluxo-bot/reports/channel_heatmap",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 404)
|
||||
self.assertEqual(response.json()["detail"], "Relatorio operacional do fluxo do bot nao encontrado.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@ -0,0 +1,126 @@
|
||||
import unittest
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from admin_app.api.dependencies import get_current_staff_principal
|
||||
from admin_app.app_factory import create_app
|
||||
from admin_app.core import AdminSettings, AuthenticatedStaffPrincipal
|
||||
from shared.contracts import StaffRole
|
||||
|
||||
|
||||
class AdminConversationTelemetryReportsWebTests(unittest.TestCase):
|
||||
def _build_client_with_role(
|
||||
self,
|
||||
role: StaffRole,
|
||||
settings: AdminSettings | None = None,
|
||||
) -> tuple[TestClient, object]:
|
||||
app = create_app(
|
||||
settings
|
||||
or AdminSettings(
|
||||
admin_auth_token_secret="test-secret",
|
||||
admin_api_prefix="/admin",
|
||||
)
|
||||
)
|
||||
app.dependency_overrides[get_current_staff_principal] = lambda: AuthenticatedStaffPrincipal(
|
||||
id=81,
|
||||
email="colaborador@empresa.com" if role == StaffRole.COLABORADOR else "diretor@empresa.com",
|
||||
display_name="Equipe de Telemetria Conversacional",
|
||||
role=role,
|
||||
is_active=True,
|
||||
)
|
||||
return TestClient(app), app
|
||||
|
||||
def test_conversation_telemetry_reports_overview_requires_authentication(self):
|
||||
app = create_app(AdminSettings(admin_auth_token_secret="test-secret", admin_api_prefix="/admin"))
|
||||
client = TestClient(app)
|
||||
|
||||
response = client.get("/admin/reports/telemetria-conversacional/overview")
|
||||
|
||||
self.assertEqual(response.status_code, 401)
|
||||
self.assertEqual(response.json()["detail"], "Autenticacao administrativa obrigatoria.")
|
||||
|
||||
def test_conversation_telemetry_reports_overview_returns_bootstrap_structure_for_colaborador(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/reports/telemetria-conversacional/overview",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()
|
||||
self.assertEqual(payload["service"], "orquestrador-admin")
|
||||
self.assertEqual(payload["area"], "telemetria_conversacional")
|
||||
self.assertEqual(payload["source_domain"], "conversation")
|
||||
self.assertEqual(payload["mode"], "conversation_telemetry_contract_bootstrap")
|
||||
self.assertEqual(payload["source_dataset_keys"], ["conversation_turns"])
|
||||
self.assertEqual(payload["materialization"]["sync_strategy"], "etl_incremental")
|
||||
self.assertEqual(len(payload["reports"]), 5)
|
||||
self.assertIn("latency_profile", [item["report_key"] for item in payload["reports"]])
|
||||
self.assertEqual(payload["metrics"][1]["value"], "5")
|
||||
|
||||
def test_conversation_telemetry_reports_catalog_returns_initial_report_definitions(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/reports/telemetria-conversacional/reports",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()
|
||||
self.assertEqual(payload["area"], "telemetria_conversacional")
|
||||
self.assertEqual(payload["source_domain"], "conversation")
|
||||
self.assertEqual(payload["source"], "shared_contract_catalog")
|
||||
self.assertEqual(len(payload["reports"]), 5)
|
||||
volume = next(item for item in payload["reports"] if item["report_key"] == "conversation_volume")
|
||||
self.assertEqual(volume["dataset_key"], "conversation_turns")
|
||||
self.assertEqual(volume["default_granularity"], "aggregate")
|
||||
self.assertEqual(volume["materialization_status"], "contract_defined_pending_snapshot_view")
|
||||
self.assertIn("domain", volume["supported_dimension_fields"])
|
||||
|
||||
def test_conversation_telemetry_report_detail_returns_metrics_filters_and_dataset_contract(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/reports/telemetria-conversacional/reports/latency_profile",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()["report"]
|
||||
self.assertEqual(payload["report_key"], "latency_profile")
|
||||
self.assertEqual(payload["dataset_key"], "conversation_turns")
|
||||
self.assertEqual(payload["default_time_field"], "completed_at")
|
||||
self.assertIn("average_latency_ms", [metric["key"] for metric in payload["metrics"]])
|
||||
self.assertIn("p95_latency_ms", [metric["key"] for metric in payload["metrics"]])
|
||||
self.assertIn("intent", [item["field_name"] for item in payload["dimensions"]])
|
||||
self.assertIn("completed_at", [item["field_name"] for item in payload["filters"]])
|
||||
self.assertFalse(payload["dataset"]["write_allowed"])
|
||||
self.assertIn("assistant_response", [field["name"] for field in payload["dataset"]["blocked_fields"]])
|
||||
|
||||
def test_conversation_telemetry_report_detail_returns_404_for_unknown_report(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/reports/telemetria-conversacional/reports/sentiment_breakdown",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 404)
|
||||
self.assertEqual(
|
||||
response.json()["detail"],
|
||||
"Relatorio de telemetria conversacional nao encontrado.",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@ -0,0 +1,65 @@
|
||||
import unittest
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from admin_app.api.dependencies import get_optional_panel_staff_context
|
||||
from admin_app.app_factory import create_app
|
||||
from admin_app.core import AdminSettings, AuthenticatedStaffContext, AuthenticatedStaffPrincipal
|
||||
from shared.contracts import StaffRole
|
||||
|
||||
|
||||
def _build_panel_context(role: StaffRole = StaffRole.COLABORADOR) -> AuthenticatedStaffContext:
|
||||
return AuthenticatedStaffContext(
|
||||
principal=AuthenticatedStaffPrincipal(
|
||||
id=41 if role == StaffRole.COLABORADOR else 42,
|
||||
email="colaborador@empresa.com" if role == StaffRole.COLABORADOR else "diretor@empresa.com",
|
||||
display_name="Equipe de Monitoramento do Bot",
|
||||
role=role,
|
||||
is_active=True,
|
||||
),
|
||||
session_id=101,
|
||||
)
|
||||
|
||||
|
||||
class AdminPanelBotMonitoringViewTests(unittest.TestCase):
|
||||
def test_bot_monitoring_page_redirects_to_login_without_session(self):
|
||||
app = create_app(AdminSettings(admin_api_prefix="/admin"))
|
||||
client = TestClient(app)
|
||||
|
||||
response = client.get("/admin/panel/monitoramento/bot", follow_redirects=False)
|
||||
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertTrue(response.headers["location"].endswith("/admin/login"))
|
||||
|
||||
def test_bot_monitoring_page_renders_for_colaborador_session(self):
|
||||
app = create_app(AdminSettings(admin_app_name="Admin Interno", admin_version="1.4.0", admin_api_prefix="/admin"))
|
||||
app.dependency_overrides[get_optional_panel_staff_context] = lambda: _build_panel_context(StaffRole.COLABORADOR)
|
||||
client = TestClient(app)
|
||||
try:
|
||||
response = client.get("/admin/panel/monitoramento/bot")
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("Monitoramento operacional do bot", response.text)
|
||||
self.assertIn('data-admin-bot-monitoring="true"', response.text)
|
||||
self.assertIn('data-bot-flow-overview-endpoint="/admin/panel/reports/fluxo-bot/overview"', response.text)
|
||||
self.assertIn('data-telemetry-overview-endpoint="/admin/panel/reports/telemetria-conversacional/overview"', response.text)
|
||||
self.assertIn("Atualizar leitura", response.text)
|
||||
|
||||
def test_dashboard_exposes_bot_monitoring_screen_link(self):
|
||||
app = create_app(AdminSettings(admin_api_prefix="/admin"))
|
||||
app.dependency_overrides[get_optional_panel_staff_context] = lambda: _build_panel_context(StaffRole.DIRETOR)
|
||||
client = TestClient(app)
|
||||
try:
|
||||
response = client.get("/admin/panel/admin")
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("/admin/panel/monitoramento/bot", response.text)
|
||||
self.assertIn("Abrir monitoramento", response.text)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@ -0,0 +1,64 @@
|
||||
import unittest
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from admin_app.api.dependencies import get_optional_panel_staff_context
|
||||
from admin_app.app_factory import create_app
|
||||
from admin_app.core import AdminSettings, AuthenticatedStaffContext, AuthenticatedStaffPrincipal
|
||||
from shared.contracts import StaffRole
|
||||
|
||||
|
||||
def _build_panel_context(role: StaffRole = StaffRole.COLABORADOR) -> AuthenticatedStaffContext:
|
||||
return AuthenticatedStaffContext(
|
||||
principal=AuthenticatedStaffPrincipal(
|
||||
id=31 if role == StaffRole.COLABORADOR else 32,
|
||||
email="colaborador@empresa.com" if role == StaffRole.COLABORADOR else "diretor@empresa.com",
|
||||
display_name="Equipe de Locacao Interna",
|
||||
role=role,
|
||||
is_active=True,
|
||||
),
|
||||
session_id=97,
|
||||
)
|
||||
|
||||
|
||||
class AdminPanelRentalReportsViewTests(unittest.TestCase):
|
||||
def test_rental_reports_page_redirects_to_login_without_session(self):
|
||||
app = create_app(AdminSettings(admin_api_prefix="/admin"))
|
||||
client = TestClient(app)
|
||||
|
||||
response = client.get("/admin/panel/relatorios/locacao", follow_redirects=False)
|
||||
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertTrue(response.headers["location"].endswith("/admin/login"))
|
||||
|
||||
def test_rental_reports_page_renders_for_colaborador_session(self):
|
||||
app = create_app(AdminSettings(admin_app_name="Admin Interno", admin_version="1.4.0", admin_api_prefix="/admin"))
|
||||
app.dependency_overrides[get_optional_panel_staff_context] = lambda: _build_panel_context(StaffRole.COLABORADOR)
|
||||
client = TestClient(app)
|
||||
try:
|
||||
response = client.get("/admin/panel/relatorios/locacao")
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("Relatorios de locacao", response.text)
|
||||
self.assertIn('data-admin-rental-reports="true"', response.text)
|
||||
self.assertIn('data-rental-overview-endpoint="/admin/panel/reports/locacao/overview"', response.text)
|
||||
self.assertIn("Atualizar leitura", response.text)
|
||||
|
||||
def test_dashboard_exposes_rental_reports_screen_link(self):
|
||||
app = create_app(AdminSettings(admin_api_prefix="/admin"))
|
||||
app.dependency_overrides[get_optional_panel_staff_context] = lambda: _build_panel_context(StaffRole.DIRETOR)
|
||||
client = TestClient(app)
|
||||
try:
|
||||
response = client.get("/admin/panel/admin")
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("/admin/panel/relatorios/locacao", response.text)
|
||||
self.assertIn("Abrir locacao", response.text)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@ -0,0 +1,117 @@
|
||||
import unittest
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from admin_app.api.dependencies import get_current_staff_principal
|
||||
from admin_app.app_factory import create_app
|
||||
from admin_app.core import AdminSettings, AuthenticatedStaffPrincipal
|
||||
from shared.contracts import StaffRole
|
||||
|
||||
|
||||
class AdminRentalReportsWebTests(unittest.TestCase):
|
||||
def _build_client_with_role(
|
||||
self,
|
||||
role: StaffRole,
|
||||
settings: AdminSettings | None = None,
|
||||
) -> tuple[TestClient, object]:
|
||||
app = create_app(
|
||||
settings
|
||||
or AdminSettings(
|
||||
admin_auth_token_secret="test-secret",
|
||||
admin_api_prefix="/admin",
|
||||
)
|
||||
)
|
||||
app.dependency_overrides[get_current_staff_principal] = lambda: AuthenticatedStaffPrincipal(
|
||||
id=61,
|
||||
email="colaborador@empresa.com" if role == StaffRole.COLABORADOR else "diretor@empresa.com",
|
||||
display_name="Equipe de Locacao",
|
||||
role=role,
|
||||
is_active=True,
|
||||
)
|
||||
return TestClient(app), app
|
||||
|
||||
def test_rental_reports_overview_requires_authentication(self):
|
||||
app = create_app(AdminSettings(admin_auth_token_secret="test-secret", admin_api_prefix="/admin"))
|
||||
client = TestClient(app)
|
||||
|
||||
response = client.get("/admin/reports/locacao/overview")
|
||||
|
||||
self.assertEqual(response.status_code, 401)
|
||||
self.assertEqual(response.json()["detail"], "Autenticacao administrativa obrigatoria.")
|
||||
|
||||
def test_rental_reports_overview_returns_bootstrap_structure_for_colaborador(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get("/admin/reports/locacao/overview", headers={"Authorization": "Bearer token"})
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()
|
||||
self.assertEqual(payload["service"], "orquestrador-admin")
|
||||
self.assertEqual(payload["area"], "locacao")
|
||||
self.assertEqual(payload["source_domain"], "rental")
|
||||
self.assertEqual(payload["mode"], "rental_contract_bootstrap")
|
||||
self.assertEqual(payload["source_dataset_keys"], ["rental_fleet", "rental_contracts"])
|
||||
self.assertEqual(payload["materialization"]["sync_strategy"], "etl_incremental")
|
||||
self.assertEqual(len(payload["reports"]), 5)
|
||||
self.assertIn("fleet_occupancy", [item["report_key"] for item in payload["reports"]])
|
||||
self.assertEqual(payload["metrics"][1]["value"], "5")
|
||||
|
||||
def test_rental_reports_catalog_returns_initial_report_definitions(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get("/admin/reports/locacao/reports", headers={"Authorization": "Bearer token"})
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()
|
||||
self.assertEqual(payload["area"], "locacao")
|
||||
self.assertEqual(payload["source_domain"], "rental")
|
||||
self.assertEqual(payload["source"], "shared_contract_catalog")
|
||||
self.assertEqual(len(payload["reports"]), 5)
|
||||
fleet = next(item for item in payload["reports"] if item["report_key"] == "fleet_availability")
|
||||
self.assertEqual(fleet["dataset_key"], "rental_fleet")
|
||||
self.assertEqual(fleet["default_granularity"], "aggregate")
|
||||
self.assertEqual(fleet["materialization_status"], "contract_defined_pending_snapshot_view")
|
||||
self.assertIn("categoria", fleet["supported_dimension_fields"])
|
||||
|
||||
def test_rental_report_detail_returns_metrics_filters_and_dataset_contract(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/reports/locacao/reports/projected_vs_final_revenue",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()["report"]
|
||||
self.assertEqual(payload["report_key"], "projected_vs_final_revenue")
|
||||
self.assertEqual(payload["dataset_key"], "rental_contracts")
|
||||
self.assertEqual(payload["default_time_field"], "updated_at")
|
||||
self.assertIn("projected_revenue", [metric["key"] for metric in payload["metrics"]])
|
||||
self.assertIn("revenue_delta", [metric["key"] for metric in payload["metrics"]])
|
||||
self.assertIn("status", [item["field_name"] for item in payload["dimensions"]])
|
||||
self.assertIn("contrato_numero", [item["field_name"] for item in payload["filters"]])
|
||||
self.assertFalse(payload["dataset"]["write_allowed"])
|
||||
self.assertIn("cpf", [field["name"] for field in payload["dataset"]["blocked_fields"]])
|
||||
|
||||
def test_rental_report_detail_returns_404_for_unknown_report(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/reports/locacao/reports/fleet_idle_heatmap",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 404)
|
||||
self.assertEqual(response.json()["detail"], "Relatorio de locacao nao encontrado.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@ -0,0 +1,114 @@
|
||||
import unittest
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from admin_app.api.dependencies import get_current_staff_principal
|
||||
from admin_app.app_factory import create_app
|
||||
from admin_app.core import AdminSettings, AuthenticatedStaffPrincipal
|
||||
from shared.contracts import StaffRole
|
||||
|
||||
|
||||
class AdminReportsWebTests(unittest.TestCase):
|
||||
def _build_client_with_role(
|
||||
self,
|
||||
role: StaffRole,
|
||||
settings: AdminSettings | None = None,
|
||||
) -> tuple[TestClient, object]:
|
||||
app = create_app(
|
||||
settings
|
||||
or AdminSettings(
|
||||
admin_auth_token_secret="test-secret",
|
||||
admin_api_prefix="/admin",
|
||||
)
|
||||
)
|
||||
app.dependency_overrides[get_current_staff_principal] = lambda: AuthenticatedStaffPrincipal(
|
||||
id=31,
|
||||
email="colaborador@empresa.com" if role == StaffRole.COLABORADOR else "diretor@empresa.com",
|
||||
display_name="Equipe de Relatorios",
|
||||
role=role,
|
||||
is_active=True,
|
||||
)
|
||||
return TestClient(app), app
|
||||
|
||||
def test_reports_overview_requires_authentication(self):
|
||||
app = create_app(AdminSettings(admin_auth_token_secret="test-secret", admin_api_prefix="/admin"))
|
||||
client = TestClient(app)
|
||||
|
||||
response = client.get("/admin/reports/overview")
|
||||
|
||||
self.assertEqual(response.status_code, 401)
|
||||
self.assertEqual(response.json()["detail"], "Autenticacao administrativa obrigatoria.")
|
||||
|
||||
def test_reports_overview_returns_contract_snapshot_for_colaborador(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get("/admin/reports/overview", headers={"Authorization": "Bearer token"})
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()
|
||||
self.assertEqual(payload["service"], "orquestrador-admin")
|
||||
self.assertEqual(payload["mode"], "shared_contract_bootstrap")
|
||||
self.assertEqual(payload["materialization"]["sync_strategy"], "etl_incremental")
|
||||
self.assertEqual(payload["materialization"]["storage_shape"], "snapshot_table")
|
||||
self.assertEqual(payload["materialization"]["query_surface"], "dedicated_view")
|
||||
self.assertIn("sales", [item["key"] for item in payload["report_families"]])
|
||||
self.assertEqual(payload["metrics"][0]["value"], "8")
|
||||
|
||||
def test_report_datasets_return_catalog_for_colaborador(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get("/admin/reports/datasets", headers={"Authorization": "Bearer token"})
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()
|
||||
self.assertEqual(payload["source"], "shared_contract_catalog")
|
||||
self.assertEqual(len(payload["datasets"]), 8)
|
||||
self.assertIn("sales_orders", [item["dataset_key"] for item in payload["datasets"]])
|
||||
self.assertIn("conversation_turns", [item["dataset_key"] for item in payload["datasets"]])
|
||||
sales = next(item for item in payload["datasets"] if item["dataset_key"] == "sales_orders")
|
||||
self.assertEqual(sales["freshness_target"], "near_real_time")
|
||||
self.assertEqual(sales["materialization_status"], "contract_defined_pending_snapshot_view")
|
||||
|
||||
def test_report_dataset_detail_returns_allowed_and_blocked_fields(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/reports/datasets/sales_orders",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()["dataset"]
|
||||
self.assertEqual(payload["dataset_key"], "sales_orders")
|
||||
self.assertEqual(payload["domain"], "sales")
|
||||
self.assertEqual(payload["read_permission"], "view_reports")
|
||||
self.assertFalse(payload["direct_product_query_allowed"])
|
||||
self.assertFalse(payload["write_allowed"])
|
||||
self.assertIn("numero_pedido", [field["name"] for field in payload["allowed_fields"]])
|
||||
self.assertIn("cpf", [field["name"] for field in payload["blocked_fields"]])
|
||||
|
||||
def test_report_dataset_detail_returns_404_for_unknown_dataset(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/reports/datasets/customers",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 404)
|
||||
self.assertEqual(
|
||||
response.json()["detail"],
|
||||
"Dataset operacional nao encontrado para relatorio.",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@ -0,0 +1,116 @@
|
||||
import unittest
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from admin_app.api.dependencies import get_current_staff_principal
|
||||
from admin_app.app_factory import create_app
|
||||
from admin_app.core import AdminSettings, AuthenticatedStaffPrincipal
|
||||
from shared.contracts import StaffRole
|
||||
|
||||
|
||||
class AdminRevenueReportsWebTests(unittest.TestCase):
|
||||
def _build_client_with_role(
|
||||
self,
|
||||
role: StaffRole,
|
||||
settings: AdminSettings | None = None,
|
||||
) -> tuple[TestClient, object]:
|
||||
app = create_app(
|
||||
settings
|
||||
or AdminSettings(
|
||||
admin_auth_token_secret="test-secret",
|
||||
admin_api_prefix="/admin",
|
||||
)
|
||||
)
|
||||
app.dependency_overrides[get_current_staff_principal] = lambda: AuthenticatedStaffPrincipal(
|
||||
id=51,
|
||||
email="colaborador@empresa.com" if role == StaffRole.COLABORADOR else "diretor@empresa.com",
|
||||
display_name="Equipe Financeira",
|
||||
role=role,
|
||||
is_active=True,
|
||||
)
|
||||
return TestClient(app), app
|
||||
|
||||
def test_revenue_reports_overview_requires_authentication(self):
|
||||
app = create_app(AdminSettings(admin_auth_token_secret="test-secret", admin_api_prefix="/admin"))
|
||||
client = TestClient(app)
|
||||
|
||||
response = client.get("/admin/reports/arrecadacao/overview")
|
||||
|
||||
self.assertEqual(response.status_code, 401)
|
||||
self.assertEqual(response.json()["detail"], "Autenticacao administrativa obrigatoria.")
|
||||
|
||||
def test_revenue_reports_overview_returns_bootstrap_structure_for_colaborador(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get("/admin/reports/arrecadacao/overview", headers={"Authorization": "Bearer token"})
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()
|
||||
self.assertEqual(payload["service"], "orquestrador-admin")
|
||||
self.assertEqual(payload["area"], "arrecadacao")
|
||||
self.assertEqual(payload["source_domain"], "rental")
|
||||
self.assertEqual(payload["mode"], "revenue_contract_bootstrap")
|
||||
self.assertEqual(payload["source_dataset_keys"], ["rental_payments"])
|
||||
self.assertEqual(payload["materialization"]["sync_strategy"], "etl_incremental")
|
||||
self.assertEqual(len(payload["reports"]), 3)
|
||||
self.assertIn("collected_amount", [item["report_key"] for item in payload["reports"]])
|
||||
self.assertEqual(payload["metrics"][1]["value"], "3")
|
||||
|
||||
def test_revenue_reports_catalog_returns_initial_report_definitions(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get("/admin/reports/arrecadacao/reports", headers={"Authorization": "Bearer token"})
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()
|
||||
self.assertEqual(payload["area"], "arrecadacao")
|
||||
self.assertEqual(payload["source_domain"], "rental")
|
||||
self.assertEqual(payload["source"], "shared_contract_catalog")
|
||||
self.assertEqual(len(payload["reports"]), 3)
|
||||
collected = next(item for item in payload["reports"] if item["report_key"] == "collected_amount")
|
||||
self.assertEqual(collected["dataset_key"], "rental_payments")
|
||||
self.assertEqual(collected["default_granularity"], "aggregate")
|
||||
self.assertEqual(collected["materialization_status"], "contract_defined_pending_snapshot_view")
|
||||
self.assertIn("contrato_numero", collected["supported_dimension_fields"])
|
||||
|
||||
def test_revenue_report_detail_returns_metrics_filters_and_dataset_contract(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/reports/arrecadacao/reports/contract_reconciliation",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()["report"]
|
||||
self.assertEqual(payload["report_key"], "contract_reconciliation")
|
||||
self.assertEqual(payload["dataset_key"], "rental_payments")
|
||||
self.assertEqual(payload["default_time_field"], "data_pagamento")
|
||||
self.assertIn("collected_amount", [metric["key"] for metric in payload["metrics"]])
|
||||
self.assertIn("contrato_numero", [item["field_name"] for item in payload["dimensions"]])
|
||||
self.assertIn("protocolo", [item["field_name"] for item in payload["filters"]])
|
||||
self.assertFalse(payload["dataset"]["write_allowed"])
|
||||
self.assertIn("identificador_comprovante", [field["name"] for field in payload["dataset"]["blocked_fields"]])
|
||||
|
||||
def test_revenue_report_detail_returns_404_for_unknown_report(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/reports/arrecadacao/reports/cash_flow_projection",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 404)
|
||||
self.assertEqual(response.json()["detail"], "Relatorio de arrecadacao nao encontrado.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@ -0,0 +1,114 @@
|
||||
import unittest
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from admin_app.api.dependencies import get_current_staff_principal
|
||||
from admin_app.app_factory import create_app
|
||||
from admin_app.core import AdminSettings, AuthenticatedStaffPrincipal
|
||||
from shared.contracts import StaffRole
|
||||
|
||||
|
||||
class AdminSalesReportsWebTests(unittest.TestCase):
|
||||
def _build_client_with_role(
|
||||
self,
|
||||
role: StaffRole,
|
||||
settings: AdminSettings | None = None,
|
||||
) -> tuple[TestClient, object]:
|
||||
app = create_app(
|
||||
settings
|
||||
or AdminSettings(
|
||||
admin_auth_token_secret="test-secret",
|
||||
admin_api_prefix="/admin",
|
||||
)
|
||||
)
|
||||
app.dependency_overrides[get_current_staff_principal] = lambda: AuthenticatedStaffPrincipal(
|
||||
id=41,
|
||||
email="colaborador@empresa.com" if role == StaffRole.COLABORADOR else "diretor@empresa.com",
|
||||
display_name="Equipe Comercial",
|
||||
role=role,
|
||||
is_active=True,
|
||||
)
|
||||
return TestClient(app), app
|
||||
|
||||
def test_sales_reports_overview_requires_authentication(self):
|
||||
app = create_app(AdminSettings(admin_auth_token_secret="test-secret", admin_api_prefix="/admin"))
|
||||
client = TestClient(app)
|
||||
|
||||
response = client.get("/admin/reports/sales/overview")
|
||||
|
||||
self.assertEqual(response.status_code, 401)
|
||||
self.assertEqual(response.json()["detail"], "Autenticacao administrativa obrigatoria.")
|
||||
|
||||
def test_sales_reports_overview_returns_bootstrap_structure_for_colaborador(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get("/admin/reports/sales/overview", headers={"Authorization": "Bearer token"})
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()
|
||||
self.assertEqual(payload["service"], "orquestrador-admin")
|
||||
self.assertEqual(payload["domain"], "sales")
|
||||
self.assertEqual(payload["mode"], "sales_contract_bootstrap")
|
||||
self.assertEqual(payload["source_dataset_keys"], ["sales_orders"])
|
||||
self.assertEqual(payload["materialization"]["sync_strategy"], "etl_incremental")
|
||||
self.assertEqual(len(payload["reports"]), 4)
|
||||
self.assertIn("average_ticket", [item["report_key"] for item in payload["reports"]])
|
||||
self.assertEqual(payload["metrics"][1]["value"], "4")
|
||||
|
||||
def test_sales_reports_catalog_returns_initial_report_definitions(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get("/admin/reports/sales/reports", headers={"Authorization": "Bearer token"})
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()
|
||||
self.assertEqual(payload["domain"], "sales")
|
||||
self.assertEqual(payload["source"], "shared_contract_catalog")
|
||||
self.assertEqual(len(payload["reports"]), 4)
|
||||
volume = next(item for item in payload["reports"] if item["report_key"] == "orders_volume")
|
||||
self.assertEqual(volume["dataset_key"], "sales_orders")
|
||||
self.assertEqual(volume["default_granularity"], "aggregate")
|
||||
self.assertEqual(volume["materialization_status"], "contract_defined_pending_snapshot_view")
|
||||
self.assertIn("status", volume["supported_dimension_fields"])
|
||||
|
||||
def test_sales_report_detail_returns_metrics_filters_and_dataset_contract(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/reports/sales/reports/cancellations_by_period",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()["report"]
|
||||
self.assertEqual(payload["report_key"], "cancellations_by_period")
|
||||
self.assertEqual(payload["dataset_key"], "sales_orders")
|
||||
self.assertEqual(payload["default_time_field"], "data_cancelamento")
|
||||
self.assertIn("cancelled_orders", [metric["key"] for metric in payload["metrics"]])
|
||||
self.assertIn("motivo_cancelamento", [item["field_name"] for item in payload["dimensions"]])
|
||||
self.assertIn("data_cancelamento", [item["field_name"] for item in payload["filters"]])
|
||||
self.assertFalse(payload["dataset"]["write_allowed"])
|
||||
self.assertIn("cpf", [field["name"] for field in payload["dataset"]["blocked_fields"]])
|
||||
|
||||
def test_sales_report_detail_returns_404_for_unknown_report(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/reports/sales/reports/orders_by_store",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 404)
|
||||
self.assertEqual(response.json()["detail"], "Relatorio de vendas nao encontrado.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@ -0,0 +1,122 @@
|
||||
import unittest
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from admin_app.api.dependencies import get_current_staff_principal
|
||||
from admin_app.app_factory import create_app
|
||||
from admin_app.core import AdminSettings, AuthenticatedStaffPrincipal
|
||||
from shared.contracts import StaffRole
|
||||
|
||||
|
||||
class AdminSystemFunctionalConfigurationWebTests(unittest.TestCase):
|
||||
def _build_client_with_role(
|
||||
self,
|
||||
role: StaffRole,
|
||||
settings: AdminSettings | None = None,
|
||||
) -> tuple[TestClient, object]:
|
||||
app = create_app(
|
||||
settings
|
||||
or AdminSettings(
|
||||
admin_auth_token_secret="test-secret",
|
||||
admin_api_prefix="/admin",
|
||||
)
|
||||
)
|
||||
app.dependency_overrides[get_current_staff_principal] = lambda: AuthenticatedStaffPrincipal(
|
||||
id=41,
|
||||
email="colaborador@empresa.com" if role == StaffRole.COLABORADOR else "diretor@empresa.com",
|
||||
display_name="Equipe de Configuracao",
|
||||
role=role,
|
||||
is_active=True,
|
||||
)
|
||||
return TestClient(app), app
|
||||
|
||||
def test_functional_configuration_catalog_requires_authentication(self):
|
||||
app = create_app(AdminSettings(admin_auth_token_secret="test-secret", admin_api_prefix="/admin"))
|
||||
client = TestClient(app)
|
||||
|
||||
response = client.get("/admin/system/configuration/functional")
|
||||
|
||||
self.assertEqual(response.status_code, 401)
|
||||
self.assertEqual(response.json()["detail"], "Autenticacao administrativa obrigatoria.")
|
||||
|
||||
def test_colaborador_can_consult_functional_configuration_catalog(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/system/configuration/functional",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()
|
||||
self.assertEqual(payload["mode"], "shared_contract_bootstrap")
|
||||
self.assertEqual(len(payload["configurations"]), 6)
|
||||
self.assertIn(
|
||||
"allowed_model_catalog",
|
||||
[item["config_key"] for item in payload["configurations"]],
|
||||
)
|
||||
self.assertIn(
|
||||
"published_runtime_state",
|
||||
[item["config_key"] for item in payload["configurations"]],
|
||||
)
|
||||
self.assertIn("atendimento_runtime_profile", payload["bot_governed_parent_config_keys"])
|
||||
self.assertNotIn("tool_generation_runtime_profile", payload["bot_governed_parent_config_keys"])
|
||||
|
||||
def test_colaborador_can_consult_bot_governed_configuration_route(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/system/configuration/functional/bot-governance",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()
|
||||
self.assertEqual(len(payload["settings"]), 15)
|
||||
self.assertIn("bot_behavior_policy", payload["parent_config_keys"])
|
||||
self.assertIn("channel_operation_policy", payload["parent_config_keys"])
|
||||
self.assertNotIn("tool_generation_runtime_profile", payload["parent_config_keys"])
|
||||
|
||||
def test_functional_configuration_detail_links_runtime_and_bot_governance(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/system/configuration/functional/atendimento_runtime_profile",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()
|
||||
self.assertEqual(payload["configuration"]["config_key"], "atendimento_runtime_profile")
|
||||
self.assertTrue(payload["managed_by_bot_governance"])
|
||||
self.assertEqual(payload["related_runtime_profile"]["runtime_target"], "atendimento")
|
||||
self.assertIn(
|
||||
"bot_tool_policy_ref",
|
||||
[item["setting_key"] for item in payload["linked_bot_settings"]],
|
||||
)
|
||||
|
||||
def test_functional_configuration_detail_returns_404_for_unknown_key(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/system/configuration/functional/segredos_infra",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 404)
|
||||
self.assertEqual(
|
||||
response.json()["detail"],
|
||||
"Configuracao funcional do sistema nao encontrada.",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@ -0,0 +1,117 @@
|
||||
import unittest
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from admin_app.api.dependencies import get_current_staff_principal
|
||||
from admin_app.app_factory import create_app
|
||||
from admin_app.core import AdminSettings, AuthenticatedStaffPrincipal
|
||||
from shared.contracts import StaffRole
|
||||
|
||||
|
||||
class AdminSystemModelRuntimeConfigurationWebTests(unittest.TestCase):
|
||||
def _build_client_with_role(
|
||||
self,
|
||||
role: StaffRole,
|
||||
settings: AdminSettings | None = None,
|
||||
) -> tuple[TestClient, object]:
|
||||
app = create_app(
|
||||
settings
|
||||
or AdminSettings(
|
||||
admin_auth_token_secret="test-secret",
|
||||
admin_api_prefix="/admin",
|
||||
admin_environment="development",
|
||||
admin_debug=True,
|
||||
)
|
||||
)
|
||||
app.dependency_overrides[get_current_staff_principal] = lambda: AuthenticatedStaffPrincipal(
|
||||
id=99,
|
||||
email="diretor@empresa.com" if role == StaffRole.DIRETOR else "colaborador@empresa.com",
|
||||
display_name="Equipe Interna",
|
||||
role=role,
|
||||
is_active=True,
|
||||
)
|
||||
return TestClient(app), app
|
||||
|
||||
def test_model_runtime_route_requires_manage_settings_permission(self):
|
||||
client, app = self._build_client_with_role(StaffRole.COLABORADOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/system/configuration/model-runtimes",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 403)
|
||||
self.assertEqual(
|
||||
response.json()["detail"],
|
||||
"Permissao administrativa insuficiente: 'manage_settings'.",
|
||||
)
|
||||
|
||||
def test_model_runtime_route_exposes_separated_atendimento_and_generation_profiles(self):
|
||||
client, app = self._build_client_with_role(StaffRole.DIRETOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/system/configuration/model-runtimes",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()["model_runtimes"]
|
||||
self.assertEqual(len(payload["runtime_profiles"]), 2)
|
||||
self.assertEqual(
|
||||
{profile["config_key"] for profile in payload["runtime_profiles"]},
|
||||
{"atendimento_runtime_profile", "tool_generation_runtime_profile"},
|
||||
)
|
||||
self.assertEqual(
|
||||
{profile["consumed_by_service"] for profile in payload["runtime_profiles"]},
|
||||
{"product", "admin"},
|
||||
)
|
||||
self.assertIn("no_implicit_propagation", payload["separation_rules"])
|
||||
self.assertEqual(
|
||||
payload["atendimento_runtime_configuration"]["config_key"],
|
||||
"atendimento_runtime_profile",
|
||||
)
|
||||
self.assertEqual(
|
||||
payload["tool_generation_runtime_configuration"]["config_key"],
|
||||
"tool_generation_runtime_profile",
|
||||
)
|
||||
self.assertIn("prompt_profile_ref", [
|
||||
field["name"] for field in payload["atendimento_runtime_configuration"]["fields"]
|
||||
])
|
||||
self.assertIn("reasoning_profile", [
|
||||
field["name"] for field in payload["tool_generation_runtime_configuration"]["fields"]
|
||||
])
|
||||
self.assertIn("atendimento_runtime_profile", payload["bot_governed_parent_config_keys"])
|
||||
self.assertNotIn("tool_generation_runtime_profile", payload["bot_governed_parent_config_keys"])
|
||||
|
||||
def test_configuration_overview_includes_model_runtime_separation_snapshot(self):
|
||||
client, app = self._build_client_with_role(StaffRole.DIRETOR)
|
||||
try:
|
||||
response = client.get(
|
||||
"/admin/system/configuration",
|
||||
headers={"Authorization": "Bearer token"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()
|
||||
self.assertEqual(
|
||||
payload["model_runtimes"]["atendimento_runtime_configuration"]["config_key"],
|
||||
"atendimento_runtime_profile",
|
||||
)
|
||||
self.assertEqual(
|
||||
payload["model_runtimes"]["tool_generation_runtime_configuration"]["config_key"],
|
||||
"tool_generation_runtime_profile",
|
||||
)
|
||||
self.assertIn(
|
||||
"model_runtime_separation",
|
||||
[item["key"] for item in payload["sources"]],
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@ -0,0 +1,60 @@
|
||||
import unittest
|
||||
|
||||
from admin_app.db.write_governance import (
|
||||
AdminWriteGovernanceViolation,
|
||||
build_admin_write_governance_payload,
|
||||
ensure_direct_admin_write_allowed,
|
||||
enforce_admin_session_write_governance,
|
||||
)
|
||||
|
||||
|
||||
class _FakeTabledObject:
|
||||
def __init__(self, table_name: str):
|
||||
self.__tablename__ = table_name
|
||||
|
||||
|
||||
class AdminWriteGovernanceTests(unittest.TestCase):
|
||||
def test_payload_exposes_internal_allowlist_and_governed_targets(self):
|
||||
payload = build_admin_write_governance_payload()
|
||||
|
||||
self.assertEqual(payload["mode"], "admin_internal_tables_only")
|
||||
self.assertEqual(
|
||||
payload["allowed_direct_write_tables"],
|
||||
["admin_audit_logs", "staff_accounts", "staff_sessions"],
|
||||
)
|
||||
self.assertIn("sales_orders", payload["blocked_operational_dataset_keys"])
|
||||
self.assertIn("orders", payload["blocked_product_source_tables"])
|
||||
self.assertIn("conversation_turns", payload["blocked_product_source_tables"])
|
||||
self.assertIn("atendimento_runtime_profile", payload["governed_configuration_keys"])
|
||||
self.assertIn("bot_behavior_policy", payload["governed_configuration_keys"])
|
||||
|
||||
def test_internal_admin_tables_are_allowed_for_direct_write(self):
|
||||
ensure_direct_admin_write_allowed("staff_accounts")
|
||||
ensure_direct_admin_write_allowed("staff_sessions")
|
||||
ensure_direct_admin_write_allowed("admin_audit_logs")
|
||||
|
||||
def test_unknown_or_product_tables_raise_governance_violation(self):
|
||||
with self.assertRaises(AdminWriteGovernanceViolation):
|
||||
ensure_direct_admin_write_allowed("orders")
|
||||
|
||||
with self.assertRaises(AdminWriteGovernanceViolation):
|
||||
ensure_direct_admin_write_allowed("conversation_turns")
|
||||
|
||||
def test_session_guard_accepts_only_internal_admin_tables(self):
|
||||
enforce_admin_session_write_governance(
|
||||
new=(_FakeTabledObject("staff_accounts"),),
|
||||
dirty=(_FakeTabledObject("staff_sessions"),),
|
||||
deleted=(_FakeTabledObject("admin_audit_logs"),),
|
||||
)
|
||||
|
||||
def test_session_guard_blocks_direct_operational_write_attempt(self):
|
||||
with self.assertRaises(AdminWriteGovernanceViolation) as context:
|
||||
enforce_admin_session_write_governance(
|
||||
new=(_FakeTabledObject("orders"),),
|
||||
)
|
||||
|
||||
self.assertIn("fluxo governado, versionado e auditavel", str(context.exception))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Reference in New Issue