You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
114 lines
4.9 KiB
Python
114 lines
4.9 KiB
Python
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() |