import json import importlib.util import subprocess import sys import unittest from pathlib import Path try: import yaml except ImportError: yaml = None ROOT = Path(__file__).resolve().parents[1] EXP_BASELINE_SEED = ROOT / "deployments" / "offline" / "runtime" / "seeds" / "exp-baseline.seed.yaml" HS07B_RUN_SEED = ROOT / "tests" / "fixtures" / "seeds" / "hs07b-workflow-action.seed.json" CAMPAIGN_OBJECT = ROOT / "packs" / "Salesforce-PingCAP-L2" / "objects" / "Campaign.json" PRICEBOOK_ENTRY_OBJECT = ROOT / "packs" / "Salesforce-PingCAP-L2" / "objects" / "PricebookEntry.json" EXP_OWNER_BINDING_BOOTSTRAP = ( ROOT / "deployments" / "offline" / "runtime" / "seeds" / "exp-hubspot-owner-bindings.json" ) PINGCAP_REGION_OWNER_RULES_SEED = ( ROOT / "deployments" / "offline" / "runtime" / "seeds" / "pingcap-region-owner-rules.seed.yaml" ) def load_exp_seed_records(): payload = load_yaml(EXP_BASELINE_SEED) return payload.get("baseline_records", []) def load_exp_owner_binding_bootstrap(): return json.loads(EXP_OWNER_BINDING_BOOTSTRAP.read_text()) def load_pingcap_region_owner_records(): payload = load_yaml(PINGCAP_REGION_OWNER_RULES_SEED) return payload.get("baseline_records", []) def load_pingcap_region_owner_lookups(): payload = load_yaml(PINGCAP_REGION_OWNER_RULES_SEED) return payload.get("lookup_records", []) def load_deploy_offline_env_module(): runtime_dir = ROOT / "deployments" / "offline" / "runtime" sys.path.insert(0, str(runtime_dir)) try: spec = importlib.util.spec_from_file_location("deploy_offline_env", runtime_dir / "deploy_offline_env.py") module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module finally: sys.path.remove(str(runtime_dir)) def load_yaml(path): if yaml is not None: return yaml.safe_load(path.read_text()) result = subprocess.run( [ "ruby", "-ryaml", "-rjson", "-e", "puts JSON.generate(YAML.load_file(ARGV.fetch(0)))", str(path), ], check=True, text=True, capture_output=True, ) return json.loads(result.stdout) class ExpBaselineSeedContractTests(unittest.TestCase): def test_exp_seed_product_catalog_is_small_currency_explicit_and_duplicate_free(self): records = load_exp_seed_records() pricebook_entry = json.loads(PRICEBOOK_ENTRY_OBJECT.read_text(encoding="utf-8"))[0] self.assertIn( "CurrencyIsoCode", {field["apiName"] for field in pricebook_entry["fields"]}, ) catalog_records = [ record for record in records if record["key"].startswith("exp_catalog_") ] by_object = {} for record in catalog_records: by_object.setdefault(record["object"], []).append(record) self.assertEqual( { object_name: len(object_records) for object_name, object_records in by_object.items() }, {"Pricebook2": 1, "PricebookEntry": 7, "Product2": 7}, ) expected_products = { "exp_catalog_product_tidb_cloud": ("C001", "TiDB Cloud"), "exp_catalog_product_tidb_cloud_dedicated": ("C002", "TiDB Cloud Dedicated"), "exp_catalog_product_tidb_cloud_starter": ("C003", "TiDB Cloud Starter"), "exp_catalog_product_tidb_cloud_essential": ("C004", "TiDB Cloud Essential"), "exp_catalog_product_tidb_cloud_premium": ("C005", "TiDB Cloud Premium"), "exp_catalog_product_tidb_cloud_lake": ("C006", "TiDB Cloud Lake"), "exp_catalog_product_tidb_cloud_byoc": ("C007", "TiDB Cloud BYOC"), } products = {record["key"]: record for record in by_object["Product2"]} self.assertEqual(set(products), set(expected_products)) for key, (product_code, name) in expected_products.items(): self.assertEqual(products[key]["match"], {"ProductCode": product_code}) self.assertEqual( products[key]["fields"], { "Name": name, "IsActive": True, "ProductCode": product_code, "Family": "TiDB Cloud", "Description": name, }, ) pricebook_reference = "{{ record.exp_catalog_standard_pricebook.id }}" expected_product_references = { f"{{{{ record.{product_key}.id }}}}" for product_key in expected_products } actual_product_references = set() for record in by_object["PricebookEntry"]: self.assertEqual(record["match"]["Pricebook2Id"], pricebook_reference) self.assertEqual(record["fields"]["Pricebook2Id"], pricebook_reference) self.assertEqual(record["match"]["CurrencyIsoCode"], "USD") self.assertEqual(record["fields"]["CurrencyIsoCode"], "USD") self.assertEqual(record["fields"]["UnitPrice"], 1) self.assertTrue(record["fields"]["IsActive"]) self.assertFalse(record["fields"]["UseStandardPrice"]) self.assertEqual( record["match"]["Product2Id"], record["fields"]["Product2Id"] ) actual_product_references.add(record["match"]["Product2Id"]) self.assertEqual(actual_product_references, expected_product_references) signatures = [ json.dumps( {"object": record.get("object"), "match": record.get("match")}, sort_keys=True, separators=(",", ":"), ) for record in records ] self.assertEqual(len(signatures), len(set(signatures))) self.assertFalse( any( record["key"].startswith( ( "original_staging_pricebook_", "original_staging_pricebook_entry_", "original_staging_product_", ) ) for record in records ) ) def test_exp_baseline_does_not_include_retired_staging_snapshots(self): blocked_prefixes = ( "original_staging_", "staging_account_", "staging_contact_", "staging_lead_", "staging_campaign_", ) blocked_records = [ record["key"] for record in load_exp_seed_records() if record["key"].startswith(blocked_prefixes) ] self.assertEqual(blocked_records, []) def test_exp_ui_connected_app_grants_scim_for_identity_sync_token_minting(self): by_key = {record["key"]: record for record in load_exp_seed_records()} connected_app = by_key["staging_watchdog_connected_app_run_as"] self.assertEqual(connected_app["object"], "_System_ConnectedApp") self.assertEqual(connected_app["match"], {"ConsumerKey": "nexuscrm-ui"}) self.assertTrue(connected_app["fields"]["IsActive"]) self.assertTrue( {"api", "refresh_token", "offline_access", "scim"}.issubset( set(connected_app["fields"]["Scopes"].split()) ) ) def test_exp_seed_contains_hubspot_default_owner_fallback(self): records = load_exp_seed_records() by_key = {record.get("key"): record for record in records} integration_user = by_key.get("exp_hubspot_integration_user") self.assertIsNotNone(integration_user) self.assertEqual(integration_user["object"], "User") self.assertEqual( integration_user["match"], {"Username": "hubspot-integration-user@example.invalid"}, ) self.assertEqual( integration_user["fields"]["Email"], "hubspot-integration-user@example.invalid", ) self.assertEqual(integration_user["fields"]["Alias"], "hsiu") self.assertEqual(integration_user["fields"]["ProfileId"], "system_admin") self.assertIs(integration_user["fields"]["IsActive"], True) default_binding = by_key.get("exp_hubspot_default_owner_binding") self.assertIsNotNone(default_binding) self.assertEqual(default_binding["object"], "HubSpotOwnerBinding__c") self.assertEqual( default_binding["match"], {"HubSpotOwnerId__c": "INTEGRATION_DEFAULT"}, ) self.assertEqual(default_binding["fields"]["HubSpotOwnerId__c"], "INTEGRATION_DEFAULT") self.assertEqual(default_binding["fields"]["BindingStatus__c"], "default") self.assertEqual(default_binding["fields"]["CRMPrincipalType__c"], "User") self.assertEqual( default_binding["fields"]["CRMPrincipalId__c"], "{{ record.exp_hubspot_integration_user.id }}", ) self.assertEqual(default_binding["fields"]["SourceOfTruth__c"], "manual") def test_hs07b_campaign_member_e2e_fixture_is_run_scoped(self): fixture = json.loads(HS07B_RUN_SEED.read_text(encoding="utf-8")) self.assertEqual(fixture["version"], 1) self.assertEqual(len(fixture["baseline_records"]), 1) campaign = fixture["baseline_records"][0] self.assertEqual(campaign["key"], "hs07b_workflow_action_campaign") self.assertEqual(campaign["object"], "Campaign") self.assertEqual( campaign["match"], {"Name": "EXP HS-07B Workflow Action Campaign {{ run_id }}"}, ) self.assertEqual( campaign["fields"]["Name"], "EXP HS-07B Workflow Action Campaign {{ run_id }}", ) self.assertEqual(campaign["fields"]["Status"], "In Progress") self.assertEqual(campaign["fields"]["Type"], "Webinar") self.assertIs(campaign["fields"]["IsActive"], True) exp_keys = {record["key"] for record in load_exp_seed_records()} self.assertNotIn("hs07b_workflow_action_campaign", exp_keys) self.assertNotIn("hs07b_workflow_action_status_sent", exp_keys) self.assertNotIn("hs07b_workflow_action_status_responded", exp_keys) campaign_object = json.loads(CAMPAIGN_OBJECT.read_text(encoding="utf-8"))[0] self.assertEqual( campaign_object["defaultChildRows"], [ { "targetObject": "CampaignMemberStatus", "targetLookupField": "CampaignId", "fieldValues": { "Label": "Sent", "IsDefault": True, "HasResponded": False, "SortOrder": 1, }, }, { "targetObject": "CampaignMemberStatus", "targetLookupField": "CampaignId", "fieldValues": { "Label": "Responded", "IsDefault": False, "HasResponded": True, "SortOrder": 2, }, }, ], ) def test_hs07b_full_e2e_does_not_preseed_campaign_members(self): hs07b_markers = ( "hs07b", "HS-07B", "EXP HS-07B Workflow Action Campaign", "{{ record.hs07b_workflow_action_campaign.id }}", ) hs07b_members = [ record for record in json.loads(HS07B_RUN_SEED.read_text())["baseline_records"] if record.get("object") == "CampaignMember" and any(marker in str(record) for marker in hs07b_markers) ] self.assertEqual(hs07b_members, []) def test_exp_baseline_has_no_orphan_report_or_dashboard_fixture(self): by_key = {record["key"]: record for record in load_exp_seed_records()} self.assertNotIn("staging_watchdog_report", by_key) self.assertNotIn("staging_watchdog_dashboard", by_key) self.assertFalse( any(record["object"] in {"Report", "Dashboard"} for record in by_key.values()) ) def test_exp_seed_does_not_write_pack_owned_approval_metadata(self): blocked_seed_objects = {"ProcessDefinition", "ProcessNode", "ProcessAction"} records = load_exp_seed_records() seeded_metadata = [ record for record in records if record.get("object") in blocked_seed_objects ] self.assertEqual(seeded_metadata, []) def test_exp_seed_does_not_write_pingcap_region_owner_rules(self): region_owner_rows = [ record for record in load_exp_seed_records() if record.get("object") == "Partner_Region_Owner_Rule__c" ] self.assertEqual(region_owner_rows, []) def test_exp_owner_binding_bootstrap_mapping_is_unique_and_complete(self): rows = load_exp_owner_binding_bootstrap() self.assertEqual(len(rows), 52) self.assertEqual(len({row["hubspot_owner_id"] for row in rows}), 52) self.assertEqual(len({row["normalized_email"] for row in rows}), 52) for row in rows: self.assertEqual(row["crm_principal_type"], "User") self.assertEqual(row["binding_status"], "bound") self.assertEqual(row["normalized_email"], row["normalized_email"].lower()) def test_pingcap_region_owner_rules_are_lookup_backed_reference_data(self): records = load_pingcap_region_owner_records() lookups = {lookup["key"]: lookup for lookup in load_pingcap_region_owner_lookups()} self.assertEqual(len(records), 6) self.assertEqual(len(lookups), 6) seen_regions = set() seen_lookup_refs = set() for record in records: self.assertEqual(record["object"], "Partner_Region_Owner_Rule__c") region = record["match"]["Region__c"] self.assertEqual(record["fields"]["Region__c"], region) self.assertNotIn(region, seen_regions) seen_regions.add(region) owner_ref = record["fields"]["Owner_Id__c"] self.assertTrue(owner_ref.startswith("{{ lookup.")) self.assertTrue(owner_ref.endswith(".id }}")) self.assertNotRegex(owner_ref, r"^005[A-Za-z0-9]{12,15}$") lookup_key = owner_ref.removeprefix("{{ lookup.").removesuffix(".id }}") self.assertIn(lookup_key, lookups) self.assertEqual(lookups[lookup_key]["object"], "User") owner_suffix = lookup_key.removeprefix("pingcap_region_owner_user_").replace("_", "-") self.assertEqual( lookups[lookup_key]["match"]["Username"], f"pingcap-region-owner-{owner_suffix}@example.invalid", ) self.assertIs(lookups[lookup_key]["match"]["IsActive"], True) seen_lookup_refs.add(lookup_key) self.assertEqual(seen_lookup_refs, set(lookups)) def test_deploy_verification_derives_pingcap_region_owner_rules_from_seed(self): module = load_deploy_offline_env_module() rules = module.load_pingcap_region_owner_rules(PINGCAP_REGION_OWNER_RULES_SEED) records = load_pingcap_region_owner_records() lookups = {lookup["key"]: lookup for lookup in load_pingcap_region_owner_lookups()} self.assertEqual(len(rules), len(records)) self.assertEqual( rules, [ { "region": record["match"]["Region__c"], "owner_name": record["fields"]["Owner_Name__c"], "owner_lookup_key": record["fields"]["Owner_Id__c"] .removeprefix("{{ lookup.") .removesuffix(".id }}"), "owner_lookup_match": lookups[ record["fields"]["Owner_Id__c"] .removeprefix("{{ lookup.") .removesuffix(".id }}") ]["match"], } for record in records ], )