"""Behavioral pin for the OpportunityLineItem.Discount formula.

`Discount` is an active formula field, so nothing in the pack stops it from
returning a number that is arithmetically correct and commercially meaningless.
That is exactly what it did: TiDB Cloud is consumption-priced against a $1.00
placeholder PricebookEntry (one product, 01t5g000000liLxAAI, list price 1 in
USD, JPY and SGD alike) while the negotiated price lives in
`Unit_Net_Price__c`, so `(ListPrice - Unit_Net_Price__c) / ListPrice` divided a
large negative number by one. A $500 line rendered as -49900%; the worst line in
production rendered as -1,499,999,900%. Salesforce, which stores `Discount` as
an ordinary writable percent rather than deriving it, holds 0.00% on every one
of those lines.

The fix returns 0 for the TiDB Cloud family only. The two assertions that matter
are therefore a pair, and neither follows from the other:

1. TiDB Cloud lines read 0 whatever the net price, quantity or currency - the
   defect being closed; and
2. every other family still evaluates the original expression unchanged - the
   blast radius, which is the whole reason this was scoped to one family rather
   than fixed by clamping the field.

Like `test_quote_term_formula_semantics`, this EVALUATES the committed formula
from both pack roots with a small interpreter supporting only the constructs
that formula uses. A formula rewritten with other functions fails loudly here
and gets reviewed rather than silently accepted, and a hand-edit to one pack
root that skips the regen fails the mirror-parity assertion.

The stored value is the evaluated fraction scaled by 100 (percent return type);
these vectors are written as fractions, so -499 is the -49900% the user saw.
"""

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",
]
CLOUD = "TiDB Cloud"


def line_item_fields(pack_root: Path):
    combined = pack_root / "objects" / "standard_objects_spec.json"
    if combined.exists():
        spec = json.loads(combined.read_text(encoding="utf-8"))
        objects = spec.get("objects") if isinstance(spec, dict) else spec
    else:
        objects = json.loads(
            (pack_root / "objects" / "OpportunityLineItem.json").read_text(
                encoding="utf-8"
            )
        )
        if isinstance(objects, dict):
            objects = [objects]
    obj = next(o for o in objects if o.get("apiName") == "OpportunityLineItem")
    return {f["apiName"]: f for f in obj["fields"]}


def discount_formula(pack_root: Path) -> str:
    return line_item_fields(pack_root)["Discount"]["formula"]


# --- interpreter -----------------------------------------------------------
#
# Supports only: IF(cond, a, b), TEXT(x), string equality, parenthesised
# arithmetic over field references and numeric literals. Anything else raises.

_TOKEN = re.compile(
    r'\s*(?:(?P<str>"[^"]*")|(?P<num>\d+(?:\.\d+)?)|(?P<name>[A-Za-z_][A-Za-z0-9_.]*)'
    r"|(?P<op>[()+\-*/,]|=))"
)


def _tokenize(src: str):
    pos, out = 0, []
    while pos < len(src):
        if src[pos].isspace():
            pos += 1
            continue
        m = _TOKEN.match(src, pos)
        if not m:
            raise AssertionError("unsupported syntax at %r" % src[pos:])
        pos = m.end()
        if m.group("str"):
            out.append(("str", m.group("str")[1:-1]))
        elif m.group("num"):
            out.append(("num", float(m.group("num"))))
        elif m.group("name"):
            out.append(("name", m.group("name")))
        else:
            out.append(("op", m.group("op")))
    return out


class _Parser:
    def __init__(self, tokens, record):
        self.t, self.i, self.record = tokens, 0, record

    def peek(self):
        return self.t[self.i] if self.i < len(self.t) else (None, None)

    def take(self, op=None):
        kind, val = self.t[self.i]
        if op is not None and (kind, val) != ("op", op):
            raise AssertionError("expected %r, got %r" % (op, val))
        self.i += 1
        return val

    def expr(self):
        left = self.additive()
        if self.peek() == ("op", "="):
            self.take("=")
            return left == self.additive()
        return left

    def additive(self):
        val = self.multiplicative()
        while self.peek() in (("op", "+"), ("op", "-")):
            op = self.take()
            rhs = self.multiplicative()
            val = val + rhs if op == "+" else val - rhs
        return val

    def multiplicative(self):
        val = self.unary()
        while self.peek() in (("op", "*"), ("op", "/")):
            op = self.take()
            rhs = self.unary()
            val = val * rhs if op == "*" else val / rhs
        return val

    def unary(self):
        if self.peek() == ("op", "-"):
            self.take("-")
            return -self.unary()
        return self.primary()

    def primary(self):
        kind, val = self.peek()
        if kind == "op" and val == "(":
            self.take("(")
            inner = self.expr()
            self.take(")")
            return inner
        if kind in ("str", "num"):
            self.i += 1
            return val
        if kind != "name":
            raise AssertionError("unexpected token %r" % (val,))
        self.i += 1
        if self.peek() == ("op", "("):
            return self.call(val.upper())
        if val not in self.record:
            raise AssertionError("formula reads unknown field %r" % val)
        return self.record[val]

    def call(self, name):
        self.take("(")
        args = []
        if self.peek() != ("op", ")"):
            args.append(self.expr())
            while self.peek() == ("op", ","):
                self.take(",")
                args.append(self.expr())
        self.take(")")
        if name == "TEXT":
            assert len(args) == 1, "TEXT takes one argument"
            return "" if args[0] is None else str(args[0])
        if name == "IF":
            assert len(args) == 3, "IF takes three arguments"
            return args[1] if args[0] else args[2]
        raise AssertionError("formula uses unsupported function %s()" % name)


