refactor: flatten padelnomics/padelnomics/ → repo root
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>
This commit is contained in:
Deeman
292
web/tests/test_credits.py
Normal file
292
web/tests/test_credits.py
Normal file
@@ -0,0 +1,292 @@
|
||||
"""
|
||||
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 == []
|
||||
Reference in New Issue
Block a user