4ae00b35d1
git mv all tracked files from the nested padelnomics/ workspace directory to the git repo root. Merged .gitignore files. No code changes — pure path rename. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
293 lines
11 KiB
Python
293 lines
11 KiB
Python
"""
|
|
Tests for the credit system (credits.py).
|
|
|
|
Pure SQL operations against real in-memory SQLite — no mocking needed.
|
|
"""
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
from padelnomics.credits import (
|
|
InsufficientCredits,
|
|
add_credits,
|
|
already_unlocked,
|
|
compute_credit_cost,
|
|
get_balance,
|
|
get_ledger,
|
|
monthly_credit_refill,
|
|
spend_credits,
|
|
unlock_lead,
|
|
)
|
|
|
|
# ── Fixtures ─────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture
|
|
async def supplier(db):
|
|
"""Supplier with credit_balance=100, monthly_credits=30, tier=growth."""
|
|
now = datetime.utcnow().isoformat()
|
|
async with db.execute(
|
|
"""INSERT INTO suppliers
|
|
(name, slug, country_code, region, category, tier,
|
|
credit_balance, monthly_credits, created_at)
|
|
VALUES ('Test Supplier', 'test-supplier', 'DE', 'Europe', 'Courts',
|
|
'growth', 100, 30, ?)""",
|
|
(now,),
|
|
) as cursor:
|
|
supplier_id = cursor.lastrowid
|
|
await db.commit()
|
|
return {"id": supplier_id}
|
|
|
|
|
|
@pytest.fixture
|
|
async def lead(db):
|
|
"""Lead request with heat_score=warm, credit_cost=20."""
|
|
now = datetime.utcnow().isoformat()
|
|
async with db.execute(
|
|
"""INSERT INTO lead_requests
|
|
(lead_type, heat_score, credit_cost, status, created_at)
|
|
VALUES ('supplier_quote', 'warm', 20, 'new', ?)""",
|
|
(now,),
|
|
) as cursor:
|
|
lead_id = cursor.lastrowid
|
|
await db.commit()
|
|
return {"id": lead_id, "credit_cost": 20}
|
|
|
|
|
|
# ── GetBalance ───────────────────────────────────────────────
|
|
|
|
|
|
class TestGetBalance:
|
|
async def test_returns_zero_for_unknown_supplier(self, db):
|
|
assert await get_balance(99999) == 0
|
|
|
|
async def test_returns_current_balance(self, db, supplier):
|
|
assert await get_balance(supplier["id"]) == 100
|
|
|
|
|
|
# ── AddCredits ───────────────────────────────────────────────
|
|
|
|
|
|
class TestAddCredits:
|
|
async def test_adds_credits_and_updates_balance(self, db, supplier):
|
|
new_balance = await add_credits(supplier["id"], 50, "pack_purchase")
|
|
assert new_balance == 150
|
|
assert await get_balance(supplier["id"]) == 150
|
|
|
|
async def test_creates_ledger_entry(self, db, supplier):
|
|
await add_credits(
|
|
supplier["id"], 50, "pack_purchase",
|
|
reference_id=42, note="Credit pack: credits_50",
|
|
)
|
|
rows = await db.execute_fetchall(
|
|
"SELECT * FROM credit_ledger WHERE supplier_id = ?",
|
|
(supplier["id"],),
|
|
)
|
|
assert len(rows) == 1
|
|
row = rows[0]
|
|
assert row[2] == 50 # delta
|
|
assert row[3] == 150 # balance_after
|
|
assert row[4] == "pack_purchase" # event_type
|
|
assert row[5] == 42 # reference_id
|
|
assert row[6] == "Credit pack: credits_50" # note
|
|
|
|
async def test_multiple_adds_accumulate(self, db, supplier):
|
|
await add_credits(supplier["id"], 10, "pack_purchase")
|
|
await add_credits(supplier["id"], 10, "pack_purchase")
|
|
await add_credits(supplier["id"], 10, "pack_purchase")
|
|
assert await get_balance(supplier["id"]) == 130
|
|
|
|
|
|
# ── SpendCredits ─────────────────────────────────────────────
|
|
|
|
|
|
class TestSpendCredits:
|
|
async def test_spends_credits_and_updates_balance(self, db, supplier):
|
|
new_balance = await spend_credits(supplier["id"], 30, "lead_unlock")
|
|
assert new_balance == 70
|
|
assert await get_balance(supplier["id"]) == 70
|
|
|
|
async def test_creates_negative_ledger_entry(self, db, supplier):
|
|
await spend_credits(supplier["id"], 30, "lead_unlock")
|
|
rows = await db.execute_fetchall(
|
|
"SELECT * FROM credit_ledger WHERE supplier_id = ?",
|
|
(supplier["id"],),
|
|
)
|
|
assert len(rows) == 1
|
|
assert rows[0][2] == -30 # delta
|
|
assert rows[0][3] == 70 # balance_after
|
|
|
|
async def test_raises_insufficient_credits(self, db, supplier):
|
|
with pytest.raises(InsufficientCredits) as exc_info:
|
|
await spend_credits(supplier["id"], 200, "lead_unlock")
|
|
assert exc_info.value.balance == 100
|
|
assert exc_info.value.required == 200
|
|
|
|
async def test_spend_exact_balance(self, db, supplier):
|
|
new_balance = await spend_credits(supplier["id"], 100, "lead_unlock")
|
|
assert new_balance == 0
|
|
assert await get_balance(supplier["id"]) == 0
|
|
|
|
|
|
# ── ComputeCreditCost ────────────────────────────────────────
|
|
|
|
|
|
class TestComputeCreditCost:
|
|
def test_hot_costs_35(self):
|
|
assert compute_credit_cost({"heat_score": "hot"}) == 35
|
|
|
|
def test_warm_costs_20(self):
|
|
assert compute_credit_cost({"heat_score": "warm"}) == 20
|
|
|
|
def test_cool_costs_8(self):
|
|
assert compute_credit_cost({"heat_score": "cool"}) == 8
|
|
|
|
def test_missing_heat_defaults_to_cool(self):
|
|
assert compute_credit_cost({}) == 8
|
|
assert compute_credit_cost({"heat_score": None}) == 8
|
|
|
|
|
|
# ── AlreadyUnlocked ──────────────────────────────────────────
|
|
|
|
|
|
class TestAlreadyUnlocked:
|
|
async def test_returns_false_when_not_unlocked(self, db, supplier, lead):
|
|
assert await already_unlocked(supplier["id"], lead["id"]) is False
|
|
|
|
async def test_returns_true_after_unlock(self, db, supplier, lead):
|
|
now = datetime.utcnow().isoformat()
|
|
await db.execute(
|
|
"""INSERT INTO lead_forwards (lead_id, supplier_id, credit_cost, created_at)
|
|
VALUES (?, ?, 20, ?)""",
|
|
(lead["id"], supplier["id"], now),
|
|
)
|
|
await db.commit()
|
|
assert await already_unlocked(supplier["id"], lead["id"]) is True
|
|
|
|
|
|
# ── UnlockLead ────────────────────────────────────────────────
|
|
|
|
|
|
class TestUnlockLead:
|
|
async def test_unlocks_lead_and_returns_details(self, db, supplier, lead):
|
|
result = await unlock_lead(supplier["id"], lead["id"])
|
|
|
|
assert result["forward_id"] is not None
|
|
assert result["credit_cost"] == 20
|
|
assert result["new_balance"] == 80
|
|
assert result["lead"]["id"] == lead["id"]
|
|
|
|
# Verify lead_forwards row
|
|
row = await db.execute_fetchall(
|
|
"SELECT * FROM lead_forwards WHERE supplier_id = ? AND lead_id = ?",
|
|
(supplier["id"], lead["id"]),
|
|
)
|
|
assert len(row) == 1
|
|
assert row[0][3] == 20 # credit_cost
|
|
|
|
# Verify ledger entry
|
|
ledger = await db.execute_fetchall(
|
|
"SELECT * FROM credit_ledger WHERE supplier_id = ? AND event_type = 'lead_unlock'",
|
|
(supplier["id"],),
|
|
)
|
|
assert len(ledger) == 1
|
|
assert ledger[0][2] == -20 # delta
|
|
assert ledger[0][3] == 80 # balance_after
|
|
|
|
# Verify unlock_count incremented
|
|
lead_row = await db.execute_fetchall(
|
|
"SELECT unlock_count FROM lead_requests WHERE id = ?", (lead["id"],),
|
|
)
|
|
assert lead_row[0][0] == 1
|
|
|
|
async def test_raises_on_duplicate_unlock(self, db, supplier, lead):
|
|
await unlock_lead(supplier["id"], lead["id"])
|
|
with pytest.raises(ValueError, match="already unlocked"):
|
|
await unlock_lead(supplier["id"], lead["id"])
|
|
|
|
async def test_raises_on_missing_lead(self, db, supplier):
|
|
with pytest.raises(ValueError, match="not found"):
|
|
await unlock_lead(supplier["id"], 99999)
|
|
|
|
async def test_raises_insufficient_credits(self, db, lead):
|
|
"""Supplier with only 5 credits tries to unlock a 20-credit lead."""
|
|
now = datetime.utcnow().isoformat()
|
|
async with db.execute(
|
|
"""INSERT INTO suppliers
|
|
(name, slug, country_code, region, category, tier,
|
|
credit_balance, monthly_credits, created_at)
|
|
VALUES ('Poor Supplier', 'poor-supplier', 'DE', 'Europe', 'Courts',
|
|
'growth', 5, 0, ?)""",
|
|
(now,),
|
|
) as cursor:
|
|
poor_id = cursor.lastrowid
|
|
await db.commit()
|
|
|
|
with pytest.raises(InsufficientCredits) as exc_info:
|
|
await unlock_lead(poor_id, lead["id"])
|
|
assert exc_info.value.balance == 5
|
|
assert exc_info.value.required == 20
|
|
|
|
|
|
# ── MonthlyRefill ─────────────────────────────────────────────
|
|
|
|
|
|
class TestMonthlyRefill:
|
|
async def test_refills_monthly_credits(self, db, supplier):
|
|
new_balance = await monthly_credit_refill(supplier["id"])
|
|
assert new_balance == 130 # 100 + 30
|
|
assert await get_balance(supplier["id"]) == 130
|
|
|
|
# Verify ledger entry
|
|
ledger = await db.execute_fetchall(
|
|
"SELECT * FROM credit_ledger WHERE supplier_id = ? AND event_type = 'monthly_allocation'",
|
|
(supplier["id"],),
|
|
)
|
|
assert len(ledger) == 1
|
|
assert ledger[0][2] == 30 # delta
|
|
|
|
async def test_noop_when_no_monthly_credits(self, db):
|
|
"""Supplier with monthly_credits=0 gets no refill."""
|
|
now = datetime.utcnow().isoformat()
|
|
async with db.execute(
|
|
"""INSERT INTO suppliers
|
|
(name, slug, country_code, region, category, tier,
|
|
credit_balance, monthly_credits, created_at)
|
|
VALUES ('Free Supplier', 'free-supplier', 'DE', 'Europe', 'Courts',
|
|
'free', 50, 0, ?)""",
|
|
(now,),
|
|
) as cursor:
|
|
free_id = cursor.lastrowid
|
|
await db.commit()
|
|
|
|
result = await monthly_credit_refill(free_id)
|
|
assert result == 0
|
|
assert await get_balance(free_id) == 50
|
|
|
|
|
|
# ── GetLedger ─────────────────────────────────────────────────
|
|
|
|
|
|
class TestGetLedger:
|
|
async def test_returns_entries_in_descending_order(self, db, supplier):
|
|
await add_credits(supplier["id"], 10, "pack_purchase", note="first")
|
|
await add_credits(supplier["id"], 20, "pack_purchase", note="second")
|
|
await add_credits(supplier["id"], 30, "pack_purchase", note="third")
|
|
|
|
entries = await get_ledger(supplier["id"])
|
|
assert len(entries) == 3
|
|
# Most recent first
|
|
assert entries[0]["note"] == "third"
|
|
assert entries[1]["note"] == "second"
|
|
assert entries[2]["note"] == "first"
|
|
|
|
async def test_respects_limit(self, db, supplier):
|
|
for i in range(5):
|
|
await add_credits(supplier["id"], 1, "pack_purchase", note=f"entry_{i}")
|
|
|
|
entries = await get_ledger(supplier["id"], limit=2)
|
|
assert len(entries) == 2
|
|
|
|
async def test_empty_for_unknown_supplier(self, db):
|
|
entries = await get_ledger(99999)
|
|
assert entries == []
|