"""L2 contract for the NetSuite revenue-voucher source-fact mirror.""" import json import re import unittest from collections import defaultdict from pathlib import Path ROOT = Path(__file__).resolve().parents[1] L2_PACK = ROOT / "packs" / "Salesforce-PingCAP-L2" OBJECT_API_NAME = "NetSuite_Revenue_Voucher_Line__c" OBJECT_PATH = L2_PACK / "objects" / f"{OBJECT_API_NAME}.json" FINAL_OBJECTS_PATH = ( ROOT / "packs" / "NOT-EDIT-final-merged-packs" / "objects" / "standard_objects_spec.json" ) KEY_PREFIX_MIGRATION_PATH = ( ROOT / "docs" / "migration" / "patch-bundles" / "20260820_netsuite_key_prefix_upgrade" / "001_converge_unused_netsuite_key_prefix.sql" ) OWNERLESS_REPLAY_RECOVERY_PATH = ( ROOT / "docs" / "migration" / "patch-bundles" / "20260821_recreate_empty_netsuite_ownerless_surface" / "001_remove_exact_empty_ownerless_surface.sql" ) BUSINESS_FIELDS = { "Transaction_Type__c": "string", "Company__c": "string", "Transaction_Date__c": "date", "Document_Number__c": "string", "Posting_Period__c": "string", "Transaction_Number__c": "string", "Description__c": "textarea", "GL_Account_Number__c": "string", "GL_Account_Name__c": "string", "Transaction_Currency__c": "string", "Transaction_Amount__c": "currency", "Exchange_Rate__c": "double", "Base_Amount__c": "currency", "Cost_Center_Code__c": "string", "Cost_Center_Formula__c": "string", "Entity_Name__c": "string", "Source_Customer_Account_Id__c": "string", "Mgmt_Report_L1_Formula__c": "string", "Mgmt_Report_L2_Formula__c": "string", "Mgmt_Report_L1__c": "string", "Mgmt_Report_L2__c": "string", "Department_No_Hierarchy__c": "string", "First_Level_Department__c": "string", "Trading_Partner_Name__c": "string", "Main_Memo__c": "textarea", "Customer_Region__c": "string", "Order_Number__c": "string", "Support_Type__c": "string", "Group_COA__c": "string", "Source_Transaction_Modified_At__c": "datetime", "Source_Line_Modified_At__c": "datetime", "GL_Account_Type__c": "string", "Product_Name__c": "string", "Product_Category__c": "string", "Cloud_Provider__c": "string", "Project__c": "string", } CONTROL_FIELD_TYPES = { "NetSuite_Source_Key__c": "string", "NetSuite_Account_Realm__c": "string", "NetSuite_Transaction_Id__c": "string", "NetSuite_Line_Unique_Key__c": "string", "NetSuite_Accounting_Book_Id__c": "string", "NetSuite_Posting_Period_Id__c": "string", "Source_Search_Id__c": "string", "Source_Schema_Version__c": "string", "Source_Payload_Hash__c": "string", "First_Seen_Run_Id__c": "string", "First_Seen_At__c": "datetime", "Last_Seen_Run_Id__c": "string", "Last_Seen_At__c": "datetime", "Is_Source_Active__c": "boolean", "CRM_Account__c": "reference", } REQUIRED_FIELD_NAMES = { "Name", "Source_Customer_Account_Id__c", "NetSuite_Source_Key__c", "NetSuite_Account_Realm__c", "NetSuite_Transaction_Id__c", "NetSuite_Line_Unique_Key__c", "NetSuite_Posting_Period_Id__c", "Source_Search_Id__c", "Source_Schema_Version__c", "Source_Payload_Hash__c", "First_Seen_Run_Id__c", "First_Seen_At__c", "Last_Seen_Run_Id__c", "Last_Seen_At__c", "Is_Source_Active__c", "CRM_Account__c", } EXPECTED_INDICES = [ {"name": "idx_nrv_account", "columns": ["CRM_Account__c"]}, { "name": "idx_nrv_period_active", "columns": [ "NetSuite_Posting_Period_Id__c", "Is_Source_Active__c", ] }, ] SEARCH_ALL_DEFAULT_FIELDS = [ "Id", "Name", "NetSuite_Source_Key__c", "Document_Number__c", "Transaction_Number__c", "Source_Customer_Account_Id__c", "Company__c", "Entity_Name__c", "GL_Account_Number__c", "GL_Account_Name__c", "Posting_Period__c", "Order_Number__c", "Product_Name__c", "Project__c", ] def load_object() -> dict: payload = json.loads(OBJECT_PATH.read_text(encoding="utf-8")) if not isinstance(payload, list) or len(payload) != 1: raise AssertionError(f"{OBJECT_PATH} must contain exactly one object spec") return payload[0] def load_final_object() -> dict: payload = json.loads(FINAL_OBJECTS_PATH.read_text(encoding="utf-8")) return next(item for item in payload if item.get("apiName") == OBJECT_API_NAME) def all_l2_objects() -> list[dict]: objects = [] for path in sorted((L2_PACK / "objects").glob("*.json")): payload = json.loads(path.read_text(encoding="utf-8")) objects.extend(payload if isinstance(payload, list) else [payload]) return objects class NetSuiteRevenueVoucherObjectContractTest(unittest.TestCase): def test_manifest_registers_the_l2_object_source(self) -> None: manifest = json.loads((L2_PACK / "pack.json").read_text(encoding="utf-8")) matches = [ component for component in manifest["components"] if component.get("kind") == "object_spec" and component.get("businessKey") == f"object:{OBJECT_API_NAME}" ] self.assertEqual( matches, [ { "kind": "object_spec", "businessKey": f"object:{OBJECT_API_NAME}", "manageability": "locked", "path": f"objects/{OBJECT_API_NAME}.json", } ], ) def test_all_36_business_fields_are_present_with_design_types(self) -> None: obj = load_object() fields = {field["apiName"]: field for field in obj["fields"]} self.assertEqual(len(BUSINESS_FIELDS), 36) self.assertEqual(len(CONTROL_FIELD_TYPES), 15) self.assertEqual( {name: fields[name]["type"] for name in BUSINESS_FIELDS}, BUSINESS_FIELDS, ) self.assertEqual( set(fields), {"Name", *BUSINESS_FIELDS, *CONTROL_FIELD_TYPES}, ) self.assertEqual(len(fields), 52) def test_identity_sharing_and_required_account_contract(self) -> None: obj = load_object() fields = {field["apiName"]: field for field in obj["fields"]} self.assertEqual(obj["category"], "crm") self.assertEqual(obj["sharingModel"], "Private") self.assertIn("customsearch1710", obj["description"]) self.assertIn("realm:transaction:lineUniqueKey", obj["description"]) self.assertIn("authenticated NexusCRM Composite external-ID API", obj["description"]) self.assertNotIn("TiDB", obj["description"]) self.assertEqual( fields["Name"], { "apiName": "Name", "label": "Name", "type": "string", "isNameField": True, "required": True, }, ) self.assertEqual( {name: fields[name]["type"] for name in CONTROL_FIELD_TYPES}, CONTROL_FIELD_TYPES, ) self.assertEqual( {name for name, field in fields.items() if field.get("required") is True}, REQUIRED_FIELD_NAMES, ) self.assertTrue(fields["Source_Customer_Account_Id__c"]["required"]) self.assertEqual(fields["NetSuite_Source_Key__c"]["type"], "string") self.assertIs(fields["NetSuite_Source_Key__c"].get("required"), True) self.assertIs(fields["NetSuite_Source_Key__c"].get("unique"), True) self.assertEqual(fields["Source_Search_Id__c"].get("default"), "customsearch1710") self.assertEqual( fields["Source_Schema_Version__c"].get("default"), "revenue-voucher-excel/v1", ) self.assertEqual(fields["Is_Source_Active__c"].get("default"), "1") self.assertEqual( fields["CRM_Account__c"], { "apiName": "CRM_Account__c", "label": "CRM Account", "type": "reference", "referenceTo": ["Account"], "relationshipName": "CRM_Account__r", "childRelationshipName": "NetSuite_Revenue_Voucher_Lines", "deleteConstraint": "Restrict", "required": True, }, ) self.assertNotIn("Account_Match_Status__c", fields) def test_required_indices_are_authored(self) -> None: indices = load_object().get("indices") self.assertEqual(indices, EXPECTED_INDICES) self.assertTrue(all(len(index["name"]) <= 64 for index in indices)) def test_generated_final_object_matches_l2_source(self) -> None: self.assertEqual(load_final_object(), load_object()) def test_key_prefix_is_valid_and_unique_in_l2_source(self) -> None: obj = load_object() # a0Z is already owned by the historically installed # HubSpotContactMergeAlias__c object in upgraded PingCAP catalogs. # This new object must use a distinct, stable tenant-authored prefix; # otherwise a clean install passes while an in-place upgrade creates # ambiguous polymorphic-reference ownership. self.assertEqual(obj["keyPrefix"], "a1Z") self.assertRegex(obj["keyPrefix"], r"^[A-Za-z0-9]{3}$") owners_by_prefix = defaultdict(list) for candidate in all_l2_objects(): prefix = candidate.get("keyPrefix") if prefix: owners_by_prefix[prefix].append(candidate["apiName"]) duplicates = { prefix: sorted(owners) for prefix, owners in owners_by_prefix.items() if len(owners) > 1 } self.assertEqual(duplicates, {}) hubspot_alias_path = ( ROOT / "packs" / "PingCAP-HubSpot-Companion" / "objects" / "HubSpotContactMergeAlias__c.json" ) hubspot_alias = json.loads(hubspot_alias_path.read_text(encoding="utf-8"))[0] self.assertEqual(hubspot_alias["keyPrefix"], "a0Z") self.assertNotEqual(hubspot_alias["keyPrefix"], obj["keyPrefix"]) def test_key_prefix_migration_proves_only_its_owned_transition(self) -> None: migration = KEY_PREFIX_MIGRATION_PATH.read_text(encoding="utf-8") self.assertIn("@final_netsuite_a0z_rows <> 0", migration) self.assertIn("@final_unexpected_a0z_owners <> 0", migration) self.assertIn("@final_unexpected_a1z_owners <> 0", migration) self.assertNotIn("final_duplicate_prefixes", migration) self.assertNotIn("GROUP BY BINARY `KeyPrefix`", migration) def test_ownerless_replay_recovery_is_exact_empty_and_fail_closed(self) -> None: recovery = OWNERLESS_REPLAY_RECOVERY_PATH.read_text(encoding="utf-8") self.assertIn( "SET @target_api_name = 'NetSuite_Revenue_Voucher_Line__c'", recovery, ) self.assertIn("BINARY `KeyPrefix` = BINARY 'a1Z'", recovery) self.assertIn("@object_rows = 1", recovery) self.assertIn("@expected_object_rows = 1", recovery) self.assertIn("@table_rows = 1", recovery) self.assertIn("@expected_table_rows = 1", recovery) self.assertIn("@field_rows > 0", recovery) self.assertIn("@physical_table_rows = 1", recovery) self.assertIn("@physical_record_rows = 0", recovery) self.assertNotIn("SELECT COUNT(*) INTO @physical_record_rows", recovery) self.assertIn( "DROP TABLE IF EXISTS `NetSuite_Revenue_Voucher_Line__c`", recovery, ) self.assertNotIn("LIKE '%NetSuite%'", recovery) def test_search_all_defaults_are_explicit_text_fields_within_runtime_cap(self) -> None: manifest = json.loads((L2_PACK / "pack.json").read_text(encoding="utf-8")) component = next( component for component in manifest["components"] if component.get("kind") == "search_all_default_fields" and component.get("businessKey") == "pingcap_search_all_default_fields" ) declarations = json.loads( (L2_PACK / component["path"]).read_text(encoding="utf-8") ) fields = next( item["fields"] for item in declarations if item.get("objectApiName") == OBJECT_API_NAME ) object_fields = {field["apiName"]: field for field in load_object()["fields"]} self.assertIn( {"kind": "object_spec", "businessKey": f"object:{OBJECT_API_NAME}"}, component["dependsOn"], ) self.assertEqual(fields, SEARCH_ALL_DEFAULT_FIELDS) self.assertLessEqual(len(fields), 32) self.assertTrue( all( field_name == "Id" or object_fields[field_name]["type"] in {"string", "textarea"} for field_name in fields ) ) if __name__ == "__main__": unittest.main()