orquestrador/tests/test_shared_contracts.py

import unittest
from datetime import datetime, timezone

from shared.contracts import (
    AdminPermission,
    BOT_GOVERNED_SETTINGS,
    GENERATED_TOOL_ENTRYPOINT,
    GENERATED_TOOLS_PACKAGE,
    MODEL_RUNTIME_PROFILES,
    MODEL_RUNTIME_SEPARATION_RULES,
    BotGovernanceArea,
    BotGovernanceMutability,
    FunctionalConfigurationDomain,
    FunctionalConfigurationMutability,
    FunctionalConfigurationPropagation,
    FunctionalConfigurationSource,
    ModelRuntimePurpose,
    ModelRuntimeSeparationRule,
    ModelRuntimeTarget,
    OperationalConsistencyModel,
    OperationalDataDomain,
    OperationalDataSensitivity,
    OperationalFreshnessTarget,
    OperationalQuerySurface,
    OperationalReadGranularity,
    OperationalReadModel,
    OperationalStorageShape,
    OperationalSyncStrategy,
    PRODUCT_OPERATIONAL_DATASETS,
    PublishedToolContract,
    ServiceName,
    StaffRole,
    SYSTEM_FUNCTIONAL_CONFIGURATIONS,
    TOOL_LIFECYCLE_STAGES,
    TOOL_LIFECYCLE_STATUS_SEQUENCE,
    ToolLifecycleStatus,
    ToolParameterContract,
    ToolParameterType,
    ToolPublicationEnvelope,
    build_generated_tool_module_name,
    build_generated_tool_module_path,
    get_bot_governed_setting,
    get_tool_lifecycle_stage,
    get_functional_configuration,
    get_model_runtime_contract,
    get_operational_dataset,
    normalize_staff_role,
    permissions_for_role,
    role_has_permission,
    role_includes,
)


class AccessControlContractTests(unittest.TestCase):
    def test_role_hierarchy_is_ordered(self):
        self.assertTrue(role_includes(StaffRole.DIRETOR, StaffRole.COLABORADOR))
        self.assertFalse(role_includes(StaffRole.COLABORADOR, StaffRole.DIRETOR))

    def test_legacy_role_aliases_are_normalized_to_portuguese_roles(self):
        self.assertEqual(normalize_staff_role("viewer"), StaffRole.COLABORADOR)
        self.assertEqual(normalize_staff_role("staff"), StaffRole.COLABORADOR)
        self.assertEqual(normalize_staff_role("admin"), StaffRole.DIRETOR)

    def test_permissions_are_inherited_by_director(self):
        self.assertIn(
            AdminPermission.VIEW_REPORTS,
            permissions_for_role(StaffRole.COLABORADOR),
        )
        self.assertTrue(
            role_has_permission(StaffRole.COLABORADOR, AdminPermission.MANAGE_TOOL_DRAFTS)
        )
        self.assertFalse(
            role_has_permission(StaffRole.COLABORADOR, AdminPermission.REVIEW_TOOL_GENERATIONS)
        )
        self.assertFalse(
            role_has_permission(StaffRole.COLABORADOR, AdminPermission.PUBLISH_TOOLS)
        )
        self.assertFalse(
            role_has_permission(StaffRole.COLABORADOR, AdminPermission.MANAGE_SETTINGS)
        )
        self.assertTrue(
            role_has_permission(StaffRole.DIRETOR, AdminPermission.MANAGE_SETTINGS)
        )
        self.assertTrue(
            role_has_permission(StaffRole.DIRETOR, AdminPermission.MANAGE_STAFF_ACCOUNTS)
        )


