"""Guard: the one-open-ledger control is keyed per credit program, and top-ups stay PoC-only.

`Credits__VR_One_Open_Ledger_Per_Org` closes the split-ledger bypass from issue
#4733: approval step criteria aggregate only within one ledger, so two sibling
ledgers on one Cloud Org are each gated on their own smaller amount and the
combined figure never reaches an approver. Adding a second credit program
(Startup) to `Credit_Type__c` forced the slot to be keyed by
(Cloud_Org__c, Credit_Type__c) rather than Cloud Org alone: a PoC engagement and
a Startup grant on one org are two legitimate grants, each entering its own
chain, and one slot for both would reject a legitimate request instead of
closing a bypass.

What these assertions are actually protecting, in the order that matters:

1. **The closure premise, not the code shape.** Keying the slot per type only
   keeps the invariant closed while `Credit_Type__c` cannot change after create.
   If it ever becomes updateable, a record can move between buckets and the
   rule's create / Cloud_Org__c-change / leaving-Cancelled triad stops being the
   complete event set - silently, because nothing else would fail. That premise
   is asserted here rather than left in prose, which is what
   `test_credit_type_is_immutable_after_create` exists for. It is the single
   most load-bearing assertion in this file.

2. **Both halves of the top-up guard.** Top-ups are PoC-only. The Add Top-up
   button's `visibilityCondition` hides a button; it is an affordance, not a
   control, and an API-direct create bypasses it entirely. The server-side rule
   is what keeps the invariant true. Asserting only one half would pass on
   exactly the shape that is broken, so both are required and they are required
   to name the same credit type - a UI that hides a different set than the rule
   rejects is its own defect.

Scope, stated plainly: while `Credit_Type__c` offers a single program, the
bucketing added here is inert - every lookup finds the same rows it found
before. That inertness is the point of shipping it ahead of the second program
rather than with it, and it is also why no assertion in this file can prove the
bucketing *works*; they prove it is *declared*. Exercising it needs a second
credit type, which the local bootstrap oracle introduces temporarily. Adding
that value to the pack early would make records of a program creatable before
its approval chain exists, which is a worse trade than a briefly inert control.

These are declaration-level checks and they do not prove runtime behaviour;
the flow-stamps-then-rule-reads ordering is proven on the local bootstrap
oracle, not here. What they do prove is that a later edit cannot quietly remove
the type predicate, flip the field to updateable, or drop one half of the
top-up guard without a red test.
"""

import json
import os
import re
import unittest

REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
L2_ROOT = os.path.join(REPO_ROOT, "packs", "Salesforce-PingCAP-L2")

CREDITS_OBJECT = os.path.join(L2_ROOT, "objects", "Credits__c.json")
LEDGER_FLOW = os.path.join(L2_ROOT, "flows", "Credits_Stamp_Open_Org_Ledger.flow.json")
VALIDATION_RULES = os.path.join(L2_ROOT, "validation", "standard_validation_rules.json")
UI_ACTIONS = os.path.join(L2_ROOT, "actions", "ui_actions.json")

LEDGER_RULE = "Credits__VR_One_Open_Ledger_Per_Org"
TOPUP_RULE = "Credits_Topup__VR_PoC_Ledgers_Only"
TOPUP_ACTION = "AddCreditsTopup"
TOPPABLE_TYPE = "PoC"
TOPUP_OBJECT = os.path.join(L2_ROOT, "objects", "Credits_Topup__c.json")


def load(path):
    with open(path, encoding="utf-8") as handle:
        return json.load(handle)


def object_spec(path):
    spec = load(path)
    return spec[0] if isinstance(spec, list) else spec


def field(spec, api_name):
    for entry in spec["fields"]:
        if entry.get("apiName") == api_name:
            return entry
    raise AssertionError("field %s is absent from %s" % (api_name, spec.get("apiName")))


def rule(name):
    for entry in load(VALIDATION_RULES)["validationRules"]:
        if entry.get("name") == name:
            return entry
    raise AssertionError("validation rule %s is absent" % name)


def ui_action(name):
    found = []

    def walk(node):
        if isinstance(node, dict):
            if node.get("name") == name:
                found.append(node)
            for value in node.values():
                walk(value)
        elif isinstance(node, list):
            for value in node:
                walk(value)

    walk(load(UI_ACTIONS))
    if not found:
        raise AssertionError("ui action %s is absent" % name)
    if len(found) > 1:
        raise AssertionError("ui action %s is declared %d times" % (name, len(found)))
    return found[0]


