Vyryi/vyryi-back
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add spin session functionality and update game mechanics

Browse Source 
- Introduced a new `spin_sessions` table to track user spin activities.
- Updated the API to include a new endpoint for spinning games, allowing users to place bets and specify spin counts.
- Enhanced game information structure and added validation for bets and spin counts.
- Implemented the `Forest Fortune` slot game mechanics, including payline evaluation and scatter functionality.
- Updated dependencies in `pyproject.toml` for development and added `httpx` for HTTP requests.
- Added unit tests for the new slot mechanics and API endpoints to ensure functionality and reliability.
This commit is contained in:
Redsandy
2026-08-12 15:27:24 +03:00
parent c9f66f330e
commit 5125b52583
18 changed files with 1023 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@@ -28,6 +28,7 @@ API docs: http://127.0.0.1:8000/docs
- `GET /me`, `POST /me/agreement`
- `GET /wallet`, `POST /wallet/deposit` (amounts: 100 / 500 / 1000)
- `GET /games`, `GET /games/{slug}`
- `POST /games/{slug}/spin` — bet + spin_count (1…1000); Forest Fortune is playable
No withdrawal endpoints by design.

40
alembic/versions/0002_spin_sessions.py Normal file
View File

@@ -0,0 +1,40 @@
"""Add spin_sessions table."""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
revision: str = "0002_spin_sessions"
down_revision: Union[str, None] = "0001_initial"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
"spin_sessions",
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
sa.Column("user_id", sa.Integer(), nullable=False),
sa.Column("slug", sa.String(length=64), nullable=False),
sa.Column("bet", sa.Numeric(18, 2), nullable=False),
sa.Column("spin_count", sa.Integer(), nullable=False),
sa.Column("total_win", sa.Numeric(18, 2), nullable=False),
sa.Column("payload_json", sa.Text(), nullable=False),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
server_default=sa.text("CURRENT_TIMESTAMP"),
nullable=False,
),
sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="CASCADE"),
sa.PrimaryKeyConstraint("id"),
)
op.create_index("ix_spin_sessions_user_id", "spin_sessions", ["user_id"], unique=False)
op.create_index("ix_spin_sessions_slug", "spin_sessions", ["slug"], unique=False)
def downgrade() -> None:
op.drop_index("ix_spin_sessions_slug", table_name="spin_sessions")
op.drop_index("ix_spin_sessions_user_id", table_name="spin_sessions")
op.drop_table("spin_sessions")

140
app/api/routes.py
View File

