import unittest from datetime import datetime, timezone from fastapi.testclient import TestClient from admin_app.api.dependencies import get_current_staff_principal, get_tool_management_service from admin_app.app_factory import create_app from admin_app.core import AdminSettings, AuthenticatedStaffPrincipal from admin_app.db.models import ToolArtifact, ToolDraft, ToolMetadata, ToolVersion from admin_app.db.models.tool_artifact import ToolArtifactKind from admin_app.services import ToolManagementService from shared.contracts import ( GENERATED_TOOL_ENTRYPOINT, StaffRole, ToolLifecycleStatus, build_generated_tool_module_name, ) class _FakeToolDraftRepository: def __init__(self): self.drafts: list[ToolDraft] = [] self.next_id = 1 def list_drafts(self, *, statuses=None) -> list[ToolDraft]: drafts = sorted( self.drafts, key=lambda draft: draft.updated_at or draft.created_at or datetime.min.replace(tzinfo=timezone.utc), reverse=True, ) if statuses: allowed = set(statuses) drafts = [draft for draft in drafts if draft.status in allowed] return drafts def get_by_tool_name(self, tool_name: str) -> ToolDraft | None: normalized = str(tool_name or "").strip().lower() for draft in self.drafts: if draft.tool_name == normalized: return draft return None def create(self, **kwargs) -> ToolDraft: now = datetime(2026, 3, 31, 16, 0, tzinfo=timezone.utc) draft = ToolDraft( id=self.next_id, draft_id=f"draft_api_{self.next_id}", created_at=now, updated_at=now, status=ToolLifecycleStatus.DRAFT, **kwargs, ) self.next_id += 1 self.drafts.append(draft) return draft def update_submission(self, draft: ToolDraft, **kwargs) -> ToolDraft: draft.display_name = kwargs["display_name"] draft.domain = kwargs["domain"] draft.description = kwargs["description"] draft.business_goal = kwargs["business_goal"] draft.summary = kwargs["summary"] draft.parameters_json = kwargs["parameters_json"] draft.required_parameter_count = kwargs["required_parameter_count"] draft.current_version_number = kwargs["current_version_number"] draft.version_count = kwargs["version_count"] draft.requires_director_approval = kwargs["requires_director_approval"] draft.owner_staff_account_id = kwargs["owner_staff_account_id"] draft.owner_display_name = kwargs["owner_display_name"] draft.updated_at = datetime(2026, 3, 31, 16, draft.current_version_number, tzinfo=timezone.utc) return draft def update_status(self, draft: ToolDraft, *, status: ToolLifecycleStatus, commit: bool = True) -> ToolDraft: draft.status = status draft.updated_at = datetime(2026, 3, 31, 16, draft.current_version_number, 30, tzinfo=timezone.utc) return draft class _FakeToolVersionRepository: def __init__(self): self.versions: list[ToolVersion] = [] self.next_id = 1 def list_versions(self, *, tool_name=None, draft_id=None, statuses=None) -> list[ToolVersion]: versions = sorted( self.versions, key=lambda version: ( version.version_number, version.updated_at or version.created_at or datetime.min.replace(tzinfo=timezone.utc), ), reverse=True, ) if tool_name: normalized = str(tool_name).strip().lower() versions = [version for version in versions if version.tool_name == normalized] if draft_id is not None: versions = [version for version in versions if version.draft_id == draft_id] if statuses: allowed = set(statuses) versions = [version for version in versions if version.status in allowed] return versions def get_next_version_number(self, tool_name: str) -> int: versions = self.list_versions(tool_name=tool_name) return (versions[0].version_number if versions else 0) + 1 def get_by_version_id(self, version_id: str) -> ToolVersion | None: normalized = str(version_id or "").strip().lower() for version in self.versions: if version.version_id == normalized: return version return None def create(self, **kwargs) -> ToolVersion: version_number = kwargs["version_number"] now = datetime(2026, 3, 31, 17, version_number, tzinfo=timezone.utc) version = ToolVersion( id=self.next_id, version_id=self.build_version_id(kwargs["tool_name"], version_number), created_at=now, updated_at=now, **kwargs, ) self.next_id += 1 self.versions.append(version) return version def update_status(self, version: ToolVersion, *, status: ToolLifecycleStatus, commit: bool = True) -> ToolVersion: version.status = status version.updated_at = datetime(2026, 3, 31, 17, version.version_number, 30, tzinfo=timezone.utc) return version @staticmethod def build_version_id(tool_name: str, version_number: int) -> str: normalized = str(tool_name or "").strip().lower() return f"tool_version::{normalized}::v{int(version_number)}" class _FakeToolMetadataRepository: def __init__(self): self.metadata_entries: list[ToolMetadata] = [] self.next_id = 1 def list_metadata(self, *, tool_name=None, statuses=None) -> list[ToolMetadata]: metadata_entries = sorted( self.metadata_entries, key=lambda metadata: ( metadata.version_number, metadata.updated_at or metadata.created_at or datetime.min.replace(tzinfo=timezone.utc), ), reverse=True, ) if tool_name: normalized = str(tool_name).strip().lower() metadata_entries = [metadata for metadata in metadata_entries if metadata.tool_name == normalized] if statuses: allowed = set(statuses) metadata_entries = [metadata for metadata in metadata_entries if metadata.status in allowed] return metadata_entries def get_by_tool_version_id(self, tool_version_id: int) -> ToolMetadata | None: for metadata in self.metadata_entries: if metadata.tool_version_id == tool_version_id: return metadata return None def create(self, **kwargs) -> ToolMetadata: version_number = kwargs["version_number"] now = datetime(2026, 3, 31, 18, version_number, tzinfo=timezone.utc) metadata = ToolMetadata( id=self.next_id, metadata_id=self.build_metadata_id(kwargs["tool_name"], version_number), created_at=now, updated_at=now, **kwargs, ) self.next_id += 1 self.metadata_entries.append(metadata) return metadata def update_metadata(self, metadata: ToolMetadata, **kwargs) -> ToolMetadata: metadata.display_name = kwargs["display_name"] metadata.domain = kwargs["domain"] metadata.description = kwargs["description"] metadata.parameters_json = kwargs["parameters_json"] metadata.status = kwargs["status"] metadata.author_staff_account_id = kwargs["author_staff_account_id"] metadata.author_display_name = kwargs["author_display_name"] metadata.updated_at = datetime(2026, 3, 31, 18, metadata.version_number, tzinfo=timezone.utc) return metadata def update_status(self, metadata: ToolMetadata, *, status: ToolLifecycleStatus, commit: bool = True) -> ToolMetadata: metadata.status = status metadata.updated_at = datetime(2026, 3, 31, 18, metadata.version_number, 30, tzinfo=timezone.utc) return metadata def upsert_version_metadata(self, **kwargs) -> ToolMetadata: existing = self.get_by_tool_version_id(kwargs["tool_version_id"]) if existing is None: return self.create(**kwargs) return self.update_metadata(existing, **kwargs) @staticmethod def build_metadata_id(tool_name: str, version_number: int) -> str: normalized = str(tool_name or "").strip().lower() return f"tool_metadata::{normalized}::v{int(version_number)}" class _FakeToolArtifactRepository: def __init__(self): self.artifacts: list[ToolArtifact] = [] self.next_id = 1 def list_artifacts(self, *, tool_name=None, tool_version_id=None, artifact_stage=None, artifact_kind=None) -> list[ToolArtifact]: artifacts = sorted( self.artifacts, key=lambda artifact: ( artifact.version_number, artifact.updated_at or artifact.created_at or datetime.min.replace(tzinfo=timezone.utc), ), reverse=True, ) if tool_name: normalized = str(tool_name).strip().lower() artifacts = [artifact for artifact in artifacts if artifact.tool_name == normalized] if tool_version_id is not None: artifacts = [artifact for artifact in artifacts if artifact.tool_version_id == tool_version_id] if artifact_stage: artifacts = [artifact for artifact in artifacts if artifact.artifact_stage == artifact_stage] if artifact_kind: artifacts = [artifact for artifact in artifacts if artifact.artifact_kind == artifact_kind] return artifacts def get_by_tool_version_and_kind(self, tool_version_id: int, artifact_kind) -> ToolArtifact | None: for artifact in self.artifacts: if artifact.tool_version_id == tool_version_id and artifact.artifact_kind == artifact_kind: return artifact return None def create(self, **kwargs) -> ToolArtifact: version_number = kwargs["version_number"] now = datetime(2026, 3, 31, 19, version_number, self.next_id, tzinfo=timezone.utc) artifact = ToolArtifact( id=self.next_id, artifact_id=self.build_artifact_id(kwargs["tool_name"], version_number, kwargs["artifact_kind"]), created_at=now, updated_at=now, checksum=kwargs.get("checksum") or f"fake-checksum-{self.next_id}", **kwargs, ) self.next_id += 1 self.artifacts.append(artifact) return artifact def update_artifact(self, artifact: ToolArtifact, **kwargs) -> ToolArtifact: artifact.artifact_status = kwargs["artifact_status"] artifact.storage_kind = kwargs.get("storage_kind", artifact.storage_kind) artifact.summary = kwargs["summary"] artifact.payload_json = kwargs["payload_json"] artifact.checksum = kwargs.get("checksum", artifact.checksum) artifact.author_staff_account_id = kwargs["author_staff_account_id"] artifact.author_display_name = kwargs["author_display_name"] artifact.updated_at = datetime(2026, 3, 31, 19, artifact.version_number, tzinfo=timezone.utc) return artifact def upsert_version_artifact(self, **kwargs) -> ToolArtifact: existing = self.get_by_tool_version_and_kind(kwargs["tool_version_id"], kwargs["artifact_kind"]) if existing is None: return self.create(**kwargs) return self.update_artifact(existing, **kwargs) @staticmethod def build_artifact_id(tool_name: str, version_number: int, artifact_kind) -> str: normalized = str(tool_name or "").strip().lower() return f"tool_artifact::{normalized}::v{int(version_number)}::{artifact_kind.value}" class AdminToolsWebTests(unittest.TestCase): def _build_client_with_role( self, role: StaffRole, settings: AdminSettings | None = None, ) -> tuple[ TestClient, object, _FakeToolDraftRepository, _FakeToolVersionRepository, _FakeToolMetadataRepository, _FakeToolArtifactRepository, ]: app = create_app( settings or AdminSettings( admin_auth_token_secret="test-secret", admin_api_prefix="/admin", ) ) draft_repository = _FakeToolDraftRepository() version_repository = _FakeToolVersionRepository() metadata_repository = _FakeToolMetadataRepository() artifact_repository = _FakeToolArtifactRepository() service = ToolManagementService( settings=app.state.admin_settings, draft_repository=draft_repository, version_repository=version_repository, metadata_repository=metadata_repository, artifact_repository=artifact_repository, ) app.dependency_overrides[get_current_staff_principal] = lambda: AuthenticatedStaffPrincipal( id=11, email="colaborador@empresa.com" if role == StaffRole.COLABORADOR else "diretor@empresa.com", display_name="Equipe de Tools", role=role, is_active=True, ) app.dependency_overrides[get_tool_management_service] = lambda: service return TestClient(app), app, draft_repository, version_repository, metadata_repository, artifact_repository def test_tools_overview_returns_metrics_workflow_and_actions_for_colaborador(self): client, app, _, _, _, _ = self._build_client_with_role(StaffRole.COLABORADOR) try: response = client.get("/admin/tools/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"], "admin_tool_draft_governance") self.assertEqual(payload["metrics"][0]["value"], "18") self.assertIn("persisted_versions", [item["key"] for item in payload["metrics"]]) self.assertIn("persisted_metadata", [item["key"] for item in payload["metrics"]]) self.assertIn("persisted_artifacts", [item["key"] for item in payload["metrics"]]) self.assertIn("active", [item["code"] for item in payload["workflow"]]) self.assertIn("/admin/tools/contracts", [item["href"] for item in payload["actions"]]) self.assertIn("/admin/tools/drafts/intake", [item["href"] for item in payload["actions"]]) self.assertNotIn("/admin/tools/review-queue", [item["href"] for item in payload["actions"]]) self.assertNotIn("/admin/tools/publications", [item["href"] for item in payload["actions"]]) self.assertIn("pipeline de geracao", payload["next_steps"][0].lower()) def test_tools_contracts_return_shared_contract_snapshot(self): client, app, _, _, _, _ = self._build_client_with_role(StaffRole.COLABORADOR) try: response = client.get("/admin/tools/contracts", headers={"Authorization": "Bearer token"}) finally: app.dependency_overrides.clear() self.assertEqual(response.status_code, 200) payload = response.json() self.assertEqual(payload["publication_source_service"], "admin") self.assertEqual(payload["publication_target_service"], "product") self.assertIn("draft", [item["code"] for item in payload["lifecycle_statuses"]]) self.assertEqual(payload["lifecycle_statuses"][0]["order"], 1) self.assertFalse(payload["lifecycle_statuses"][0]["terminal"]) self.assertTrue(payload["lifecycle_statuses"][-1]["terminal"]) self.assertIn("string", [item["code"] for item in payload["parameter_types"]]) self.assertIn("published_tool", payload["publication_fields"]) def test_tools_draft_intake_blocks_tool_name_reserved_by_core_catalog(self): client, app, draft_repository, version_repository, metadata_repository, artifact_repository = self._build_client_with_role(StaffRole.COLABORADOR) try: response = client.post( "/admin/tools/drafts/intake", headers={"Authorization": "Bearer token"}, json={ "domain": "vendas", "tool_name": "consultar_estoque", "display_name": "Consultar estoque paralelo", "description": "Tentativa de sobrescrever a tool core de estoque com uma versao administrativa.", "business_goal": "Validar que a API bloqueia nomes reservados do runtime core.", "parameters": [], }, ) finally: app.dependency_overrides.clear() self.assertEqual(response.status_code, 422) self.assertIn("catalogo core do sistema", response.json()["detail"]) self.assertEqual(len(draft_repository.drafts), 0) self.assertEqual(len(version_repository.versions), 0) self.assertEqual(len(metadata_repository.metadata_entries), 0) self.assertEqual(len(artifact_repository.artifacts), 0) def test_tools_drafts_return_single_root_draft_with_current_version_after_reintake(self): client, app, _, _, _, _ = self._build_client_with_role(StaffRole.COLABORADOR) try: first_response = client.post( "/admin/tools/drafts/intake", headers={"Authorization": "Bearer token"}, json={ "domain": "vendas", "tool_name": "consultar_resumo_financeiro", "display_name": "Consultar resumo financeiro", "description": "Consulta o resumo financeiro consolidado para analise do time administrativo.", "business_goal": "Ajudar a equipe interna a priorizar a leitura dos principais indicadores financeiros.", "parameters": [], }, ) second_response = client.post( "/admin/tools/drafts/intake", headers={"Authorization": "Bearer token"}, json={ "domain": "vendas", "tool_name": "consultar_resumo_financeiro", "display_name": "Consultar resumo financeiro", "description": "Consulta o resumo financeiro consolidado com detalhamento adicional para analise administrativa.", "business_goal": "Ajudar a equipe interna a priorizar indicadores financeiros com contexto extra e leitura mais acionavel.", "parameters": [], }, ) response = client.get("/admin/tools/drafts", headers={"Authorization": "Bearer token"}) finally: app.dependency_overrides.clear() self.assertEqual(first_response.status_code, 200) self.assertEqual(second_response.status_code, 200) self.assertEqual(response.status_code, 200) payload = response.json() self.assertEqual(payload["storage_status"], "admin_database") self.assertEqual(len(payload["drafts"]), 1) self.assertEqual(payload["drafts"][0]["tool_name"], "consultar_resumo_financeiro") self.assertEqual(payload["drafts"][0]["current_version_number"], 2) self.assertEqual(payload["drafts"][0]["version_count"], 2) self.assertEqual(payload["supported_statuses"], ["draft"]) def test_tools_draft_intake_persists_admin_draft_with_version_metadata_and_artifacts(self): client, app, draft_repository, version_repository, metadata_repository, artifact_repository = self._build_client_with_role(StaffRole.COLABORADOR) try: response = client.post( "/admin/tools/drafts/intake", headers={"Authorization": "Bearer token"}, json={ "domain": "orquestracao", "tool_name": "priorizar_contato_quente", "display_name": "Priorizar contato quente", "description": "Classifica contatos mais quentes para orientar o proximo passo do atendimento.", "business_goal": "Dar mais foco comercial ao time interno ao identificar oportunidades mais urgentes.", "parameters": [ { "name": "score_interesse", "parameter_type": "number", "description": "Pontuacao atual de interesse do lead.", "required": True, } ], }, ) finally: app.dependency_overrides.clear() self.assertEqual(response.status_code, 200) payload = response.json() self.assertEqual(payload["storage_status"], "admin_database") self.assertEqual(payload["submission_policy"]["mode"], "draft_only") self.assertEqual(payload["submission_policy"]["submitter_role"], "colaborador") self.assertFalse(payload["submission_policy"]["submitter_can_publish_now"]) self.assertFalse(payload["submission_policy"]["submitter_can_authorize_generation_now"]) self.assertTrue(payload["submission_policy"]["direct_publication_blocked"]) self.assertTrue(payload["submission_policy"]["requires_generation_authorization"]) self.assertEqual(payload["submission_policy"]["required_approver_role"], "diretor") self.assertEqual(payload["submission_policy"]["required_generation_permission"], "review_tool_generations") self.assertEqual(payload["submission_policy"]["required_publish_permission"], "publish_tools") self.assertEqual(payload["draft_preview"]["status"], "draft") self.assertEqual(payload["draft_preview"]["domain"], "orquestracao") self.assertEqual(payload["draft_preview"]["version_id"], "tool_version::priorizar_contato_quente::v1") self.assertEqual(payload["draft_preview"]["version_number"], 1) self.assertEqual(payload["draft_preview"]["version_count"], 1) self.assertTrue(payload["draft_preview"]["requires_director_approval"]) self.assertEqual(payload["draft_preview"]["owner_name"], "Equipe de Tools") self.assertGreaterEqual(len(payload["warnings"]), 1) self.assertEqual(len(draft_repository.drafts), 1) self.assertEqual(len(version_repository.versions), 1) self.assertEqual(len(metadata_repository.metadata_entries), 1) self.assertEqual(len(artifact_repository.artifacts), 2) artifact_kinds = {artifact.artifact_kind for artifact in artifact_repository.artifacts} self.assertEqual(artifact_kinds, {ToolArtifactKind.GENERATION_REQUEST, ToolArtifactKind.VALIDATION_REPORT}) def test_tools_review_queue_requires_director_review_permission(self): client, app, _, _, _, _ = self._build_client_with_role(StaffRole.COLABORADOR) try: response = client.get("/admin/tools/review-queue", headers={"Authorization": "Bearer token"}) finally: app.dependency_overrides.clear() self.assertEqual(response.status_code, 403) self.assertEqual( response.json()["detail"], "Permissao administrativa insuficiente: 'review_tool_generations'.", ) def test_tools_review_action_requires_director_review_permission(self): client, app, _, _, _, _ = self._build_client_with_role(StaffRole.COLABORADOR) try: intake_response = client.post( "/admin/tools/drafts/intake", headers={"Authorization": "Bearer token"}, json={ "domain": "revisao", "tool_name": "consultar_revisao_aberta", "display_name": "Consultar revisao aberta", "description": "Consulta revisoes abertas com filtros administrativos para a oficina.", "business_goal": "Ajudar o time a localizar revisoes abertas com mais contexto operacional.", "parameters": [], }, ) version_id = intake_response.json()["draft_preview"]["version_id"] response = client.post( f"/admin/tools/review-queue/{version_id}/review", headers={"Authorization": "Bearer token"}, json={ "decision_notes": "Parecer inicial da diretoria para a revisao humana.", "reviewed_generated_code": True, }, ) finally: app.dependency_overrides.clear() self.assertEqual(intake_response.status_code, 200) self.assertEqual(response.status_code, 403) self.assertEqual( response.json()["detail"], "Permissao administrativa insuficiente: 'review_tool_generations'.", ) def test_tools_review_queue_is_available_for_diretor(self): client, app, _, _, _, _ = self._build_client_with_role(StaffRole.DIRETOR) try: intake_response = client.post( "/admin/tools/drafts/intake", headers={"Authorization": "Bearer token"}, json={ "domain": "revisao", "tool_name": "consultar_revisao_aberta", "display_name": "Consultar revisao aberta", "description": "Consulta revisoes abertas com filtros administrativos para a oficina.", "business_goal": "Ajudar o time a localizar revisoes abertas com mais contexto operacional.", "parameters": [], }, ) response = client.get("/admin/tools/review-queue", headers={"Authorization": "Bearer token"}) finally: app.dependency_overrides.clear() self.assertEqual(intake_response.status_code, 200) self.assertEqual(response.status_code, 200) payload = response.json() self.assertEqual(payload["queue_mode"], "governed_admin_queue") self.assertEqual(len(payload["items"]), 1) self.assertEqual(payload["items"][0]["status"], "draft") self.assertEqual(payload["items"][0]["gate"], "generation_pipeline_required") self.assertEqual(payload["items"][0]["version_number"], 1) self.assertIn("approved", payload["supported_statuses"]) def test_tools_review_detail_returns_generated_source_for_diretor(self): client, app, _, _, _, _ = self._build_client_with_role(StaffRole.DIRETOR) try: intake_response = client.post( "/admin/tools/drafts/intake", headers={"Authorization": "Bearer token"}, json={ "domain": "revisao", "tool_name": "consultar_revisao_aberta", "display_name": "Consultar revisao aberta", "description": "Consulta revisoes abertas com filtros administrativos para a oficina.", "business_goal": "Ajudar o time a localizar revisoes abertas com mais contexto operacional.", "parameters": [ { "name": "placa", "parameter_type": "string", "description": "Placa usada na busca da revisao.", "required": True, } ], }, ) version_id = intake_response.json()["draft_preview"]["version_id"] pipeline_response = client.post( f"/admin/tools/pipeline/{version_id}/run", headers={"Authorization": "Bearer token"}, ) response = client.get( f"/admin/tools/review-queue/{version_id}", headers={"Authorization": "Bearer token"}, ) finally: app.dependency_overrides.clear() self.assertEqual(intake_response.status_code, 200) self.assertEqual(pipeline_response.status_code, 200) self.assertEqual(response.status_code, 200) payload = response.json() self.assertEqual(payload["tool_name"], "consultar_revisao_aberta") self.assertTrue(payload["human_gate"]["review_action_available"]) self.assertFalse(payload["human_gate"]["run_pipeline_action_available"]) self.assertEqual(payload["generation_context"]["latest_generation_iteration"], 1) self.assertEqual(payload["generation_context"]["generation_iterations_count"], 1) self.assertEqual(payload["generation_context"]["latest_generation_mode"], "initial_generation") self.assertIn("async def run", payload["generated_source_code"]) self.assertEqual(len(payload["automated_validations"]), 4) def test_tools_collaborator_cannot_run_generation_pipeline_before_director_authorization(self): client, app, _, _, _, _ = self._build_client_with_role(StaffRole.COLABORADOR) try: intake_response = client.post( "/admin/tools/drafts/intake", headers={"Authorization": "Bearer token"}, json={ "domain": "revisao", "tool_name": "consultar_revisao_aberta", "display_name": "Consultar revisao aberta", "description": "Consulta revisoes abertas com filtros administrativos para a oficina.", "business_goal": "Ajudar o time a localizar revisoes abertas com mais contexto operacional.", "parameters": [], }, ) version_id = intake_response.json()["draft_preview"]["version_id"] response = client.post( f"/admin/tools/pipeline/{version_id}/run", headers={"Authorization": "Bearer token"}, ) finally: app.dependency_overrides.clear() self.assertEqual(intake_response.status_code, 200) self.assertEqual(response.status_code, 409) self.assertIn("autorizacao de diretor", response.json()["detail"].lower()) def test_tools_publications_require_director_publication_permission(self): client, app, _, _, _, _ = self._build_client_with_role(StaffRole.COLABORADOR) try: response = client.get("/admin/tools/publications", headers={"Authorization": "Bearer token"}) finally: app.dependency_overrides.clear() self.assertEqual(response.status_code, 403) self.assertEqual( response.json()["detail"], "Permissao administrativa insuficiente: 'publish_tools'.", ) def test_tools_publish_action_requires_director_publication_permission(self): client, app, _, _, _, _ = self._build_client_with_role(StaffRole.COLABORADOR) try: intake_response = client.post( "/admin/tools/drafts/intake", headers={"Authorization": "Bearer token"}, json={ "domain": "revisao", "tool_name": "consultar_revisao_aberta", "display_name": "Consultar revisao aberta", "description": "Consulta revisoes abertas com filtros administrativos para a oficina.", "business_goal": "Ajudar o time a localizar revisoes abertas com mais contexto operacional.", "parameters": [], }, ) version_id = intake_response.json()["draft_preview"]["version_id"] response = client.post( f"/admin/tools/publications/{version_id}/publish", headers={"Authorization": "Bearer token"}, ) finally: app.dependency_overrides.clear() self.assertEqual(intake_response.status_code, 200) self.assertEqual(response.status_code, 403) self.assertEqual( response.json()["detail"], "Permissao administrativa insuficiente: 'publish_tools'.", ) def test_tools_publications_return_bootstrap_catalog_for_diretor(self): client, app, _, _, _, _ = self._build_client_with_role(StaffRole.DIRETOR) try: response = client.get("/admin/tools/publications", headers={"Authorization": "Bearer token"}) finally: app.dependency_overrides.clear() self.assertEqual(response.status_code, 200) payload = response.json() self.assertEqual(payload["source"], "bootstrap_catalog") self.assertEqual(payload["target_service"], "product") self.assertGreaterEqual(len(payload["publications"]), 10) self.assertIn("consultar_estoque", [item["tool_name"] for item in payload["publications"]]) first = payload["publications"][0] self.assertEqual(first["status"], "active") self.assertEqual(first["implementation_module"], "app.services.tools.handlers") def test_tools_publications_keep_bootstrap_catalog_after_intake(self): client, app, _, _, _, _ = self._build_client_with_role(StaffRole.DIRETOR) try: intake_response = client.post( "/admin/tools/drafts/intake", headers={"Authorization": "Bearer token"}, json={ "domain": "revisao", "tool_name": "consultar_revisao_aberta", "display_name": "Consultar revisao aberta", "description": "Consulta revisoes abertas com filtros administrativos para a oficina.", "business_goal": "Ajudar o time a localizar revisoes abertas com mais contexto operacional.", "parameters": [ { "name": "placa", "parameter_type": "string", "description": "Placa usada na busca da revisao.", "required": True, } ], }, ) response = client.get("/admin/tools/publications", headers={"Authorization": "Bearer token"}) finally: app.dependency_overrides.clear() self.assertEqual(intake_response.status_code, 200) self.assertEqual(response.status_code, 200) payload = response.json() self.assertEqual(payload["source"], "bootstrap_catalog") self.assertGreaterEqual(len(payload["publications"]), 10) self.assertNotIn("consultar_revisao_aberta", [item["tool_name"] for item in payload["publications"]]) def test_tools_director_workflow_reviews_approves_and_publishes_before_activation(self): client, app, _, _, _, _ = self._build_client_with_role(StaffRole.DIRETOR) try: intake_response = client.post( "/admin/tools/drafts/intake", headers={"Authorization": "Bearer token"}, json={ "domain": "revisao", "tool_name": "consultar_revisao_aberta", "display_name": "Consultar revisao aberta", "description": "Consulta revisoes abertas com filtros administrativos para a oficina.", "business_goal": "Ajudar o time a localizar revisoes abertas com mais contexto operacional.", "parameters": [ { "name": "placa", "parameter_type": "string", "description": "Placa usada na busca da revisao.", "required": True, } ], }, ) version_id = intake_response.json()["draft_preview"]["version_id"] publish_before_approval = client.post( f"/admin/tools/publications/{version_id}/publish", headers={"Authorization": "Bearer token"}, ) review_before_pipeline = client.post( f"/admin/tools/review-queue/{version_id}/review", headers={"Authorization": "Bearer token"}, json={ "decision_notes": "Tentativa de revisao antes da pipeline.", "reviewed_generated_code": True, }, ) pipeline_response = client.post( f"/admin/tools/pipeline/{version_id}/run", headers={"Authorization": "Bearer token"}, ) review_response = client.post( f"/admin/tools/review-queue/{version_id}/review", headers={"Authorization": "Bearer token"}, json={ "decision_notes": "Analisei o codigo completo gerado antes da validacao humana.", "reviewed_generated_code": True, }, ) approve_response = client.post( f"/admin/tools/review-queue/{version_id}/approve", headers={"Authorization": "Bearer token"}, json={ "decision_notes": "Aprovacao formal da diretoria para seguir com a publicacao.", "reviewed_generated_code": True, }, ) pre_publications = client.get("/admin/tools/publications", headers={"Authorization": "Bearer token"}) publish_response = client.post( f"/admin/tools/publications/{version_id}/publish", headers={"Authorization": "Bearer token"}, ) final_publications = client.get("/admin/tools/publications", headers={"Authorization": "Bearer token"}) finally: app.dependency_overrides.clear() self.assertEqual(intake_response.status_code, 200) self.assertEqual(publish_before_approval.status_code, 409) self.assertIn("approved", publish_before_approval.json()["detail"]) self.assertEqual(review_before_pipeline.status_code, 409) self.assertIn("generated", review_before_pipeline.json()["detail"]) self.assertEqual(pipeline_response.status_code, 200) self.assertEqual(pipeline_response.json()["status"], "generated") self.assertEqual(pipeline_response.json()["queue_entry"]["gate"], "validation_required") self.assertEqual(pipeline_response.json()["queue_entry"]["automated_validation_status"], "passed") self.assertEqual(pipeline_response.json()["queue_entry"]["automated_validation_summary"], "4/4 validacoes automaticas passaram antes da revisao humana.") self.assertEqual(len(pipeline_response.json()["automated_validations"]), 4) self.assertTrue(all(check["status"] == "passed" for check in pipeline_response.json()["automated_validations"])) self.assertEqual(review_response.status_code, 200) self.assertEqual(review_response.json()["status"], "validated") self.assertEqual(review_response.json()["queue_entry"]["gate"], "director_approval_required") self.assertEqual(approve_response.status_code, 200) self.assertEqual(approve_response.json()["status"], "approved") self.assertEqual(approve_response.json()["queue_entry"]["gate"], "director_publication_required") self.assertEqual(pre_publications.status_code, 200) self.assertEqual(pre_publications.json()["source"], "bootstrap_catalog") self.assertNotIn("consultar_revisao_aberta", [item["tool_name"] for item in pre_publications.json()["publications"]]) self.assertEqual(publish_response.status_code, 200) self.assertEqual(publish_response.json()["status"], "active") self.assertIsNone(publish_response.json()["queue_entry"]) self.assertEqual(publish_response.json()["publication"]["tool_name"], "consultar_revisao_aberta") self.assertEqual(final_publications.status_code, 200) payload = final_publications.json() self.assertEqual(payload["source"], "hybrid_runtime_catalog") self.assertGreaterEqual(len(payload["publications"]), 11) publication = next(item for item in payload["publications"] if item["tool_name"] == "consultar_revisao_aberta") self.assertEqual(publication["publication_id"], "tool_metadata::consultar_revisao_aberta::v1") self.assertEqual(publication["status"], "active") self.assertEqual(publication["parameter_count"], 1) self.assertEqual(publication["author_name"], "Equipe de Tools") self.assertEqual(publication["implementation_module"], build_generated_tool_module_name("consultar_revisao_aberta")) self.assertEqual(publication["implementation_callable"], GENERATED_TOOL_ENTRYPOINT) self.assertEqual(publication["parameters"][0]["name"], "placa") self.assertEqual(publication["parameters"][0]["parameter_type"], "string") def test_tools_director_can_deactivate_active_publication(self): client, app, _, _, _, _ = self._build_client_with_role(StaffRole.DIRETOR) try: intake_response = client.post( "/admin/tools/drafts/intake", headers={"Authorization": "Bearer token"}, json={ "domain": "revisao", "tool_name": "consultar_revisao_aberta", "display_name": "Consultar revisao aberta", "description": "Consulta revisoes abertas com filtros administrativos para a oficina.", "business_goal": "Ajudar o time a localizar revisoes abertas com mais contexto operacional.", "parameters": [], }, ) version_id = intake_response.json()["draft_preview"]["version_id"] client.post(f"/admin/tools/pipeline/{version_id}/run", headers={"Authorization": "Bearer token"}) client.post( f"/admin/tools/review-queue/{version_id}/review", headers={"Authorization": "Bearer token"}, json={"decision_notes": "Analisei o codigo completo antes da ativacao.", "reviewed_generated_code": True}, ) client.post( f"/admin/tools/review-queue/{version_id}/approve", headers={"Authorization": "Bearer token"}, json={"decision_notes": "Aprovacao formal para disponibilizar a ferramenta."}, ) client.post(f"/admin/tools/publications/{version_id}/publish", headers={"Authorization": "Bearer token"}) deactivate_response = client.post( f"/admin/tools/publications/{version_id}/deactivate", headers={"Authorization": "Bearer token"}, json={"decision_notes": "Desativacao controlada da ferramenta ativa apos teste concluido."}, ) publications_response = client.get("/admin/tools/publications", headers={"Authorization": "Bearer token"}) finally: app.dependency_overrides.clear() self.assertEqual(deactivate_response.status_code, 200) self.assertEqual(deactivate_response.json()["status"], "archived") self.assertIsNone(deactivate_response.json()["queue_entry"]) self.assertEqual(publications_response.status_code, 200) self.assertNotIn("consultar_revisao_aberta", [item["tool_name"] for item in publications_response.json()["publications"]]) def test_tools_director_can_rollback_active_publication(self): client, app, _, _, _, _ = self._build_client_with_role(StaffRole.DIRETOR) try: first_intake = client.post( "/admin/tools/drafts/intake", headers={"Authorization": "Bearer token"}, json={ "domain": "revisao", "tool_name": "consultar_revisao_aberta", "display_name": "Consultar revisao aberta", "description": "Consulta revisoes abertas com filtros administrativos para a oficina.", "business_goal": "Ajudar o time a localizar revisoes abertas com mais contexto operacional.", "parameters": [], }, ) first_version_id = first_intake.json()["draft_preview"]["version_id"] client.post(f"/admin/tools/pipeline/{first_version_id}/run", headers={"Authorization": "Bearer token"}) client.post( f"/admin/tools/review-queue/{first_version_id}/review", headers={"Authorization": "Bearer token"}, json={"decision_notes": "Primeira revisao completa do codigo gerado.", "reviewed_generated_code": True}, ) client.post( f"/admin/tools/review-queue/{first_version_id}/approve", headers={"Authorization": "Bearer token"}, json={"decision_notes": "Primeira aprovacao formal da diretoria."}, ) client.post(f"/admin/tools/publications/{first_version_id}/publish", headers={"Authorization": "Bearer token"}) second_intake = client.post( "/admin/tools/drafts/intake", headers={"Authorization": "Bearer token"}, json={ "domain": "revisao", "tool_name": "consultar_revisao_aberta", "display_name": "Consultar revisao aberta", "description": "Consulta revisoes abertas com mais contexto operacional para a oficina.", "business_goal": "Ajudar o time a localizar revisoes abertas com filtros extras.", "parameters": [], }, ) second_version_id = second_intake.json()["draft_preview"]["version_id"] client.post(f"/admin/tools/pipeline/{second_version_id}/run", headers={"Authorization": "Bearer token"}) client.post( f"/admin/tools/review-queue/{second_version_id}/review", headers={"Authorization": "Bearer token"}, json={"decision_notes": "Segunda revisao completa do codigo gerado.", "reviewed_generated_code": True}, ) client.post( f"/admin/tools/review-queue/{second_version_id}/approve", headers={"Authorization": "Bearer token"}, json={"decision_notes": "Segunda aprovacao formal da diretoria."}, ) client.post(f"/admin/tools/publications/{second_version_id}/publish", headers={"Authorization": "Bearer token"}) rollback_response = client.post( f"/admin/tools/publications/{second_version_id}/rollback", headers={"Authorization": "Bearer token"}, json={"decision_notes": "Rollback controlado para restaurar a versao anterior estavel."}, ) publications_response = client.get("/admin/tools/publications", headers={"Authorization": "Bearer token"}) finally: app.dependency_overrides.clear() self.assertEqual(rollback_response.status_code, 200) self.assertEqual(rollback_response.json()["status"], "active") self.assertEqual(rollback_response.json()["version_id"], first_version_id) publication = next(item for item in publications_response.json()["publications"] if item["tool_name"] == "consultar_revisao_aberta") self.assertEqual(publication["version_id"], first_version_id) self.assertTrue(publication["deactivation_action_available"]) if __name__ == "__main__": unittest.main()