"""Tests for the pluggable driver registry + metadata.""" from __future__ import annotations import pytest from app.integrations.driver_base import ( ChatApprovalDriver, CloudDriver, DataAccessDriver, HrisDriver, IntegrationDriver, SoftwareProvisioningDriver, ) from app.integrations.drivers.bigquery_data_access import BigQueryDataAccessDriver from app.integrations.drivers.gcp_cloud import GcpCloudDriver from app.integrations.registry import ( _DRIVERS, get_driver_class, kinds, list_drivers, providers_for_kind, register_driver, ) from app.integrations.schemas import IntegrationFieldInfo def test_all_builtin_drivers_registered(): """Every driver shipped in this PR is discoverable at runtime.""" registered = {(m.kind, m.provider) for m in list_drivers()} expected = { ("hris", "successfactors"), ("chat_approval", "feishu"), ("cloud", "aws"), ("cloud", "aliyun"), ("cloud", "tencent"), ("data_access", "bigquery"), ("software_provisioning", "google_workspace"), ("software_provisioning", "feishu_group"), ("software_provisioning", "manual"), } assert expected.issubset(registered), ( f"Missing drivers: {expected - registered}" ) def test_kinds_lists_distinct_kinds(): ks = set(kinds()) assert { "hris", "chat_approval", "cloud", "data_access", "software_provisioning", }.issubset(ks) def test_providers_for_kind_filters_correctly(): cloud = list(providers_for_kind("cloud")) assert "aws" in cloud and "aliyun" in cloud and "tencent" in cloud assert "feishu" not in cloud assert "bigquery" not in cloud chat = list(providers_for_kind("chat_approval")) assert "feishu" in chat assert "aws" not in chat assert list(providers_for_kind("data_access")) == ["bigquery"] def test_get_driver_class_returns_class(): cls = get_driver_class("hris", "successfactors") assert cls is not None assert issubclass(cls, HrisDriver) cls = get_driver_class("chat_approval", "feishu") assert cls is not None assert issubclass(cls, ChatApprovalDriver) cls = get_driver_class("data_access", "bigquery") assert cls is not None assert issubclass(cls, DataAccessDriver) def test_get_driver_class_returns_none_for_unknown(): assert get_driver_class("hris", "workday") is None assert get_driver_class("unknown_kind", "whatever") is None def test_metadata_includes_binding_fields_for_software_provisioning(): gw_meta = get_driver_class("software_provisioning", "google_workspace").metadata() assert gw_meta.binding_fields, "Google Workspace must declare binding fields" op_field = next((f for f in gw_meta.binding_fields if f.key == "operation"), None) assert op_field is not None assert op_field.options is not None and len(op_field.options) >= 2 # HRIS / chat_approval / cloud / data_access — no binding fields for (k, p) in [ ("hris", "successfactors"), ("chat_approval", "feishu"), ("cloud", "aws"), ("data_access", "bigquery"), ]: meta = get_driver_class(k, p).metadata() assert meta.binding_fields == [], f"{k}/{p} should not declare binding_fields" def test_bigquery_data_access_driver_is_not_gcp_cloud_driver(): cls = get_driver_class("data_access", "bigquery") assert cls is BigQueryDataAccessDriver assert cls is not GcpCloudDriver assert issubclass(cls, DataAccessDriver) assert not issubclass(cls, CloudDriver) meta = cls.metadata() assert meta.kind == "data_access" assert meta.provider == "bigquery" field_by_key = {field.key: field for field in meta.fields} assert set(field_by_key) == { "service_account_json", "apply_mode_enabled", # Added by #3171 (personal-employee DAX apply gate). This test's # subject is "the bigquery data_access driver is NOT the GCP cloud # driver"; the field set is incidental scaffolding, so a legitimate # new config field just needs listing here — it had been left red # on main. "personal_apply_enabled", } assert field_by_key["service_account_json"].is_secret is True assert field_by_key["service_account_json"].required is True assert field_by_key["apply_mode_enabled"].options == [ {"label": "Disabled", "value": "false"}, {"label": "Enabled", "value": "true"}, ] assert field_by_key["apply_mode_enabled"].default == "false" assert meta.binding_fields == [] assert meta.supports_test_connection is False driver = cls(integration_id="dummy", config={}) assert not isinstance(driver, CloudDriver) for cloud_method in ( "list_scopes", "grant_access", "revoke_access", "list_scope_grants", ): assert not hasattr(driver, cloud_method) def test_registering_duplicate_driver_logs_and_keeps_original(caplog): """Pre-launch hardening 2026-05-05: a duplicate ``register_driver`` decorator (typically a copy-paste during development that forgot to change ``provider``) must NOT take down the whole app at import time. Instead, log critical and skip the new class. The original is the source of truth — startup tolerance > registry purity. The duplicate-class case still produces a crit-level log so a careful reviewer notices. """ import logging class _Dup(IntegrationDriver): kind = "hris" provider = "successfactors" display_name = "dup" description = "" config_schema = [] # No raise — registration is a no-op + logged. with caplog.at_level(logging.CRITICAL, logger="app.integrations.registry"): register_driver(_Dup) assert any( "Driver collision" in r.getMessage() or "already registered" in r.getMessage() for r in caplog.records ) # Original driver still wins. cls = get_driver_class("hris", "successfactors") assert cls.__name__ != "_Dup" def test_registering_driver_without_kind_or_provider_raises(): class _MissingKind(IntegrationDriver): provider = "something" config_schema = [] with pytest.raises(ValueError, match="missing"): register_driver(_MissingKind) def test_custom_driver_extensibility(): """Prove that adding a new provider is a pure decorator call (no DB migration, no code edit to the registry/service).""" try: @register_driver class MyNewHris(HrisDriver): provider = "test_fake_workday" display_name = "Test Fake Workday" description = "Just to prove HrisDriver is extensible" config_schema = [ IntegrationFieldInfo(key="host", label="Host", is_secret=False, required=True), ] async def list_active_users(self): return [] async def list_hires_in_range(self, start_date, end_date): return [] async def list_terminations_in_range(self, start_date, end_date): return [] metas = {(m.kind, m.provider) for m in list_drivers()} assert ("hris", "test_fake_workday") in metas finally: # Clean up so other tests see the stable registry. _DRIVERS.pop(("hris", "test_fake_workday"), None) def test_software_provisioning_driver_declares_grant_revoke(): """Every SoftwareProvisioningDriver must at least be instantiable and its base implementations should raise NotImplementedError for unsupported operations (sanity: manual driver handles all four ops).""" manual = get_driver_class("software_provisioning", "manual") assert manual is not None instance = manual(integration_id="dummy", config={}) assert isinstance(instance, SoftwareProvisioningDriver) assert instance.supports_seat is True