class ToolPublicationContractTests(unittest.TestCase):
    def test_tool_publication_envelope_is_built_with_shared_contract(self):
        published_tool = PublishedToolContract(
            tool_name="consultar_financiamento",
            display_name="Consultar Financiamento",
            description="Consulta opcoes de financiamento.",
            version=2,
            status=ToolLifecycleStatus.APPROVED,
            parameters=(
                ToolParameterContract(
                    name="valor_veiculo",
                    parameter_type=ToolParameterType.NUMBER,
                    description="Valor do veiculo em reais.",
                    required=True,
                ),
            ),
            implementation_module="generated_tools.consultar_financiamento",
            implementation_callable="run",
            checksum="sha256:abc123",
            published_at=datetime(2026, 3, 26, 12, 0, tzinfo=timezone.utc),
            published_by="diretor:1",
        )

        envelope = ToolPublicationEnvelope(
            source_service=ServiceName.ADMIN,
            target_service=ServiceName.PRODUCT,
            publication_id="pub_001",
            published_tool=published_tool,
            emitted_at=datetime(2026, 3, 26, 12, 1, tzinfo=timezone.utc),
        )

        self.assertEqual(envelope.source_service, ServiceName.ADMIN)
        self.assertEqual(envelope.target_service, ServiceName.PRODUCT)
        self.assertEqual(envelope.published_tool.tool_name, "consultar_financiamento")
        self.assertEqual(
            envelope.published_tool.parameters[0].parameter_type,
            ToolParameterType.NUMBER,
        )

    def test_generated_tool_namespace_uses_isolated_package(self):
        self.assertEqual(GENERATED_TOOLS_PACKAGE, "generated_tools")
        self.assertEqual(GENERATED_TOOL_ENTRYPOINT, "run")
        self.assertEqual(
            build_generated_tool_module_name("consultar_financiamento"),
            "generated_tools.consultar_financiamento",
        )
        self.assertEqual(
            build_generated_tool_module_path("consultar_financiamento"),
            "generated_tools/consultar_financiamento.py",
        )

    def test_lifecycle_catalog_exposes_expected_states_in_order(self):
        self.assertEqual(
            TOOL_LIFECYCLE_STATUS_SEQUENCE,
            (
                ToolLifecycleStatus.DRAFT,
                ToolLifecycleStatus.GENERATED,
                ToolLifecycleStatus.VALIDATED,
                ToolLifecycleStatus.APPROVED,
                ToolLifecycleStatus.ACTIVE,
                ToolLifecycleStatus.FAILED,
                ToolLifecycleStatus.ARCHIVED,
            ),
        )
        self.assertEqual([stage.order for stage in TOOL_LIFECYCLE_STAGES], [1, 2, 3, 4, 5, 6, 7])

    def test_get_tool_lifecycle_stage_returns_terminal_metadata(self):
        approved_stage = get_tool_lifecycle_stage("approved")
        failed_stage = get_tool_lifecycle_stage(ToolLifecycleStatus.FAILED)
        archived_stage = get_tool_lifecycle_stage(ToolLifecycleStatus.ARCHIVED)

        self.assertEqual(approved_stage.label, "Approved")
        self.assertFalse(approved_stage.terminal)
        self.assertTrue(failed_stage.terminal)
        self.assertTrue(archived_stage.terminal)


