active subscriptions with nothing to charge on renewal

"""Report Stripe subscriptions with no payment method in any of the four slots.

Read only. GET requests only, no writes: give this a RESTRICTED key with read
access to Subscriptions and Customers. The repair is printed, never performed,
because this script holds a credential to a live payments account.
"""
import argparse
import logging
import os
import sys

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("stripe_sub_payment_method")

API = "https://api.stripe.com/v1"


def verdict(sub):
    """Walk Stripe's payment-method resolution order for one subscription.

    Pure, so the order can be tested against the documented one without a network.
    The order is: subscription.default_payment_method, subscription.default_source,
    customer.invoice_settings.default_payment_method, customer.default_source.
    """
    if sub.get("default_payment_method"):
        return ("subscription", "charges subscription.default_payment_method")
    if sub.get("default_source"):
        return ("subscription",
                "charges subscription.default_source, a legacy source object")
    customer = sub.get("customer")
    if not isinstance(customer, dict):
        return ("unknown",
                "customer was not expanded, so the two customer-level defaults "
                "cannot be read; re-run with expand[]=data.customer")
    settings = customer.get("invoice_settings") or {}
    if settings.get("default_payment_method"):
        return ("customer",
                "falls back to customer.invoice_settings.default_payment_method")
    if customer.get("default_source"):
        return ("customer",
                "falls back to customer.default_source, a legacy source object")
    return ("unchargeable",
            "all four resolution slots are null, so the renewal invoice cannot be "
            "paid and Stripe schedules no retry")


def get(session, path, **params):
    r = session.get(API + path, params=params, timeout=30)
    if r.status_code == 401:
        raise SystemExit("401 from Stripe: the key is wrong, or is for the other mode")
    r.raise_for_status()
    return r.json()


def page_subscriptions(session, status, limit):
    """Walk one status page by page. Read only; every call here is a GET."""
    out = []
    params = {"status": status, "limit": 100, "expand[]": "data.customer"}
    while True:
        page = get(session, "/subscriptions", **params)
        out.extend(page.get("data", []))
        if not page.get("has_more") or len(out) >= limit:
            break
        params["starting_after"] = page["data"][-1]["id"]
    return out[:limit]


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--status", action="append", default=None,
                    help="subscription status to check (repeatable)")
    ap.add_argument("--max", type=int, default=1000,
                    help="stop after this many subscriptions per status")
    args = ap.parse_args()
    statuses = args.status or ["active", "trialing"]

    key = os.environ.get("STRIPE_API_KEY")
    if not key:
        log.error("set STRIPE_API_KEY (use a restricted, read-only key)")
        return 2

    s = requests.Session()
    s.headers.update({"Authorization": "Bearer " + key})

    checked = 0
    counts = {}
    for status in statuses:
        for sub in page_subscriptions(s, status, args.max):
            checked += 1
            state, detail = verdict(sub)
            counts[state] = counts.get(state, 0) + 1
            if state == "subscription":
                continue
            line = "%-13s %s (%s)  %s" % (state, sub.get("id", "?"), status, detail)
            if state == "customer":
                log.info(line)
                continue
            log.warning(line)
            cus = sub.get("customer")
            cus_id = cus.get("id") if isinstance(cus, dict) else cus
            log.warning("  repair: collect a card with a SetupIntent or the billing "
                        "portal, then POST %s/customers/%s "
                        "-d invoice_settings[default_payment_method]=pm_...",
                        API, cus_id or "cus_...")
            log.warning("  and pin it to the subscription too: POST %s/subscriptions/%s "
                        "-d default_payment_method=pm_...", API, sub.get("id"))

    log.info("%d subscription(s) checked, %d unchargeable, %d relying on a "
             "customer-level default", checked, counts.get("unchargeable", 0),
             counts.get("customer", 0))
    if counts.get("unknown"):
        log.warning("%d row(s) could not be classified: re-run with the customer "
                    "expanded", counts["unknown"])
    return 1 if counts.get("unchargeable") or counts.get("unknown") else 0


if __name__ == "__main__":
    sys.exit(main())

/**
 * Report Stripe subscriptions with no payment method in any of the four slots.
 *
 * Read only. GET requests only, no writes: give this a RESTRICTED key with read
 * access to Subscriptions and Customers. The repair is printed, never performed.
 */
const API = 'https://api.stripe.com/v1';

/**
 * Walk Stripe's payment-method resolution order for one subscription.
 * Pure, so the order can be tested against the documented one without a network.
 */
export function verdict(sub) {
  if (sub.default_payment_method) {
    return ['subscription', 'charges subscription.default_payment_method'];
  }
  if (sub.default_source) {
    return ['subscription', 'charges subscription.default_source, a legacy source object'];
  }
  const customer = sub.customer;
  if (customer === null || typeof customer !== 'object') {
    return ['unknown',
      'customer was not expanded, so the two customer-level defaults cannot be ' +
      'read; re-run with expand[]=data.customer'];
  }
  const settings = customer.invoice_settings ?? {};
  if (settings.default_payment_method) {
    return ['customer', 'falls back to customer.invoice_settings.default_payment_method'];
  }
  if (customer.default_source) {
    return ['customer', 'falls back to customer.default_source, a legacy source object'];
  }
  return ['unchargeable',
    'all four resolution slots are null, so the renewal invoice cannot be paid ' +
    'and Stripe schedules no retry'];
}

