"""Cross-artifact invariants for the quote-side term derivation. Every assertion here compares two live artifacts inside the pack against each other - none authors an expected business value in the test and compares it back to the JSON (the caller-free snapshot shape deleted by #5008). Behavior (clearing on incomplete dates, the gate actually blocking a submit) is owned by the local bootstrap oracle, not by this file. The invariants and why each is relational rather than copied: 1. The line Flow and Quote header formula must both call the same pack-owned SERVICE_WINDOW_TERM_MONTHS function with their corresponding start/end fields. The bundle is compared across source and final pack roots, while the companion semantic test evaluates its committed body against an independent calendar reference. This keeps one business-rule owner without copying the leap-year/month-end implementation into each consumer. 2. The derivation must run after every QuoteLineItem autofill flow and before every OTHER QuoteLineItem flow that reads the stamped term - readers are discovered from flow content ($Record.Term_Months__c), not from a hardcoded filename, so a future second reader is bound automatically. Flows are discovered by start.object, not by filename glob. The derivation must also have no upstream writer of its own date inputs (asserted empty so a future date-writing flow forces this test to be revisited rather than silently ordering wherever it lands). 3. Every COUNT rollup on Quote that counts a nullable column must carry a filter that is exactly " IS NOT NULL". The rollup engine has two execution paths that disagree about NULLs: the incremental delta path counts every linked child matching the FILTER and never consults the summary field, while the full-recompute path runs SQL COUNT over the column, which skips NULLs. Filter == summary-field-not-null makes the two paths agree by construction (observed live: without the filter the delta path counted a NULL-Net_fees line, so both gate counts equalled the line count and the submit gate never fired). The two gate rollups' summary fields are pinned to the columns the derivation and Net Fees flows actually assign, so retargeting either flow or flipping a summaryField to Id breaks the test. 4. The submit gate may reference only physical columns: every identifier in its condition that resolves to a Quote field must not be an active-mode formula field (the NX#3913 regression class), every __c identifier it names must resolve (the build sanitizer silently DROPS rollup fields missing from the wave6 allowlist - a gate referencing a dropped column would never match), and the gate must read at least two rollupsummary fields (so it cannot be gutted to a Status-only tautology and stay green). 5. The derivation flow, the gate rule, the line lock rule and the three count rollupConfigs must compare EQUAL across the L2 source and the final merged pack. Per-root internal consistency does not imply cross-root agreement: both roots can drift together-internally-consistent (e.g. the merged copy alone rebuilt with a different divisor) and every per-root invariant above stays green. Deliberately absent: a guard on the Create-Quote mapping no longer copying the term. createable: false does not stop a flow-pipeline write (only REST writes are stripped), so a resurrected copy would land - and then be overwritten in the same save by the order-150 derivation, which is the actual defense. The mapping removal is hygiene; the behavioral guarantee belongs to the bootstrap oracle, not a static snapshot. """ import json import re import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[1] PACK_ROOTS = [ ROOT / "packs" / "Salesforce-PingCAP-L2", ROOT / "packs" / "NOT-EDIT-final-merged-packs", ] DERIVATION_FLOW = "QuoteLineItem_Term_Derivation.flow.json" GATE_NAME = "Quote_Submit_Requires_Complete_Lines" LOCK_NAME = "QuoteLineItem_Locked_After_Quote_Leaves_Draft" def load_json(path: Path): return json.loads(path.read_text(encoding="utf-8")) def object_spec(pack_root: Path, api_name: str) -> dict: combined = pack_root / "objects" / "standard_objects_spec.json" if combined.exists(): spec = load_json(combined) objects = spec.get("objects") if isinstance(spec, dict) else spec return next(o for o in objects if o.get("apiName") == api_name) payload = load_json(pack_root / "objects" / f"{api_name}.json") return payload[0] if isinstance(payload, list) else payload def field_map(obj: dict) -> dict: return {f.get("apiName"): f for f in obj.get("fields", []) if f.get("apiName")} def validation_rules(pack_root: Path) -> list: return load_json( pack_root / "validation" / "standard_validation_rules.json" )["validationRules"] def rule_by_name(pack_root: Path, name: str) -> dict: return next(r for r in validation_rules(pack_root) if r.get("name") == name) def qli_flows(pack_root: Path) -> dict: """Every flow whose start.object is QuoteLineItem, discovered by content.""" found = {} for path in sorted((pack_root / "flows").glob("*.flow.json")): flow = load_json(path) if flow.get("start", {}).get("object") == "QuoteLineItem": found[path.name] = flow return found def stamp_target(flow: dict, assignment_name: str) -> str: assignment = next(a for a in flow["assignments"] if a["name"] == assignment_name) ref = assignment["assignmentItems"][0]["assignToReference"] assert ref.startswith("$Record."), ref return ref[len("$Record."):] def assigned_final_formula(flow: dict) -> str: """The formula the Stamp_Term assignment actually writes - never [-1].""" assignment = next(a for a in flow["assignments"] if a["name"] == "Stamp_Term") name = assignment["assignmentItems"][0]["value"]["elementReference"] return next(f["expression"] for f in flow["formulas"] if f["name"] == name) class QuoteTermDerivationInvariants(unittest.TestCase): def test_object_component_does_not_depend_on_later_phase_function_bundle(self): dependency = { "kind": "formula_function_bundle", "businessKey": "pingcap_quote_functions", } for pack_root, object_key in ( (PACK_ROOTS[0], "object:Quote"), (PACK_ROOTS[1], "standard_objects_spec:crm"), ): with self.subTest(pack=pack_root.name): manifest = load_json(pack_root / "pack.json") component = next( item for item in manifest["components"] if item.get("kind") == "object_spec" and item.get("businessKey") == object_key ) self.assertNotIn(dependency, component.get("dependsOn", [])) def test_line_and_header_use_the_single_pack_owned_term_function(self): for pack_root in PACK_ROOTS: with self.subTest(pack=pack_root.name): header = field_map(object_spec(pack_root, "Quote"))["Term_months__c"]["formula"] flow = load_json(pack_root / "flows" / DERIVATION_FLOW) final = assigned_final_formula(flow) self.assertEqual( final, "SERVICE_WINDOW_TERM_MONTHS({!$Record.Service_Start_Date__c}, {!$Record.Service_End_Date__c})", ) self.assertEqual( header, "SERVICE_WINDOW_TERM_MONTHS(Overall_Start_Date__c, Overall_End_Date__c)", ) bundle = load_json( pack_root / "formula_functions" / "pingcap_quote_functions.json" ) function = next( f for f in bundle["functions"] if f.get("name") == "SERVICE_WINDOW_TERM_MONTHS" ) self.assertEqual(function["minArgs"], 2) self.assertEqual(function["maxArgs"], 2) self.assertEqual(function["return"], {"kind": "fixed", "type": "number"}) self.assertEqual( function["args"], [{"dateLikeLiteral": True}, {"dateLikeLiteral": True}], ) def test_derivation_runs_between_autofill_and_every_term_reader(self): for pack_root in PACK_ROOTS: with self.subTest(pack=pack_root.name): flows = qli_flows(pack_root) self.assertIn(DERIVATION_FLOW, flows) flow = flows[DERIVATION_FLOW] start = flow["start"] self.assertEqual(flow["status"], "Active") self.assertEqual(start["triggerType"], "RecordBeforeSave") self.assertEqual(start["recordTriggerType"], "CreateAndUpdate") order = start["triggerOrder"] readers, date_writers, autofill_orders = [], [], [] for name, other in flows.items(): if name == DERIVATION_FLOW: continue raw = json.dumps(other) if "$Record.Term_Months__c" in raw: readers.append((name, other["start"]["triggerOrder"])) for assignment in other.get("assignments", []): for item in assignment.get("assignmentItems", []): if item.get("assignToReference") in ( "$Record.Service_Start_Date__c", "$Record.Service_End_Date__c", ): date_writers.append(name) if "Autofill" in name: autofill_orders.append(other["start"]["triggerOrder"]) self.assertTrue(readers, "no QuoteLineItem flow reads the stamped term") self.assertLess( order, min(o for _, o in readers), f"derivation must run before every term reader: {readers}", ) self.assertEqual( date_writers, [], "a QuoteLineItem flow now WRITES a service date - the " "derivation's lower bound must be re-derived, not assumed", ) self.assertTrue(autofill_orders, "QuoteLineItem autofill chain not found") self.assertGreater(order, max(autofill_orders)) def test_count_rollup_filter_matches_summary_field(self): for pack_root in PACK_ROOTS: with self.subTest(pack=pack_root.name): fields = field_map(object_spec(pack_root, "Quote")) count_rollups = { f["apiName"]: f for f in fields.values() if str(f.get("type", "")).lower() == "rollupsummary" and str(f.get("rollupConfig", {}).get("calcType", "")).upper() == "COUNT" } self.assertTrue(count_rollups, "no COUNT rollups found on Quote") for f in count_rollups.values(): config = f["rollupConfig"] summary_field = config.get("summaryField", "") if summary_field in ("", "Id"): continue # counting rows, not a nullable column self.assertEqual( config.get("filter"), f"{summary_field} IS NOT NULL", f"{f['apiName']}: COUNT delta path counts filter matches " "and ignores summary-field NULLs; the filter must mirror " "the summary field for the two engine paths to agree", ) flows = qli_flows(pack_root) derivation_target = stamp_target(flows[DERIVATION_FLOW], "Stamp_Term") self.assertEqual( count_rollups["Termed_Line_Item_Count__c"]["rollupConfig"]["summaryField"], derivation_target, "Termed count must count exactly the column the derivation stamps", ) net_fees_targets = { item["assignToReference"][len("$Record."):] for other in flows.values() for assignment in other.get("assignments", []) for item in assignment.get("assignmentItems", []) if item.get("assignToReference", "").startswith("$Record.Net_fees") } self.assertEqual(len(net_fees_targets), 1, net_fees_targets) self.assertEqual( count_rollups["Priced_Line_Item_Count__c"]["rollupConfig"]["summaryField"], net_fees_targets.pop(), "Priced count must count exactly the column the money writer stamps", ) def test_submit_gate_references_no_virtual_formula_field(self): for pack_root in PACK_ROOTS: with self.subTest(pack=pack_root.name): rule = rule_by_name(pack_root, GATE_NAME) self.assertTrue(rule["active"]) self.assertEqual(rule["objectApiName"], "Quote") quote_fields = field_map(object_spec(pack_root, "Quote")) identifiers = set(re.findall( r"[A-Za-z_][A-Za-z0-9_]*__c|\b[A-Za-z]\w*\b", rule["condition"] )) custom = {n for n in identifiers if n.endswith("__c")} self.assertEqual( custom - set(quote_fields), set(), "gate names a Quote field that does not exist - the build " "sanitizer drops non-allowlisted rollups SILENTLY, so an " "unresolvable identifier here means the shipped gate can " "never match", ) referenced = {name for name in identifiers if name in quote_fields} rollups_read = { n for n in referenced if str(quote_fields[n].get("type", "")).lower() == "rollupsummary" } self.assertGreaterEqual( len(rollups_read), 2, "gate must compare at least two count rollups; fewer means " "it degenerated into a Status-only tautology", ) for name in referenced: field = quote_fields[name] self.assertNotEqual( str(field.get("formulaMode", "")).strip().lower(), "active", f"gate reads virtual formula field {name}; it must check " "the derivation's stored inputs, not its virtual output", ) def test_artifacts_identical_across_pack_roots(self): l2, merged = PACK_ROOTS self.assertEqual( load_json(l2 / "flows" / DERIVATION_FLOW), load_json(merged / "flows" / DERIVATION_FLOW), "derivation flow drifted between L2 source and merged pack", ) self.assertEqual( load_json(l2 / "formula_functions" / "pingcap_quote_functions.json"), load_json(merged / "formula_functions" / "pingcap_quote_functions.json"), "term function drifted between pack roots", ) for name in (GATE_NAME, LOCK_NAME): self.assertEqual( rule_by_name(l2, name), rule_by_name(merged, name), f"validation rule {name} drifted between pack roots", ) l2_fields = field_map(object_spec(l2, "Quote")) merged_fields = field_map(object_spec(merged, "Quote")) self.assertEqual( l2_fields["Term_months__c"]["formula"], merged_fields["Term_months__c"]["formula"], "Term_months__c header formula drifted between pack roots", ) for api_name in ( "Line_Item_Count__c", "Priced_Line_Item_Count__c", "Termed_Line_Item_Count__c", ): self.assertEqual( l2_fields[api_name]["rollupConfig"], merged_fields[api_name]["rollupConfig"], f"{api_name} rollupConfig drifted between pack roots", ) if __name__ == "__main__": unittest.main()