import unittest from datetime import datetime, timezone from admin_app.core import AdminSettings from admin_app.db.models import ToolArtifact, ToolDraft, ToolMetadata, ToolVersion from admin_app.db.models.tool_artifact import ToolArtifactKind from admin_app.services.tool_management_service import ToolManagementService from shared.contracts import ( GENERATED_TOOL_ENTRYPOINT, GENERATED_TOOLS_PACKAGE, ToolLifecycleStatus, ToolParameterType, build_generated_tool_module_name, build_generated_tool_module_path, ) 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, *, tool_name: str, display_name: str, domain: str, description: str, business_goal: str, summary: str, parameters_json: list[dict], required_parameter_count: int, current_version_number: int, version_count: int, owner_staff_account_id: int, owner_display_name: str, requires_director_approval: bool = True, commit: bool = True, ) -> ToolDraft: now = datetime(2026, 3, 31, 15, 0, tzinfo=timezone.utc) draft = ToolDraft( id=self.next_id, draft_id=f"draft_fake_{self.next_id}", tool_name=tool_name, display_name=display_name, domain=domain, description=description, business_goal=business_goal, status=ToolLifecycleStatus.DRAFT, summary=summary, parameters_json=parameters_json, required_parameter_count=required_parameter_count, current_version_number=current_version_number, version_count=version_count, requires_director_approval=requires_director_approval, owner_staff_account_id=owner_staff_account_id, owner_display_name=owner_display_name, created_at=now, updated_at=now, ) self.next_id += 1 self.drafts.append(draft) return draft def update_submission( self, draft: ToolDraft, *, display_name: str, domain: str, description: str, business_goal: str, summary: str, parameters_json: list[dict], required_parameter_count: int, current_version_number: int, version_count: int, owner_staff_account_id: int, owner_display_name: str, requires_director_approval: bool = True, commit: bool = True, ) -> ToolDraft: draft.display_name = display_name draft.domain = domain draft.description = description draft.business_goal = business_goal draft.status = ToolLifecycleStatus.DRAFT draft.summary = summary draft.parameters_json = parameters_json draft.required_parameter_count = required_parameter_count draft.current_version_number = current_version_number draft.version_count = version_count draft.requires_director_approval = requires_director_approval draft.owner_staff_account_id = owner_staff_account_id draft.owner_display_name = owner_display_name draft.updated_at = datetime(2026, 3, 31, 15, current_version_number, 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 create( self, *, draft_id: int, tool_name: str, version_number: int, summary: str, description: str, business_goal: str, parameters_json: list[dict], required_parameter_count: int, owner_staff_account_id: int, owner_display_name: str, status: ToolLifecycleStatus = ToolLifecycleStatus.DRAFT, requires_director_approval: bool = True, commit: bool = True, ) -> ToolVersion: now = datetime(2026, 3, 31, 16, version_number, tzinfo=timezone.utc) version = ToolVersion( id=self.next_id, version_id=self.build_version_id(tool_name, version_number), draft_id=draft_id, tool_name=tool_name, version_number=version_number, status=status, summary=summary, description=description, business_goal=business_goal, parameters_json=parameters_json, required_parameter_count=required_parameter_count, requires_director_approval=requires_director_approval, owner_staff_account_id=owner_staff_account_id, owner_display_name=owner_display_name, created_at=now, updated_at=now, ) self.next_id += 1 self.versions.append(version) 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, 17, 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, 17, metadata.version_number, 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, 18, 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["storage_kind"] artifact.summary = kwargs["summary"] artifact.payload_json = kwargs["payload_json"] artifact.checksum = kwargs["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, 18, 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 AdminToolManagementServiceTests(unittest.TestCase): def setUp(self): self.draft_repository = _FakeToolDraftRepository() self.version_repository = _FakeToolVersionRepository() self.metadata_repository = _FakeToolMetadataRepository() self.artifact_repository = _FakeToolArtifactRepository() self.service = ToolManagementService( settings=AdminSettings(admin_api_prefix="/admin"), draft_repository=self.draft_repository, version_repository=self.version_repository, metadata_repository=self.metadata_repository, artifact_repository=self.artifact_repository, ) def test_create_draft_submission_persists_initial_tool_version_metadata_and_artifacts(self): payload = self.service.create_draft_submission( { "domain": "vendas", "tool_name": "consultar_vendas_periodo", "display_name": "Consultar vendas por periodo", "description": "Consulta vendas consolidadas por periodo informado no painel.", "business_goal": "Ajudar o time interno a acompanhar o desempenho comercial com mais agilidade.", "parameters": [ { "name": "periodo_inicio", "parameter_type": "string", "description": "Data inicial usada no filtro.", "required": True, }, { "name": "periodo_fim", "parameter_type": "string", "description": "Data final usada no filtro.", "required": True, }, ], }, owner_staff_account_id=7, owner_name="Equipe Interna", ) self.assertEqual(payload["storage_status"], "admin_database") self.assertEqual(payload["draft_preview"]["draft_id"], "draft_fake_1") self.assertEqual(payload["draft_preview"]["version_id"], "tool_version::consultar_vendas_periodo::v1") self.assertEqual(payload["draft_preview"]["version_number"], 1) self.assertEqual(payload["draft_preview"]["version_count"], 1) self.assertEqual(payload["draft_preview"]["status"], ToolLifecycleStatus.DRAFT) self.assertEqual(payload["draft_preview"]["owner_name"], "Equipe Interna") self.assertEqual(len(self.draft_repository.drafts), 1) self.assertEqual(len(self.version_repository.versions), 1) self.assertEqual(len(self.metadata_repository.metadata_entries), 1) self.assertEqual(self.metadata_repository.metadata_entries[0].author_display_name, "Equipe Interna") self.assertEqual(self.metadata_repository.metadata_entries[0].version_number, 1) self.assertEqual(len(self.artifact_repository.artifacts), 2) artifacts_by_kind = { artifact.artifact_kind: artifact for artifact in self.artifact_repository.artifacts } generation_artifact = artifacts_by_kind[ToolArtifactKind.GENERATION_REQUEST] validation_artifact = artifacts_by_kind[ToolArtifactKind.VALIDATION_REPORT] self.assertEqual(generation_artifact.tool_name, "consultar_vendas_periodo") self.assertEqual(generation_artifact.payload_json["target_package"], GENERATED_TOOLS_PACKAGE) self.assertEqual( generation_artifact.payload_json["target_module"], build_generated_tool_module_name("consultar_vendas_periodo"), ) self.assertEqual( generation_artifact.payload_json["target_file_path"], build_generated_tool_module_path("consultar_vendas_periodo"), ) self.assertEqual(generation_artifact.payload_json["target_callable"], GENERATED_TOOL_ENTRYPOINT) self.assertEqual(generation_artifact.payload_json["reserved_lifecycle_target"], "generated") self.assertEqual(generation_artifact.payload_json["parameters"][0]["name"], "periodo_inicio") self.assertEqual(validation_artifact.payload_json["validation_status"], "passed") self.assertEqual(validation_artifact.payload_json["parameter_count"], 2) self.assertEqual(validation_artifact.author_display_name, "Equipe Interna") self.assertTrue(generation_artifact.checksum) self.assertTrue(validation_artifact.checksum) def test_create_draft_submission_blocks_tool_name_reserved_by_core_catalog(self): with self.assertRaisesRegex(ValueError, "catalogo core do sistema"): self.service.create_draft_submission( { "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 o admin nao permite reutilizar nomes reservados do runtime core.", "parameters": [], }, owner_staff_account_id=11, owner_name="Equipe Interna", ) self.assertEqual(len(self.draft_repository.drafts), 0) self.assertEqual(len(self.version_repository.versions), 0) self.assertEqual(len(self.metadata_repository.metadata_entries), 0) self.assertEqual(len(self.artifact_repository.artifacts), 0) def test_create_draft_submission_reuses_root_draft_and_increments_version(self): self.service.create_draft_submission( { "domain": "locacao", "tool_name": "emitir_resumo_locacao", "display_name": "Emitir resumo de locacao", "description": "Resume o contrato atual de locacao para consulta administrativa.", "business_goal": "Dar visibilidade rapida ao status do contrato e dos dados principais.", "parameters": [], }, owner_staff_account_id=3, owner_name="Analista de Locacao", ) payload = self.service.create_draft_submission( { "domain": "locacao", "tool_name": "emitir_resumo_locacao", "display_name": "Emitir resumo de locacao", "description": "Resume o contrato atual de locacao e os principais eventos administrativos.", "business_goal": "Dar visibilidade rapida ao status do contrato, do pagamento e dos dados principais.", "parameters": [ { "name": "contrato_id", "parameter_type": "string", "description": "Identificador do contrato consultado.", "required": True, } ], }, owner_staff_account_id=4, owner_name="Coordenacao de Locacao", ) self.assertEqual(payload["draft_preview"]["version_id"], "tool_version::emitir_resumo_locacao::v2") self.assertEqual(payload["draft_preview"]["version_number"], 2) self.assertEqual(payload["draft_preview"]["version_count"], 2) self.assertEqual(len(self.draft_repository.drafts), 1) self.assertEqual(len(self.version_repository.versions), 2) self.assertEqual(len(self.metadata_repository.metadata_entries), 2) self.assertEqual(len(self.artifact_repository.artifacts), 4) self.assertEqual(self.draft_repository.drafts[0].current_version_number, 2) self.assertEqual(self.draft_repository.drafts[0].version_count, 2) self.assertEqual(self.draft_repository.drafts[0].owner_display_name, "Coordenacao de Locacao") def test_build_drafts_payload_returns_versioned_draft_summaries(self): self.service.create_draft_submission( { "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": [], }, owner_staff_account_id=5, owner_name="Equipe de Tools", ) self.service.create_draft_submission( { "domain": "orquestracao", "tool_name": "priorizar_contato_quente", "display_name": "Priorizar contato quente", "description": "Classifica contatos mais quentes com sinais adicionais para orientar o atendimento.", "business_goal": "Dar mais foco comercial ao time interno ao identificar oportunidades quentes com mais contexto.", "parameters": [], }, owner_staff_account_id=6, owner_name="Diretoria Comercial", ) payload = self.service.build_drafts_payload() self.assertEqual(payload["storage_status"], "admin_database") self.assertEqual(len(payload["drafts"]), 1) self.assertEqual(payload["drafts"][0]["tool_name"], "priorizar_contato_quente") self.assertEqual(payload["drafts"][0]["current_version_number"], 2) self.assertEqual(payload["drafts"][0]["version_count"], 2) self.assertEqual(payload["drafts"][0]["owner_name"], "Diretoria Comercial") self.assertEqual(payload["supported_statuses"], [ToolLifecycleStatus.DRAFT]) def test_build_publications_payload_prefers_persisted_metadata_catalog(self): self.service.create_draft_submission( { "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, } ], }, owner_staff_account_id=8, owner_name="Operacao de Oficina", ) payload = self.service.build_publications_payload() self.assertEqual(payload["source"], "admin_metadata_catalog") self.assertEqual(payload["target_service"], "product") self.assertEqual(len(payload["publications"]), 1) publication = payload["publications"][0] self.assertEqual(publication["publication_id"], "tool_metadata::consultar_revisao_aberta::v1") self.assertEqual(publication["tool_name"], "consultar_revisao_aberta") self.assertEqual(publication["version"], 1) self.assertEqual(publication["status"], ToolLifecycleStatus.DRAFT) self.assertEqual(publication["author_name"], "Operacao de Oficina") self.assertEqual(publication["implementation_module"], build_generated_tool_module_name("consultar_revisao_aberta")) self.assertEqual(publication["implementation_callable"], GENERATED_TOOL_ENTRYPOINT) self.assertEqual(publication["parameter_count"], 1) self.assertEqual(publication["parameters"][0]["parameter_type"], ToolParameterType.STRING) if __name__ == "__main__": unittest.main()