class CreditTypeImmutabilityTests(unittest.TestCase):
    """The premise the per-type slot key rests on."""

    def test_credit_type_is_immutable_after_create(self):
        credit_type = field(object_spec(CREDITS_OBJECT), "Credit_Type__c")
        self.assertIs(
            credit_type.get("updateable"),
            False,
            "Credit_Type__c must stay updateable:false. The one-open-ledger rule "
            "keys its slot on (Cloud_Org__c, Credit_Type__c) and fires on create, "
            "Cloud_Org__c change and leaving Cancelled. A mutable credit type adds "
            "a fourth way to move between slots that no clause covers, so a record "
            "could be re-typed into an occupied slot with the rule never firing - "
            "reopening the #4733 split-ledger bypass across programs, silently.",
        )

    def test_credit_type_can_never_be_null(self):
        credit_type = field(object_spec(CREDITS_OBJECT), "Credit_Type__c")
        self.assertIs(credit_type.get("required"), True)
        self.assertTrue(
            credit_type.get("default"),
            "The ledger lookup filters Credit_Type__c EqualTo $Record.Credit_Type__c. "
            "A Flow filter compiles to SQL, where = NULL matches no row, so a null "
            "credit type would stamp 'no open ledger' on every save and silently "
            "disable the control - fail OPEN, the worst failure mode for a spend "
            "gate. required plus a default is what keeps null unreachable: schema "
            "required is enforced ahead of RecordBeforeSave dispatch, and the field "
            "is updateable:false so an existing value cannot be cleared.",
        )

    def test_the_toppable_program_is_a_declared_credit_type(self):
        values = [
            option.get("value")
            for option in field(object_spec(CREDITS_OBJECT), "Credit_Type__c").get("options", [])
        ]
        self.assertIn(
            TOPPABLE_TYPE,
            values,
            "Both top-up guards allowlist this exact string. If the program is "
            "renamed or dropped from the picklist without updating them, the "
            "server-side rule rejects every top-up and the button disappears - "
            "the guards would still pass each other's consistency check while "
            "blocking the only path they exist to protect.",
        )


class LedgerBucketingTests(unittest.TestCase):
    """The stamp answers a per-(org, type) question, and the rule reads that stamp."""

    def lookup(self):
        flow = load(LEDGER_FLOW)
        lookups = [entry for entry in flow["recordLookups"] if entry["name"] == "Get_Open_Org_Ledger"]
        self.assertEqual(len(lookups), 1, "Get_Open_Org_Ledger must be declared exactly once")
        return lookups[0]

    def test_open_ledger_lookup_is_scoped_to_the_same_credit_type(self):
        filters = {entry["field"]: entry for entry in self.lookup()["filters"]}
        self.assertIn(
            "Credit_Type__c",
            filters,
            "Without this filter the stamp answers 'this org has any open ledger', "
            "so a Startup grant on an org that already runs a PoC engagement is "
            "blocked - rejecting a legitimate request rather than closing a bypass.",
        )
        type_filter = filters["Credit_Type__c"]
        self.assertEqual(type_filter["operator"], "EqualTo")
        self.assertEqual(
            type_filter["value"].get("elementReference"),
            "$Record.Credit_Type__c",
            "The filter must compare against the saving record's own type. A "
            "hard-coded literal would scope the control to one program and leave "
            "every other program ungated.",
        )

    def test_open_ledger_lookup_keeps_its_original_predicates(self):
        filters = {entry["field"]: entry for entry in self.lookup()["filters"]}
        for required in ("Cloud_Org__c", "Credit_Lifecycle__c", "Id"):
            self.assertIn(
                required,
                filters,
                "%s is part of what the stamp means; bucketing by credit type "
                "narrows the question, it does not replace it." % required,
            )
        self.assertEqual(filters["Credit_Lifecycle__c"]["operator"], "NotEqualTo")
        self.assertEqual(filters["Credit_Lifecycle__c"]["value"].get("stringValue"), "Cancelled")
        self.assertEqual(
            filters["Id"]["operator"],
            "NotEqualTo",
            "On update the record is already persisted and would match its own "
            "lookup and block itself.",
        )

    def test_ledger_rule_still_reads_the_stamp_and_is_active(self):
        entry = rule(LEDGER_RULE)
        self.assertIs(entry.get("active"), True)
        self.assertIn(
            "Has_Open_Org_Ledger__c",
            entry["condition"],
            "The rule reads the stamp; the bucketing lives in the flow that "
            "computes it. A rule that stopped reading the stamp would be a "
            "different control entirely.",
        )