class ProductOperationalDataContractTests(unittest.TestCase):
    def test_catalog_exposes_expected_operational_domains(self):
        self.assertEqual(len(PRODUCT_OPERATIONAL_DATASETS), 8)
        self.assertEqual(
            {dataset.domain for dataset in PRODUCT_OPERATIONAL_DATASETS},
            {
                OperationalDataDomain.INVENTORY,
                OperationalDataDomain.SALES,
                OperationalDataDomain.REVIEW,
                OperationalDataDomain.RENTAL,
                OperationalDataDomain.CONVERSATION,
                OperationalDataDomain.INTEGRATION,
            },
        )

    def test_all_datasets_use_async_admin_read_model_without_direct_product_query(self):
        for dataset in PRODUCT_OPERATIONAL_DATASETS:
            self.assertEqual(dataset.read_permission, AdminPermission.VIEW_REPORTS)
            self.assertEqual(dataset.report_read_model, OperationalReadModel.ADMIN_READ_MODEL)
            self.assertEqual(dataset.consistency_model, OperationalConsistencyModel.EVENTUAL)
            self.assertEqual(dataset.sync_strategy, OperationalSyncStrategy.ETL_INCREMENTAL)
            self.assertEqual(dataset.storage_shape, OperationalStorageShape.SNAPSHOT_TABLE)
            self.assertEqual(dataset.query_surface, OperationalQuerySurface.DEDICATED_VIEW)
            self.assertFalse(dataset.uses_product_replica)
            self.assertFalse(dataset.direct_product_query_allowed)
            self.assertFalse(dataset.write_allowed)

    def test_sales_orders_dataset_is_read_only_and_blocks_customer_identity(self):
        dataset = get_operational_dataset("sales_orders")

        self.assertIsNotNone(dataset)
        self.assertEqual(
            dataset.freshness_target,
            OperationalFreshnessTarget.NEAR_REAL_TIME,
        )
        self.assertEqual(
            dataset.allowed_granularities,
            (
                OperationalReadGranularity.AGGREGATE,
                OperationalReadGranularity.RECORD,
            ),
        )
        self.assertIn("numero_pedido", [field.name for field in dataset.allowed_fields])

        blocked_fields = {field.name: field.sensitivity for field in dataset.blocked_fields}
        self.assertEqual(
            blocked_fields["cpf"],
            OperationalDataSensitivity.CUSTOMER_IDENTIFIER,
        )
        self.assertEqual(
            blocked_fields["user_id"],
            OperationalDataSensitivity.INTERNAL_IDENTIFIER,
        )

    def test_conversation_turns_dataset_exposes_telemetry_without_raw_messages(self):
        dataset = get_operational_dataset("conversation_turns")

        self.assertIsNotNone(dataset)
        self.assertEqual(
            dataset.freshness_target,
            OperationalFreshnessTarget.INTRA_HOUR,
        )
        self.assertIn("tool_name", [field.name for field in dataset.allowed_fields])

        blocked_names = {field.name for field in dataset.blocked_fields}
        self.assertIn("user_message", blocked_names)
        self.assertIn("assistant_response", blocked_names)
        self.assertIn("tool_arguments", blocked_names)
        self.assertIn("error_detail", blocked_names)

    def test_rental_payment_dataset_blocks_receipt_identifiers(self):
        dataset = get_operational_dataset("rental_payments")

        self.assertIsNotNone(dataset)
        blocked_fields = {field.name: field.sensitivity for field in dataset.blocked_fields}
        self.assertEqual(
            blocked_fields["identificador_comprovante"],
            OperationalDataSensitivity.SECRET,
        )
        self.assertNotIn("identificador_comprovante", [field.name for field in dataset.allowed_fields])

    def test_unknown_operational_dataset_returns_none(self):
        self.assertIsNone(get_operational_dataset("customers"))