@@ -1,3 +1,4 @@
import json
from datetime import datetime, timezone
from decimal import Decimal
@@ -8,41 +9,28 @@ from sqlalchemy.orm import Session
from app.core.database import get_db
from app.core.security import create_access_token, decode_access_token
from app.models import Deposit, User
from app.models import Deposit, SpinSession, User
from app.schemas import (
DepositRequest,
DepositResponse,
DevLoginRequest,
GameStub,
GameInfo,
PaylineWinResponse,
SessionSpinResponse,
SpinOutcomeResponse,
SpinRequest,
TelegramAuthRequest,
TokenResponse,
UserResponse,
WalletResponse,
)
from app.slots.registry import get_slot, list_slots
router = APIRouter()
security = HTTPBearer()
ALLOWED_DEPOSIT_AMOUNTS = {Decimal("100"), Decimal("500"), Decimal("1000")}
GAMES: list[GameStub] = [
GameStub(
slug="raven-reels",
title="Вороньи Барабаны",
description="Слоты у врат Вырия. Скоро откроются.",
),
GameStub(
slug="golden-gate",
title="Золотые Врата",
description="Портал удачи. Пока запечатан.",
),
GameStub(
slug="forest-fortune",
title="Лесная Удача",
description="Древние символы ещё не проснулись.",
),
]
def user_to_response(user: User) -> UserResponse:
return UserResponse(
@@ -71,6 +59,11 @@ def get_current_user(
return user
def slot_to_info(slot) -> GameInfo:
data = slot.info()
return GameInfo(**data)
@router.post("/auth/dev-login", response_model=TokenResponse)
def dev_login(payload: DevLoginRequest, db: Session = Depends(get_db)) -> TokenResponse:
user = db.execute(select(User).where(User.telegram_id == payload.telegram_id)).scalar_one_or_none()
@@ -176,14 +169,105 @@ def deposit_stub(
)
@router.get("/games", response_model=list[GameStub])
def list_games() -> list[GameStub]:
return GAMES
@router.get("/games", response_model=list[GameInfo])
def list_games() -> list[GameInfo]:
return [slot_to_info(slot) for slot in list_slots()]
@router.get("/games/{slug}", response_model=GameStub)
def get_game(slug: str) -> GameStub:
for game in GAMES:
if game.slug == slug:
return game
@router.get("/games/{slug}", response_model=GameInfo)
def get_game(slug: str) -> GameInfo:
slot = get_slot(slug)
if slot is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Game not found")
return slot_to_info(slot)
@router.post("/games/{slug}/spin", response_model=SessionSpinResponse)
def spin_game(
slug: str,
payload: SpinRequest,
user: User = Depends(get_current_user),
db: Session = Depends(get_db),
) -> SessionSpinResponse:
if user.agreement_accepted_at is None:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Accept user agreement first",
)
slot = get_slot(slug)
if slot is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Game not found")
if slot.status != "available":
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="Game is not available yet",
)
bet = payload.bet.quantize(Decimal("0.01"))
spin_count = payload.spin_count
try:
slot.validate_bet(bet)
slot.validate_spin_count(spin_count)
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
total_bet = (bet * spin_count).quantize(Decimal("0.01"))
balance = user.balance or Decimal("0.00")
if balance < total_bet:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Insufficient balance")
user.balance = balance - total_bet
try:
session_result = slot.play_session(bet, spin_count)
except Exception:
db.rollback()
raise
user.balance = (user.balance or Decimal("0.00")) + session_result.total_win
record = SpinSession(
user_id=user.id,
slug=slug,
bet=bet,
spin_count=spin_count,
total_win=session_result.total_win,
payload_json=json.dumps(session_result.to_payload(), ensure_ascii=False),
)
db.add(record)
db.commit()
db.refresh(user)
return SessionSpinResponse(
slug=slug,
bet=session_result.bet,
spin_count=session_result.spin_count,
total_bet=session_result.total_bet,
total_win=session_result.total_win,
max_level=session_result.max_level,
trigger_count=session_result.trigger_count,
balance=user.balance,
spins=[
SpinOutcomeResponse(
index=s.index,
grid=s.grid,
line_wins=[
PaylineWinResponse(
line_index=w.line_index,
symbol=w.symbol,
count=w.count,
amount=w.amount,
path=w.path,
)
for w in s.line_wins
],
scatter_count=s.scatter_count,
scatter_win=s.scatter_win,
total_win=s.total_win,
level=s.level,
multiplier=s.multiplier,
meta_triggered=s.meta_triggered,
)
for s in session_result.spins
],
)

2
app/main.py
View File

@@ -6,7 +6,7 @@ from fastapi.middleware.cors import CORSMiddleware
from app.api.routes import router
from app.core.config import get_settings
from app.core.database import Base, engine
from app.models import Deposit, User # noqa: F401
from app.models import Deposit, SpinSession, User # noqa: F401
@asynccontextmanager

20
app/models/__init__.py
View File

@@ -1,7 +1,7 @@
from datetime import datetime
from decimal import Decimal
from sqlalchemy import DateTime, ForeignKey, Numeric, String, BigInteger, func
from sqlalchemy import DateTime, ForeignKey, Numeric, String, BigInteger, Text, Integer, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.core.database import Base
@@ -25,6 +25,9 @@ class User(Base):
)
deposits: Mapped[list["Deposit"]] = relationship(back_populates="user", cascade="all, delete-orphan")
spin_sessions: Mapped[list["SpinSession"]] = relationship(
back_populates="user", cascade="all, delete-orphan"
)
class Deposit(Base):
@@ -36,3 +39,18 @@ class Deposit(Base):
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), nullable=False)
user: Mapped[User] = relationship(back_populates="deposits")
class SpinSession(Base):
__tablename__ = "spin_sessions"
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True, nullable=False)
slug: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
bet: Mapped[Decimal] = mapped_column(Numeric(18, 2), nullable=False)
spin_count: Mapped[int] = mapped_column(Integer, nullable=False)
total_win: Mapped[Decimal] = mapped_column(Numeric(18, 2), nullable=False)
payload_json: Mapped[str] = mapped_column(Text, nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), nullable=False)
user: Mapped[User] = relationship(back_populates="spin_sessions")

47
app/schemas/__init__.py
View File

@@ -49,8 +49,53 @@ class WalletResponse(BaseModel):
deposits: list[DepositResponse]
class GameStub(BaseModel):
class GameInfo(BaseModel):
slug: str
title: str
description: str
status: str = "coming_soon"
min_bet: Decimal = Decimal("1.00")
max_bet: Decimal = Decimal("1000.00")
min_spin_count: int = 1
max_spin_count: int = 1000
# Backwards-compatible alias
GameStub = GameInfo
class SpinRequest(BaseModel):
bet: Decimal = Field(..., gt=0)
spin_count: int = Field(..., ge=1, le=1000)
class PaylineWinResponse(BaseModel):
line_index: int
symbol: str
count: int
amount: Decimal
path: list[int]
class SpinOutcomeResponse(BaseModel):
index: int
grid: list[list[str]]
line_wins: list[PaylineWinResponse]
scatter_count: int
scatter_win: Decimal
total_win: Decimal
level: int
multiplier: int
meta_triggered: bool
class SessionSpinResponse(BaseModel):
slug: str
bet: Decimal
spin_count: int
total_bet: Decimal
total_win: Decimal
max_level: int
trigger_count: int
balance: Decimal
spins: list[SpinOutcomeResponse]

