X-Financial/server/tests/test_commercial_models.py

from __future__ import annotations

from sqlalchemy import ForeignKeyConstraint

from app.db.base import Base
from app.models.commercial import (
    CommercialCostEvent,
    CommercialEntitlement,
    TenantCommercialPlan,
    TenantSubscription,
    UsageMeterEvent,
)

COMMERCIAL_TABLES = {
    "tenant_commercial_plans",
    "tenant_subscriptions",
    "commercial_entitlements",
    "usage_meter_events",
    "commercial_cost_events",
}


def _constraint_names(model: type[object]) -> set[str]:
    return {
        str(constraint.name)
        for constraint in model.__table__.constraints  # type: ignore[attr-defined]
        if constraint.name is not None
    }


def _foreign_key_targets(model: type[object]) -> set[str]:
    return {
        element.target_fullname.split(".", maxsplit=1)[0]
        for constraint in model.__table__.constraints  # type: ignore[attr-defined]
        if isinstance(constraint, ForeignKeyConstraint)
        for element in constraint.elements
    }


def test_commercial_models_are_registered_as_separate_tenant_domain() -> None:
    models = (
        TenantCommercialPlan,
        TenantSubscription,
        CommercialEntitlement,
        UsageMeterEvent,
        CommercialCostEvent,
    )
    assert {model.__table__.name for model in models} == COMMERCIAL_TABLES
    assert COMMERCIAL_TABLES.issubset(Base.metadata.tables)
    assert all("tenant_id" in model.__table__.c for model in models)


def test_plan_subscription_and_entitlement_declare_contract_invariants() -> None:
    assert {
        "uq_tenant_commercial_plans_tenant_id",
        "uq_tenant_commercial_plans_tenant_code_version",
        "ck_tenant_commercial_plans_pricing_model",
        "ck_tenant_commercial_plans_billing_interval",
        "ck_tenant_commercial_plans_status",
        "ck_tenant_commercial_plans_values",
        "ck_tenant_commercial_plans_effective_window",
    }.issubset(_constraint_names(TenantCommercialPlan))
    assert {
        "uq_tenant_subscriptions_tenant_id",
        "uq_tenant_subscriptions_tenant_key",
        "uq_tenant_subscriptions_external_ref",
        "fk_tenant_subscriptions_tenant_plan",
        "ck_tenant_subscriptions_status",
        "ck_tenant_subscriptions_period",
        "ck_tenant_subscriptions_external_pair",
        "ck_tenant_subscriptions_cancellation",
    }.issubset(_constraint_names(TenantSubscription))
    assert {
        "uq_commercial_entitlements_tenant_id",
        "uq_commercial_entitlements_tenant_subscription_id",
        "uq_commercial_entitlements_subscription_key",
        "fk_commercial_entitlements_tenant_subscription",
        "ck_commercial_entitlements_type",
        "ck_commercial_entitlements_quota_shape",
        "ck_commercial_entitlements_overage_policy",
        "ck_commercial_entitlements_effective_window",
    }.issubset(_constraint_names(CommercialEntitlement))

    active_plan_index = next(
        index
        for index in TenantCommercialPlan.__table__.indexes
        if index.name == "uq_tenant_commercial_plans_active_code"
    )
    assert active_plan_index.unique is True
    assert "status = 'active'" in str(
        active_plan_index.dialect_options["postgresql"]["where"]
    )
    current_subscription_index = next(
        index
        for index in TenantSubscription.__table__.indexes
        if index.name == "uq_tenant_subscriptions_current"
    )
    assert current_subscription_index.unique is True
    assert "trialing" in str(
        current_subscription_index.dialect_options["postgresql"]["where"]
    )


def test_usage_and_cost_events_declare_idempotency_and_tenant_safe_links() -> None:
    assert {
        "uq_usage_meter_events_tenant_id",
        "uq_usage_meter_events_tenant_subscription_id",
        "uq_usage_meter_events_entitlement_id",
        "uq_usage_meter_events_source_request",
        "fk_usage_meter_events_tenant_subscription",
        "fk_usage_meter_events_tenant_entitlement",
        "fk_usage_meter_events_tenant_reversal",
        "ck_usage_meter_events_quantity",
        "ck_usage_meter_events_reversal",
        "ck_usage_meter_events_keys",
    }.issubset(_constraint_names(UsageMeterEvent))
    assert {
        "uq_commercial_cost_events_tenant_id",
        "uq_commercial_cost_events_source_request",
        "fk_commercial_cost_events_tenant_subscription",
        "fk_commercial_cost_events_tenant_usage",
        "fk_commercial_cost_events_tenant_reversal",
        "ck_commercial_cost_events_category",
        "ck_commercial_cost_events_amount_direction",
        "ck_commercial_cost_events_usage_pair",
        "ck_commercial_cost_events_keys",
    }.issubset(_constraint_names(CommercialCostEvent))

    usage_constraints = {
        constraint.name: constraint
        for constraint in UsageMeterEvent.__table__.constraints
        if isinstance(constraint, ForeignKeyConstraint)
    }
    assert tuple(
        usage_constraints["fk_usage_meter_events_tenant_entitlement"].column_keys
    ) == ("tenant_id", "subscription_id", "entitlement_id")
    assert tuple(usage_constraints["fk_usage_meter_events_tenant_reversal"].column_keys) == (
        "tenant_id",
        "subscription_id",
        "entitlement_id",
        "reversal_of_event_id",
    )
    cost_constraints = {
        constraint.name: constraint
        for constraint in CommercialCostEvent.__table__.constraints
        if isinstance(constraint, ForeignKeyConstraint)
    }
    assert tuple(cost_constraints["fk_commercial_cost_events_tenant_usage"].column_keys) == (
        "tenant_id",
        "subscription_id",
        "usage_event_id",
    )


def test_internal_cost_facts_are_physically_separate_from_customer_value_facts() -> None:
    savings_tables = {
        "profile_baseline_snapshots",
        "savings_opportunities",
        "savings_realizations",
        "savings_evidence_links",
        "savings_events",
    }
    assert _foreign_key_targets(CommercialCostEvent).isdisjoint(savings_tables)
    assert _foreign_key_targets(UsageMeterEvent).isdisjoint(savings_tables)
    assert "updated_at" not in UsageMeterEvent.__table__.c
    assert "updated_at" not in CommercialCostEvent.__table__.c