"""Runtime oracle: evaluate Credits_Topup__VR_PoC_Ledgers_Only on the real L0 engine. The declaration-level assertions in `test_credits_ledger_bucketing` prove the rule is *declared* the way it was designed. They cannot prove it *behaves* that way, and for this rule the gap is the whole risk: the condition traverses to the parent engagement through `Credits__r`, and validation-formula relation resolution returns `nil, nil` for anything it cannot resolve - a missing FK, an unknown relation, a repository miss. Combined with an allowlist condition (block unless the parent is PoC) that means an unresolvable parent fails CLOSED. Fail-closed is the right direction for a spend control, but it also means a resolution bug does not read as a bug: it reads as every top-up being rejected. So the assertion that matters most here is not that the rule blocks. It is that a top-up on an ordinary PoC ledger is still ALLOWED. That one does not follow from the blocking cases passing, and no static check can produce it. This mirrors `test_credits_topup_poc_only_formula_runtime`'s sibling, `test_hubspot_disqualified_date_formula_runtime`: the committed L2 condition is read out of the pack, injected into a Go test overlaid onto the kernel's formula package, and executed against whichever NexusCRM checkout NEXUSCRM_ROOT names - so the evidence is about the engine under test, not about a description of it. Set NEXUSCRM_EXPECTED_HEAD to pin that checkout when the answer needs to be attributable to an exact commit. """ import json import os import subprocess import tempfile import textwrap import unittest from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] RULE_NAME = "Credits_Topup__VR_PoC_Ledgers_Only" GO_RUNTIME_TEST = r''' package formula import "testing" func TestL2CreditsTopupPoCOnlyCondition(t *testing.T) { const condition = `__L2_CONDITION__` engine := NewEngine() context := func(parent map[string]interface{}, prior map[string]interface{}) *Context { return &Context{ Record: map[string]interface{}{"Credits__c": "parent-id", "Topup_Amount__c": 100}, Prior: prior, // Mirrors the non-admin caller: the bypass key resolves false. Variables: map[string]interface{}{ "__Permission": map[string]interface{}{"ModifyAllData": false}, }, Fetcher: func(record map[string]interface{}, relationName string) (map[string]interface{}, error) { if relationName == "Credits__r" { return parent, nil } return nil, nil }, } } poc := map[string]interface{}{"Credit_Type__c": "PoC"} startup := map[string]interface{}{"Credit_Type__c": "Startup"} unchanged := map[string]interface{}{"Credits__c": "parent-id", "Topup_Amount__c": 50} reparented := map[string]interface{}{"Credits__c": "other-id", "Topup_Amount__c": 100} for _, tc := range []struct { name string ctx *Context fires bool why string }{ { name: "top-up created on a PoC ledger is allowed", ctx: context(poc, nil), fires: false, why: "the case that proves the rule does not over-block; an unresolvable " + "traversal would land here as a rejection of every legitimate top-up", }, {name: "top-up created on a Startup ledger is blocked", ctx: context(startup, nil), fires: true, why: "the invariant: Startup is one grant per org with no top-up path"}, {name: "unresolvable parent fails closed", ctx: context(nil, nil), fires: true, why: "allowlist semantics - documented, and asserted so a later change to " + "a denylist shows up here rather than in production"}, {name: "editing an existing top-up is not blocked", ctx: context(startup, unchanged), fires: false, why: "the rule gates on the events that can newly create the violation, so " + "an already-existing row stays editable"}, {name: "reparenting onto a Startup ledger is blocked", ctx: context(startup, reparented), fires: true, why: "reparenting is the second of the two events, and is why ISNEW alone " + "would not close the invariant"}, {name: "admin bypass lets the save through", ctx: adminContext(startup), fires: false, why: "$Permission.ModifyAllData is the deliberate cleanup path - and the " + "reason acceptance testing on a real environment must not use an admin"}, } { t.Run(tc.name, func(t *testing.T) { got, err := engine.Evaluate(condition, tc.ctx) if err != nil { t.Fatalf("Evaluate returned %v; the condition must evaluate cleanly", err) } fired, ok := got.(bool) if !ok { t.Fatalf("Evaluate = %#v (%T), want a bool; a non-boolean condition " + "cannot gate a save", got, got) } if fired != tc.fires { t.Fatalf("rule fired = %v, want %v (%s)", fired, tc.fires, tc.why) } }) } } func adminContext(parent map[string]interface{}) *Context { return &Context{ Record: map[string]interface{}{"Credits__c": "parent-id", "Topup_Amount__c": 100}, Variables: map[string]interface{}{ "__Permission": map[string]interface{}{"ModifyAllData": true}, }, Fetcher: func(record map[string]interface{}, relationName string) (map[string]interface{}, error) { if relationName == "Credits__r" { return parent, nil } return nil, nil }, } } ''' class CreditsTopupPoCOnlyRuntimeTest(unittest.TestCase): def condition(self): path = REPO_ROOT / "packs" / "Salesforce-PingCAP-L2" / "validation" / "standard_validation_rules.json" rules = json.loads(path.read_text(encoding="utf-8"))["validationRules"] for rule in rules: if rule.get("name") == RULE_NAME: return rule["condition"] self.fail("validation rule %s is absent" % RULE_NAME) def test_committed_condition_behaves_on_the_current_l0_evaluator(self): 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(), "formula package not found under %s" % 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, "behaviour must be attributed to a known build, not to whatever the " "checkout happens to be on", ) condition = self.condition() self.assertNotIn( "`", condition, "a backtick would terminate the Go raw string literal" ) with tempfile.TemporaryDirectory(prefix="nexuscrm-l2-topup-") as temp_dir: temp = Path(temp_dir) backing_test = temp / "l2_credits_topup_poc_only_test.go" backing_test.write_text( textwrap.dedent(GO_RUNTIME_TEST).replace("__L2_CONDITION__", condition), encoding="utf-8", ) overlay_target = formula_dir / "l2_credits_topup_poc_only_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", "-overlay=%s" % overlay, "./pkg/formula", "-run", "^TestL2CreditsTopupPoCOnlyCondition$", "-count=1", "-v", ], cwd=upstream / "nexuscrm-kernel", env={**os.environ, "GOWORK": "off"}, text=True, capture_output=True, check=False, ) self.assertEqual( completed.returncode, 0, "the committed condition did not behave as designed on this engine:\n%s\n%s" % (completed.stdout, completed.stderr), ) if __name__ == "__main__": unittest.main()