5
app/slots/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
from app.slots.base import BaseSlot
from app.slots.forest_fortune import ForestFortuneSlot
from app.slots.registry import get_slot, list_slots
__all__ = ["BaseSlot", "ForestFortuneSlot", "get_slot", "list_slots"]

117
app/slots/base.py Normal file
View File

@@ -0,0 +1,117 @@
from __future__ import annotations
from abc import ABC, abstractmethod
from decimal import Decimal
from app.slots.rng import Rng, system_rng
from app.slots.types import Grid, SessionMetaState, SessionResult, SpinOutcome, SpinResult
class BaseSlot(ABC):
slug: str
title: str
description: str
status: str = "coming_soon"
min_bet: Decimal = Decimal("1.00")
max_bet: Decimal = Decimal("1000.00")
min_spin_count: int = 1
max_spin_count: int = 1000
@abstractmethod
def build_grid(self, rng: Rng) -> Grid:
raise NotImplementedError
@abstractmethod
def evaluate(self, grid: Grid, bet: Decimal, multiplier: int) -> SpinOutcome:
raise NotImplementedError
@abstractmethod
def session_meta_on_spin(self, state: SessionMetaState, outcome: SpinOutcome) -> SessionMetaState:
raise NotImplementedError
def validate_bet(self, bet: Decimal) -> None:
if bet < self.min_bet or bet > self.max_bet:
raise ValueError(f"Bet must be between {self.min_bet} and {self.max_bet}")
def validate_spin_count(self, spin_count: int) -> None:
if spin_count < self.min_spin_count or spin_count > self.max_spin_count:
raise ValueError(
f"spin_count must be between {self.min_spin_count} and {self.max_spin_count}"
)
def play_session(
self,
bet: Decimal,
spin_count: int,
rng: Rng | None = None,
) -> SessionResult:
if self.status != "available":
raise RuntimeError(f"Game '{self.slug}' is not available")
bet = bet.quantize(Decimal("0.01"))
self.validate_bet(bet)
self.validate_spin_count(spin_count)
rng = rng or system_rng()
meta = SessionMetaState()
spins: list[SpinResult] = []
total_win = Decimal("0.00")
trigger_count = 0
max_level = 1
meta_enabled = spin_count > 1
for index in range(spin_count):
level = meta.level
multiplier = meta.multiplier
grid = self.build_grid(rng)
outcome = self.evaluate(grid, bet, multiplier)
if meta_enabled and outcome.meta_triggered:
next_meta = self.session_meta_on_spin(meta, outcome)
if next_meta.triggered:
trigger_count += 1
meta = SessionMetaState(
level=next_meta.level,
multiplier=next_meta.multiplier,
triggered=False,
)
max_level = max(max_level, meta.level)
triggered = True
else:
triggered = False
spin = SpinResult(
index=index,
grid=outcome.grid,
line_wins=outcome.line_wins,
scatter_count=outcome.scatter_count,
scatter_win=outcome.scatter_win,
total_win=outcome.total_win,
level=level,
multiplier=multiplier,
meta_triggered=triggered,
)
spins.append(spin)
total_win += outcome.total_win
return SessionResult(
bet=bet,
spin_count=spin_count,
total_bet=(bet * spin_count).quantize(Decimal("0.01")),
total_win=total_win.quantize(Decimal("0.01")),
max_level=max_level,
trigger_count=trigger_count,
spins=spins,
)
def info(self) -> dict:
return {
"slug": self.slug,
"title": self.title,
"description": self.description,
"status": self.status,
"min_bet": self.min_bet,
"max_bet": self.max_bet,
"min_spin_count": self.min_spin_count,
"max_spin_count": self.max_spin_count,
}

29
app/slots/coming_soon.py Normal file
View File

@@ -0,0 +1,29 @@
from __future__ import annotations
from decimal import Decimal
from app.slots.base import BaseSlot
from app.slots.rng import Rng
from app.slots.types import Grid, SessionMetaState, SpinOutcome
class ComingSoonSlot(BaseSlot):
status = "coming_soon"
min_bet = Decimal("1.00")
max_bet = Decimal("100.00")
min_spin_count = 1
max_spin_count = 1
def __init__(self, slug: str, title: str, description: str) -> None:
self.slug = slug
self.title = title
self.description = description
def build_grid(self, rng: Rng) -> Grid:
raise RuntimeError(f"Game '{self.slug}' is not available")
def evaluate(self, grid: Grid, bet: Decimal, multiplier: int) -> SpinOutcome:
raise RuntimeError(f"Game '{self.slug}' is not available")
def session_meta_on_spin(self, state: SessionMetaState, outcome: SpinOutcome) -> SessionMetaState:
return state