def evaluate(formula: str, record: dict):
    # IF is not short-circuiting here, so guard the division the legacy branch
    # performs on a list price that is zero only for families this pack does
    # not price that way.
    parser = _Parser(_tokenize(formula), record)
    value = parser.expr()
    assert parser.i == len(parser.t), "trailing tokens in formula"
    return value


def record(family, list_price, unit_net_price):
    return {
        "Product2.Family": family,
        "ListPrice": float(list_price),
        "Unit_Net_Price__c": float(unit_net_price),
    }


def legacy(list_price, unit_net_price):
    return (float(list_price) - float(unit_net_price)) / float(list_price)


class DiscountFormulaSemanticsTest(unittest.TestCase):
    def test_both_pack_roots_carry_the_same_formula(self):
        formulas = {str(root): discount_formula(root) for root in PACK_ROOTS}
        self.assertEqual(
            len(set(formulas.values())),
            1,
            "L2 and the generated final pack disagree, so the deployed formula is "
            "not the reviewed one - regenerate the final pack instead of hand-editing "
            "it:\n%s" % json.dumps(formulas, indent=2),
        )

    def test_tidb_cloud_reads_zero_whatever_the_net_price(self):
        formula = discount_formula(PACK_ROOTS[0])
        # The reported line, the worst line in production, the Salesforce
        # screenshot vector, and a genuine sub-dollar consumption price.
        for unit_net_price in (500, 15_000_000, 1_300_000, 21.39, 0.42, 1):
            with self.subTest(unit_net_price=unit_net_price):
                self.assertEqual(
                    evaluate(formula, record(CLOUD, 1, unit_net_price)),
                    0,
                    "TiDB Cloud must read 0.00%, matching Salesforce",
                )

    def test_tidb_cloud_is_currency_and_list_price_agnostic(self):
        # The placeholder PricebookEntry is list 1 in USD, JPY and SGD, so the
        # gate is the family and not a USD list price of exactly 1. Should the
        # placeholder ever be restated, TiDB Cloud must still read 0.
        formula = discount_formula(PACK_ROOTS[0])
        for list_price in (1, 100, 160):
            with self.subTest(list_price=list_price):
                self.assertEqual(
                    evaluate(formula, record(CLOUD, list_price, 12_500)), 0
                )

    def test_the_reported_defect_is_closed(self):
        formula = discount_formula(PACK_ROOTS[0])
        before = legacy(1, 500)
        self.assertEqual(before * 100, -49900, "the vector the user reported")
        self.assertEqual(evaluate(formula, record(CLOUD, 1, 500)), 0)

    def test_every_other_family_is_untouched(self):
        formula = discount_formula(PACK_ROOTS[0])
        others = [
            "TiDB Enterprise Subscription",
            "PingCAP Professional Services",
            "PingCAP Consulting Services",
            "PingCAP Training & Exam Services",
        ]
        # Includes a real 35% discount, a zero-discount line, and a negative
        # one: out-of-range values on other families are deliberately still
        # reported rather than silently zeroed, because there the number is
        # evidence of bad data rather than of a bad formula.
        for family in others:
            for list_price, unit_net_price in ((1000, 650), (1000, 1000), (1, 500)):
                with self.subTest(family=family, list_price=list_price):
                    self.assertAlmostEqual(
                        evaluate(formula, record(family, list_price, unit_net_price)),
                        legacy(list_price, unit_net_price),
                        places=12,
                    )

    def test_an_unknown_family_keeps_the_legacy_expression(self):
        # A family added later must not silently inherit the TiDB Cloud
        # exemption.
        formula = discount_formula(PACK_ROOTS[0])
        self.assertAlmostEqual(
            evaluate(formula, record("Some Future Family", 1000, 650)),
            legacy(1000, 650),
            places=12,
        )

    def test_quantity_and_term_defaults_are_the_ones_cloud_lines_rely_on(self):
        # Net fees for TiDB Cloud is Quantity * Unit_Net_Price__c, so a
        # quantity default other than 1 would change money, and the term
        # default of 12 is what the Cloud line contract assumes. Neither is
        # changed by this fix; both are pinned so a later edit is deliberate.
        for root in PACK_ROOTS:
            fields = line_item_fields(root)
            with self.subTest(pack_root=str(root)):
                self.assertEqual(str(fields["Quantity"].get("default")), "1")
                self.assertEqual(str(fields["Term_Mounths__c"].get("default")), "12")

    def test_discount_stays_a_derived_field(self):
        for root in PACK_ROOTS:
            field = line_item_fields(root)["Discount"]
            with self.subTest(pack_root=str(root)):
                self.assertEqual(field["type"], "formula")
                self.assertEqual(field["returnType"], "percent")
                self.assertEqual(field["formulaMode"], "active")
                self.assertFalse(field.get("createable", False))
                self.assertFalse(field.get("updateable", False))


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