colin/colibre
1
0
Fork 0
Code Issues 30 Pull Requests Actions Packages Projects Releases 33 Activity

feat(subscriptions): historique multi-lignes + subscriber_state + handle abo-{user_id}-N

Browse Source 
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Colin Maudry
2026-07-01 14:29:47 +02:00
parent f52b2b7b71
commit 64158e9080
2 changed files with 367 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/subscriptions/db.py
+205 -67
View File
@@ -5,73 +5,209 @@ from src.auth.db import get_conn
SUBSCRIPTIONS_SCHEMA = """
CREATE TABLE IF NOT EXISTS subscriptions (
user_id INTEGER PRIMARY KEY,
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
frisbii_customer_handle TEXT,
frisbii_subscription_handle TEXT,
plan TEXT,
prix_ht REAL,
status TEXT,
current_period_end TEXT,
trial_used INTEGER NOT NULL DEFAULT 0,
votes_balance INTEGER NOT NULL DEFAULT 0,
votes_last_credited_at TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idx_subscriptions_customer
ON subscriptions(frisbii_customer_handle);
CREATE INDEX IF NOT EXISTS idx_subscriptions_user ON subscriptions(user_id);
CREATE UNIQUE INDEX IF NOT EXISTS idx_subscriptions_handle
ON subscriptions(frisbii_subscription_handle);
CREATE TABLE IF NOT EXISTS subscriber_state (
user_id INTEGER PRIMARY KEY,
trial_used INTEGER NOT NULL DEFAULT 0,
votes_balance INTEGER NOT NULL DEFAULT 0,
votes_last_credited_at TEXT,
updated_at TEXT NOT NULL,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
);
"""
_ACCESS_STATUSES = ("trial", "active")
def _now() -> str:
return datetime.now(timezone.utc).isoformat()
INITIAL_VOTES = 3
VOTES_PER_WEEK = 3
WEEK_SECONDS = 7 * 24 * 3600
def _now() -> str:
return datetime.now(timezone.utc).isoformat()
def init_schema() -> None:
get_conn().executescript(SUBSCRIPTIONS_SCHEMA)
conn = get_conn()
conn.executescript(SUBSCRIPTIONS_SCHEMA)
_migrate(conn)
def create_pending(
user_id: int, customer_handle: str, plan: str, prix_ht: float | None = None
) -> None:
now = _now()
get_conn().execute(
"INSERT INTO subscriptions "
"(user_id, frisbii_customer_handle, plan, prix_ht, status, created_at, updated_at) "
"VALUES (?, ?, ?, ?, 'pending', ?, ?) "
"ON CONFLICT(user_id) DO UPDATE SET "
"frisbii_customer_handle=excluded.frisbii_customer_handle, "
"plan=excluded.plan, prix_ht=excluded.prix_ht, "
"status='pending', updated_at=excluded.updated_at",
(user_id, customer_handle, plan, prix_ht, now, now),
def _migrate(conn: sqlite3.Connection) -> None:
cols = {row["name"] for row in conn.execute("PRAGMA table_info(subscriptions)")}
if "id" not in cols:
_rebuild_subscriptions_history(conn)
def _rebuild_subscriptions_history(conn: sqlite3.Connection) -> None:
# L'ancien schéma avait user_id en PRIMARY KEY (1 ligne par utilisateur) et
# portait aussi l'état cumulatif (votes, essai). SQLite ne peut ni retirer
# une PRIMARY KEY ni répartir des colonnes vers une autre table via ALTER :
# on reconstruit la table. foreign_keys OFF pour éviter le cascade-delete
# pendant le DROP.
conn.execute("PRAGMA foreign_keys = OFF")
try:
conn.execute("BEGIN")
conn.execute(
"""
CREATE TABLE subscriptions_new (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
frisbii_customer_handle TEXT,
frisbii_subscription_handle TEXT,
plan TEXT,
prix_ht REAL,
status TEXT,
current_period_end TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
)
"""
)
conn.execute(
"INSERT INTO subscriptions_new (user_id, frisbii_customer_handle, "
"frisbii_subscription_handle, plan, prix_ht, status, "
"current_period_end, created_at, updated_at) "
"SELECT user_id, frisbii_customer_handle, frisbii_subscription_handle, "
"plan, prix_ht, status, current_period_end, created_at, updated_at "
"FROM subscriptions"
)
conn.execute(
"INSERT INTO subscriber_state (user_id, trial_used, votes_balance, "
"votes_last_credited_at, updated_at) "
"SELECT user_id, trial_used, votes_balance, votes_last_credited_at, "
"updated_at FROM subscriptions"
)
conn.execute("DROP TABLE subscriptions")
conn.execute("ALTER TABLE subscriptions_new RENAME TO subscriptions")
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_subscriptions_user "
"ON subscriptions(user_id)"
)
conn.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS idx_subscriptions_handle "
"ON subscriptions(frisbii_subscription_handle)"
)
conn.execute("COMMIT")
except Exception:
conn.execute("ROLLBACK")
raise
finally:
conn.execute("PRAGMA foreign_keys = ON")
def get_by_user(user_id: int) -> sqlite3.Row | None:
def get_current(user_id: int) -> sqlite3.Row | None:
return (
get_conn()
.execute("SELECT * FROM subscriptions WHERE user_id = ?", (user_id,))
.execute(
"SELECT * FROM subscriptions WHERE user_id = ? ORDER BY id DESC LIMIT 1",
(user_id,),
)
.fetchone()
)
def get_by_customer(customer_handle: str) -> sqlite3.Row | None:
def get_by_handle(subscription_handle: str) -> sqlite3.Row | None:
return (
get_conn()
.execute(
"SELECT * FROM subscriptions WHERE frisbii_customer_handle = ?",
"SELECT * FROM subscriptions WHERE frisbii_subscription_handle = ?",
(subscription_handle,),
)
.fetchone()
)
def customer_known(customer_handle: str) -> bool:
row = (
get_conn()
.execute(
"SELECT 1 FROM subscriptions WHERE frisbii_customer_handle = ? LIMIT 1",
(customer_handle,),
)
.fetchone()
)
return row is not None
def _next_handle(user_id: int) -> str:
prefix = f"abo-{user_id}-"
rows = (
get_conn()
.execute(
"SELECT frisbii_subscription_handle FROM subscriptions "
"WHERE user_id = ? AND frisbii_subscription_handle LIKE ?",
(user_id, f"{prefix}%"),
)
.fetchall()
)
n = 0
for row in rows:
suffix = row["frisbii_subscription_handle"][len(prefix) :]
if suffix.isdigit():
n = max(n, int(suffix))
return f"{prefix}{n + 1}"
def create_pending(
user_id: int, customer_handle: str, plan: str, prix_ht: float | None = None
) -> tuple[str, int]:
"""Crée une nouvelle ligne d'historique en statut 'pending'.
Renvoie (handle, subscription_id). Le handle est choisi et enregistré ici,
avant tout appel à l'API Frisbii.
"""
now = _now()
handle = _next_handle(user_id)
conn = get_conn()
conn.execute(
"INSERT OR IGNORE INTO subscriber_state (user_id, updated_at) VALUES (?, ?)",
(user_id, now),
)
cur = conn.execute(
"INSERT INTO subscriptions "
"(user_id, frisbii_customer_handle, frisbii_subscription_handle, plan, "
"prix_ht, status, created_at, updated_at) "
"VALUES (?, ?, ?, ?, ?, 'pending', ?, ?)",
(user_id, customer_handle, handle, plan, prix_ht, now, now),
)
return handle, cur.lastrowid
def mark_failed(subscription_id: int) -> None:
"""Marque une ligne comme échouée (échec de l'appel Frisbii après create_pending).
Le handle reste enregistré (pas de suppression de la ligne) pour ne
jamais être réutilisé par _next_handle.
"""
get_conn().execute(
"UPDATE subscriptions SET status = 'failed', updated_at = ? WHERE id = ?",
(_now(), subscription_id),
)
def _get_state(user_id: int) -> sqlite3.Row | None:
return (
get_conn()
.execute("SELECT * FROM subscriber_state WHERE user_id = ?", (user_id,))
.fetchone()
)
def freeze_votes_cursor(user_id: int) -> None:
@@ -80,59 +216,60 @@ def freeze_votes_cursor(user_id: int) -> None:
Ne fait rien si le curseur est NULL (première activation jamais atteinte) :
les +2 initiaux restent gérés par credit_pending. Ne re-crédite jamais.
"""
row = get_by_user(user_id)
row = _get_state(user_id)
if row is None or row["votes_last_credited_at"] is None:
return
now = _now()
get_conn().execute(
"UPDATE subscriptions SET votes_last_credited_at = ?, updated_at = ? "
"UPDATE subscriber_state SET votes_last_credited_at = ?, updated_at = ? "
"WHERE user_id = ?",
(now, now, user_id),
)
def update_from_webhook(
customer_handle: str,
subscription_handle: str | None,
subscription_handle: str,
status: str,
current_period_end: str | None,
) -> None:
prev = get_by_customer(customer_handle)
if prev is not None and prev["status"] == "active" and status != "active":
prev = get_by_handle(subscription_handle)
if prev is None:
return
if prev["status"] == "active" and status != "active":
credit_pending(prev["user_id"]) # banque les semaines acquises avant gel
trial_flag = 1 if status in _ACCESS_STATUSES else 0
get_conn().execute(
"UPDATE subscriptions SET "
"frisbii_subscription_handle = COALESCE(?, frisbii_subscription_handle), "
"status = ?, current_period_end = ?, "
"trial_used = max(trial_used, ?), updated_at = ? "
"WHERE frisbii_customer_handle = ?",
(
subscription_handle,
status,
current_period_end,
trial_flag,
_now(),
customer_handle,
),
"UPDATE subscriptions SET status = ?, current_period_end = ?, "
"updated_at = ? WHERE id = ?",
(status, current_period_end, _now(), prev["id"]),
)
if prev is not None and prev["status"] != "active" and status == "active":
if status in _ACCESS_STATUSES:
get_conn().execute(
"UPDATE subscriber_state SET trial_used = 1, updated_at = ? "
"WHERE user_id = ?",
(_now(), prev["user_id"]),
)
if prev["status"] != "active" and status == "active":
freeze_votes_cursor(prev["user_id"])
def set_cancelled(user_id: int, current_period_end: str | None) -> None:
credit_pending(
user_id
) # banque les semaines pleines acquises (statut encore actif)
def set_cancelled(subscription_id: int, current_period_end: str | None) -> None:
row = (
get_conn()
.execute("SELECT user_id FROM subscriptions WHERE id = ?", (subscription_id,))
.fetchone()
)
if row is None:
return
credit_pending(row["user_id"]) # banque les semaines pleines acquises
get_conn().execute(
"UPDATE subscriptions SET status = 'cancelled', current_period_end = ?, "
"updated_at = ? WHERE user_id = ?",
(current_period_end, _now(), user_id),
"updated_at = ? WHERE id = ?",
(current_period_end, _now(), subscription_id),
)
def has_active_subscription(user_id: int) -> bool:
row = get_by_user(user_id)
row = get_current(user_id)
if row is None:
return False
if row["status"] in _ACCESS_STATUSES:
@@ -149,13 +286,13 @@ def has_active_subscription(user_id: int) -> bool:
def has_used_trial(user_id: int) -> bool:
row = get_by_user(user_id)
row = _get_state(user_id)
return bool(row and row["trial_used"])
def _set_votes(user_id: int, balance: int, cursor_iso: str) -> None:
get_conn().execute(
"UPDATE subscriptions SET votes_balance = ?, votes_last_credited_at = ?, "
"UPDATE subscriber_state SET votes_balance = ?, votes_last_credited_at = ?, "
"updated_at = ? WHERE user_id = ?",
(balance, cursor_iso, _now(), user_id),
)
@@ -168,14 +305,15 @@ def credit_pending(user_id: int) -> int:
pleine. Le solde est cappé à VOTES_PER_WEEK (pas d'accumulation).
Idempotent : ne crédite que des semaines pleines.
"""
row = get_by_user(user_id)
if row is None:
state = _get_state(user_id)
if state is None:
return 0
balance = row["votes_balance"] or 0
if row["status"] != "active":
balance = state["votes_balance"] or 0
current = get_current(user_id)
if current is None or current["status"] != "active":
return balance
now = datetime.now(timezone.utc)
cursor = row["votes_last_credited_at"]
cursor = state["votes_last_credited_at"]
if cursor is None:
balance = min(balance + INITIAL_VOTES, VOTES_PER_WEEK)
_set_votes(user_id, balance, now.isoformat())
@@ -192,8 +330,8 @@ def credit_pending(user_id: int) -> int:
def spend_vote(user_id: int) -> bool:
"""Débite 1 vote si le solde le permet. Renvoie True si un vote a été débité."""
cur = get_conn().execute(
"UPDATE subscriptions SET votes_balance = votes_balance - 1, updated_at = ? "
"WHERE user_id = ? AND votes_balance > 0",
"UPDATE subscriber_state SET votes_balance = votes_balance - 1, "
"updated_at = ? WHERE user_id = ? AND votes_balance > 0",
(_now(), user_id),
)
return cur.rowcount > 0
@@ -201,7 +339,7 @@ def spend_vote(user_id: int) -> bool:
def next_recharge_at(user_id: int) -> datetime | None:
"""Retourne la date du prochain rechargement de votes, ou None si non applicable."""
row = get_by_user(user_id)
row = _get_state(user_id)
if not row or not row["votes_last_credited_at"]:
return None
cursor = datetime.fromisoformat(row["votes_last_credited_at"])
tests/subscriptions/test_db.py
+153 -50
View File
@@ -1,6 +1,7 @@
from datetime import datetime, timedelta, timezone
from src.auth import db as auth_db
from src.auth.db import get_conn
from src.subscriptions import db
@@ -19,14 +20,17 @@ def _make_user(email="u@ex.fr"):
def _activate(uid, cursor_iso=None):
"""Met l'abonnement en statut actif avec un curseur d'accumulation donné."""
db.get_conn().execute(
"UPDATE subscriptions SET status = 'active', votes_last_credited_at = ? "
"WHERE user_id = ?",
row = db.get_current(uid)
get_conn().execute(
"UPDATE subscriptions SET status = 'active' WHERE id = ?", (row["id"],)
)
get_conn().execute(
"UPDATE subscriber_state SET votes_last_credited_at = ? WHERE user_id = ?",
(cursor_iso, uid),
)
def test_init_schema_creates_table(users_db_path):
def test_init_schema_creates_tables(users_db_path):
db.init_schema()
conn = auth_db.get_conn()
tables = {
@@ -34,55 +38,136 @@ def test_init_schema_creates_table(users_db_path):
for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")
}
assert "subscriptions" in tables
assert "subscriber_state" in tables
def test_create_pending_and_get_by_user(users_db_path):
def test_init_schema_migrates_old_single_row_subscriptions(users_db_path):
auth_db.init_schema()
uid = auth_db.create_user("legacy@ex.fr", "hash")
conn = get_conn()
conn.execute("DROP TABLE IF EXISTS subscriptions")
conn.execute("DROP TABLE IF EXISTS subscriber_state")
conn.execute(
"""
CREATE TABLE subscriptions (
user_id INTEGER PRIMARY KEY,
frisbii_customer_handle TEXT,
frisbii_subscription_handle TEXT,
plan TEXT,
prix_ht REAL,
status TEXT,
current_period_end TEXT,
trial_used INTEGER NOT NULL DEFAULT 0,
votes_balance INTEGER NOT NULL DEFAULT 0,
votes_last_credited_at TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
)
"""
)
conn.execute(
"INSERT INTO subscriptions (user_id, frisbii_customer_handle, "
"frisbii_subscription_handle, plan, status, trial_used, votes_balance, "
"votes_last_credited_at, created_at, updated_at) "
"VALUES (?, 'colibre-legacy', 'sub_legacy', 'simple', 'active', 1, 2, "
"'2026-01-01T00:00:00+00:00', '2026-01-01T00:00:00+00:00', "
"'2026-01-01T00:00:00+00:00')",
(uid,),
)
db.init_schema()
cols = {row["name"] for row in conn.execute("PRAGMA table_info(subscriptions)")}
assert "id" in cols
row = db.get_current(uid)
assert row["frisbii_subscription_handle"] == "sub_legacy"
assert row["status"] == "active"
state = conn.execute(
"SELECT * FROM subscriber_state WHERE user_id = ?", (uid,)
).fetchone()
assert state["trial_used"] == 1
assert state["votes_balance"] == 2
assert state["votes_last_credited_at"] == "2026-01-01T00:00:00+00:00"
def test_create_pending_and_get_current(users_db_path):
db.init_schema()
uid = _make_user()
db.create_pending(uid, "colibre-1", "simple")
row = db.get_by_user(uid)
handle, subscription_id = db.create_pending(uid, "colibre-1", "simple")
assert handle == f"abo-{uid}-1"
row = db.get_current(uid)
assert row["id"] == subscription_id
assert row["status"] == "pending"
assert row["plan"] == "simple"
assert row["frisbii_customer_handle"] == "colibre-1"
assert row["frisbii_subscription_handle"] == handle
def test_update_from_webhook_sets_status_and_handle(users_db_path):
def test_create_pending_generates_incrementing_handle_per_user(users_db_path):
db.init_schema()
uid = _make_user()
db.create_pending(uid, "colibre-1", "simple")
db.update_from_webhook("colibre-1", "sub_42", "trial", _future())
row = db.get_by_user(uid)
first_handle, first_id = db.create_pending(uid, "colibre-1", "simple")
db.mark_failed(first_id)
second_handle, _ = db.create_pending(uid, "colibre-1", "simple")
assert first_handle == f"abo-{uid}-1"
assert second_handle == f"abo-{uid}-2"
def test_mark_failed_sets_status_and_keeps_handle(users_db_path):
db.init_schema()
uid = _make_user()
handle, subscription_id = db.create_pending(uid, "colibre-1", "simple")
db.mark_failed(subscription_id)
row = db.get_current(uid)
assert row["status"] == "failed"
assert row["frisbii_subscription_handle"] == handle
def test_update_from_webhook_sets_status(users_db_path):
db.init_schema()
uid = _make_user()
handle, _ = db.create_pending(uid, "colibre-1", "simple")
db.update_from_webhook(handle, "trial", _future())
row = db.get_current(uid)
assert row["status"] == "trial"
assert row["frisbii_subscription_handle"] == "sub_42"
assert db.get_by_customer("colibre-1")["user_id"] == uid
assert row["frisbii_subscription_handle"] == handle
assert db.customer_known("colibre-1") is True
def test_customer_known_false_for_unknown_customer(users_db_path):
db.init_schema()
assert db.customer_known("colibre-inconnu") is False
def test_has_active_subscription_by_status(users_db_path):
db.init_schema()
uid = _make_user()
db.create_pending(uid, "colibre-1", "simple")
handle, _ = db.create_pending(uid, "colibre-1", "simple")
# pending → faux
assert db.has_active_subscription(uid) is False
for status in ("trial", "active"):
db.update_from_webhook("colibre-1", "sub_42", status, _future())
db.update_from_webhook(handle, status, _future())
assert db.has_active_subscription(uid) is True
# cancelled futur → vrai, cancelled passé → faux
db.update_from_webhook("colibre-1", "sub_42", "cancelled", _future())
db.update_from_webhook(handle, "cancelled", _future())
assert db.has_active_subscription(uid) is True
db.update_from_webhook("colibre-1", "sub_42", "cancelled", _past())
db.update_from_webhook(handle, "cancelled", _past())
assert db.has_active_subscription(uid) is False
db.update_from_webhook("colibre-1", "sub_42", "expired", None)
db.update_from_webhook(handle, "expired", None)
assert db.has_active_subscription(uid) is False
def test_set_cancelled(users_db_path):
db.init_schema()
uid = _make_user()
db.create_pending(uid, "colibre-1", "simple")
db.update_from_webhook("colibre-1", "sub_42", "active", _future())
handle, subscription_id = db.create_pending(uid, "colibre-1", "simple")
db.update_from_webhook(handle, "active", _future())
end = _future()
db.set_cancelled(uid, end)
row = db.get_by_user(uid)
db.set_cancelled(subscription_id, end)
row = db.get_current(uid)
assert row["status"] == "cancelled"
assert row["current_period_end"] == end
@@ -91,11 +176,10 @@ def test_has_active_subscription_z_suffix_datetime(users_db_path):
"""Fix 2 : suffix 'Z' dans current_period_end doit être parsé correctement."""
db.init_schema()
uid = _make_user()
db.create_pending(uid, "colibre-1", "simple")
# Frisbii peut renvoyer des dates avec 'Z' au lieu de '+00:00'.
db.update_from_webhook("colibre-1", "sub_42", "cancelled", "2099-12-31T23:59:59Z")
handle, _ = db.create_pending(uid, "colibre-1", "simple")
db.update_from_webhook(handle, "cancelled", "2099-12-31T23:59:59Z")
assert db.has_active_subscription(uid) is True
db.update_from_webhook("colibre-1", "sub_42", "cancelled", "2020-01-01T00:00:00Z")
db.update_from_webhook(handle, "cancelled", "2020-01-01T00:00:00Z")
assert db.has_active_subscription(uid) is False
@@ -103,14 +187,12 @@ def test_has_active_subscription_bad_datetime_returns_false(users_db_path):
"""Fix 2 : une date invalide ne doit pas lever d'exception, juste retourner False."""
db.init_schema()
uid = _make_user()
db.create_pending(uid, "colibre-1", "simple")
handle, subscription_id = db.create_pending(uid, "colibre-1", "simple")
# Injection directe d'une valeur invalide via update bas-niveau.
from src.auth.db import get_conn
get_conn().execute(
"UPDATE subscriptions SET status='cancelled', current_period_end='not-a-date' "
"WHERE user_id=?",
(uid,),
"WHERE id=?",
(subscription_id,),
)
assert db.has_active_subscription(uid) is False
@@ -118,24 +200,28 @@ def test_has_active_subscription_bad_datetime_returns_false(users_db_path):
def test_trial_used_is_sticky_across_resubscribe(users_db_path):
db.init_schema()
uid = _make_user()
db.create_pending(uid, "colibre-1", "simple")
handle, _ = db.create_pending(uid, "colibre-1", "simple")
assert db.has_used_trial(uid) is False
# l'abonnement entre en essai → trial_used positionné
db.update_from_webhook("colibre-1", "sub_42", "trial", _future())
db.update_from_webhook(handle, "trial", _future())
assert db.has_used_trial(uid) is True
# essai abandonné, puis nouvelle souscription : trial_used reste vrai
db.update_from_webhook("colibre-1", "sub_42", "expired", _past())
db.update_from_webhook(handle, "expired", _past())
db.create_pending(uid, "colibre-1", "soutien")
assert db.has_used_trial(uid) is True
def test_init_schema_creates_votes_columns(users_db_path):
def test_create_pending_initializes_subscriber_state(users_db_path):
db.init_schema()
uid = _make_user()
db.create_pending(uid, "colibre-1", "simple")
row = db.get_by_user(uid)
assert row["votes_balance"] == 0
assert row["votes_last_credited_at"] is None
state = (
get_conn()
.execute("SELECT * FROM subscriber_state WHERE user_id = ?", (uid,))
.fetchone()
)
assert state["votes_balance"] == 0
assert state["votes_last_credited_at"] is None
def test_credit_pending_grants_initial_two_on_first_active(users_db_path):
@@ -145,7 +231,15 @@ def test_credit_pending_grants_initial_two_on_first_active(users_db_path):
_activate(uid, cursor_iso=None)
balance = db.credit_pending(uid)
assert balance == db.INITIAL_VOTES
assert db.get_by_user(uid)["votes_last_credited_at"] is not None
state = (
get_conn()
.execute(
"SELECT votes_last_credited_at FROM subscriber_state WHERE user_id = ?",
(uid,),
)
.fetchone()
)
assert state["votes_last_credited_at"] is not None
def test_credit_pending_is_idempotent_same_day(users_db_path):
@@ -181,7 +275,12 @@ def test_spend_vote_decrements_when_balance_positive(users_db_path):
_activate(uid, cursor_iso=None)
db.credit_pending(uid) # solde = INITIAL_VOTES
assert db.spend_vote(uid) is True
assert db.get_by_user(uid)["votes_balance"] == db.INITIAL_VOTES - 1
state = (
get_conn()
.execute("SELECT votes_balance FROM subscriber_state WHERE user_id = ?", (uid,))
.fetchone()
)
assert state["votes_balance"] == db.INITIAL_VOTES - 1
def test_spend_vote_refused_when_balance_zero(users_db_path):
@@ -196,21 +295,25 @@ def test_spend_vote_refused_when_balance_zero(users_db_path):
def test_reactivation_resets_cursor_without_regranting(users_db_path):
db.init_schema()
uid = _make_user()
db.create_pending(uid, "colibre-1", "simple")
handle, _ = db.create_pending(uid, "colibre-1", "simple")
_activate(uid, cursor_iso=None)
db.credit_pending(uid) # +3, curseur posé
# désabonnement
db.update_from_webhook("colibre-1", "sub_1", "cancelled", _future())
db.update_from_webhook(handle, "cancelled", _future())
# période sans abonnement simulée : on recule artificiellement le curseur
old_cursor = (datetime.now(timezone.utc) - timedelta(days=30)).isoformat()
db.get_conn().execute(
"UPDATE subscriptions SET votes_last_credited_at = ? WHERE user_id = ?",
get_conn().execute(
"UPDATE subscriber_state SET votes_last_credited_at = ? WHERE user_id = ?",
(old_cursor, uid),
)
# réabonnement
db.update_from_webhook("colibre-1", "sub_1", "active", _future())
row = db.get_by_user(uid)
assert row["votes_balance"] == db.INITIAL_VOTES # pas de re-crédit des +3
db.update_from_webhook(handle, "active", _future())
state = (
get_conn()
.execute("SELECT votes_balance FROM subscriber_state WHERE user_id = ?", (uid,))
.fetchone()
)
assert state["votes_balance"] == db.INITIAL_VOTES # pas de re-crédit des +3
# le curseur a été remis ~à maintenant → pas de crédit du gap de 30 jours
assert db.credit_pending(uid) == db.INITIAL_VOTES
@@ -218,8 +321,8 @@ def test_reactivation_resets_cursor_without_regranting(users_db_path):
def test_trial_to_active_does_not_reset_then_grants_two(users_db_path):
db.init_schema()
uid = _make_user()
db.create_pending(uid, "colibre-1", "simple")
db.update_from_webhook("colibre-1", "sub_1", "trial", _future())
db.update_from_webhook("colibre-1", "sub_1", "active", _future())
handle, _ = db.create_pending(uid, "colibre-1", "simple")
db.update_from_webhook(handle, "trial", _future())
db.update_from_webhook(handle, "active", _future())
# fin d'essai : credit_pending accorde les +INITIAL_VOTES initiaux
assert db.credit_pending(uid) == db.INITIAL_VOTES