async function get(key, path, params = {}) {
  const url = new URL(API + path);
  for (const [k, v] of Object.entries(params)) url.searchParams.set(k, v);
  const res = await fetch(url, { headers: { Authorization: `Bearer ${key}` } });
  if (res.status === 401) {
    throw new Error('401 from Stripe: the key is wrong, or is for the other mode');
  }
  if (!res.ok) throw new Error(`${res.status} from ${url.pathname}`);
  return res.json();
}

async function pageSubscriptions(key, status, limit) {
  const out = [];
  const q = { status, limit: 100, 'expand[]': 'data.customer' };
  for (;;) {
    const page = await get(key, '/subscriptions', q);
    out.push(...(page.data ?? []));
    if (!page.has_more || out.length >= limit) break;
    q.starting_after = page.data[page.data.length - 1].id;
  }
  return out.slice(0, limit);
}

async function main() {
  const key = process.env.STRIPE_API_KEY;
  if (!key) {
    console.error('set STRIPE_API_KEY (use a restricted, read-only key)');
    process.exitCode = 2;
    return;
  }

  let checked = 0;
  const counts = new Map();
  for (const status of ['active', 'trialing']) {
    for (const sub of await pageSubscriptions(key, status, 1000)) {
      checked += 1;
      const [state, detail] = verdict(sub);
      counts.set(state, (counts.get(state) ?? 0) + 1);
      if (state === 'subscription') continue;
      const line = `${state.padEnd(13)} ${sub.id ?? '?'} (${status})  ${detail}`;
      if (state === 'customer') { console.log(line); continue; }
      console.warn(line);
      const cus = typeof sub.customer === 'object' && sub.customer !== null
        ? sub.customer.id : sub.customer;
      console.warn(`  repair: collect a card with a SetupIntent or the billing portal, ` +
        `then POST ${API}/customers/${cus ?? 'cus_...'} ` +
        `-d invoice_settings[default_payment_method]=pm_...`);
      console.warn(`  and pin it to the subscription too: ` +
        `POST ${API}/subscriptions/${sub.id} -d default_payment_method=pm_...`);
    }
  }

  console.log(`${checked} subscription(s) checked, ` +
    `${counts.get('unchargeable') ?? 0} unchargeable, ` +
    `${counts.get('customer') ?? 0} relying on a customer-level default`);
  if (counts.get('unknown')) {
    console.warn(`${counts.get('unknown')} row(s) could not be classified: re-run ` +
      'with the customer expanded');
  }
  process.exitCode = (counts.get('unchargeable') ?? 0) + (counts.get('unknown') ?? 0)
    ? 1 : 0;
}

// Only run when invoked directly. The test file imports this module, and without
// the guard main() would run there too, fail on the missing key, and set a
// non-zero exit code that fails the whole test file even as every test passes.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from stripe_sub_payment_method import verdict


def test_subscription_level_payment_method_wins():
    state, detail = verdict({"default_payment_method": "pm_1", "customer": {}})
    assert state == "subscription"
    assert "subscription.default_payment_method" in detail


def test_legacy_subscription_source_is_still_chargeable():
    state, detail = verdict({"default_source": "card_1", "customer": {}})
    assert state == "subscription"
    assert "legacy" in detail


def test_customer_invoice_settings_are_the_third_slot():
    sub = {"customer": {"invoice_settings": {"default_payment_method": "pm_2"}}}
    state, detail = verdict(sub)
    assert state == "customer"
    assert "invoice_settings" in detail


def test_customer_default_source_is_the_fourth_slot():
    state, _ = verdict({"customer": {"default_source": "card_2"}})
    assert state == "customer"


def test_all_four_null_is_unchargeable_and_says_no_retry():
    sub = {"customer": {"invoice_settings": {"default_payment_method": None},
                        "default_source": None}}
    state, detail = verdict(sub)
    assert state == "unchargeable"
    assert "no retry" in detail


def test_unexpanded_customer_is_not_reported_as_unchargeable():
    # A bare id string looks identical to an absent default. Saying "unchargeable"
    # here would point someone at every healthy subscription in the account.
    state, detail = verdict({"customer": "cus_123"})
    assert state == "unknown"
    assert "expand" in detail

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { verdict } from './stripe-sub-payment-method.mjs';

test('subscription level payment method wins', () => {
  const [state, detail] = verdict({ default_payment_method: 'pm_1', customer: {} });
  assert.equal(state, 'subscription');
  assert.match(detail, /subscription\.default_payment_method/);
});

test('legacy subscription source is still chargeable', () => {
  const [state, detail] = verdict({ default_source: 'card_1', customer: {} });
  assert.equal(state, 'subscription');
  assert.match(detail, /legacy/);
});

test('customer invoice settings are the third slot', () => {
  const [state, detail] = verdict({
    customer: { invoice_settings: { default_payment_method: 'pm_2' } },
  });
  assert.equal(state, 'customer');
  assert.match(detail, /invoice_settings/);
});

test('customer default source is the fourth slot', () => {
  assert.equal(verdict({ customer: { default_source: 'card_2' } })[0], 'customer');
});

test('all four null is unchargeable and says no retry', () => {
  const [state, detail] = verdict({
    customer: { invoice_settings: { default_payment_method: null }, default_source: null },
  });
  assert.equal(state, 'unchargeable');
  assert.match(detail, /no retry/);
});

test('unexpanded customer is not reported as unchargeable', () => {
  const [state, detail] = verdict({ customer: 'cus_123' });
  assert.equal(state, 'unknown');
  assert.match(detail, /expand/);
});