192
app/slots/forest_fortune.py Normal file
View File

@@ -0,0 +1,192 @@
from __future__ import annotations
from decimal import Decimal
from app.slots.base import BaseSlot
from app.slots.rng import Rng
from app.slots.types import Grid, PaylineWin, SessionMetaState, SpinOutcome
REELS = 5
ROWS = 5
LEAF = "leaf"
MUSHROOM = "mushroom"
PINE = "pine"
DEER = "deer"
FOX = "fox"
OAK = "oak"
WILD = "wild"
SCATTER = "scatter"
SYMBOLS = (LEAF, MUSHROOM, PINE, DEER, FOX, OAK, WILD, SCATTER)
# Weights per reel (index aligns with SYMBOLS)
REEL_WEIGHTS: list[list[float]] = [
[28, 24, 18, 12, 8, 5, 3, 2],
[26, 24, 18, 12, 8, 6, 4, 2],
[24, 22, 18, 14, 10, 6, 4, 2],
[26, 24, 18, 12, 8, 6, 4, 2],
[28, 24, 18, 12, 8, 5, 3, 2],
]
# Multipliers of total spin bet
PAYTABLE: dict[str, dict[int, Decimal]] = {
LEAF: {3: Decimal("0.20"), 4: Decimal("0.50"), 5: Decimal("1.50")},
MUSHROOM: {3: Decimal("0.25"), 4: Decimal("0.60"), 5: Decimal("2.00")},
PINE: {3: Decimal("0.35"), 4: Decimal("0.80"), 5: Decimal("2.50")},
DEER: {3: Decimal("0.50"), 4: Decimal("1.20"), 5: Decimal("4.00")},
FOX: {3: Decimal("0.70"), 4: Decimal("1.80"), 5: Decimal("6.00")},
OAK: {3: Decimal("1.00"), 4: Decimal("3.00"), 5: Decimal("10.00")},
WILD: {3: Decimal("1.50"), 4: Decimal("4.00"), 5: Decimal("15.00")},
}
SCATTER_TABLE: dict[int, Decimal] = {
3: Decimal("1.00"),
4: Decimal("3.00"),
5: Decimal("8.00"),
}
# Cap scatter pay at 5+ using the 5-table entry; more scatters still use 5 rate
SCATTER_TRIGGER = 3
PAYLINES: list[list[int]] = [
[0, 0, 0, 0, 0],
[1, 1, 1, 1, 1],
[2, 2, 2, 2, 2],
[3, 3, 3, 3, 3],
[4, 4, 4, 4, 4],
[0, 1, 2, 3, 4],
[4, 3, 2, 1, 0],
[0, 0, 1, 0, 0],
[4, 4, 3, 4, 4],
[1, 2, 3, 2, 1],
[3, 2, 1, 2, 3],
[0, 1, 0, 1, 0],
[4, 3, 4, 3, 4],
[2, 1, 0, 1, 2],
[2, 3, 4, 3, 2],
[0, 1, 2, 1, 0],
[4, 3, 2, 3, 4],
[1, 1, 2, 3, 3],
[3, 3, 2, 1, 1],
[0, 2, 4, 2, 0],
]
def _line_symbols(grid: Grid, path: list[int]) -> list[str]:
return [grid[reel][path[reel]] for reel in range(REELS)]
def evaluate_payline(symbols: list[str]) -> tuple[str, int] | None:
"""Return (symbol, count) for best left-to-right win, or None."""
if not symbols:
return None
# All wilds
if all(s == WILD for s in symbols):
return WILD, len(symbols)
# Find target symbol: first non-wild, non-scatter from the left
target: str | None = None
for s in symbols:
if s == SCATTER:
break
if s != WILD:
target = s
break
if target is None:
# Leading wilds then scatter/end — pay as wild streak only if length >= 3
count = 0
for s in symbols:
if s == WILD:
count += 1
else:
break
if count >= 3:
return WILD, count
return None
count = 0
for s in symbols:
if s == SCATTER:
break
if s == target or s == WILD:
count += 1
else:
break
if count >= 3:
return target, count
return None
def scatter_pay_key(count: int) -> int | None:
if count < 3:
return None
return min(count, 5)
class ForestFortuneSlot(BaseSlot):
slug = "forest-fortune"
title = "Лесная Удача"
description = "5×5 лесной слот. В пачке спинов духи поднимают множитель."
status = "available"
min_bet = Decimal("1.00")
max_bet = Decimal("1000.00")
min_spin_count = 1
max_spin_count = 1000
def build_grid(self, rng: Rng) -> Grid:
grid: Grid = []
for reel in range(REELS):
column = [
rng.choices(list(SYMBOLS), weights=REEL_WEIGHTS[reel], k=1)[0]
for _ in range(ROWS)
]
grid.append(column)
return grid
def evaluate(self, grid: Grid, bet: Decimal, multiplier: int) -> SpinOutcome:
line_wins: list[PaylineWin] = []
for line_index, path in enumerate(PAYLINES):
symbols = _line_symbols(grid, path)
result = evaluate_payline(symbols)
if result is None:
continue
symbol, count = result
rate = PAYTABLE.get(symbol, {}).get(count)
if rate is None:
continue
amount = (bet * rate * Decimal(multiplier)).quantize(Decimal("0.01"))
line_wins.append(
PaylineWin(
line_index=line_index,
symbol=symbol,
count=count,
amount=amount,
path=list(path),
)
)
scatter_count = sum(1 for reel in grid for cell in reel if cell == SCATTER)
scatter_win = Decimal("0.00")
key = scatter_pay_key(scatter_count)
if key is not None:
rate = SCATTER_TABLE[key]
scatter_win = (bet * rate * Decimal(multiplier)).quantize(Decimal("0.01"))
total = sum((w.amount for w in line_wins), Decimal("0.00")) + scatter_win
return SpinOutcome(
grid=grid,
line_wins=line_wins,
scatter_count=scatter_count,
scatter_win=scatter_win,
total_win=total.quantize(Decimal("0.01")),
meta_triggered=scatter_count >= SCATTER_TRIGGER,
)
def session_meta_on_spin(self, state: SessionMetaState, outcome: SpinOutcome) -> SessionMetaState:
if outcome.meta_triggered:
return state.after_trigger()
return SessionMetaState(level=state.level, multiplier=state.multiplier, triggered=False)

