from __future__ import annotations import importlib.util import json from pathlib import Path import unittest from scripts import apply_deploy_seed REPO_ROOT = Path(__file__).resolve().parents[1] SCRIPT_PATH = REPO_ROOT / "scripts" / "apply_identity_sync_policy.py" SEED_PATH = REPO_ROOT / "packs" / "Salesforce-PingCAP-L2" / "identity-sync-policy.seed.json" CATALOG_PATH = ( REPO_ROOT / "deployments" / "offline" / "runtime" / "seeds" / "deploy-seed-catalog.json" ) SPEC = importlib.util.spec_from_file_location("apply_identity_sync_policy", SCRIPT_PATH) assert SPEC is not None and SPEC.loader is not None apply_identity_sync_policy = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(apply_identity_sync_policy) class IdentitySyncPolicySeedContractTests(unittest.TestCase): def test_seed_declares_dedicated_zero_access_onboarding_profile(self) -> None: seed = apply_identity_sync_policy.load_seed(REPO_ROOT, [str(SEED_PATH)]) self.assertEqual(seed["singleton"]["defaultProfileName"], "SCIM Zero Access") self.assertNotEqual( seed["singleton"]["defaultProfileName"], seed["singleton"]["serviceProfileName"], ) self.assertEqual( seed["onboardingProfile"], { "description": ( "Zero-access profile for first-time Helix SCIM onboarding; " "business access is assigned separately." ), "isActive": True, "isSystem": False, "userType": "Standard", }, ) def test_seed_declares_exact_native_scim_service_profile_contract(self) -> None: seed = apply_identity_sync_policy.load_seed(REPO_ROOT, [str(SEED_PATH)]) self.assertEqual(seed["singleton"]["serviceProfileName"], "SCIM Provisioner") self.assertEqual( seed["serviceProfile"], { "description": ( "Dedicated least-privilege service profile for native SCIM provisioning." ), "isActive": True, "isSystem": False, "userType": "Standard", "systemPermissions": ["ManageUsers"], "objectPermissions": [ { "object": "IdentitySyncDeactivationTicket", "read": True, "create": True, "edit": False, "delete": False, "viewAllRecords": False, "modifyAllRecords": False, }, { "object": "IdentitySyncSetting", "read": True, "create": False, "edit": False, "delete": False, "viewAllRecords": True, "modifyAllRecords": False, }, { "object": "Profile", "read": True, "create": False, "edit": False, "delete": False, "viewAllRecords": False, "modifyAllRecords": False, }, { "object": "User", "read": True, "create": True, "edit": True, "delete": False, "viewAllRecords": False, "modifyAllRecords": False, }, ], "fieldPermissions": [ { "object": "IdentitySyncDeactivationTicket", "fields": ["Bucket", "Name", "Slot"], "read": True, "edit": True, }, { "object": "IdentitySyncSetting", "fields": [ "BreakerCooldownSeconds", "BreakerFailureThreshold", "DefaultProfileName", "EmailEncodingKey", "ExemptPrincipals", "LanguageLocaleKey", "LocaleSidKey", "Name", "ServiceProfileName", "ServiceUsername", "TimeZoneSidKey", "TokenTtlHours", ], "read": True, "edit": False, }, { "object": "Profile", "fields": ["Id", "Name"], "read": True, "edit": False, }, { "object": "User", "fields": ["Id"], "read": True, "edit": False, }, { "object": "User", "fields": [ "Alias", "Department", "Email", "EmailEncodingKey", "FirstName", "IsActive", "LanguageLocaleKey", "LastName", "LocaleSidKey", "MobilePhone", "Phone", "ProfileId", "TimeZoneSidKey", "Title", "UserRoleId", "Username", ], "read": True, "edit": True, }, ], "denyUnlistedFieldsForObjects": [ "IdentitySyncDeactivationTicket", "IdentitySyncSetting", "Profile", "User", ], }, ) def test_seed_declares_exact_native_scim_service_user_contract(self) -> None: seed = apply_identity_sync_policy.load_seed(REPO_ROOT, [str(SEED_PATH)]) self.assertEqual( seed["serviceUser"], { "lastName": "SCIM Provisioning", "alias": "scimprov", "isActive": True, "userRoleId": None, }, ) def test_service_object_permissions_obey_salesforce_dependencies(self) -> None: seed = apply_identity_sync_policy.load_seed(REPO_ROOT, [str(SEED_PATH)]) for permission in seed["serviceProfile"]["objectPermissions"]: with self.subTest(object=permission["object"]): if any( permission[field] for field in ( "create", "edit", "delete", "viewAllRecords", "modifyAllRecords", ) ): self.assertTrue(permission["read"]) def test_service_field_permissions_obey_salesforce_dependencies(self) -> None: seed = apply_identity_sync_policy.load_seed(REPO_ROOT, [str(SEED_PATH)]) for permission in seed["serviceProfile"]["fieldPermissions"]: if permission["edit"]: with self.subTest( object=permission["object"], fields=permission["fields"] ): self.assertTrue(permission["read"]) def test_catalog_and_adapter_counts_are_exactly_aligned(self) -> None: catalog = json.loads(CATALOG_PATH.read_text(encoding="utf-8")) self.assertEqual(catalog["catalog_version"], "2026-08-29.1") self.assertEqual( catalog["modules"]["identity_sync_policy"]["declared_stats"], apply_identity_sync_policy.PLANNED_COUNTS, ) def test_dispatcher_forwards_only_required_runtime_profile_inputs(self) -> None: environment = { "PATH": "/usr/bin:/bin", "TIDB_HOST": "db.example.invalid", "TIDB_PORT": "4000", "TIDB_USER": "seed-user", "TIDB_PASSWORD": "opaque-password", "TIDB_DATABASE": "crm", "TIDB_TLS": "true", "BACKEND_PORT": "3001", "VERIFY_USERNAME": "admin@example.invalid", "NEXUS_ADMIN_PASSWORD": "opaque-admin-password", "STAGING_INSPECTION_OAUTH_CLIENT_ID": "salesforce-compat-client", "STAGING_INSPECTION_OAUTH_CLIENT_SECRET": "programmatic-secret", "UNRELATED_SECRET": "must-not-pass", } outer_environment = ( apply_identity_sync_policy.deploy_offline_env.deploy_seed_dispatch_env( environment, profile="staging", nexuscrm_worktree=None, ) ) dispatcher_environment = apply_deploy_seed._dispatcher_environment( "staging", outer_environment, ) adapter_environment = apply_deploy_seed._adapter_environment( "identity_sync_policy", dispatcher_environment, ) for name in ( "TIDB_HOST", "TIDB_PORT", "TIDB_USER", "TIDB_PASSWORD", "TIDB_DATABASE", "TIDB_TLS", "BACKEND_PORT", "VERIFY_USERNAME", "NEXUS_ADMIN_PASSWORD", "STAGING_INSPECTION_OAUTH_CLIENT_ID", "STAGING_INSPECTION_OAUTH_CLIENT_SECRET", ): self.assertEqual(adapter_environment[name], environment[name]) self.assertNotIn("UNRELATED_SECRET", adapter_environment) class NativeSCIMServiceProfileTests(unittest.TestCase): PROFILE_ID = "profile-scim-service" PERMISSION_SET_ID = "profile-scim-service-permissions" @staticmethod def field_catalog() -> set[tuple[str, str]]: fields = { (permission["object"], field) for permission in apply_identity_sync_policy.SERVICE_PROFILE_FIELD_PERMISSIONS for field in permission["fields"] } fields.update( { ("IdentitySyncDeactivationTicket", "CreatedDate"), ("IdentitySyncSetting", "CreatedDate"), ("Profile", "Description"), ("User", "CompanyName"), ("User", "Division"), ("User", "ManagerId"), } ) return fields def test_unlisted_runtime_fields_are_materialized_as_explicit_denies(self) -> None: expected = apply_identity_sync_policy._expected_service_field_permissions( self.field_catalog() ) for field in ("CompanyName", "Division", "ManagerId"): self.assertEqual(expected[("User", field)], ("0", "0")) self.assertEqual(expected[("User", "Username")], ("1", "1")) if __name__ == "__main__": unittest.main()