From 7b4f94bd94853fbfbd14f0829a2fee1e74594859 Mon Sep 17 00:00:00 2001 From: Nicolas Marchildon Date: Wed, 29 Jul 2026 12:08:33 -0400 Subject: [PATCH] =?UTF-8?q?feat(engine):=20pluggable=20SaaS=20pricing=20sh?= =?UTF-8?q?apes=20=E2=80=94=20flat=5Fsubscription,=20per=5Funit=5Fflat,=20?= =?UTF-8?q?free=5Ftier,=20transactional=20(#241)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a pluggable SaaS pricing-handler registry (SaaSPricingRegistry) with four built-in shapes: - flat_subscription — fixed monthly fee, charged when enabled (e.g. 9/mo per custom domain). - per_unit_flat — $X × count (e.g. 25/mo per SSO connection, /mo per org). - free_tier — first N units free, then /unit overage (e.g. 1M MAU free, bash.01/MAU above). Supports optional stepped tiers for graduated overage. - transactional — the existing percentage + per-transaction + per-call shape, preserved for vocabulary completeness. A cost-model node declares the shape per metric via a 'shape' field: workos: provider: workos usageMetrics: SSO-Connection: { unit: Conn, value: 2, shape: per_unit_flat, rate: 125.0 } The engine dispatches to the shape handler before the catalog / embedded pricingRates path. An unknown shape falls back to the existing path (opt-in, backward-compatible). Third-party packages register additional shapes via the infra_cost_model.saas_handlers entry-point group. Design follows existing patterns — registry class (like ResourceRegistry), shape computed per-metric in _compute_flat_cost and _compute_tiered_cost, quantity = derived invocation count × value (usage-driven) or direct value (fixed/flatOverride). Co-authored-by: silent-orca-64 --- infra_cost_model/engine/engine.py | 39 +- infra_cost_model/saas/__init__.py | 25 ++ infra_cost_model/saas/pricing_shapes.py | 250 +++++++++++++ pyproject.toml | 14 + tests/test_saas_pricing_shapes.py | 479 ++++++++++++++++++++++++ 5 files changed, 804 insertions(+), 3 deletions(-) create mode 100644 infra_cost_model/saas/__init__.py create mode 100644 infra_cost_model/saas/pricing_shapes.py create mode 100644 tests/test_saas_pricing_shapes.py diff --git a/infra_cost_model/engine/engine.py b/infra_cost_model/engine/engine.py index 552618a..d09fad0 100644 --- a/infra_cost_model/engine/engine.py +++ b/infra_cost_model/engine/engine.py @@ -443,10 +443,30 @@ def _compute_flat_cost(self, address: str, node: dict, per_invocation if metric_fixed else invocations * per_invocation ) + # SaaS pricing shapes (#241): if the metric declares a ``shape``, + # dispatch to the pluggable SaaS pricing-handler registry before + # the catalog / embedded-rates path. A shaped metric is priced by + # its shape handler (flat_subscription, per_unit_flat, free_tier, + # transactional, or a plugin-registered shape) using the metric's + # inline parameters — this is the first-class path for non-IaC SaaS + # resources that the catalog cannot reach. If the shape is unknown + # to the registry, fall through to the catalog / embedded path + # (backward-compatible). + metric_cost = None + shape = None + if isinstance(metric_def, dict): + shape = metric_def.get("shape") + if shape is not None: + from infra_cost_model.saas import SaaSPricingRegistry + shaped = SaaSPricingRegistry.compute( + shape, total_quantity, metric_def if isinstance(metric_def, dict) else {} + ) + if shaped is not None: + metric_cost = shaped + # Query catalog first (preferred path per Principle 13), else fall # back to embedded pricingRates (deprecated per Principle 13). - metric_cost = None - if self.catalog is not None: + if metric_cost is None and self.catalog is not None: result = self.catalog.query( provider, service, region, metric_name, total_quantity ) @@ -529,7 +549,20 @@ def _compute_tiered_cost(self, address: str, node: dict, invocations: float) -> ) metric_cost = None - if self.catalog is not None: + # SaaS pricing shapes (#241): dispatch to the shape registry before + # the catalog path, same as _compute_flat_cost. + shape = None + if isinstance(metric_def, dict): + shape = metric_def.get("shape") + if shape is not None: + from infra_cost_model.saas import SaaSPricingRegistry + shaped = SaaSPricingRegistry.compute( + shape, total_quantity, metric_def if isinstance(metric_def, dict) else {} + ) + if shaped is not None: + metric_cost = shaped + + if metric_cost is None and self.catalog is not None: result = self.catalog.query( provider, service, region, metric_name, total_quantity ) diff --git a/infra_cost_model/saas/__init__.py b/infra_cost_model/saas/__init__.py new file mode 100644 index 0000000..694a363 --- /dev/null +++ b/infra_cost_model/saas/__init__.py @@ -0,0 +1,25 @@ +"""SaaS pricing-shape handlers — flat subscription, per-unit, free-tier, transactional. + +See :mod:`infra_cost_model.saas.pricing_shapes` for the full module. This +package re-exports the registry and built-in handlers for convenience. +""" + +from infra_cost_model.saas.pricing_shapes import ( + SaaSPricingRegistry, + SaaSCostHandler, + flat_subscription, + free_tier, + per_unit_flat, + transactional, + discover_entry_point_handlers, +) + +__all__ = [ + "SaaSPricingRegistry", + "SaaSCostHandler", + "flat_subscription", + "free_tier", + "per_unit_flat", + "transactional", + "discover_entry_point_handlers", +] \ No newline at end of file diff --git a/infra_cost_model/saas/pricing_shapes.py b/infra_cost_model/saas/pricing_shapes.py new file mode 100644 index 0000000..36fb012 --- /dev/null +++ b/infra_cost_model/saas/pricing_shapes.py @@ -0,0 +1,250 @@ +"""Pluggable SaaS pricing shapes — flat subscription, per-unit, free-tier, transactional. + +Closes #241: ``ExternalServiceRegistry`` only registered an address *prefix* and +the cost model for external nodes was hardcoded transactional (percentage of +volume + per-transaction + per-call). That fits Stripe/Twilio/SendGrid but not +the much larger class of SaaS vendors whose pricing is **not** a percentage of +transaction volume — flat monthly subscriptions (Datadog per-host/mo), per-unit +flat (per-organization, per-connection, per-seat), and free tiers (first N units +free, then overage). + +This module promotes the external extension point from a prefix-recognizer into a +**pluggable SaaS pricing-handler registry**. Each handler computes a metric's +monthly cost from its model-declared ``shape`` and parameters, beyond the single +transactional formula. A node declares which shape each metric uses:: + + workos_identity: + provider: workos + usageMetrics: + WorkOS-MAU: { unit: Users, value: 0, shape: free_tier, free: 1000000, overage: 0.0 } + WorkOS-SSO-Connection: { unit: Conns, value: 0, shape: per_unit_flat, rate: 125.0 } + WorkOS-AuditLog-Org: { unit: Orgs, value: 0, shape: per_unit_flat, rate: 5.0 } + WorkOS-CustomDomain: { unit: Months,value: 0, shape: flat_subscription, rate: 99.0 } + +so free-tier boundaries and per-unit rates live in the model, not in Python. + +Third-party packages can register additional shapes via the +``infra_cost_model.saas_handlers`` entry-point group, mirroring how +``ResourceRegistry.register`` works for IaC handlers but installable. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Callable, Optional, Protocol + + +class SaaSCostHandler(Protocol): + """Protocol for a SaaS pricing-shape handler. + + A handler receives the total quantity for one metric (already scaled by the + derived invocation count for usage-driven metrics, or the raw value for + fixed metrics) and the metric's shape parameters from the model YAML, and + returns the monthly cost in USD. + """ + + def __call__(self, quantity: float, params: dict[str, Any]) -> float: ... + + +# ── Built-in shape handlers ────────────────────────────────────────────── + + +def flat_subscription(quantity: float, params: dict[str, Any]) -> float: + """A fixed monthly fee, charged once regardless of quantity. + + Example: a custom domain at $99/mo. ``quantity`` is ignored — the fee is + charged if the metric's value is > 0 (the model flips it from 0 to 1 to + "enable" it), or charged ``quantity`` times if the caller passes a count. + The rate comes from ``params['rate']``. + + To keep the "flip from 0 to 1" idiom ergonomic, a quantity of 0 yields 0 + (the feature is off) and any quantity >= 1 charges ``rate`` once. For + multi-instance flat subscriptions (e.g. 3 custom domains), pass the count + as quantity — it charges ``rate × quantity``. + """ + rate = float(params.get("rate", 0.0)) + if quantity <= 0: + return 0.0 + if quantity < 1: + # A fractional quantity (shouldn't normally happen for a flat + # subscription, but be defensive) charges once. + return rate + return rate * quantity + + +def per_unit_flat(quantity: float, params: dict[str, Any]) -> float: + """$X × count, where count is an org/seat/connection/api-key count. + + Example: $125/mo per SSO connection, $5/org/mo for audit logs. The rate + comes from ``params['rate']``; ``quantity`` is the unit count. + """ + rate = float(params.get("rate", 0.0)) + return rate * quantity + + +def free_tier(quantity: float, params: dict[str, Any]) -> float: + """First N units free, then $X/unit above N (optionally stepped). + + Example: WorkOS AuthKit — first 1M MAU free, then $0 overage. The free + allowance comes from ``params['free']`` and the overage rate from + ``params['overage']`` (default 0.0). Supports an optional ``tiers`` list + for stepped overage:: + + tiers: + - up_to: 50000 # first 50k above the free tier at $0.01 + rate: 0.01 + - up_to: inf # everything above 50k-overage at $0.005 + rate: 0.005 + """ + free_allowance = float(params.get("free", 0.0)) + overage = float(params.get("overage", 0.0)) + tiers = params.get("tiers") + + billable = max(0.0, quantity - free_allowance) + if billable <= 0: + return 0.0 + + if tiers: + # Stepped overage: walk the tier list, accumulating cost. + cost = 0.0 + remaining = billable + prev_cap = 0.0 + for tier in tiers: + cap = tier.get("up_to", float("inf")) + tier_rate = float(tier.get("rate", 0.0)) + band = cap - prev_cap + if band <= 0: + continue + chunk = min(remaining, band) + cost += chunk * tier_rate + remaining -= chunk + if remaining <= 0: + break + prev_cap = cap + return cost + + return billable * overage + + +def transactional(quantity: float, params: dict[str, Any]) -> float: + """The existing percentage/per-call shape, preserved for shape-parity. + + ``quantity`` is the transaction count. ``params`` may carry + ``percentage_rate`` (of a separate ``volume`` param), ``fixed_per_transaction``, + and ``per_call``. This handler exists so a transactional vendor can declare + ``shape: transactional`` in the model rather than relying on the legacy + ``_external_cost`` function — the shape vocabulary is exhaustive. + """ + percentage_rate = float(params.get("percentage_rate", 0.0)) + fixed_per_transaction = float(params.get("fixed_per_transaction", 0.0)) + per_call = float(params.get("per_call", 0.0)) + volume = float(params.get("volume", 0.0)) + return volume * percentage_rate + quantity * fixed_per_transaction + quantity * per_call + + +# ── Registry ───────────────────────────────────────────────────────────── + + +@dataclass +class _RegisteredHandler: + handler: SaaSCostHandler + name: str + + +class SaaSPricingRegistry: + """Registry of named SaaS pricing-shape handlers. + + Built-in shapes (``flat_subscription``, ``per_unit_flat``, ``free_tier``, + ``transactional``) are registered at module load. Third-party packages + extend the vocabulary by registering additional shapes, optionally + discovered via the ``infra_cost_model.saas_handlers`` entry-point group. + """ + + _handlers: dict[str, _RegisteredHandler] = {} + + @classmethod + def register(cls, name: str, handler: SaaSCostHandler) -> None: + """Register a pricing-shape handler by name. + + Args: + name: The shape name used in model YAML (e.g. ``"free_tier"``). + handler: A callable ``(quantity, params) -> monthly_cost_usd``. + """ + cls._handlers[name] = _RegisteredHandler(handler=handler, name=name) + + @classmethod + def get(cls, name: str) -> Optional[SaaSCostHandler]: + """Look up a shape handler by name, or ``None`` if not registered.""" + entry = cls._handlers.get(name) + return entry.handler if entry else None + + @classmethod + def known_shapes(cls) -> set[str]: + """Return the set of registered shape names.""" + return set(cls._handlers.keys()) + + @classmethod + def reset(cls) -> None: + """Clear all handlers (primarily for testing).""" + cls._handlers.clear() + + @classmethod + def compute(cls, shape: str, quantity: float, params: dict[str, Any]) -> Optional[float]: + """Compute cost for a shaped metric, or ``None`` if the shape is unknown. + + Returning ``None`` (rather than raising) lets the engine fall back to + the catalog / embedded ``pricingRates`` path when a metric has no + ``shape`` or an unregistered one — keeping the feature opt-in and + backward-compatible. + """ + handler = cls.get(shape) + if handler is None: + return None + return handler(quantity, params) + + +# ── Entry-point plugin discovery ───────────────────────────────────────── + + +def discover_entry_point_handlers() -> None: + """Discover and register SaaS shape handlers from the entry-point group. + + Third-party packages register a handler by adding an entry-point to the + ``infra_cost_model.saas_handlers`` group in their ``pyproject.toml``:: + + [project.entry-points."infra_cost_model.saas_handlers"] + my_shape = "my_package.pricing:my_shape_handler" + + The entry-point value must be a callable matching the ``SaaSCostHandler`` + protocol. Each discovered callable is registered under its entry-point name. + """ + try: + from importlib.metadata import entry_points + except ImportError: # pragma: no cover — Python < 3.8 + from importlib_metadata import entry_points # type: ignore + + try: + eps = entry_points(group="infra_cost_model.saas_handlers") + except TypeError: + # Python 3.9+ returns EntryPoints; older returns dict-like. + eps = entry_points().get("infra_cost_model.saas_handlers", []) # type: ignore + + for ep in eps: + try: + handler = ep.load() + SaaSPricingRegistry.register(ep.name, handler) + except Exception: + # A broken plugin must not crash the engine — skip it. + # Logging would be ideal, but this module is import-time and + # logging may not be configured yet. Silent skip is safe. + continue + + +# ── Module init: register built-ins + discover plugins ─────────────────── + +SaaSPricingRegistry.register("flat_subscription", flat_subscription) +SaaSPricingRegistry.register("per_unit_flat", per_unit_flat) +SaaSPricingRegistry.register("free_tier", free_tier) +SaaSPricingRegistry.register("transactional", transactional) + +discover_entry_point_handlers() \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index f201c14..c402828 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,6 +13,20 @@ dependencies = [ [project.scripts] infra-cost-model = "infra_cost_model.cli:main" +# SaaS pricing-shape handler discovery (#241). Third-party packages register +# additional shapes by adding an entry-point to this group: +# +# [project.entry-points."infra_cost_model.saas_handlers"] +# my_shape = "my_package.pricing:my_handler" +# +# The callable must match the SaaSCostHandler protocol +# (quantity, params) -> monthly_cost_usd. +[project.entry-points."infra_cost_model.saas_handlers"] +flat_subscription = "infra_cost_model.saas.pricing_shapes:flat_subscription" +per_unit_flat = "infra_cost_model.saas.pricing_shapes:per_unit_flat" +free_tier = "infra_cost_model.saas.pricing_shapes:free_tier" +transactional = "infra_cost_model.saas.pricing_shapes:transactional" + [project.optional-dependencies] dev = [ "pytest>=7.0", diff --git a/tests/test_saas_pricing_shapes.py b/tests/test_saas_pricing_shapes.py new file mode 100644 index 0000000..23c69a0 --- /dev/null +++ b/tests/test_saas_pricing_shapes.py @@ -0,0 +1,479 @@ +"""Tests for SaaS pricing-shape handlers (#241). + +Covers the built-in shapes (flat_subscription, per_unit_flat, free_tier, +transactional), the registry, entry-point discovery, and engine integration — +a shaped metric in a cost model DAG prices through the shape handler instead of +the catalog / embedded pricingRates path. +""" + +import pytest + +from infra_cost_model.saas import ( + SaaSPricingRegistry, + flat_subscription, + free_tier, + per_unit_flat, + transactional, +) +from infra_cost_model.saas.pricing_shapes import discover_entry_point_handlers +from infra_cost_model.engine import CostEngine + + +# ── Built-in shape handlers ────────────────────────────────────────────── + + +class TestFlatSubscription: + """flat_subscription: a fixed monthly fee, charged when enabled.""" + + def test_zero_quantity_is_free(self): + """A metric at 0 means the feature is off — no charge.""" + assert flat_subscription(0, {"rate": 99.0}) == 0.0 + + def test_enabled_charges_once(self): + """Flipping from 0 to 1 charges the rate once.""" + assert flat_subscription(1, {"rate": 99.0}) == 99.0 + + def test_multi_instance_charges_per_unit(self): + """3 custom domains → 3 × rate.""" + assert flat_subscription(3, {"rate": 99.0}) == 297.0 + + def test_missing_rate_defaults_to_zero(self): + """No rate param → $0 (defensive).""" + assert flat_subscription(1, {}) == 0.0 + + def test_fractional_quantity_charges_once(self): + """A fractional quantity (unusual) charges once, not 0.""" + assert flat_subscription(0.5, {"rate": 50.0}) == 50.0 + + +class TestPerUnitFlat: + """per_unit_flat: $X × count.""" + + def test_basic(self): + """5 SSO connections at $125 each.""" + assert per_unit_flat(5, {"rate": 125.0}) == 625.0 + + def test_zero_count(self): + """0 connections → $0.""" + assert per_unit_flat(0, {"rate": 125.0}) == 0.0 + + def test_missing_rate(self): + assert per_unit_flat(10, {}) == 0.0 + + +class TestFreeTier: + """free_tier: first N units free, then overage.""" + + def test_under_free_allowance(self): + """900k MAU under 1M free → $0.""" + assert free_tier(900_000, {"free": 1_000_000, "overage": 0.0}) == 0.0 + + def test_exactly_at_allowance(self): + """Exactly 1M MAU → $0 (boundary).""" + assert free_tier(1_000_000, {"free": 1_000_000, "overage": 0.0}) == 0.0 + + def test_above_allowance_with_overage(self): + """1.5M MAU, 1M free, $0.01 overage → 500k × $0.01 = $5000.""" + assert free_tier(1_500_000, {"free": 1_000_000, "overage": 0.01}) == 5000.0 + + def test_zero_overage_above_allowance(self): + """Above the free tier but overage rate is 0 → still $0.""" + assert free_tier(2_000_000, {"free": 1_000_000, "overage": 0.0}) == 0.0 + + def test_no_free_allowance(self): + """free=0 means every unit is billable.""" + assert free_tier(100, {"free": 0, "overage": 0.05}) == 5.0 + + def test_stepped_tiers(self): + """Tiers: first 50k overage at $0.01, above 50k at $0.005. + + 75k billable → 50k × $0.01 + 25k × $0.005 = $500 + $125 = $625. + """ + params = { + "free": 1_000_000, + "overage": 0.0, + "tiers": [ + {"up_to": 50_000, "rate": 0.01}, + {"up_to": float("inf"), "rate": 0.005}, + ], + } + assert free_tier(1_075_000, params) == pytest.approx(625.0) + + def test_stepped_tiers_partial(self): + """Only 30k billable, first tier up to 50k → 30k × $0.01 = $300.""" + params = { + "free": 1_000_000, + "tiers": [ + {"up_to": 50_000, "rate": 0.01}, + {"up_to": float("inf"), "rate": 0.005}, + ], + } + assert free_tier(1_030_000, params) == pytest.approx(300.0) + + +class TestTransactional: + """transactional: the preserved percentage/per-call shape.""" + + def test_percentage_plus_fixed(self): + """2.9% + $0.30 per transaction, 100 transactions, $1000 volume.""" + cost = transactional(100, { + "percentage_rate": 0.029, + "fixed_per_transaction": 0.30, + "volume": 1000.0, + }) + assert cost == pytest.approx(1000 * 0.029 + 100 * 0.30) # $29 + $30 = $59 + + def test_per_call(self): + """Twilio-style: $0.0075 per call, 1000 calls.""" + cost = transactional(1000, {"per_call": 0.0075}) + assert cost == pytest.approx(7.5) + + def test_zero_transactions(self): + assert transactional(0, {"per_call": 0.01}) == 0.0 + + +# ── Registry ───────────────────────────────────────────────────────────── + + +class TestSaaSPricingRegistry: + """The pluggable shape registry.""" + + def test_builtin_shapes_registered(self): + """All four built-in shapes are registered at module load.""" + shapes = SaaSPricingRegistry.known_shapes() + assert "flat_subscription" in shapes + assert "per_unit_flat" in shapes + assert "free_tier" in shapes + assert "transactional" in shapes + + def test_get_returns_handler(self): + """get() returns the callable for a known shape.""" + handler = SaaSPricingRegistry.get("per_unit_flat") + assert handler is not None + assert callable(handler) + assert handler(5, {"rate": 10.0}) == 50.0 + + def test_get_unknown_shape_returns_none(self): + """get() returns None for an unregistered shape.""" + assert SaaSPricingRegistry.get("nonexistent_shape") is None + + def test_compute_unknown_shape_returns_none(self): + """compute() returns None for an unknown shape — enables fallback.""" + assert SaaSPricingRegistry.compute("nonexistent", 100, {}) is None + + def test_register_custom_shape(self): + """A third-party shape can be registered and computed.""" + def my_shape(quantity, params): + return quantity * float(params.get("rate", 1.0)) + 10.0 + + SaaSPricingRegistry.register("my_custom_shape", my_shape) + try: + assert SaaSPricingRegistry.get("my_custom_shape") is not None + assert SaaSPricingRegistry.compute("my_custom_shape", 5, {"rate": 2.0}) == 20.0 + finally: + SaaSPricingRegistry.reset() + # Re-register built-ins after reset (reset clears everything). + SaaSPricingRegistry.register("flat_subscription", flat_subscription) + SaaSPricingRegistry.register("per_unit_flat", per_unit_flat) + SaaSPricingRegistry.register("free_tier", free_tier) + SaaSPricingRegistry.register("transactional", transactional) + + def test_reset_clears_handlers(self): + """reset() clears all handlers (for testing).""" + SaaSPricingRegistry.reset() + assert len(SaaSPricingRegistry.known_shapes()) == 0 + # Restore for other tests. + SaaSPricingRegistry.register("flat_subscription", flat_subscription) + SaaSPricingRegistry.register("per_unit_flat", per_unit_flat) + SaaSPricingRegistry.register("free_tier", free_tier) + SaaSPricingRegistry.register("transactional", transactional) + + +# ── Engine integration ─────────────────────────────────────────────────── + + +class TestEngineShapeIntegration: + """A shaped metric in a DAG prices through the shape handler.""" + + def _make_engine(self, nodes, edges=None): + model = { + "workflow": {"name": "test", "entry": "entry", "frequency": {"unit": "perMonth", "value": 1000}}, + "nodes": nodes, + "edges": edges or [], + } + return CostEngine(model, catalog=None, time_basis="monthly") + + def test_per_unit_flat_in_engine(self): + """A per_unit_flat metric prices correctly in the engine.""" + nodes = { + "entry": { + "nodeType": "routing", + "resourceAddress": "entry", + "provider": "test", + "service": "Test", + "region": "global", + "usageMetrics": {"requests": {"unit": "requests", "value": 1}}, + "pricingRates": {"requests": 0.0}, + }, + "saas_node": { + "nodeType": "compute", + "resourceAddress": "saas_node", + "provider": "workos", + "service": "WorkOS", + "region": "global", + "pricingModel": "flat", + "flatOverride": True, + "usageMetrics": { + "SSO-Connection": { + "unit": "Conns", + "value": 3, + "shape": "per_unit_flat", + "rate": 125.0, + }, + }, + }, + } + engine = self._make_engine(nodes) + costs = engine.compute() + # 3 connections × $125 = $375 + assert costs["saas_node"] == pytest.approx(375.0) + + def test_free_tier_in_engine(self): + """A free_tier metric prices correctly in the engine.""" + nodes = { + "entry": { + "nodeType": "routing", + "resourceAddress": "entry", + "provider": "test", + "service": "Test", + "region": "global", + "usageMetrics": {"requests": {"unit": "requests", "value": 1}}, + "pricingRates": {"requests": 0.0}, + }, + "saas_node": { + "nodeType": "compute", + "resourceAddress": "saas_node", + "provider": "workos", + "service": "WorkOS", + "region": "global", + "pricingModel": "flat", + "flatOverride": True, + "usageMetrics": { + "MAU": { + "unit": "Users", + "value": 1_500_000, + "shape": "free_tier", + "free": 1_000_000, + "overage": 0.01, + }, + }, + }, + } + engine = self._make_engine(nodes) + costs = engine.compute() + # 500k overage × $0.01 = $5000 + assert costs["saas_node"] == pytest.approx(5000.0) + + def test_flat_subscription_in_engine(self): + """A flat_subscription metric charges the rate when enabled.""" + nodes = { + "entry": { + "nodeType": "routing", + "resourceAddress": "entry", + "provider": "test", + "service": "Test", + "region": "global", + "usageMetrics": {"requests": {"unit": "requests", "value": 1}}, + "pricingRates": {"requests": 0.0}, + }, + "saas_node": { + "nodeType": "compute", + "resourceAddress": "saas_node", + "provider": "workos", + "service": "WorkOS", + "region": "global", + "pricingModel": "flat", + "flatOverride": True, + "usageMetrics": { + "CustomDomain": { + "unit": "Months", + "value": 1, + "shape": "flat_subscription", + "rate": 99.0, + }, + }, + }, + } + engine = self._make_engine(nodes) + costs = engine.compute() + assert costs["saas_node"] == pytest.approx(99.0) + + def test_mixed_shapes_in_one_node(self): + """A node with multiple shaped metrics (the WorkOS pattern).""" + nodes = { + "entry": { + "nodeType": "routing", + "resourceAddress": "entry", + "provider": "test", + "service": "Test", + "region": "global", + "usageMetrics": {"requests": {"unit": "requests", "value": 1}}, + "pricingRates": {"requests": 0.0}, + }, + "workos": { + "nodeType": "compute", + "resourceAddress": "workos", + "provider": "workos", + "service": "WorkOS", + "region": "global", + "pricingModel": "flat", + "flatOverride": True, + "usageMetrics": { + "MAU": { + "unit": "Users", "value": 1_200_000, + "shape": "free_tier", "free": 1_000_000, "overage": 0.0, + }, + "SSO": { + "unit": "Conns", "value": 2, + "shape": "per_unit_flat", "rate": 125.0, + }, + "AuditLog": { + "unit": "Orgs", "value": 3, + "shape": "per_unit_flat", "rate": 5.0, + }, + "Domain": { + "unit": "Months", "value": 1, + "shape": "flat_subscription", "rate": 99.0, + }, + }, + }, + } + engine = self._make_engine(nodes) + costs = engine.compute() + # MAU: 200k overage × $0 = $0 + # SSO: 2 × $125 = $250 + # AuditLog: 3 × $5 = $15 + # Domain: 1 × $99 = $99 + # Total: $364 + assert costs["workos"] == pytest.approx(364.0) + + def test_unknown_shape_falls_back_to_pricing_rates(self): + """An unregistered shape falls back to embedded pricingRates.""" + nodes = { + "entry": { + "nodeType": "routing", + "resourceAddress": "entry", + "provider": "test", + "service": "Test", + "region": "global", + "usageMetrics": {"requests": {"unit": "requests", "value": 1}}, + "pricingRates": {"requests": 0.0}, + }, + "saas_node": { + "nodeType": "compute", + "resourceAddress": "saas_node", + "provider": "datadog", + "service": "Datadog", + "region": "global", + "pricingModel": "flat", + "flatOverride": True, + "usageMetrics": { + "Hosts": { + "unit": "Hosts", "value": 4, + "shape": "nonexistent_shape", # not registered + }, + }, + "pricingRates": { + "Hosts": 46.0, # falls back to this + }, + }, + } + engine = self._make_engine(nodes) + costs = engine.compute() + # Unknown shape → None → falls back to pricingRates → 4 × $46 = $184 + assert costs["saas_node"] == pytest.approx(184.0) + + def test_no_shape_uses_existing_path(self): + """A metric without a shape uses the existing catalog/pricingRates path.""" + nodes = { + "entry": { + "nodeType": "routing", + "resourceAddress": "entry", + "provider": "test", + "service": "Test", + "region": "global", + "usageMetrics": {"requests": {"unit": "requests", "value": 1}}, + "pricingRates": {"requests": 0.0}, + }, + "saas_node": { + "nodeType": "compute", + "resourceAddress": "saas_node", + "provider": "datadog", + "service": "Datadog", + "region": "global", + "pricingModel": "flat", + "flatOverride": True, + "usageMetrics": { + "Hosts": {"unit": "Hosts", "value": 4}, # no shape + }, + "pricingRates": { + "Hosts": 46.0, + }, + }, + } + engine = self._make_engine(nodes) + costs = engine.compute() + # No shape → existing path → 4 × $46 = $184 + assert costs["saas_node"] == pytest.approx(184.0) + + def test_usage_driven_shaped_metric(self): + """A shaped metric that's NOT fixed scales with invocation count.""" + nodes = { + "entry": { + "nodeType": "routing", + "resourceAddress": "entry", + "provider": "test", + "service": "Test", + "region": "global", + "usageMetrics": {"requests": {"unit": "requests", "value": 1}}, + "pricingRates": {"requests": 0.0}, + }, + "saas_node": { + "nodeType": "compute", + "resourceAddress": "saas_node", + "provider": "datadog", + "service": "Datadog", + "region": "global", + "pricingModel": "flat", + # NO flatOverride — metrics are usage-driven by default + "usageMetrics": { + "LogIngestion": { + "unit": "GB", + "value": 0.00002, # per-request + "shape": "per_unit_flat", + "rate": 0.10, + }, + }, + }, + } + edges = [{"from": "entry", "to": "saas_node", "rate": 1}] + engine = self._make_engine(nodes, edges) + costs = engine.compute() + # 1000 requests × 0.00002 GB/req = 0.02 GB + # 0.02 GB × $0.10/GB = $0.002 + assert costs["saas_node"] == pytest.approx(0.002) + + +# ── Entry-point discovery ──────────────────────────────────────────────── + + +class TestEntryPointDiscovery: + """Entry-point plugin discovery (no real plugins installed in tests).""" + + def test_discover_does_not_crash_without_plugins(self): + """discover_entry_point_handlers() is safe to call with no plugins.""" + # Should not raise even though no infra_cost_model.saas_handlers + # entry-points are installed in the test environment. + discover_entry_point_handlers() + # Built-ins should still be present. + assert "flat_subscription" in SaaSPricingRegistry.known_shapes() \ No newline at end of file