29
app/slots/registry.py Normal file
View File

@@ -0,0 +1,29 @@
from __future__ import annotations
from app.slots.base import BaseSlot
from app.slots.coming_soon import ComingSoonSlot
from app.slots.forest_fortune import ForestFortuneSlot
_SLOTS: list[BaseSlot] = [
ComingSoonSlot(
slug="raven-reels",
title="Вороньи Барабаны",
description="Слоты у врат Вырия. Скоро откроются.",
),
ComingSoonSlot(
slug="golden-gate",
title="Золотые Врата",
description="Портал удачи. Пока запечатан.",
),
ForestFortuneSlot(),
]
_BY_SLUG: dict[str, BaseSlot] = {slot.slug: slot for slot in _SLOTS}
def list_slots() -> list[BaseSlot]:
return list(_SLOTS)
def get_slot(slug: str) -> BaseSlot | None:
return _BY_SLUG.get(slug)

24
app/slots/rng.py Normal file
View File

@@ -0,0 +1,24 @@
from __future__ import annotations
import random
from typing import Any, Protocol, Sequence
class Rng(Protocol):
def random(self) -> float: ...
def choices(
self,
population: Sequence[Any],
weights: Sequence[float] | None = None,
*,
k: int = 1,
) -> list[Any]: ...
def system_rng() -> random.SystemRandom:
return random.SystemRandom()
def seeded_rng(seed: int | str | bytes | bytearray) -> random.Random:
return random.Random(seed)

96
app/slots/types.py Normal file
View File

@@ -0,0 +1,96 @@
from __future__ import annotations
from dataclasses import dataclass, field
from decimal import Decimal
from typing import Any
# grid[reel][row], reel 0..4, row 0..4 (0 = top)
Grid = list[list[str]]
@dataclass
class PaylineWin:
line_index: int
symbol: str
count: int
amount: Decimal
path: list[int]
@dataclass
class SpinOutcome:
grid: Grid
line_wins: list[PaylineWin]
scatter_count: int
scatter_win: Decimal
total_win: Decimal
meta_triggered: bool
@dataclass
class SessionMetaState:
level: int = 1
multiplier: int = 1
triggered: bool = False
def after_trigger(self) -> SessionMetaState:
level = self.level + 1
return SessionMetaState(level=level, multiplier=level, triggered=True)
@dataclass
class SpinResult:
index: int
grid: Grid
line_wins: list[PaylineWin]
scatter_count: int
scatter_win: Decimal
total_win: Decimal
level: int
multiplier: int
meta_triggered: bool
@dataclass
class SessionResult:
bet: Decimal
spin_count: int
total_bet: Decimal
total_win: Decimal
max_level: int
trigger_count: int
spins: list[SpinResult] = field(default_factory=list)
def to_payload(self) -> dict[str, Any]:
return {
"bet": str(self.bet),
"spin_count": self.spin_count,
"total_bet": str(self.total_bet),
"total_win": str(self.total_win),
"max_level": self.max_level,
"trigger_count": self.trigger_count,
"spins": [
{
"index": s.index,
"grid": s.grid,
"line_wins": [
{
"line_index": w.line_index,
"symbol": w.symbol,
"count": w.count,
"amount": str(w.amount),
"path": w.path,
}
for w in s.line_wins
],
"scatter_count": s.scatter_count,
"scatter_win": str(s.scatter_win),
"total_win": str(s.total_win),
"level": s.level,
"multiplier": s.multiplier,
"meta_triggered": s.meta_triggered,
}
for s in self.spins
],
}

