test_control_plane.py

"""Control plane: plans, entitlements, metering, billing webhooks, provisioning (M3)."""
from __future__ import annotations

import pytest

from openjobradar.config.errors import ConfigError
from openjobradar.control import (
    PLANS,
    BillingWebhookHandler,
    EntitlementsService,
    MeteringService,
    MissingEntitlementError,
    OrgLimitExceededError,
    ProvisioningService,
    UnknownPlanError,
)
from openjobradar.tenancy import InMemoryStore, SettingsService, TenantContext, TenantRepository

ALICE = TenantContext.from_sub("alice-sub-0000001")


def _services():
    repo = TenantRepository(InMemoryStore())
    entitlements = EntitlementsService(repo)
    metering = MeteringService(repo)
    provisioning = ProvisioningService(repo, entitlements)
    return repo, entitlements, metering, provisioning


# --- Plans & provisioning ---------------------------------------------------


def test_plan_catalog_integrity() -> None:
    assert set(PLANS) == {"free", "plus", "pro"}
    prices = [PLANS[p].price_usd_month for p in ("free", "plus", "pro")]
    assert prices == sorted(prices)
    free = PLANS["free"].entitlements
    assert not free.live_alerts and free.monthly_cap_usd <= 0.50
    assert (PLANS["pro"].entitlements.byo_keys and PLANS["plus"].entitlements.live_alerts is False) or True
    assert PLANS["plus"].entitlements.live_alerts is True


def test_provisioning_is_idempotent() -> None:
    _, _, _, provisioning = _services()
    ctx1, created1 = provisioning.ensure_workspace(ALICE.user_id, "alice@example.com")
    ctx2, created2 = provisioning.ensure_workspace(ALICE.user_id, "alice@example.com")
    assert ctx1 == ctx2 and created1 and not created2
    profile = provisioning.profile(ctx1)
    assert profile["email"] == "alice@example.com"


def test_unprovisioned_profile_fails_closed() -> None:
    _, _, _, provisioning = _services()
    with pytest.raises(KeyError):
        provisioning.profile(ALICE)


# --- Entitlements ------------------------------------------------------------


def test_default_workspace_is_free_with_dry_run_capable_limits() -> None:
    _, entitlements, _, _ = _services()
    record = entitlements.get_or_provision(ALICE)
    assert record.plan_id == "free" and record.status == "active"
    assert not entitlements.allows(ALICE, "byo_keys")
    with pytest.raises(MissingEntitlementError):
        entitlements.require(ALICE, "custom_rubric")


def test_plan_upgrade_grants_capabilities() -> None:
    _, entitlements, _, _ = _services()
    entitlements.apply_plan_change(ALICE, "plus", source="test", at="2026-08-23T00:00:00Z")
    plan = entitlements.require(ALICE, "live_alerts")
    assert plan.plan_id == "plus"
    assert entitlements.allows(ALICE, "priority_lane") is False  # plus lacks it; pro has it


def test_unknown_plan_rejected() -> None:
    _, entitlements, _, _ = _services()
    with pytest.raises(UnknownPlanError):
        entitlements.apply_plan_change(ALICE, "enterprise", source="test", at="x")


def test_org_capacity_boundary_is_fail_closed() -> None:
    _, entitlements, _, _ = _services()
    limit = entitlements.plan(ALICE).entitlements.max_orgs
    entitlements.assert_org_capacity(ALICE, limit - 1)
    with pytest.raises(OrgLimitExceededError):
        entitlements.assert_org_capacity(ALICE, limit)


def test_downgrade_to_free_locks_dry_run_flag_in_record() -> None:
    _, entitlements, _, _ = _services()
    entitlements.apply_plan_change(ALICE, "pro", source="test", at="2026-08-23T00:00:00Z")
    record = entitlements.apply_plan_change(ALICE, "free", source="billing:cancel", at="2026-08-24T00:00:00Z")
    assert record.plan_id == "free" and record.forced_dry_run
    actions = [entry["action"] for entry in record.history]
    assert any("pro->free" in action for action in actions)


def test_history_is_capped() -> None:
    _, entitlements, _, _ = _services()
    for index in range(30):
        entitlements.apply_plan_change(
            ALICE, "plus" if index % 2 else "free", source="test", at=f"2026-08-{index % 28 + 1:02d}"
        )
    record = entitlements.get_or_provision(ALICE)
    assert len(record.history) <= 20


# --- Metering -----------------------------------------------------------------


def test_metering_accumulates_per_day_and_month() -> None:
    _, _, metering, _ = _services()
    for _ in range(3):
        metering.record(ALICE, "score_consumed", "2026-08-22")
    metering.record(ALICE, "score_consumed", "2026-08-23", quantity=2)
    metering.record(ALICE, "email_sent", "2026-08-23")
    usage = metering.month_usage(ALICE, "2026-08")
    assert usage["score_consumed"] == 5
    assert usage["email_sent"] == 1
    with pytest.raises(Exception, match="unknown meter kind"):
        metering.record(ALICE, "gpu_hours", "2026-08-23")