class SystemFunctionalConfigurationContractTests(unittest.TestCase):
    def test_catalog_exposes_expected_configuration_domains(self):
        self.assertEqual(len(SYSTEM_FUNCTIONAL_CONFIGURATIONS), 6)
        self.assertEqual(
            {configuration.domain for configuration in SYSTEM_FUNCTIONAL_CONFIGURATIONS},
            {
                FunctionalConfigurationDomain.MODEL_CATALOG,
                FunctionalConfigurationDomain.ATENDIMENTO_RUNTIME,
                FunctionalConfigurationDomain.TOOL_GENERATION_RUNTIME,
                FunctionalConfigurationDomain.BOT_POLICY,
                FunctionalConfigurationDomain.CHANNEL_OPERATION,
                FunctionalConfigurationDomain.CONFIG_PUBLICATION,
            },
        )

    def test_director_governed_configurations_require_manage_settings(self):
        for configuration in SYSTEM_FUNCTIONAL_CONFIGURATIONS:
            self.assertEqual(configuration.read_permission, AdminPermission.VIEW_SYSTEM)
            self.assertFalse(configuration.direct_product_write_allowed)

            if configuration.mutability == FunctionalConfigurationMutability.DIRECTOR_GOVERNED:
                self.assertEqual(configuration.write_permission, AdminPermission.MANAGE_SETTINGS)
                self.assertEqual(
                    configuration.propagation,
                    FunctionalConfigurationPropagation.VERSIONED_PUBLICATION,
                )
            else:
                self.assertIsNone(configuration.write_permission)
                self.assertEqual(
                    configuration.propagation,
                    FunctionalConfigurationPropagation.OBSERVATION_ONLY,
                )

    def test_atendimento_and_tool_generation_profiles_are_separated(self):
        atendimento = get_functional_configuration("atendimento_runtime_profile")
        tool_generation = get_functional_configuration("tool_generation_runtime_profile")

        self.assertIsNotNone(atendimento)
        self.assertIsNotNone(tool_generation)
        self.assertEqual(
            atendimento.domain,
            FunctionalConfigurationDomain.ATENDIMENTO_RUNTIME,
        )
        self.assertEqual(
            tool_generation.domain,
            FunctionalConfigurationDomain.TOOL_GENERATION_RUNTIME,
        )
        self.assertNotEqual(atendimento.config_key, tool_generation.config_key)
        self.assertEqual(atendimento.source, FunctionalConfigurationSource.ADMIN_GOVERNED_STATE)
        self.assertEqual(tool_generation.source, FunctionalConfigurationSource.ADMIN_GOVERNED_STATE)

    def test_read_only_configuration_surfaces_are_catalog_and_effective_runtime_state(self):
        catalog = get_functional_configuration("allowed_model_catalog")
        published_state = get_functional_configuration("published_runtime_state")

        self.assertIsNotNone(catalog)
        self.assertIsNotNone(published_state)
        self.assertEqual(catalog.mutability, FunctionalConfigurationMutability.READ_ONLY)
        self.assertEqual(published_state.mutability, FunctionalConfigurationMutability.READ_ONLY)
        self.assertEqual(catalog.source, FunctionalConfigurationSource.PLATFORM_CATALOG)
        self.assertEqual(
            published_state.source,
            FunctionalConfigurationSource.PRODUCT_EFFECTIVE_STATE,
        )
        self.assertTrue(all(not field.writable for field in catalog.fields))
        self.assertTrue(all(not field.writable for field in published_state.fields))

    def test_unknown_functional_configuration_returns_none(self):
        self.assertIsNone(get_functional_configuration("database_credentials"))


class BotGovernedConfigurationContractTests(unittest.TestCase):
    def test_catalog_exposes_expected_bot_governance_areas(self):
        self.assertEqual(len(BOT_GOVERNED_SETTINGS), 15)
        self.assertEqual(
            {setting.area for setting in BOT_GOVERNED_SETTINGS},
            {
                BotGovernanceArea.MODEL_SELECTION,
                BotGovernanceArea.RESPONSE_GENERATION,
                BotGovernanceArea.TOOL_USAGE,
                BotGovernanceArea.FALLBACK_AND_HANDOFF,
                BotGovernanceArea.CHANNEL_OPERATION,
            },
        )

    def test_all_bot_settings_are_director_governed_and_versioned(self):
        for setting in BOT_GOVERNED_SETTINGS:
            self.assertEqual(setting.read_permission, AdminPermission.VIEW_SYSTEM)
            self.assertEqual(setting.write_permission, AdminPermission.MANAGE_SETTINGS)
            self.assertEqual(
                setting.mutability,
                BotGovernanceMutability.DIRECTOR_GOVERNED,
            )
            self.assertTrue(setting.versioned_publication_required)
            self.assertFalse(setting.direct_product_write_allowed)

    def test_bot_governance_excludes_tool_generation_runtime(self):
        self.assertEqual(
            {setting.parent_config_key for setting in BOT_GOVERNED_SETTINGS},
            {
                "atendimento_runtime_profile",
                "bot_behavior_policy",
                "channel_operation_policy",
            },
        )
        self.assertNotIn(
            "tool_generation_runtime_profile",
            {setting.parent_config_key for setting in BOT_GOVERNED_SETTINGS},
        )

    def test_governed_settings_cover_model_policy_and_channel_controls(self):
        governed_keys = {setting.setting_key for setting in BOT_GOVERNED_SETTINGS}
        self.assertIn("bot_model_provider", governed_keys)
        self.assertIn("bot_prompt_profile_ref", governed_keys)
        self.assertIn("bot_tool_policy_ref", governed_keys)
        self.assertIn("bot_handoff_enabled", governed_keys)
        self.assertIn("channel_maintenance_mode", governed_keys)

    def test_unknown_bot_governed_setting_returns_none(self):
        self.assertIsNone(get_bot_governed_setting("provider_api_key"))