5
pyproject.toml
View File

@@ -19,3 +19,8 @@ dependencies = [
[tool.uv]
package = false
[dependency-groups]
dev = [
"httpx>=0.28.1",
]

1
tests/__init__.py Normal file
View File

@@ -0,0 +1 @@
# Test package

149
tests/test_forest_fortune.py Normal file
View File

@@ -0,0 +1,149 @@
"""Unit tests for Forest Fortune slot mechanics and session meta."""
from __future__ import annotations
import unittest
from decimal import Decimal
from app.slots.forest_fortune import (
FOX,
OAK,
SCATTER,
WILD,
ForestFortuneSlot,
evaluate_payline,
)
from app.slots.rng import seeded_rng
from app.slots.types import SessionMetaState, SpinOutcome
class PaylineTests(unittest.TestCase):
def test_three_of_kind(self) -> None:
self.assertEqual(evaluate_payline([OAK, OAK, OAK, FOX, OAK]), (OAK, 3))
def test_break_stops_chain(self) -> None:
self.assertIsNone(evaluate_payline([OAK, OAK, FOX, OAK, OAK]))
def test_wild_extends(self) -> None:
self.assertEqual(evaluate_payline([OAK, WILD, OAK, FOX, FOX]), (OAK, 3))
def test_scatter_breaks_line(self) -> None:
self.assertIsNone(evaluate_payline([OAK, OAK, SCATTER, OAK, OAK]))
def test_all_wilds(self) -> None:
self.assertEqual(evaluate_payline([WILD, WILD, WILD, WILD, WILD]), (WILD, 5))
class SessionMetaTests(unittest.TestCase):
def setUp(self) -> None:
self.slot = ForestFortuneSlot()
def _forced_outcome(self, scatter_count: int, multiplier: int = 1) -> SpinOutcome:
grid = [["leaf"] * 5 for _ in range(5)]
placed = 0
for reel in range(5):
for row in range(5):
if placed < scatter_count:
grid[reel][row] = SCATTER
placed += 1
return self.slot.evaluate(grid, Decimal("10.00"), multiplier)
def test_single_spin_no_level_up_effect(self) -> None:
# Even with scatters, spin_count=1 keeps meta disabled in play_session
class ForceRng:
def __init__(self) -> None:
self.n = 0
def random(self) -> float:
return 0.5
def choices(self, population, weights=None, *, k=1):
# Force many scatters
return [SCATTER] * k
result = self.slot.play_session(Decimal("1.00"), 1, rng=ForceRng())
self.assertEqual(result.spin_count, 1)
self.assertEqual(result.max_level, 1)
self.assertEqual(result.trigger_count, 0)
self.assertFalse(result.spins[0].meta_triggered)
def test_pack_trigger_applies_to_next_spins(self) -> None:
scatters_on = {4} # 0-based index 4 => 5th spin
class ScriptedRng:
def __init__(self) -> None:
self.spin = -1
self.cell = 0
def random(self) -> float:
return 0.5
def choices(self, population, weights=None, *, k=1):
# New spin starts every 25 cells
if self.cell % 25 == 0:
self.spin += 1
self.cell += 1
if self.spin in scatters_on:
# first 3 cells of that spin are scatters, rest leaf
pos = (self.cell - 1) % 25
return [SCATTER if pos < 3 else "leaf"] * k
return ["leaf"] * k
result = self.slot.play_session(Decimal("1.00"), 10, rng=ScriptedRng())
self.assertEqual(result.trigger_count, 1)
self.assertEqual(result.max_level, 2)
# spins 0..4 at level 1; spin 4 triggers; spins 5..9 at level 2
self.assertEqual(result.spins[4].level, 1)
self.assertTrue(result.spins[4].meta_triggered)
self.assertEqual(result.spins[5].level, 2)
self.assertEqual(result.spins[5].multiplier, 2)
self.assertEqual(result.spins[9].multiplier, 2)
def test_second_trigger_triples(self) -> None:
scatters_on = {2, 5}
class ScriptedRng:
def __init__(self) -> None:
self.spin = -1
self.cell = 0
def random(self) -> float:
return 0.5
def choices(self, population, weights=None, *, k=1):
if self.cell % 25 == 0:
self.spin += 1
self.cell += 1
if self.spin in scatters_on:
pos = (self.cell - 1) % 25
return [SCATTER if pos < 3 else "leaf"] * k
return ["leaf"] * k
result = self.slot.play_session(Decimal("1.00"), 8, rng=ScriptedRng())
self.assertEqual(result.trigger_count, 2)
self.assertEqual(result.max_level, 3)
self.assertEqual(result.spins[3].multiplier, 2)
self.assertEqual(result.spins[6].multiplier, 3)
def test_evaluate_scatter_pay_and_trigger_flag(self) -> None:
outcome = self._forced_outcome(3, multiplier=2)
self.assertTrue(outcome.meta_triggered)
self.assertEqual(outcome.scatter_count, 3)
self.assertEqual(outcome.scatter_win, Decimal("20.00")) # 10 * 1.00 * 2
def test_meta_state_after_trigger(self) -> None:
state = SessionMetaState()
outcome = self._forced_outcome(3)
nxt = self.slot.session_meta_on_spin(state, outcome)
self.assertEqual(nxt.level, 2)
self.assertEqual(nxt.multiplier, 2)
self.assertTrue(nxt.triggered)
def test_seeded_session_runs(self) -> None:
result = self.slot.play_session(Decimal("5.00"), 5, rng=seeded_rng(42))
self.assertEqual(len(result.spins), 5)
self.assertEqual(result.total_bet, Decimal("25.00"))
if __name__ == "__main__":
unittest.main()

111
tests/test_spin_api.py Normal file
View File

@@ -0,0 +1,111 @@
"""API tests for spin endpoint and wallet debit/credit."""
from __future__ import annotations
import unittest
from decimal import Decimal
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from app.core.database import Base, get_db
from app.main import create_app
class SpinApiTests(unittest.TestCase):
def setUp(self) -> None:
engine = create_engine(
"sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
TestingSession = sessionmaker(bind=engine, autocommit=False, autoflush=False)
Base.metadata.create_all(bind=engine)
def override_get_db():
db = TestingSession()
try:
yield db
finally:
db.close()
self.app = create_app()
self.app.dependency_overrides[get_db] = override_get_db
self.client = TestClient(self.app)
self.TestingSession = TestingSession
login = self.client.post(
"/auth/dev-login",
json={"telegram_id": 424242, "username": "tester", "first_name": "Test"},
)
self.assertEqual(login.status_code, 200)
token = login.json()["access_token"]
self.headers = {"Authorization": f"Bearer {token}"}
agree = self.client.post("/me/agreement", headers=self.headers)
self.assertEqual(agree.status_code, 200)
deposit = self.client.post("/wallet/deposit", headers=self.headers, json={"amount": 1000})
self.assertEqual(deposit.status_code, 200)
self.assertEqual(Decimal(deposit.json()["balance"]), Decimal("1000.00"))
def test_list_games_includes_forest(self) -> None:
res = self.client.get("/games")
self.assertEqual(res.status_code, 200)
slugs = {g["slug"]: g for g in res.json()}
self.assertIn("forest-fortune", slugs)
self.assertEqual(slugs["forest-fortune"]["status"], "available")
self.assertEqual(slugs["forest-fortune"]["max_spin_count"], 1000)
def test_spin_debit_and_credit(self) -> None:
res = self.client.post(
"/games/forest-fortune/spin",
headers=self.headers,
json={"bet": 10, "spin_count": 3},
)
self.assertEqual(res.status_code, 200, res.text)
body = res.json()
self.assertEqual(body["spin_count"], 3)
self.assertEqual(Decimal(body["total_bet"]), Decimal("30.00"))
self.assertEqual(len(body["spins"]), 3)
expected = Decimal("1000.00") - Decimal(body["total_bet"]) + Decimal(body["total_win"])
self.assertEqual(Decimal(body["balance"]), expected)
def test_reject_spin_count_over_limit(self) -> None:
res = self.client.post(
"/games/forest-fortune/spin",
headers=self.headers,
json={"bet": 1, "spin_count": 1001},
)
self.assertEqual(res.status_code, 422)
def test_reject_spin_count_zero_via_validation(self) -> None:
res = self.client.post(
"/games/forest-fortune/spin",
headers=self.headers,
json={"bet": 1, "spin_count": 0},
)
self.assertEqual(res.status_code, 422)
def test_insufficient_balance(self) -> None:
res = self.client.post(
"/games/forest-fortune/spin",
headers=self.headers,
json={"bet": 500, "spin_count": 10},
)
self.assertEqual(res.status_code, 400)
self.assertIn("Insufficient", res.json()["detail"])
def test_coming_soon_unavailable(self) -> None:
res = self.client.post(
"/games/raven-reels/spin",
headers=self.headers,
json={"bet": 1, "spin_count": 1},
)
self.assertEqual(res.status_code, 503)
if __name__ == "__main__":
unittest.main()

45
uv.lock generated
View File

@@ -38,6 +38,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/da/35/f2287558c17e29fafc8ef3daf819bb9834061cfa43bff8014f7df7f63bdc/anyio-4.14.2-py3-none-any.whl", hash = "sha256:9f505dda5ac9f0c8309b5e8bd445a8c2bf7246f3ce950121e45ea15bc41d1494", size = 125813, upload-time = "2026-07-12T20:29:05.763Z" },
]
[[package]]
name = "certifi"
version = "2026.7.22"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/a3/c2/24167ea9858356b47a87a50d39908bfdb72ceeefe0041586e704e5376b3a/certifi-2026.7.22.tar.gz", hash = "sha256:741e2c3b351ddf169a738da9f2c048608ff7f2c5cc02f1ebc6b118bb090d5d55", size = 138112, upload-time = "2026-07-22T03:35:12.644Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/0b/a7/71ac2cff56fec219ed242bb11b8efb69fcc4bec75db06fb7bfe35de520e6/certifi-2026.7.22-py3-none-any.whl", hash = "sha256:62f22742b58a1a33014a2b6b706588a8d7e2a88ae7bd1a6ebe8c992928483775", size = 136983, upload-time = "2026-07-22T03:35:11.276Z" },
]
[[package]]
name = "cffi"
version = "2.1.1"
@@ -262,6 +271,19 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/04/4b/29cac41a4d98d144bf5f6d33995617b185d14b22401f75ca86f384e87ff1/h11-0.16.0-py3-none-any.whl", hash = "sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86", size = 37515, upload-time = "2025-04-24T03:35:24.344Z" },
]
[[package]]
name = "httpcore"
version = "1.0.9"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "certifi" },
{ name = "h11" },
]
sdist = { url = "https://files.pythonhosted.org/packages/06/94/82699a10bca87a5556c9c59b5963f2d039dbd239f25bc2a63907a05a14cb/httpcore-1.0.9.tar.gz", hash = "sha256:6e34463af53fd2ab5d807f399a9b45ea31c3dfa2276f15a2c3f00afff6e176e8", size = 85484, upload-time = "2025-04-24T22:06:22.219Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/7e/f5/f66802a942d491edb555dd61e3a9961140fd64c90bce1eafd741609d334d/httpcore-1.0.9-py3-none-any.whl", hash = "sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55", size = 78784, upload-time = "2025-04-24T22:06:20.566Z" },
]
[[package]]
name = "httptools"
version = "0.8.0"
@@ -298,6 +320,21 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/48/63/b906c01e53f50d432c0defe43ce52764a111dc1bdd028bafbeb54dcfd008/httptools-0.8.0-cp314-cp314t-win_amd64.whl", hash = "sha256:384c17174464c8e873398b7af24f0b1f44d992c820328413951a625323155d77", size = 108209, upload-time = "2026-05-25T22:17:39.473Z" },
]
[[package]]
name = "httpx"
version = "0.28.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "anyio" },
{ name = "certifi" },
{ name = "httpcore" },
{ name = "idna" },
]
sdist = { url = "https://files.pythonhosted.org/packages/b1/df/48c586a5fe32a0f01324ee087459e112ebb7224f646c0b5023f5e79e9956/httpx-0.28.1.tar.gz", hash = "sha256:75e98c5f16b0f35b567856f597f06ff2270a374470a5c2392242528e3e3e42fc", size = 141406, upload-time = "2024-12-06T15:37:23.222Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/2a/39/e50c7c3a983047577ee07d2a9e53faf5a69493943ec3f6a384bdc792deb2/httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad", size = 73517, upload-time = "2024-12-06T15:37:21.509Z" },
]
[[package]]
name = "idna"
version = "3.18"
@@ -724,6 +761,11 @@ dependencies = [
{ name = "uvicorn", extra = ["standard"] },
]
[package.dev-dependencies]
dev = [
{ name = "httpx" },
]
[package.metadata]
requires-dist = [
{ name = "alembic", specifier = "==1.14.0" },
@@ -738,6 +780,9 @@ requires-dist = [
{ name = "uvicorn", extras = ["standard"], specifier = "==0.34.0" },
]
[package.metadata.requires-dev]
dev = [{ name = "httpx", specifier = ">=0.28.1" }]
[[package]]
name = "watchfiles"
version = "1.2.0"