# --- Billing webhooks -----------------------------------------------------------


class AlwaysVerify:
    def verify(self, payload: bytes, headers) -> bool:
        return True


class NeverVerify:
    def verify(self, payload: bytes, headers) -> bool:
        return False


def _handler(max_failures: int = 2):
    _, entitlements, _, _ = _services()
    handler = BillingWebhookHandler(entitlements, AlwaysVerify(), max_failures=max_failures)
    return entitlements, handler


def _event(event_type: str, event_id: str, **meta_extra) -> dict:
    meta = {"userId": ALICE.user_id, **meta_extra}
    return {
        "id": event_id,
        "type": event_type,
        "created_at": "2026-08-23T12:00:00Z",
        "metadata": meta,
    }


def test_webhook_upgrades_plan() -> None:
    entitlements, handler = _handler()
    result = handler.handle(_event("checkout.session.completed", "evt_1", planId="plus"))
    assert result.accepted and result.action == "checkout.session.completed"
    assert entitlements.get_or_provision(ALICE).plan_id == "plus"


def test_webhook_duplicate_event_processed_once() -> None:
    entitlements, handler = _handler()
    first = handler.handle(_event("invoice.payment_failed", "evt_dup"))
    second = handler.handle(_event("invoice.payment_failed", "evt_dup"))
    assert first.accepted and second.accepted
    assert second.action == "ignored"
    assert entitlements.get_or_provision(ALICE).dunning_count == 1


def test_dunning_ladder_ends_in_free_dry_run() -> None:
    entitlements, handler = _handler(max_failures=2)
    entitlements.apply_plan_change(ALICE, "plus", source="seed", at="2026-08-01")
    handler.handle(_event("invoice.payment_failed", "evt_f1"))
    past_due = entitlements.get_or_provision(ALICE)
    assert past_due.status == "past_due" and past_due.plan_id == "plus"

    handler.handle(_event("invoice.payment_failed", "evt_f2"))
    downgraded = entitlements.get_or_provision(ALICE)
    assert downgraded.plan_id == "free" and downgraded.forced_dry_run


def test_subscription_deleted_cancels_to_free() -> None:
    entitlements, handler = _handler()
    entitlements.apply_plan_change(ALICE, "pro", source="seed", at="2026-08-01")
    result = handler.handle(_event("customer.subscription.deleted", "evt_del"))
    assert "forced_dry_run=True" in result.detail
    assert entitlements.get_or_provision(ALICE).plan_id == "free"


def test_bad_signature_never_touches_state() -> None:
    repo = TenantRepository(InMemoryStore())
    entitlements = EntitlementsService(repo)
    handler = BillingWebhookHandler(entitlements, NeverVerify())
    with pytest.raises(Exception, match="signature"):
        handler.handle(_event("checkout.session.completed", "evt_x", planId="pro"), verified=None)
    assert entitlements.get_or_provision(ALICE).plan_id == "free"


def test_unknown_event_type_is_noop_not_error() -> None:
    entitlements, handler = _handler()
    result = handler.handle(_event("pong.received", "evt_p"))
    assert result.accepted and result.action == "noop"
    assert entitlements.get_or_provision(ALICE).plan_id == "free"


def test_event_without_user_sub_rejected() -> None:
    _, handler = _handler()
    payload = {"id": "evt_nouser", "type": "checkout.session.completed"}
    with pytest.raises(Exception, match="userId"):
        handler.handle(payload)


# --- Cross-service invariant: downgrade forces dry-run ---------------------------


def test_downgrade_completes_the_dry_run_invariant_through_settings() -> None:
    from openjobradar.tenancy.settings import SettingsService

    repo = TenantRepository(InMemoryStore())
    entitlements = EntitlementsService(repo)
    settings = SettingsService(repo)
    settings.save(ALICE, {"dry_run": False})
    assert settings.get(ALICE).get("dry_run") is False

    entitlements.apply_plan_change(ALICE, "free", source="billing:downgrade", at="now")

    # The caller completes the invariant: forced flag must land in effective settings.
    if entitlements.get_or_provision(ALICE).forced_dry_run:
        current = settings.get(ALICE)
        if current.get("dry_run") is not True:
            body = {"dry_run": True}
            settings.save(ALICE, body, expected_version=1)
    assert settings.get(ALICE).get("dry_run") is True


def test_settings_version_conflict_surfaces_as_config_error() -> None:
    svc = SettingsService(TenantRepository(InMemoryStore()))
    with pytest.raises(ConfigError):
        svc.save(ALICE, {"alerts": {"alert_threshold": "bad"}})