import json import os import subprocess import tempfile import textwrap import unittest from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] FLOW_PATH = ( REPO_ROOT / "packs" / "PingCAP-HubSpot-Companion" / "flows" / "HubSpot_Contact_Lead_Upsert.flow.json" ) GO_RUNTIME_TEST = r''' package formula import "testing" func TestL2HubSpotPortalIDNormalization(t *testing.T) { engine := NewEngine() const expression = `__L2_EXPRESSION__` for _, tc := range []struct { name string portalID interface{} want string }{ {name: "JSON number", portalID: float64(12345678), want: "12345678"}, {name: "JSON number with trailing zeros", portalID: float64(12000000), want: "12000000"}, {name: "JSON number with e+08 exponent", portalID: float64(123456789), want: "123456789"}, {name: "maximum exactly representable JSON integer", portalID: float64(9007199254740991), want: "9007199254740991"}, {name: "JSON number below scientific threshold", portalID: float64(123456), want: "123456"}, {name: "string", portalID: "12345678", want: "12345678"}, } { t.Run(tc.name, func(t *testing.T) { got, err := engine.Evaluate(expression, &Context{ Variables: map[string]interface{}{ "payload": map[string]interface{}{"portalId": tc.portalID}, }, }) if err != nil { t.Fatalf("Evaluate(%T(%v)) error = %v", tc.portalID, tc.portalID, err) } if got != tc.want { t.Fatalf("Evaluate(%T(%v)) = %#v, want %q", tc.portalID, tc.portalID, got, tc.want) } }) } } ''' GO_PROGRAM_CHECK_TEST = r''' package programchecker import ( "context" "testing" "github.com/nexuscrm/kernel/pkg/formula" "github.com/nexuscrm/shared/pkg/models" ) func TestL2HubSpotPortalIDNormalizationPassesFlowProgramCheck(t *testing.T) { formulas, err := formula.ExecutableFunctionRegistryWithBundles(nil) if err != nil { t.Fatalf("ExecutableFunctionRegistryWithBundles: %v", err) } const expression = `__L2_EXPRESSION__` diags := CheckFlowMetadata(context.Background(), models.FlowMetadata{ Variables: []models.FlowVariable{ {Name: "payload", DataType: "Any", IsInput: true}, }, Formulas: []models.FlowFormula{ {Name: "normalizedPortalId", DataType: "String", Expression: expression}, }, }, FlowCheckOptions{ Metadata: newFakeProgramMetadataReader(), Formula: NewFormulaCatalogReader(formulas), ComponentPath: "flows/HubSpot_Contact_Lead_Upsert.flow.json", OwnerPack: "pingcap-hubspot-companion", }) if len(diags) != 0 { t.Fatalf("CheckFlowMetadata diagnostics = %#v, want none", diags) } } ''' class HubSpotPortalIdNormalizationTest(unittest.TestCase): def test_numeric_portal_id_is_normalized_by_current_l0_formula_runtime(self): flow = json.loads(FLOW_PATH.read_text(encoding="utf-8")) formula = next( item for item in flow["formulas"] if item["name"] == "normalizedPortalId" ) self.assertEqual(formula["dataType"], "String") nexuscrm_root = os.environ.get("NEXUSCRM_ROOT") if not nexuscrm_root: self.fail("NEXUSCRM_ROOT must point to the exact NexusCRM checkout under test") upstream = Path(nexuscrm_root).resolve() formula_dir = upstream / "nexuscrm-kernel" / "pkg" / "formula" self.assertTrue(formula_dir.is_dir(), f"formula package not found under {upstream}") expected_head = os.environ.get("NEXUSCRM_EXPECTED_HEAD") if expected_head: actual_head = subprocess.check_output( ["git", "rev-parse", "HEAD"], cwd=upstream, text=True ).strip() self.assertEqual(actual_head, expected_head) with tempfile.TemporaryDirectory(prefix="nexuscrm-l2-portal-id-") as temp_dir: temp = Path(temp_dir) backing_test = temp / "l2_hubspot_portal_id_test.go" backing_test.write_text( textwrap.dedent(GO_RUNTIME_TEST).replace( "__L2_EXPRESSION__", formula["expression"] ), encoding="utf-8", ) overlay_target = formula_dir / "l2_hubspot_portal_id_test.go" overlay = temp / "overlay.json" overlay.write_text( json.dumps({"Replace": {str(overlay_target): str(backing_test)}}), encoding="utf-8", ) completed = subprocess.run( [ "go", "test", f"-overlay={overlay}", "./pkg/formula", "-run", "^TestL2HubSpotPortalIDNormalization$", "-count=1", ], cwd=upstream / "nexuscrm-kernel", env={**os.environ, "GOWORK": "off"}, text=True, capture_output=True, check=False, ) self.assertEqual( completed.returncode, 0, f"go test failed\nstdout:\n{completed.stdout}\nstderr:\n{completed.stderr}", ) def test_formula_passes_current_l0_flow_program_checker(self): flow = json.loads(FLOW_PATH.read_text(encoding="utf-8")) formula_definition = next( item for item in flow["formulas"] if item["name"] == "normalizedPortalId" ) nexuscrm_root = os.environ.get("NEXUSCRM_ROOT") if not nexuscrm_root: self.fail("NEXUSCRM_ROOT must point to the exact NexusCRM checkout under test") upstream = Path(nexuscrm_root).resolve() checker_dir = upstream / "nexuscrm-metadata" / "programchecker" self.assertTrue(checker_dir.is_dir(), f"programchecker package not found under {upstream}") expected_head = os.environ.get("NEXUSCRM_EXPECTED_HEAD") if expected_head: actual_head = subprocess.check_output( ["git", "rev-parse", "HEAD"], cwd=upstream, text=True ).strip() self.assertEqual(actual_head, expected_head) with tempfile.TemporaryDirectory(prefix="nexuscrm-l2-portal-id-check-") as temp_dir: temp = Path(temp_dir) backing_test = temp / "l2_hubspot_portal_id_program_check_test.go" backing_test.write_text( textwrap.dedent(GO_PROGRAM_CHECK_TEST).replace( "__L2_EXPRESSION__", formula_definition["expression"] ), encoding="utf-8", ) overlay_target = checker_dir / "l2_hubspot_portal_id_program_check_test.go" overlay = temp / "overlay.json" overlay.write_text( json.dumps({"Replace": {str(overlay_target): str(backing_test)}}), encoding="utf-8", ) completed = subprocess.run( [ "go", "test", f"-overlay={overlay}", "./programchecker", "-run", "^TestL2HubSpotPortalIDNormalizationPassesFlowProgramCheck$", "-count=1", ], cwd=upstream / "nexuscrm-metadata", env={**os.environ, "GOWORK": "off"}, text=True, capture_output=True, check=False, ) self.assertEqual( completed.returncode, 0, f"go test failed\nstdout:\n{completed.stdout}\nstderr:\n{completed.stderr}", ) def test_all_portal_id_consumers_use_the_normalized_string_formula(self): flow = json.loads(FLOW_PATH.read_text(encoding="utf-8")) normalized_reference = {"elementReference": "normalizedPortalId"} binding_lookup = next( item for item in flow["recordLookups"] if item["name"] == "FindActiveSalesforceIdentityBindingForShell" ) portal_filter = next( item for item in binding_lookup["filters"] if item["field"] == "PortalId__c" ) self.assertEqual(portal_filter["value"], normalized_reference) claim_subflows = { item["name"]: item for item in flow["subflows"] if item["name"] in {"ClaimLeadSalesforceIdentity", "ClaimContactSalesforceIdentity"} } self.assertEqual( set(claim_subflows), {"ClaimLeadSalesforceIdentity", "ClaimContactSalesforceIdentity"}, ) for name, subflow in claim_subflows.items(): with self.subTest(subflow=name): portal_input = next( item for item in subflow["inputAssignments"] if item["name"] == "portalId" ) self.assertEqual(portal_input["value"], normalized_reference) element_references = [] def collect_element_references(value): if isinstance(value, dict): if "elementReference" in value: element_references.append(value["elementReference"]) for child in value.values(): collect_element_references(child) elif isinstance(value, list): for child in value: collect_element_references(child) collect_element_references(flow) self.assertNotIn("payload.portalId", element_references) self.assertEqual(element_references.count("normalizedPortalId"), 3) if __name__ == "__main__": unittest.main()