class TopupPocOnlyTests(unittest.TestCase):
    """Top-ups are PoC-only, and both guards say so about the same type."""

    def test_server_side_rule_blocks_top_ups_on_other_programs(self):
        entry = rule(TOPUP_RULE)
        self.assertEqual(entry["objectApiName"], "Credits_Topup__c")
        self.assertIs(entry.get("active"), True)
        condition = entry["condition"]
        self.assertIn(
            "Credits__r.Credit_Type__c",
            condition,
            "Credits_Topup__c carries no credit type of its own - the program is a "
            "property of the parent engagement, so the rule must traverse to it.",
        )
        self.assertIn(
            '!= "%s"' % TOPPABLE_TYPE,
            condition.replace("'", '"'),
            "The condition must be an allowlist - block unless the parent is the "
            "toppable program. A denylist naming one bad type silently admits "
            "every program added after this rule was written.",
        )

    def test_parent_traversal_in_the_rule_resolves_by_naming_convention(self):
        condition = rule(TOPUP_RULE)["condition"]
        traversals = set(re.findall(r"([A-Za-z_][A-Za-z0-9_]*)__r\.", condition))
        self.assertTrue(traversals, "the rule is expected to read the parent engagement")
        declared = {
            entry.get("apiName") for entry in object_spec(TOPUP_OBJECT)["fields"]
        }
        for relation in sorted(traversals):
            self.assertIn(
                "%s__c" % relation,
                declared,
                "The relationship resolver tries the schema's declared "
                "relationshipName first and only then falls back to the __r -> __c "
                "naming convention. Credits_Topup__c declares relationshipName "
                "'Topups', so that first tier does not match and %s__r resolves "
                "purely because a field is literally named %s__c. Rename that field "
                "and the traversal returns no parent, which against an allowlist "
                "condition fails closed - every top-up rejected, with nothing in "
                "the rule text looking wrong." % (relation, relation),
            )

    def test_top_up_button_is_hidden_off_the_toppable_program(self):
        condition = ui_action(TOPUP_ACTION)["config"]["visibilityCondition"]
        self.assertIn(
            "record.Credit_Type__c",
            condition,
            "The button must not offer a path the server-side rule rejects.",
        )
        self.assertIn(TOPPABLE_TYPE, condition)

    def test_button_visibility_binds_the_type_check_to_every_lifecycle_branch(self):
        condition = ui_action(TOPUP_ACTION)["config"]["visibilityCondition"]
        self.assertRegex(
            condition,
            r"\(\s*record\.Credit_Lifecycle__c\b[^)]*\|\|[^)]*\)\s*&&",
            "The lifecycle branches are OR-ed and the type check is AND-ed onto "
            "them. The expression evaluator splits on || before &&, so without "
            "parentheses around the OR group the type check binds to the last "
            "lifecycle branch alone and the button stays visible for the others - "
            "a guard that silently covers one third of what it reads as covering.",
        )

    def test_both_guards_name_the_same_toppable_program(self):
        ui_condition = ui_action(TOPUP_ACTION)["config"]["visibilityCondition"]
        rule_condition = rule(TOPUP_RULE)["condition"]
        ui_types = set(re.findall(r"record\.Credit_Type__c\s*[=!]=\s*['\"]([^'\"]+)['\"]", ui_condition))
        rule_types = set(
            re.findall(r"Credit_Type__c\)\s*(?:!=|<>)\s*['\"]([^'\"]+)['\"]", rule_condition)
        )
        # Both extractions must find something before comparing them. Two empty
        # sets are equal, so without this the assertion passes at its loudest -
        # when BOTH guards have been deleted - which is precisely the case it
        # exists to catch. It reads as covering the pair while covering nothing.
        self.assertTrue(ui_types, "no credit-type check found in %s" % TOPUP_ACTION)
        self.assertTrue(rule_types, "no credit-type check found in %s" % TOPUP_RULE)
        self.assertEqual(
            ui_types,
            rule_types,
            "The button hides what the rule rejects. If these drift, the UI either "
            "offers a path that always errors, or hides a path the server allows.",
        )


if __name__ == "__main__":
    unittest.main()