feat: unlocks, inventory items, sigils, drops, and essence (Workstream C)
Implements the backend REST surface and WS wiring for
docs/superpowers/specs/2026-07-23-character-depth-ghost-log-design.md's
Workstream C:
- New models: UnlockRecord (unlocks), InventoryItem (inventory_items),
Sigil (sigils) — brand-new tables, picked up by main.py's existing
create_all.
- New app/inventory.py: unlock price table, item drop table/odds,
essence economy constants, sigil design validation, and an atomic
(row-locked) purchase_unlock() that guards against double-spend races.
- New app/routes/inventory.py: GET unlocks/items/sigils, POST sigils
(validates the placeholder {points, rune} shape, points capped at 12),
POST unlocks/{unlock_key} (402 on insufficient essence, 404 on unknown
key, idempotent re-buy).
- GET /auth/me now includes unlocks: list[str] and essence: int.
- ws.py: wires essence trickle + item_drop rolls into the one trigger
point that exists in this worktree today (_handle_summon, covering
every successful summon plus high-rarity summons); the other two
contract trigger points (correct judgment, successful ritual) belong
to Workstream B's not-yet-landed ritual/judgment WS handlers, which
should call app.inventory's same helpers once they land.
- User.essence: int added (Workstream B owns this column per the spec;
added here per orchestrator instruction so this workstream is
independently testable — merge controller reconciles the duplicate
edit).
Also fast-forwarded this worktree's branch onto master (it had fallen
behind several commits) so the files this workstream depends on
(shop.py, ws.py, entities.py, etc.) were actually present to build
against.
Tests: 109 passed (drop-roll statistical sanity with seeded RNG,
inventory/sigil CRUD, purchase success/insufficient-funds/idempotency/
unknown-key paths, /auth/me shape, ws summon-trickle and item-drop
wiring).
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Indiana
352
backend/tests/test_inventory.py
Normal file
352
backend/tests/test_inventory.py
Normal file
@@ -0,0 +1,352 @@
|
||||
import random
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.inventory import (
|
||||
DROP_CHANCES,
|
||||
HIGH_RARITY_TIERS,
|
||||
ITEM_POOLS,
|
||||
SIGIL_MAX_POINTS,
|
||||
UNLOCK_PRICES,
|
||||
InsufficientEssenceError,
|
||||
UnknownUnlockError,
|
||||
credit_essence,
|
||||
purchase_unlock,
|
||||
roll_item_drop,
|
||||
summon_drop_trigger,
|
||||
validate_sigil_design,
|
||||
)
|
||||
from app.models.unlock import UnlockRecord
|
||||
from app.models.user import User
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Drop-roll probability sanity — seeded RNG, so this is a deterministic
|
||||
# statistical check, not a flaky one.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_roll_item_drop_statistical_sanity_summon_mythic():
|
||||
rng = random.Random(1234)
|
||||
trials = 20_000
|
||||
hits = sum(1 for _ in range(trials) if roll_item_drop("summon_mythic", rng) is not None)
|
||||
empirical = hits / trials
|
||||
expected = DROP_CHANCES["summon_mythic"]
|
||||
assert abs(empirical - expected) < 0.02
|
||||
|
||||
|
||||
def test_roll_item_drop_statistical_sanity_judgment():
|
||||
rng = random.Random(5678)
|
||||
trials = 20_000
|
||||
hits = sum(1 for _ in range(trials) if roll_item_drop("judgment", rng) is not None)
|
||||
empirical = hits / trials
|
||||
expected = DROP_CHANCES["judgment"]
|
||||
assert abs(empirical - expected) < 0.02
|
||||
|
||||
|
||||
def test_roll_item_drop_is_deterministic_given_same_seed():
|
||||
results_a = [roll_item_drop("ritual", random.Random(42)) for _ in range(50)]
|
||||
results_b = [roll_item_drop("ritual", random.Random(42)) for _ in range(50)]
|
||||
assert results_a == results_b
|
||||
|
||||
|
||||
def test_roll_item_drop_unknown_trigger_returns_none():
|
||||
assert roll_item_drop("not_a_real_trigger", random.Random(1)) is None
|
||||
|
||||
|
||||
def test_roll_item_drop_shape_matches_item_pool():
|
||||
# A high roll chance with a fixed seed known to hit, to check the shape.
|
||||
rng = random.Random(1234)
|
||||
item = None
|
||||
for _ in range(200):
|
||||
item = roll_item_drop("ritual", rng)
|
||||
if item is not None:
|
||||
break
|
||||
assert item is not None
|
||||
item_type, keys = ITEM_POOLS["ritual"]
|
||||
assert item["item_type"] == item_type
|
||||
assert item["item_key"] in keys
|
||||
assert item["payload"] == {"trigger": "ritual"}
|
||||
|
||||
|
||||
def test_summon_drop_trigger_only_fires_for_high_rarity():
|
||||
assert summon_drop_trigger("common") is None
|
||||
assert summon_drop_trigger("uncommon") is None
|
||||
assert summon_drop_trigger("rare") == "summon_rare"
|
||||
assert summon_drop_trigger("mythic") == "summon_mythic"
|
||||
assert HIGH_RARITY_TIERS == {"rare", "mythic"}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# credit_essence
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_credit_essence_adds_and_floors_at_zero():
|
||||
user = User(username="x", password_hash="x")
|
||||
user.essence = 5
|
||||
assert credit_essence(user, 3) == 8
|
||||
assert credit_essence(user, -100) == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Sigil design validation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_validate_sigil_design_accepts_valid_shape():
|
||||
design = {"points": [[0, 0], [1.5, 2.5], [3, -3]], "rune": "eld"}
|
||||
normalized = validate_sigil_design(design)
|
||||
assert normalized == {"points": [[0.0, 0.0], [1.5, 2.5], [3.0, -3.0]], "rune": "eld"}
|
||||
|
||||
|
||||
def test_validate_sigil_design_rejects_too_many_points():
|
||||
design = {"points": [[i, i] for i in range(SIGIL_MAX_POINTS + 1)], "rune": "eld"}
|
||||
assert validate_sigil_design(design) is None
|
||||
|
||||
|
||||
def test_validate_sigil_design_accepts_max_points():
|
||||
design = {"points": [[i, i] for i in range(SIGIL_MAX_POINTS)], "rune": "eld"}
|
||||
assert validate_sigil_design(design) is not None
|
||||
|
||||
|
||||
def test_validate_sigil_design_rejects_empty_points():
|
||||
assert validate_sigil_design({"points": [], "rune": "eld"}) is None
|
||||
|
||||
|
||||
def test_validate_sigil_design_rejects_bad_point_shape():
|
||||
assert validate_sigil_design({"points": [[1, 2, 3]], "rune": "eld"}) is None
|
||||
assert validate_sigil_design({"points": [["a", "b"]], "rune": "eld"}) is None
|
||||
assert validate_sigil_design({"points": "not-a-list", "rune": "eld"}) is None
|
||||
|
||||
|
||||
def test_validate_sigil_design_rejects_missing_or_bad_rune():
|
||||
assert validate_sigil_design({"points": [[0, 0]]}) is None
|
||||
assert validate_sigil_design({"points": [[0, 0]], "rune": ""}) is None
|
||||
assert validate_sigil_design({"points": [[0, 0]], "rune": 5}) is None
|
||||
assert validate_sigil_design({"points": [[0, 0]], "rune": "x" * 33}) is None
|
||||
|
||||
|
||||
def test_validate_sigil_design_rejects_non_dict():
|
||||
assert validate_sigil_design("nope") is None
|
||||
assert validate_sigil_design(None) is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# purchase_unlock — success, insufficient funds, unknown unlock, idempotency,
|
||||
# race safety.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_purchase_unlock_success_deducts_essence_and_records(db_session):
|
||||
user = User(username="buyer", password_hash="x", essence=100)
|
||||
db_session.add(user)
|
||||
await db_session.commit()
|
||||
await db_session.refresh(user)
|
||||
|
||||
price = UNLOCK_PRICES["listening_tool"]
|
||||
record = await purchase_unlock(db_session, user.id, "listening_tool")
|
||||
assert record.unlock_key == "listening_tool"
|
||||
|
||||
await db_session.refresh(user)
|
||||
assert user.essence == 100 - price
|
||||
|
||||
rows = (
|
||||
await db_session.execute(
|
||||
select(UnlockRecord).where(UnlockRecord.user_id == user.id)
|
||||
)
|
||||
).scalars().all()
|
||||
assert len(rows) == 1
|
||||
assert rows[0].unlock_key == "listening_tool"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_purchase_unlock_insufficient_funds_raises_and_does_not_charge(db_session):
|
||||
user = User(username="pauper", password_hash="x", essence=5)
|
||||
db_session.add(user)
|
||||
await db_session.commit()
|
||||
await db_session.refresh(user)
|
||||
|
||||
with pytest.raises(InsufficientEssenceError):
|
||||
await purchase_unlock(db_session, user.id, "listening_tool")
|
||||
|
||||
await db_session.refresh(user)
|
||||
assert user.essence == 5
|
||||
rows = (
|
||||
await db_session.execute(
|
||||
select(UnlockRecord).where(UnlockRecord.user_id == user.id)
|
||||
)
|
||||
).scalars().all()
|
||||
assert rows == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_purchase_unlock_unknown_key_raises(db_session):
|
||||
user = User(username="curious", password_hash="x", essence=1000)
|
||||
db_session.add(user)
|
||||
await db_session.commit()
|
||||
await db_session.refresh(user)
|
||||
|
||||
with pytest.raises(UnknownUnlockError):
|
||||
await purchase_unlock(db_session, user.id, "does_not_exist")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_purchase_unlock_is_idempotent_does_not_double_charge(db_session):
|
||||
user = User(username="rebuyer", password_hash="x", essence=100)
|
||||
db_session.add(user)
|
||||
await db_session.commit()
|
||||
await db_session.refresh(user)
|
||||
|
||||
price = UNLOCK_PRICES["listening_tool"]
|
||||
await purchase_unlock(db_session, user.id, "listening_tool")
|
||||
await db_session.refresh(user)
|
||||
assert user.essence == 100 - price
|
||||
|
||||
# Buying again must not charge a second time.
|
||||
await purchase_unlock(db_session, user.id, "listening_tool")
|
||||
await db_session.refresh(user)
|
||||
assert user.essence == 100 - price
|
||||
|
||||
rows = (
|
||||
await db_session.execute(
|
||||
select(UnlockRecord).where(UnlockRecord.user_id == user.id)
|
||||
)
|
||||
).scalars().all()
|
||||
assert len(rows) == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# HTTP layer: purchase endpoint, list endpoints, sigil save, /auth/me shape.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def _register_and_login(client, username="seeker"):
|
||||
await client.post(
|
||||
"/auth/register", json={"username": username, "password": "spookyspooky"}
|
||||
)
|
||||
resp = await client.post(
|
||||
"/auth/login", json={"username": username, "password": "spookyspooky"}
|
||||
)
|
||||
return resp.json()["id"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_buy_unlock_success_over_http(client, db_session):
|
||||
user_id = await _register_and_login(client, "affluent")
|
||||
user = await db_session.get(User, uuid.UUID(user_id))
|
||||
user.essence = 100
|
||||
await db_session.commit()
|
||||
|
||||
resp = await client.post("/api/inventory/unlocks/listening_tool")
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["unlock_key"] == "listening_tool"
|
||||
assert body["essence"] == 100 - UNLOCK_PRICES["listening_tool"]
|
||||
|
||||
me = await client.get("/auth/me")
|
||||
assert me.json()["unlocks"] == ["listening_tool"]
|
||||
assert me.json()["essence"] == 100 - UNLOCK_PRICES["listening_tool"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_buy_unlock_insufficient_funds_over_http(client, db_session):
|
||||
await _register_and_login(client, "broke")
|
||||
# A fresh user starts with 0 essence.
|
||||
resp = await client.post("/api/inventory/unlocks/listening_tool")
|
||||
assert resp.status_code == 402
|
||||
|
||||
me = await client.get("/auth/me")
|
||||
assert me.json()["unlocks"] == []
|
||||
assert me.json()["essence"] == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_buy_unlock_unknown_key_over_http(client):
|
||||
await _register_and_login(client, "explorer")
|
||||
resp = await client.post("/api/inventory/unlocks/nonexistent_thing")
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_buy_unlock_requires_auth(client):
|
||||
resp = await client.post("/api/inventory/unlocks/listening_tool")
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_inventory_list_endpoints_require_auth(client):
|
||||
assert (await client.get("/api/inventory/unlocks")).status_code == 401
|
||||
assert (await client.get("/api/inventory/items")).status_code == 401
|
||||
assert (await client.get("/api/inventory/sigils")).status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_unlocks_and_items_empty_then_populated(client, db_session):
|
||||
user_id = await _register_and_login(client, "collector")
|
||||
|
||||
empty = await client.get("/api/inventory/unlocks")
|
||||
assert empty.status_code == 200
|
||||
assert empty.json() == []
|
||||
|
||||
user = await db_session.get(User, uuid.UUID(user_id))
|
||||
user.essence = 100
|
||||
await db_session.commit()
|
||||
await client.post("/api/inventory/unlocks/listening_tool")
|
||||
|
||||
populated = await client.get("/api/inventory/unlocks")
|
||||
assert len(populated.json()) == 1
|
||||
assert populated.json()[0]["unlock_key"] == "listening_tool"
|
||||
assert "unlocked_at" in populated.json()[0]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_save_and_list_sigil(client):
|
||||
await _register_and_login(client, "sigilsmith")
|
||||
|
||||
resp = await client.post(
|
||||
"/api/inventory/sigils",
|
||||
json={"name": "ward-of-quiet", "design": {"points": [[0, 0], [1, 1]], "rune": "eld"}},
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
body = resp.json()
|
||||
assert body["name"] == "ward-of-quiet"
|
||||
assert body["design"]["rune"] == "eld"
|
||||
|
||||
listed = await client.get("/api/inventory/sigils")
|
||||
assert len(listed.json()) == 1
|
||||
assert listed.json()[0]["name"] == "ward-of-quiet"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_save_sigil_rejects_too_many_points(client):
|
||||
await _register_and_login(client, "greedy-sigilsmith")
|
||||
resp = await client.post(
|
||||
"/api/inventory/sigils",
|
||||
json={
|
||||
"name": "overreach",
|
||||
"design": {"points": [[i, i] for i in range(13)], "rune": "eld"},
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_save_sigil_rejects_missing_rune(client):
|
||||
await _register_and_login(client, "runeless")
|
||||
resp = await client.post(
|
||||
"/api/inventory/sigils",
|
||||
json={"name": "blank", "design": {"points": [[0, 0]]}},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_auth_me_includes_essence_and_unlocks_default(client):
|
||||
await _register_and_login(client, "freshuser")
|
||||
resp = await client.get("/auth/me")
|
||||
body = resp.json()
|
||||
assert body["essence"] == 0
|
||||
assert body["unlocks"] == []
|
||||
Reference in New Issue
Block a user