class ModelRuntimeSeparationContractTests(unittest.TestCase):
    def test_runtime_profiles_exist_for_attendimento_and_tool_generation(self):
        self.assertEqual(len(MODEL_RUNTIME_PROFILES), 2)
        self.assertEqual(
            {runtime_contract.runtime_target for runtime_contract in MODEL_RUNTIME_PROFILES},
            {
                ModelRuntimeTarget.ATENDIMENTO,
                ModelRuntimeTarget.TOOL_GENERATION,
            },
        )

    def test_attendimento_and_generation_have_distinct_consumers_and_purposes(self):
        atendimento = get_model_runtime_contract("atendimento")
        tool_generation = get_model_runtime_contract("tool_generation")

        self.assertIsNotNone(atendimento)
        self.assertIsNotNone(tool_generation)
        self.assertEqual(atendimento.config_key, "atendimento_runtime_profile")
        self.assertEqual(tool_generation.config_key, "tool_generation_runtime_profile")
        self.assertEqual(atendimento.consumed_by_service, ServiceName.PRODUCT)
        self.assertEqual(tool_generation.consumed_by_service, ServiceName.ADMIN)
        self.assertEqual(atendimento.purpose, ModelRuntimePurpose.CUSTOMER_RESPONSE)
        self.assertEqual(
            tool_generation.purpose,
            ModelRuntimePurpose.TOOL_GENERATION_AND_VALIDATION,
        )
        self.assertTrue(atendimento.affects_customer_response)
        self.assertFalse(tool_generation.affects_customer_response)
        self.assertFalse(atendimento.can_generate_code)
        self.assertTrue(tool_generation.can_generate_code)

    def test_runtime_profiles_require_independent_publication_and_rollback(self):
        for runtime_contract in MODEL_RUNTIME_PROFILES:
            self.assertEqual(runtime_contract.read_permission, AdminPermission.VIEW_SYSTEM)
            self.assertEqual(runtime_contract.write_permission, AdminPermission.MANAGE_SETTINGS)
            self.assertTrue(runtime_contract.published_independently)
            self.assertTrue(runtime_contract.rollback_independently)
            self.assertFalse(runtime_contract.cross_target_propagation_allowed)
            self.assertEqual(runtime_contract.catalog_runtime_target, runtime_contract.runtime_target)

    def test_separation_rules_forbid_implicit_cross_runtime_propagation(self):
        self.assertEqual(
            set(MODEL_RUNTIME_SEPARATION_RULES),
            {
                ModelRuntimeSeparationRule.SEPARATE_CONFIG_KEYS,
                ModelRuntimeSeparationRule.SEPARATE_CATALOG_TARGETS,
                ModelRuntimeSeparationRule.INDEPENDENT_PUBLICATION,
                ModelRuntimeSeparationRule.INDEPENDENT_ROLLBACK,
                ModelRuntimeSeparationRule.NO_IMPLICIT_PROPAGATION,
            },
        )

    def test_unknown_model_runtime_returns_none(self):
        self.assertIsNone(get_model_runtime_contract("shared_runtime"))


if __name__ == "__main__":
    unittest.main()