diff --git a/clawmetry/entitlements.py b/clawmetry/entitlements.py index fd595e29ac..c97021f569 100644 --- a/clawmetry/entitlements.py +++ b/clawmetry/entitlements.py @@ -13001,6 +13001,239 @@ def _capacity_row(kind: str, raw, resolver) -> dict: return empty +def has_all_breakdown( + *, + features=None, + runtimes=None, + channels: int | None = None, + retention_days: int | None = None, + nodes: int | None = None, +) -> dict: + """Per-axis breakdown of :func:`has_all` with the *blocking* axis + (or axes) called out. + + Boolean-fold twin of :func:`min_tier_for_all_breakdown`. Where the + aggregate scalar :func:`has_all` collapses the answer to a single + ``has_all`` bool, this helper preserves the per-axis contribution + and identifies which axis (or axes) is *blocking* the grant. A + diagnostics tile can then render an honest "you can't have this + because your Starter tier caps channels at 5 (you asked for 8)" + tooltip off ONE round-trip instead of five ``has_*`` calls + a + client-side which-axis-is-false walk. + + Pairs directly with :func:`min_tier_for_all_breakdown` on the same + input: a UI wiring the paywall CTA can render "denied here BECAUSE + of axis Y" (from this helper's ``blocking_axes``) alongside "the + cheapest tier that would grant it is Z BECAUSE of axis W" (from + the reverse-lookup breakdown's ``binding_axes``). The two calls + accept identical kwargs and echo the axes on parallel row shapes. + + Same input semantics as :func:`has_all` (same five-axis kwargs, + same per-axis ``None`` "not supplied" sentinel, same + ``retention_days=None means unset, not unlimited`` posture, same + never-raise contract). + + Response shape:: + + { + "has_all": , # matches has_all(**kwargs) + "axes": { + "features": | None, # None iff axis unsupplied + "runtimes": | None, + "channels": | None, + "retention_days": | None, + "nodes": | None, + }, + "blocking_axes": ["channels"] | [], # axis ids whose per-axis + # ``has`` is False when + # the aggregate is False + } + + Each ```` carries ``kind``, ``supplied`` (``True`` -- an + unsupplied axis short-circuits to ``None`` at the envelope level), + ``has`` (the per-axis singular scalar's answer), and ``blocking`` + (``True`` iff the aggregate ``has_all`` is ``False`` AND this + axis' ``has`` is ``False``). Grant axes additionally carry + ``items`` (the normalised known ids the fold saw) and ``unknown`` + (typo tokens the fold saw -- the singular + :func:`has_features` / :func:`has_runtimes` scalars collapse the + axis to ``False`` on any unknown, and the split lets a tooltip + render "typo *Fleeet*" instead of a bare denial). Capacity axes + additionally carry ``value`` (the parsed int, or the raw input if + it did not parse, mirroring :func:`min_tier_for_all_breakdown`). + + ``blocking_axes`` is the list of axis keys (``"features"`` / + ``"runtimes"`` / ``"channels"`` / ``"retention_days"`` / + ``"nodes"``) whose per-axis ``has`` is ``False``, in envelope order + (``features``, ``runtimes``, ``channels``, ``retention_days``, + ``nodes``). When ``has_all`` is ``True`` -- every supplied axis + grants -- ``blocking_axes`` is the empty list and every axis row's + ``blocking`` is ``False``. When no axes are supplied at all, + :func:`has_all` collapses to ``False`` (its empty-``False`` typo + posture) BUT no axis is *blocking* since none was asked about; + ``blocking_axes`` is the empty list on that branch too and every + axis in ``axes`` is ``None``. + + Grace posture mirrors :func:`has_all` byte-for-byte: while + ``ent.grace`` is ``True`` every singular ``has_*`` delegate + returns ``True`` for its fully-known input, so this helper reports + ``has_all=True`` and empty ``blocking_axes`` for every + fully-known bundle. Wiring this into a paywall diagnostics tile + today surfaces NOTHING (matches the ``has_all=True`` grace + answer on the same bundle). Unknown grant tokens, empty grant + axes, and non-int capacity axes still collapse the fold to + ``False`` even in grace (matches the singular scalars' strict + callsite-typo posture), and the corresponding axis is surfaced + in ``blocking_axes``. + + Post-enforcement: for every supplied axis, the row's ``has`` is + the exact answer the resolved entitlement's live grant would give + to the singular ``has_*`` call, and ``blocking_axes`` names every + axis the live grant denies. + + Never raises: any delegate failure logs a warning and short- + circuits to the empty envelope (``has_all=False``, every axis + ``None``, empty ``blocking_axes``) so a caller can bind this into + a diagnostics dict without a try/except. + """ + empty = { + "has_all": False, + "axes": { + "features": None, + "runtimes": None, + "channels": None, + "retention_days": None, + "nodes": None, + }, + "blocking_axes": [], + } + try: + axes: dict = { + "features": None, + "runtimes": None, + "channels": None, + "retention_days": None, + "nodes": None, + } + + if features is not None: + feats = _normalise_csv(features) + known = [f for f in feats if f in ALL_FEATURES] + unknown = [f for f in feats if f not in ALL_FEATURES] + # Delegate the axis-level fold to the singular scalar so + # the empty-[] and unknown-token postures inherit + # ``has_features``'s strict-``False`` typo semantics + # (feats == [] -> False; any unknown -> False even under + # grace) without a divergent code path here. + axis_has = bool(has_features(feats)) + axes["features"] = { + "kind": "features", + "supplied": True, + "items": known, + "unknown": unknown, + "has": axis_has, + "blocking": False, + } + + if runtimes is not None: + raw_rts = _normalise_csv(runtimes) + seen: set[str] = set() + canon_rts: list[str] = [] + unknown_rts: list[str] = [] + for raw in raw_rts: + rt = canonical_runtime(raw) + key = rt if rt else raw + if key in seen: + continue + seen.add(key) + if rt and rt in ALL_RUNTIMES: + canon_rts.append(rt) + else: + unknown_rts.append(raw) + axis_has = bool(has_runtimes(raw_rts)) + axes["runtimes"] = { + "kind": "runtimes", + "supplied": True, + "items": canon_rts, + "unknown": unknown_rts, + "has": axis_has, + "blocking": False, + } + + def _capacity_row(kind: str, raw, gate) -> dict: + try: + n = int(raw) + parsed = True + except (TypeError, ValueError): + n = None + parsed = False + if parsed: + axis_has = bool(gate(n)) + else: + # Non-int capacity input collapses the axis-level gate + # to False -- matches the singular capacity scalars' + # strict-``False`` typo posture (and matches + # :func:`has_all` which folds the same input to False). + axis_has = False + return { + "kind": kind, + "supplied": True, + "value": n if parsed else raw, + "has": axis_has, + "blocking": False, + } + + if channels is not None: + axes["channels"] = _capacity_row( + "channels", channels, has_channel_count + ) + if retention_days is not None: + axes["retention_days"] = _capacity_row( + "retention_days", retention_days, has_retention_window + ) + if nodes is not None: + axes["nodes"] = _capacity_row( + "nodes", nodes, has_node_count + ) + + # Aggregate fold matches :func:`has_all` byte-for-byte on the + # SAME kwargs, but resolved off the per-axis rows we already + # built (rather than a second round of delegate calls) so the + # scalar seat and the axis seats cannot drift. + supplied_rows = [axis for axis in axes.values() if axis is not None] + if not supplied_rows: + # Nothing supplied. :func:`has_all` collapses to False on + # this branch (its empty-``False`` typo posture), but no + # axis is *blocking* because none was asked about -- so + # ``blocking_axes`` stays empty and every axis row stays + # None. Diverges from ``has_all_bundle`` and friends only + # in that we surface the aggregate scalar directly (False) + # so a paired call to ``has_all`` on the same kwargs sees + # byte-identical fold answers. + return empty + aggregate = all(row["has"] for row in supplied_rows) + if aggregate: + return { + "has_all": True, + "axes": axes, + "blocking_axes": [], + } + blocking_axes: list[str] = [] + for key in ("features", "runtimes", "channels", "retention_days", "nodes"): + axis = axes[key] + if axis and not axis["has"]: + axis["blocking"] = True + blocking_axes.append(key) + return { + "has_all": False, + "axes": axes, + "blocking_axes": blocking_axes, + } + except Exception as exc: + logger.warning("entitlements: has_all_breakdown failed: %s", exc) + return empty + + def affordable_tiers( *, features=None, diff --git a/routes/entitlement.py b/routes/entitlement.py index e0401db29e..203309a8df 100644 --- a/routes/entitlement.py +++ b/routes/entitlement.py @@ -13715,6 +13715,184 @@ def api_entitlement_required_tier_breakdown(): ) +@bp_entitlement.route("/api/entitlement/has-all-breakdown") +def api_entitlement_has_all_breakdown(): + """``GET /api/entitlement/has-all-breakdown?features=a,b&runtimes=x,y + &channels=N&retention_days=K&nodes=M`` -- per-axis boolean-fold + breakdown sibling of ``/api/entitlement/required-tier-breakdown``. + + Wraps :func:`entitlements.has_all_breakdown`. Where the reverse- + lookup breakdown identifies which axis (or axes, on a tie) is + *binding* the aggregate min-tier floor, this endpoint identifies + which axis (or axes) is *blocking* the LIVE aggregate grant -- so + a paywall diagnostics tile can render "denied here BECAUSE of + channels (Starter caps at 5, you asked for 8)" off ONE round-trip + instead of five ``/api/entitlement/has-*`` calls + a client-side + which-axis-is-false walk. Pairs directly with + ``/required-tier-breakdown`` on the same query args so a UI can + render "denied because axis Y" alongside "cheapest tier that + would grant it is Z because axis W". + + At least one of ``features=`` / ``runtimes=`` / ``channels=`` / + ``retention_days=`` / ``nodes=`` must be supplied (non-empty / + parseable after normalisation) -- otherwise 400. The three + capacity axes accept a single int each; a blank or non-int value + still surfaces the axis in the response (with ``has=false`` and + the raw input in ``value`` so the caller can flag the typo in a + tooltip), matching the never-crash posture of the singular + ``/has-*`` endpoints. ``retention_days=`` mirrors the strict + :func:`has_all` posture: an unset param is *unset*, NOT + *unlimited* -- asking about the unlimited-retention live grant is + the singular ``/api/entitlement/has-retention-window`` call's job. + + Response body:: + + { + "features": [], + "runtimes": [], + "channels": | null, + "retention_days": | null, + "nodes": | null, + "has_all": , + "current_tier": , + "current_tier_rank": , + "grace": , + "enforced": , + "axes": { | null }, + "blocking_axes": ["channels"] # ordered; empty when has_all=true + } + + ``axes`` and ``blocking_axes`` come straight from + :func:`has_all_breakdown`; see that helper's docstring for the + row shape. The ``current_tier*`` / ``grace`` / ``enforced`` + fields match the sibling ``/has-*`` endpoints so a caller + migrating from the singular endpoints can adopt the breakdown + without reshaping its diagnostics payload. + + Never 5xxs: the OSS-fallback shape is returned on any resolver + failure (``has_all=false``, empty ``blocking_axes``, every axis + ``null``). + """ + try: + from clawmetry import entitlements as _ent + + features = _parse_csv_arg("features") + runtimes = _parse_csv_arg("runtimes") + (channels_present, channels_ok, channels_n, channels_raw) = _parse_capacity_arg( + "channels" + ) + ( + retention_present, + retention_ok, + retention_n, + retention_raw, + ) = _parse_capacity_arg("retention_days") + (nodes_present, nodes_ok, nodes_n, nodes_raw) = _parse_capacity_arg("nodes") + + if ( + not features + and not runtimes + and not channels_present + and not retention_present + and not nodes_present + ): + return ( + jsonify( + { + "error": ( + "supply at least one of features=, " + "runtimes=, channels=, " + "retention_days=, or nodes=" + ) + } + ), + 400, + ) + + # A capacity arg that is present-but-unparseable still routes to + # the helper (as the raw string) so the axis row surfaces with + # ``has=false`` and ``value=``. Matches the never-crash + # posture of the singular ``/has-*`` endpoints: a typo returns + # a shape a UI can render, not a 400 wall. + def _capacity_kw(present: bool, ok: bool, n: int | None, raw: str): + if not present: + return None + if ok: + return n + return raw + + breakdown = _ent.has_all_breakdown( + features=features or None, + runtimes=runtimes or None, + channels=_capacity_kw(channels_present, channels_ok, channels_n, channels_raw), + retention_days=_capacity_kw( + retention_present, retention_ok, retention_n, retention_raw + ), + nodes=_capacity_kw(nodes_present, nodes_ok, nodes_n, nodes_raw), + ) + + ent = _ent.get_entitlement() + cur_rank = _ent.tier_rank(ent.tier) + + return jsonify( + { + "features": features, + "runtimes": runtimes, + "channels": channels_n if channels_ok else (channels_raw if channels_present else None), + "retention_days": retention_n if retention_ok else (retention_raw if retention_present else None), + "nodes": nodes_n if nodes_ok else (nodes_raw if nodes_present else None), + "has_all": bool(breakdown.get("has_all")), + "current_tier": ent.tier, + "current_tier_rank": cur_rank, + "grace": bool(getattr(ent, "grace", False)), + "enforced": not bool(getattr(ent, "grace", False)), + "axes": breakdown.get("axes") + or { + "features": None, + "runtimes": None, + "channels": None, + "retention_days": None, + "nodes": None, + }, + "blocking_axes": breakdown.get("blocking_axes") or [], + } + ) + except Exception as exc: + logger.warning("api_entitlement_has_all_breakdown: error: %s", exc) + (channels_present, channels_ok, channels_n, channels_raw) = _parse_capacity_arg( + "channels" + ) + ( + retention_present, + retention_ok, + retention_n, + retention_raw, + ) = _parse_capacity_arg("retention_days") + (nodes_present, nodes_ok, nodes_n, nodes_raw) = _parse_capacity_arg("nodes") + return jsonify( + { + "features": _parse_csv_arg("features"), + "runtimes": _parse_csv_arg("runtimes"), + "channels": channels_n if channels_ok else (channels_raw if channels_present else None), + "retention_days": retention_n if retention_ok else (retention_raw if retention_present else None), + "nodes": nodes_n if nodes_ok else (nodes_raw if nodes_present else None), + "has_all": False, + "current_tier": "oss", + "current_tier_rank": 0, + "grace": True, + "enforced": False, + "axes": { + "features": None, + "runtimes": None, + "channels": None, + "retention_days": None, + "nodes": None, + }, + "blocking_axes": [], + } + ) + + @bp_entitlement.route("/api/entitlement/feature-catalog-at") def api_entitlement_feature_catalog_at(): """``GET /api/entitlement/feature-catalog-at?tier=`` -- what-if diff --git a/tests/test_entitlement_has_all_breakdown.py b/tests/test_entitlement_has_all_breakdown.py new file mode 100644 index 0000000000..361f24618e --- /dev/null +++ b/tests/test_entitlement_has_all_breakdown.py @@ -0,0 +1,502 @@ +"""Tests for :func:`clawmetry.entitlements.has_all_breakdown` and the +``/api/entitlement/has-all-breakdown`` endpoint. + +The breakdown helper is the boolean-fold twin of +:func:`min_tier_for_all_breakdown`. Where the reverse-lookup breakdown +identifies which axis is *binding* the aggregate required-tier floor, +this helper identifies which axis (or axes) is *blocking* the LIVE +aggregate grant. A paywall diagnostics tile can then render "denied +here BECAUSE of channels (Starter caps at 5, you asked for 8)" off +one round-trip alongside "the cheapest tier that would grant it is +Pro BECAUSE of channels" from the reverse-lookup companion. + +This file pins: + +* Parity with :func:`has_all` on the aggregate fold across the five + capacity axes -- so a future tier shuffle or singular-scalar + posture change breaks loudly here. +* The per-axis row shape (``kind``, ``supplied``, ``has``, ``blocking``, + plus ``items`` / ``unknown`` for grants and ``value`` for capacities). +* ``blocking_axes`` identification -- single-axis denials, multi-axis + denials, empty on ``has_all=True``, empty on the "nothing supplied" + edge. +* Grace vs enforce -- grace turns every fully-known bundle into + ``has_all=True`` / empty ``blocking_axes`` (matches :func:`has_all`); + unknown / empty / non-int input still collapses the fold in grace + (matches the singular scalars' strict-``False`` typo posture). +* The never-raise contract on unknown ids, non-int capacity input, + and the "nothing supplied" edge. +* HTTP envelope shape (``current_tier`` / ``grace`` / ``enforced`` + mirror the sibling ``/has-*`` endpoints; the breakdown fields sit + alongside). +* Pairing invariants with the reverse-lookup breakdown + (:func:`min_tier_for_all_breakdown`) on the same inputs -- axes + echoes match and the two envelopes render coherently in the same + paywall tooltip. +""" +from __future__ import annotations + +import importlib + +import pytest +from flask import Flask + + +AXIS_KEYS = ("features", "runtimes", "channels", "retention_days", "nodes") + + +@pytest.fixture +def ent(monkeypatch, tmp_path): + monkeypatch.delenv("CLAWMETRY_ENFORCE", raising=False) + monkeypatch.setenv("HOME", str(tmp_path)) + import clawmetry.entitlements as e + + importlib.reload(e) + e.invalidate() + yield e + e.invalidate() + + +@pytest.fixture +def client(ent): + from routes.entitlement import bp_entitlement + + app = Flask(__name__) + app.register_blueprint(bp_entitlement) + return app.test_client() + + +# ── envelope shape + null / empty edges ──────────────────────────────────── + + +def test_no_constraints_returns_empty_envelope(ent): + """Mirrors :func:`has_all`: "nothing asked" collapses the aggregate + to ``False`` (empty-``False`` typo posture) but no axis is + *blocking* since none was asked about.""" + out = ent.has_all_breakdown() + assert out["has_all"] is False + for k in AXIS_KEYS: + assert out["axes"][k] is None + assert out["blocking_axes"] == [] + + +def test_all_axes_none_returns_empty_envelope(ent): + out = ent.has_all_breakdown( + features=None, + runtimes=None, + channels=None, + retention_days=None, + nodes=None, + ) + assert out["has_all"] is False + for k in AXIS_KEYS: + assert out["axes"][k] is None + assert out["blocking_axes"] == [] + + +def test_envelope_top_level_keys(ent): + out = ent.has_all_breakdown(features=["fleet"]) + assert set(out.keys()) == {"has_all", "axes", "blocking_axes"} + assert set(out["axes"].keys()) == set(AXIS_KEYS) + + +# ── parity with has_all across the five axes ─────────────────────────────── + + +def test_parity_features_only(ent): + for feats in (["fleet"], ["sso"], ["fleet", "sso"]): + out = ent.has_all_breakdown(features=feats) + assert out["has_all"] == ent.has_all(features=feats), feats + + +def test_parity_runtimes_only(ent): + for rts in (["openclaw"], ["claude_code"], ["openclaw", "claude_code"]): + out = ent.has_all_breakdown(runtimes=rts) + assert out["has_all"] == ent.has_all(runtimes=rts), rts + + +def test_parity_capacity_only(ent): + for kw in ({"channels": 5}, {"retention_days": 30}, {"nodes": 2}): + out = ent.has_all_breakdown(**kw) + assert out["has_all"] == ent.has_all(**kw), kw + + +def test_parity_mixed_bundle(ent): + bundle = dict( + features=["fleet"], + runtimes=["claude_code"], + channels=8, + retention_days=30, + nodes=2, + ) + out = ent.has_all_breakdown(**bundle) + assert out["has_all"] == ent.has_all(**bundle) + + +def test_parity_empty_grant_axis_collapses_to_false(ent): + """``features=[]`` collapses :func:`has_all` to False (its empty- + ``False`` typo posture); the breakdown must fold the same way and + surface the features axis as blocking.""" + out = ent.has_all_breakdown(features=[]) + assert out["has_all"] is False + assert out["has_all"] == ent.has_all(features=[]) + row = out["axes"]["features"] + assert row is not None + assert row["has"] is False + assert row["blocking"] is True + assert "features" in out["blocking_axes"] + + +def test_parity_unknown_grant_token_collapses_to_false(ent): + """An unknown token in features collapses :func:`has_all` to False + even under grace (matches :func:`has_features` strict typo posture); + the breakdown must fold the same way and surface the axis as + blocking with the typo in ``unknown``.""" + out = ent.has_all_breakdown(features=["fleet", "bogus-feature"]) + assert out["has_all"] is False + assert out["has_all"] == ent.has_all(features=["fleet", "bogus-feature"]) + row = out["axes"]["features"] + assert row["has"] is False + assert row["blocking"] is True + assert "bogus-feature" in row["unknown"] + assert "fleet" in row["items"] + + +def test_parity_non_int_capacity_collapses_to_false(ent): + """Non-int capacity value collapses :func:`has_all` to False; the + breakdown must fold the same way and surface the axis as blocking + with the raw value in ``value``.""" + out = ent.has_all_breakdown(channels="five") + assert out["has_all"] is False + assert out["has_all"] == ent.has_all(channels="five") + row = out["axes"]["channels"] + assert row["has"] is False + assert row["blocking"] is True + assert row["value"] == "five" + + +# ── blocking-axis identification ─────────────────────────────────────────── + + +def test_blocking_axes_empty_when_has_all_true(ent): + """Grace grants every fully-known bundle; blocking_axes must be + empty and every axis row's ``blocking`` must be False.""" + out = ent.has_all_breakdown( + features=["fleet"], + runtimes=["openclaw"], + channels=1, + retention_days=7, + nodes=1, + ) + assert out["has_all"] is True + assert out["blocking_axes"] == [] + for k in AXIS_KEYS: + row = out["axes"][k] + if row is not None: + assert row["blocking"] is False + + +def test_blocking_axes_single_axis_denial(ent): + """When only one axis' :func:`has_*` is False, only that axis is + blocking. Uses ``channels=-1`` to force a per-axis False without + depending on tier caps (grace-independent typo path via a + non-positive channel count).""" + # Force the fold to False by giving features=[] alone -- only the + # features axis can be blocking. + out = ent.has_all_breakdown( + features=[], + runtimes=["openclaw"], + ) + assert out["has_all"] is False + assert out["blocking_axes"] == ["features"] + assert out["axes"]["features"]["blocking"] is True + assert out["axes"]["runtimes"]["blocking"] is False + + +def test_blocking_axes_multi_axis_denial_ordered(ent): + """When several axes deny, ``blocking_axes`` lists them in envelope + order (features / runtimes / channels / retention_days / nodes).""" + out = ent.has_all_breakdown( + features=[], runtimes=[], channels="five" + ) + assert out["has_all"] is False + # features + runtimes + channels all block; envelope order preserved. + assert out["blocking_axes"] == ["features", "runtimes", "channels"] + + +def test_blocking_axes_null_when_nothing_supplied(ent): + """Empty envelope: has_all=False (empty-``False`` typo posture) + but no axis was asked about, so ``blocking_axes`` stays empty.""" + out = ent.has_all_breakdown() + assert out["has_all"] is False + assert out["blocking_axes"] == [] + + +# ── per-axis row shape ───────────────────────────────────────────────────── + + +def test_grant_axis_row_shape(ent): + out = ent.has_all_breakdown(features=["fleet"]) + row = out["axes"]["features"] + assert row["kind"] == "features" + assert row["supplied"] is True + assert row["items"] == ["fleet"] + assert row["unknown"] == [] + assert isinstance(row["has"], bool) + assert isinstance(row["blocking"], bool) + + +def test_runtime_axis_row_shape(ent): + out = ent.has_all_breakdown(runtimes=["openclaw"]) + row = out["axes"]["runtimes"] + assert row["kind"] == "runtimes" + assert row["supplied"] is True + assert row["items"] == ["openclaw"] + assert row["unknown"] == [] + assert isinstance(row["has"], bool) + + +def test_capacity_axis_row_shape(ent): + out = ent.has_all_breakdown(channels=5) + row = out["axes"]["channels"] + assert row["kind"] == "channels" + assert row["supplied"] is True + assert row["value"] == 5 + assert isinstance(row["has"], bool) + + +def test_capacity_row_preserves_bad_input(ent): + """A non-int capacity value keeps ``supplied=True`` and echoes the + raw input in ``value`` so a tooltip can flag the typo. ``has`` is + False and the axis is blocking.""" + out = ent.has_all_breakdown(channels="five") + row = out["axes"]["channels"] + assert row["supplied"] is True + assert row["value"] == "five" + assert row["has"] is False + assert row["blocking"] is True + + +def test_runtime_axis_canonicalises_aliases(ent): + """``claude-code`` -> ``claude_code`` matches the singular + :func:`has_runtimes` posture (aliases collapse).""" + out = ent.has_all_breakdown(runtimes=["claude-code"]) + row = out["axes"]["runtimes"] + assert "claude_code" in row["items"] + + +def test_grant_axis_splits_known_from_unknown(ent): + out = ent.has_all_breakdown(features=["fleet", "bogus1", "bogus2"]) + row = out["axes"]["features"] + assert row["items"] == ["fleet"] + assert set(row["unknown"]) == {"bogus1", "bogus2"} + + +def test_unsupplied_axes_are_none(ent): + """Axes the caller did not supply short-circuit to ``None`` at the + envelope level (mirrors :func:`min_tier_for_all_breakdown`).""" + out = ent.has_all_breakdown(features=["fleet"]) + assert out["axes"]["runtimes"] is None + assert out["axes"]["channels"] is None + assert out["axes"]["retention_days"] is None + assert out["axes"]["nodes"] is None + + +# ── grace / enforce posture ──────────────────────────────────────────────── + + +def test_grace_grants_fully_known_bundle(ent): + """Grace posture: every fully-known bundle folds to ``has_all=True`` + with empty ``blocking_axes`` (matches :func:`has_all` grace).""" + out = ent.has_all_breakdown( + features=["fleet", "sso"], + runtimes=["openclaw", "claude_code"], + channels=100, + retention_days=90, + nodes=100, + ) + assert out["has_all"] is True + assert out["blocking_axes"] == [] + + +def test_grace_still_denies_unknown_token(ent): + """Grace does NOT rescue an unknown grant token (:func:`has_features` + strict typo posture is preserved even in grace); the breakdown + surfaces the axis as blocking with the typo in ``unknown``.""" + out = ent.has_all_breakdown(features=["Fleeet"]) + assert out["has_all"] is False + row = out["axes"]["features"] + assert row["has"] is False + assert row["blocking"] is True + assert "fleeet" in row["unknown"] or "Fleeet" in row["unknown"] + + +def test_grace_still_denies_non_int_capacity(ent): + out = ent.has_all_breakdown(channels="five") + assert out["has_all"] is False + assert "channels" in out["blocking_axes"] + + +# ── pair-invariants with min_tier_for_all_breakdown ──────────────────────── + + +def test_axes_echo_matches_reverse_lookup_breakdown(ent): + """The two breakdowns must agree on axis-level ``supplied`` / echo + slots so a paywall tooltip pairing them renders coherently.""" + bundle = dict( + features=["fleet"], + runtimes=["claude_code"], + channels=8, + retention_days=30, + nodes=2, + ) + fwd = ent.has_all_breakdown(**bundle) + rev = ent.min_tier_for_all_breakdown(**bundle) + for key in AXIS_KEYS: + f_row = fwd["axes"][key] + r_row = rev["axes"][key] + assert (f_row is None) == (r_row is None), key + if f_row is None: + continue + assert f_row["kind"] == r_row["kind"] + assert f_row["supplied"] == r_row["supplied"] + if "items" in f_row and "items" in r_row: + assert f_row["items"] == r_row["items"], key + if "value" in f_row and "value" in r_row: + assert f_row["value"] == r_row["value"], key + + +# ── never-raise contract ─────────────────────────────────────────────────── + + +def test_never_raises_on_garbage(ent): + """Every bad-input shape must fall through to the empty envelope + rather than 500-ing / raising.""" + for bad in ( + {"features": object()}, + {"runtimes": 42}, + {"channels": {"nope": "dict"}}, + {"retention_days": ["not", "int"]}, + {"nodes": None}, + ): + out = ent.has_all_breakdown(**bad) + assert isinstance(out, dict) + assert "has_all" in out + assert "axes" in out + assert "blocking_axes" in out + + +def test_never_raises_when_delegate_boom(ent, monkeypatch): + """A delegate raising mid-fold must short-circuit to the empty + envelope, not propagate.""" + def _boom(*a, **kw): + raise RuntimeError("boom") + + monkeypatch.setattr(ent, "has_features", _boom) + out = ent.has_all_breakdown(features=["fleet"]) + assert out["has_all"] is False + assert out["blocking_axes"] == [] + assert out["axes"]["features"] is None + + +# ── HTTP endpoint ────────────────────────────────────────────────────────── + + +def test_endpoint_400_on_no_args(client): + resp = client.get("/api/entitlement/has-all-breakdown") + assert resp.status_code == 400 + body = resp.get_json() + assert "error" in body + + +def test_endpoint_features_only_returns_shape(ent, client): + resp = client.get("/api/entitlement/has-all-breakdown?features=fleet") + assert resp.status_code == 200 + body = resp.get_json() + for k in ( + "features", + "runtimes", + "channels", + "retention_days", + "nodes", + "has_all", + "current_tier", + "current_tier_rank", + "grace", + "enforced", + "axes", + "blocking_axes", + ): + assert k in body, f"missing key {k}" + assert body["features"] == ["fleet"] + assert body["has_all"] == ent.has_all(features=["fleet"]) + + +def test_endpoint_mixed_bundle_matches_helper(ent, client): + resp = client.get( + "/api/entitlement/has-all-breakdown" + "?features=fleet&runtimes=claude_code&channels=8" + "&retention_days=30&nodes=2" + ) + assert resp.status_code == 200 + body = resp.get_json() + expected = ent.has_all_breakdown( + features=["fleet"], + runtimes=["claude_code"], + channels=8, + retention_days=30, + nodes=2, + ) + assert body["has_all"] == expected["has_all"] + assert body["blocking_axes"] == expected["blocking_axes"] + assert body["axes"] == expected["axes"] + + +def test_endpoint_bad_capacity_value_still_200(ent, client): + """Non-int capacity doesn't 400 -- the axis still surfaces with + ``has=false`` and the raw input in ``value`` (matches ``/has-*`` + never-crash posture).""" + resp = client.get( + "/api/entitlement/has-all-breakdown?features=fleet&channels=five" + ) + assert resp.status_code == 200 + body = resp.get_json() + assert body["has_all"] is False + assert body["axes"]["channels"] is not None + assert body["axes"]["channels"]["value"] == "five" + assert "channels" in body["blocking_axes"] + + +def test_endpoint_unknown_feature_flags_blocking(client): + resp = client.get("/api/entitlement/has-all-breakdown?features=bogus") + assert resp.status_code == 200 + body = resp.get_json() + assert body["has_all"] is False + assert "features" in body["blocking_axes"] + row = body["axes"]["features"] + assert row is not None + assert "bogus" in row["unknown"] + + +def test_endpoint_grace_flag_true_on_default_install(client): + """The default OSS install ships in grace -- the ``grace`` flag + MUST be True and ``enforced`` MUST be False in the response.""" + resp = client.get("/api/entitlement/has-all-breakdown?features=fleet") + body = resp.get_json() + assert body["grace"] is True + assert body["enforced"] is False + + +def test_endpoint_current_tier_defaults_to_oss(client): + resp = client.get("/api/entitlement/has-all-breakdown?features=fleet") + body = resp.get_json() + assert body["current_tier"] == "oss" + assert body["current_tier_rank"] == 0 + + +def test_endpoint_capacity_only_returns_shape(ent, client): + resp = client.get("/api/entitlement/has-all-breakdown?nodes=1") + assert resp.status_code == 200 + body = resp.get_json() + assert body["nodes"] == 1 + assert body["has_all"] == ent.has_all(nodes=1)