This commit is contained in:
2026-02-11 23:54:42 +01:00 Unverified
commit ace6f5a022
47 changed files with 5315 additions and 0 deletions
+216
View File
@@ -0,0 +1,216 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[codz]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py.cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
# Pipfile.lock
# UV
# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# uv.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
# poetry.lock
# poetry.toml
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
# pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python.
# https://pdm-project.org/en/latest/usage/project/#working-with-version-control
# pdm.lock
# pdm.toml
.pdm-python
.pdm-build/
# pixi
# Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control.
# pixi.lock
# Pixi creates a virtual environment in the .pixi directory, just like venv module creates one
# in the .venv directory. It is recommended not to include this directory in version control.
.pixi
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# Redis
*.rdb
*.aof
*.pid
# RabbitMQ
mnesia/
rabbitmq/
rabbitmq-data/
# ActiveMQ
activemq-data/
# SageMath parsed files
*.sage.py
# Environments
.env
.envrc
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
# .idea/
# Abstra
# Abstra is an AI-powered process automation framework.
# Ignore directories containing user credentials, local state, and settings.
# Learn more at https://abstra.io/docs
.abstra/
# Visual Studio Code
# Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore
# that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore
# and can be added to the global gitignore or merged into this file. However, if you prefer,
# you could uncomment the following to ignore the entire vscode folder
# .vscode/
# Ruff stuff:
.ruff_cache/
# PyPI configuration file
.pypirc
# Marimo
marimo/_static/
marimo/_lsp/
__marimo__/
# Streamlit
.streamlit/secrets.toml
View File
+1
View File
@@ -0,0 +1 @@
# backend/app/__init__.py
+34
View File
@@ -0,0 +1,34 @@
# backend/app/constants.py
from enum import Enum
class TournamentTypes(str, Enum):
SINGLE = "Single"
DOUBLE = "Double"
ROUND_ROBIN = "Round_Robin"
class BracketType(str, Enum):
WINNERS = "Winners"
LOSERS = "Losers"
FINALS = "Finals"
class MatchSourceType(str, Enum):
WINNER = "Winner"
LOSER = "Loser"
class MatchStatus(str, Enum):
PENDING = "Pending"
SCHEDULED = "Scheduled"
FINISHED = "Finished"
class WinnerSide(str, Enum):
P1 = "p1"
P2 = "p2"
NONE = "none"
SUCCESS = {"status": "ok"}
+1
View File
@@ -0,0 +1 @@
# backend\app\core\__init__.py
+62
View File
@@ -0,0 +1,62 @@
# backend/app/core/auth.py
from datetime import datetime, timedelta, timezone
from typing import Optional
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
import jwt
from jwt.exceptions import PyJWTError
from .config import SECRET_KEY, ALGORITHM, ADMIN_USER
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")
oauth2_scheme_optional = OAuth2PasswordBearer(tokenUrl="auth/token", auto_error=False)
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.now(timezone.utc) + expires_delta
else:
expire = datetime.now(timezone.utc) + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username = payload.get("sub")
if username is None or username != ADMIN_USER:
raise credentials_exception
except PyJWTError:
raise credentials_exception
return username
async def get_optional_user(
token: Optional[str] = Depends(oauth2_scheme_optional),
) -> Optional[str]:
if not token:
return None
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username = payload.get("sub")
if username == ADMIN_USER:
return username
except Exception:
pass
return None
+26
View File
@@ -0,0 +1,26 @@
# backend/app/core/config.py
import os
import secrets
from pwdlib import PasswordHash
DB_PATH = os.getenv("DB_PATH", "./tournaments.db")
# Security Config
SECRET_KEY = os.getenv("SECRET_KEY", secrets.token_hex(32))
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 # 24 hours
# Admin Credentials
ADMIN_USER = os.getenv("ADMIN_USER", "admin")
ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "admin")
password_hash = PasswordHash.recommended()
ADMIN_HASH = password_hash.hash(ADMIN_PASSWORD)
def verify_password(plain_password, hashed_password):
return password_hash.verify(plain_password, hashed_password)
def get_password_hash(password):
return password_hash.hash(password)
+1
View File
@@ -0,0 +1 @@
# backend\app\core\utils.py
+40
View File
@@ -0,0 +1,40 @@
# backend/app/core/websocket_manager.py
import asyncio
from fastapi import WebSocket
class ConnectionManager:
def __init__(self):
self.active_connections: set[WebSocket] = set()
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.add(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.discard(websocket)
async def broadcast(self, message: dict):
if not self.active_connections:
return
connections_snapshot = list(self.active_connections)
async def send_safe(ws: WebSocket):
try:
await ws.send_json(message)
except Exception:
self.disconnect(ws)
await asyncio.gather(*(send_safe(ws) for ws in connections_snapshot))
manager = ConnectionManager()
async def send_ws_update(id: str):
await asyncio.gather(
manager.broadcast({"type": "dashboard_update"}),
manager.broadcast({"type": "tournament_update", "id": id}),
)
+236
View File
@@ -0,0 +1,236 @@
# backend/app/crud.py
from sqlalchemy.orm import Session
from uuid import uuid4
from . import models, schemas, logic
# --- HELPER ---
def _rebuild_bracket(db: Session, t: models.Tournament):
"""
Internal helper to regenerate matches, refresh the bracket logic,
and update the schedule. Used whenever teams or type changes.
"""
db.refresh(t)
current_team_names = [team.name for team in t.teams]
match_data = logic.generate_structure(current_team_names, t.type)
t.matches = [models.Match(**m, tournament_id=t.id) for m in match_data]
# 4. Run Logic
logic.refresh_bracket(t)
logic.update_schedule(t)
# --- TOURNAMENTS ---
def get_tournaments(db: Session):
return db.query(models.Tournament).all()
def get_tournament(db: Session, tournament_id: str):
return (
db.query(models.Tournament)
.filter(models.Tournament.id == tournament_id)
.first()
)
def create_tournament(db: Session, data: schemas.TournamentCreate):
t_id = str(uuid4())[:8]
new_t = models.Tournament(
id=t_id,
name=data.name,
code=data.code,
timestamp=data.timestamp,
duration=data.duration,
type=data.type,
)
new_t.teams = [models.Team(name=n) for n in data.teams]
new_t.courts = [models.Court(name=n) for n in data.courts]
match_data = logic.generate_structure(data.teams, data.type)
new_t.matches = [models.Match(**m, tournament_id=t_id) for m in match_data]
logic.refresh_bracket(new_t)
logic.update_schedule(new_t)
db.add(new_t)
db.commit()
db.refresh(new_t)
return new_t
def delete_tournament(db: Session, tournament_id: str) -> bool:
t = get_tournament(db, tournament_id)
if not t:
return False
db.delete(t)
db.commit()
return True
def update_tournament_details(
db: Session, tournament_id: str, data: schemas.TournamentUpdate
):
t = get_tournament(db, tournament_id)
if not t:
return None
update_data = data.model_dump(exclude_unset=True)
incoming_type = update_data.get("type")
type_changed = incoming_type and incoming_type != t.type
for key, value in update_data.items():
setattr(t, key, value)
if type_changed:
_rebuild_bracket(db, t)
else:
logic.update_schedule(t)
db.commit()
db.refresh(t)
return t
def update_tournament_teams(db: Session, tournament_id: str, new_team_names: list[str]):
t = get_tournament(db, tournament_id)
if not t:
return None
current_team_names = [team.name for team in t.teams]
if new_team_names == current_team_names:
return t
t.teams = [models.Team(name=n, tournament_id=t.id) for n in new_team_names]
db.flush()
_rebuild_bracket(db, t)
db.commit()
db.refresh(t)
return t
def update_tournament_courts(
db: Session, tournament_id: str, new_court_names: list[str]
):
t = get_tournament(db, tournament_id)
if not t:
return None
current_court_names = [c.name for c in t.courts]
if set(new_court_names) == set(current_court_names):
return t
t.courts = [models.Court(name=c, tournament_id=t.id) for c in new_court_names]
logic.update_schedule(t)
db.commit()
db.refresh(t)
return t
def get_tournament_matches(db: Session, tournament_id: str):
return (
db.query(models.Match)
.filter(models.Match.tournament_id == tournament_id)
.order_by(models.Match.timestamp, models.Match.court_name)
.all()
)
def get_match(db: Session, tournament_id: str, match_id: str):
return (
db.query(models.Match)
.filter(models.Match.tournament_id == tournament_id)
.filter(models.Match.id == match_id)
.first()
)
# --- TEAMS ---
def get_teams(db: Session, tournament_id: str):
return (
db.query(models.Team).filter(models.Team.tournament_id == tournament_id).all()
)
def create_team(db: Session, tournament_id: str, team_data: schemas.TeamCreate):
t = get_tournament(db, tournament_id)
if not t:
return None
new_team = models.Team(name=team_data.name, tournament_id=tournament_id)
db.add(new_team)
db.flush()
_rebuild_bracket(db, t)
db.commit()
db.refresh(new_team)
return new_team
def delete_team(db: Session, tournament_id: str, team_id: int):
t = get_tournament(db, tournament_id)
if not t:
return None
team = db.get(models.Team, team_id)
if not team or team.tournament_id != tournament_id:
return None
db.delete(team)
db.flush()
_rebuild_bracket(db, t)
db.commit()
return True
# --- COURTS ---
def get_courts(db: Session, tournament_id: str):
return (
db.query(models.Court).filter(models.Court.tournament_id == tournament_id).all()
)
def create_court(db: Session, tournament_id: str, court_data: schemas.CourtCreate):
t = get_tournament(db, tournament_id)
if not t:
return None
new_court = models.Court(name=court_data.name, tournament_id=tournament_id)
db.add(new_court)
db.flush()
db.refresh(t)
logic.update_schedule(t)
db.commit()
db.refresh(new_court)
return new_court
def delete_court(db: Session, tournament_id: str, court_id: int):
t = get_tournament(db, tournament_id)
if not t:
return None
court = db.get(models.Court, court_id)
if not court or court.tournament_id != tournament_id:
return None
db.delete(court)
db.flush()
db.refresh(t)
logic.update_schedule(t)
db.commit()
return True
+26
View File
@@ -0,0 +1,26 @@
# backend/app/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
from .core.config import DB_PATH
SQLALCHEMY_DATABASE_URL = f"sqlite:///{DB_PATH}"
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False},
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
class Base(DeclarativeBase):
pass
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
+341
View File
@@ -0,0 +1,341 @@
# backend/app/logic.py
import math
from datetime import datetime, timedelta
from typing import List, Dict, Any
from .constants import BracketType, MatchSourceType, MatchStatus, TournamentTypes
from .models import Tournament
def get_seeded_positions(num_slots, teams):
seeds = [1, 2]
while len(seeds) < num_slots:
next_seeds = []
for s in seeds:
next_seeds.append(s)
next_seeds.append(2 * len(seeds) + 1 - s)
seeds = next_seeds
return [teams[s - 1] if s <= len(teams) else "BYE" for s in seeds]
def generate_structure(
teams: List[str], type: TournamentTypes = TournamentTypes.DOUBLE
) -> List[Dict[str, Any]]:
count = len(teams)
if count < 2:
return []
power = math.ceil(math.log2(count)) if count > 0 else 1
size = 2**power
seeded_teams = get_seeded_positions(size, teams)
class Node:
def __init__(self, id, bracket: BracketType, round_n: int):
self.id = str(id)
self.bracket = bracket
self.round = round_n
self.p1: str | None = None
self.p2: str | None = None
self.winner_next_match_id: str | None = None
self.loser_next_match_id: str | None = None
self.previous_match_p1_id: str | None = None
self.previous_match_p2_id: str | None = None
self.source_p1_type: MatchSourceType | None = None
self.source_p2_type: MatchSourceType | None = None
def to_dict(self):
return {
"id": self.id,
# Pass Enum OBJECTS, not strings. SQLAlchemy handles the rest.
"bracket": self.bracket,
"round": self.round,
"p1_name": self.p1,
"p2_name": self.p2,
"status": MatchStatus.PENDING,
"previous_match_p1_id": self.previous_match_p1_id,
"previous_match_p2_id": self.previous_match_p2_id,
"source_p1_type": self.source_p1_type,
"source_p2_type": self.source_p2_type,
"winner_next_match_id": self.winner_next_match_id,
"loser_next_match_id": self.loser_next_match_id,
}
nodes: List[Node] = []
match_counter = 1
def create_node(bracket: BracketType, round_n: int):
nonlocal match_counter
n = Node(match_counter, bracket, round_n)
match_counter += 1
nodes.append(n)
return n
# --- Winners Bracket ---
wb_rounds = power
wb_matches = {r: [] for r in range(1, wb_rounds + 1)}
for r in range(1, wb_rounds + 1):
for _ in range(size // (2**r)):
wb_matches[r].append(create_node(BracketType.WINNERS, r))
# Link Winners
for r in range(1, wb_rounds):
for i, m in enumerate(wb_matches[r]):
target = wb_matches[r + 1][i // 2]
m.winner_next_match_id = target.id
if i % 2 == 0:
target.previous_match_p1_id = m.id
target.source_p1_type = MatchSourceType.WINNER
else:
target.previous_match_p2_id = m.id
target.source_p2_type = MatchSourceType.WINNER
for i, m in enumerate(wb_matches[1]):
m.p1 = seeded_teams[i * 2]
m.p2 = seeded_teams[i * 2 + 1]
# --- Losers Bracket ---
if type == TournamentTypes.DOUBLE and size >= 4:
lb_rounds = (wb_rounds - 1) * 2
lb_matches = {r: [] for r in range(1, lb_rounds + 1)}
current_count = size // 4
for r in range(1, lb_rounds + 1):
for _ in range(current_count):
lb_matches[r].append(create_node(BracketType.LOSERS, r))
if r % 2 == 0:
current_count //= 2
# Link Losers Internal
for r in range(1, lb_rounds):
for i, m in enumerate(lb_matches[r]):
target = (
lb_matches[r + 1][i] if r % 2 != 0 else lb_matches[r + 1][i // 2]
)
m.winner_next_match_id = target.id
if r % 2 != 0:
target.previous_match_p1_id = m.id
target.source_p1_type = MatchSourceType.WINNER
else:
if i % 2 == 0:
target.previous_match_p1_id = m.id
target.source_p1_type = MatchSourceType.WINNER
else:
target.previous_match_p2_id = m.id
target.source_p2_type = MatchSourceType.WINNER
# Link Losers Drop-down
for r in range(1, wb_rounds):
drop_round = 1 if r == 1 else (r - 1) * 2
wb_layer = wb_matches[r]
lb_layer = lb_matches[drop_round]
for i, wb_m in enumerate(wb_layer):
target = (
lb_layer[i // 2]
if r == 1
else (lb_layer[i] if i < len(lb_layer) else lb_layer[-1])
)
slot = "p1" if (r == 1 and i % 2 == 0) else "p2"
wb_m.loser_next_match_id = target.id
if slot == "p1":
target.previous_match_p1_id = wb_m.id
target.source_p1_type = MatchSourceType.LOSER
else:
target.previous_match_p2_id = wb_m.id
target.source_p2_type = MatchSourceType.LOSER
# Finals Linking
wb_final = wb_matches[wb_rounds][0]
lb_final = lb_matches[lb_rounds][0]
wb_final.loser_next_match_id = lb_final.id
lb_final.previous_match_p2_id = wb_final.id
lb_final.source_p2_type = MatchSourceType.LOSER
final = create_node(BracketType.FINALS, 1)
wb_final.winner_next_match_id = final.id
lb_final.winner_next_match_id = final.id
final.previous_match_p1_id = wb_final.id
final.source_p1_type = MatchSourceType.WINNER
final.previous_match_p2_id = lb_final.id
final.source_p2_type = MatchSourceType.WINNER
return [n.to_dict() for n in nodes]
def refresh_bracket(t_obj: Tournament):
matches_map = {m.id: m for m in t_obj.matches}
for _ in range(20):
for m in t_obj.matches:
def resolve(src_id, type_):
if not src_id or src_id not in matches_map:
return None
src = matches_map[src_id]
if type_ == MatchSourceType.WINNER:
return src.winner
if type_ == MatchSourceType.LOSER:
if src.winner == "BYE":
return "BYE"
if src.winner:
return src.p1_name if src.winner == src.p2_name else src.p2_name
return None
return None
if m.previous_match_p1_id:
m.p1_name = resolve(m.previous_match_p1_id, m.source_p1_type)
if m.previous_match_p2_id:
m.p2_name = resolve(m.previous_match_p2_id, m.source_p2_type)
# BYE Auto-Win
if not m.winner and (m.p1_name == "BYE" or m.p2_name == "BYE"):
if m.p1_name == "BYE" and m.p2_name == "BYE":
m.winner = "BYE"
elif m.p1_name == "BYE":
m.winner = m.p2_name
else:
m.winner = m.p1_name
m.status = MatchStatus.FINISHED
# Reset Logic
if m.status == MatchStatus.FINISHED and m.winner != "BYE":
has_p1 = bool(m.p1_name)
has_p2 = bool(m.p2_name)
if (
not has_p1
or not has_p2
or (m.winner != m.p1_name and m.winner != m.p2_name)
):
m.winner = None
m.status = MatchStatus.PENDING
m.sets = []
# Numbering
display_counter = 1
sorted_matches = sorted(
t_obj.matches, key=lambda x: int(x.id) if x.id.isdigit() else 999
)
for m in sorted_matches:
if m.winner == "BYE" or m.p1_name == "BYE" or m.p2_name == "BYE":
m.number = None
else:
m.number = display_counter
display_counter += 1
# NO LABEL GENERATION HERE - FRONTEND HANDLES IT
def update_schedule(t_obj: Tournament):
match_map = {m.id: m for m in t_obj.matches}
depth_cache = {}
def get_depth(mid):
if mid not in match_map:
return 0
if mid in depth_cache:
return depth_cache[mid]
m = match_map[mid]
d = 1 + max(
get_depth(m.winner_next_match_id) if m.winner_next_match_id else 0,
get_depth(m.loser_next_match_id) if m.loser_next_match_id else 0,
)
depth_cache[mid] = d
return d
criticality_map = {}
for m in t_obj.matches:
criticality_map[m.id] = get_depth(m.id)
start_time = t_obj.timestamp
duration = t_obj.duration
finish_times: Dict[str, datetime] = {}
court_timers: Dict[str, datetime] = {c.name: start_time for c in t_obj.courts}
unscheduled = []
# 1. Initialize
for m in t_obj.matches:
if m.winner == "BYE" or m.p1_name == "BYE" or m.p2_name == "BYE":
finish_times[m.id] = start_time
m.status = MatchStatus.FINISHED
elif m.status == MatchStatus.FINISHED:
match_start = m.timestamp if m.timestamp else start_time
fin = match_start + timedelta(minutes=duration)
finish_times[m.id] = fin
if m.court_name and m.court_name in court_timers:
if fin > court_timers[m.court_name]:
court_timers[m.court_name] = fin
else:
m.timestamp = None
m.court_name = None
m.status = MatchStatus.PENDING
unscheduled.append(m)
if not court_timers:
return
# 2. Schedule
loop = len(t_obj.matches) * 2
while unscheduled and loop > 0:
loop -= 1
best_court = min(court_timers, key=lambda k: court_timers[k])
current_time_slot = court_timers[best_court]
ready = []
for m in unscheduled:
p1_r = (
finish_times.get(m.previous_match_p1_id, start_time)
if m.previous_match_p1_id
else start_time
)
p2_r = (
finish_times.get(m.previous_match_p2_id, start_time)
if m.previous_match_p2_id
else start_time
)
if max(p1_r, p2_r) <= current_time_slot:
ready.append(m)
if ready:
ready.sort(key=lambda x: (-criticality_map.get(x.id, 0), x.round))
cand = ready[0]
cand.court_name = best_court
cand.timestamp = current_time_slot
cand.status = MatchStatus.SCHEDULED
fin = current_time_slot + timedelta(minutes=duration)
finish_times[cand.id] = fin
court_timers[best_court] = fin
unscheduled.remove(cand)
else:
next_wake = None
for m in unscheduled:
p1_r = (
finish_times.get(m.previous_match_p1_id, start_time)
if m.previous_match_p1_id
else start_time
)
p2_r = (
finish_times.get(m.previous_match_p2_id, start_time)
if m.previous_match_p2_id
else start_time
)
ready_at = max(p1_r, p2_r)
if ready_at > current_time_slot:
if next_wake is None or ready_at < next_wake:
next_wake = ready_at
if next_wake:
court_timers[best_court] = next_wake
else:
break
+30
View File
@@ -0,0 +1,30 @@
# backend/app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from .database import Base, engine
from .routes import auth, tournaments, websocket
@asynccontextmanager
async def lifespan(app: FastAPI):
Base.metadata.create_all(bind=engine)
yield
app = FastAPI(title="Tournament Bracket API", lifespan=lifespan)
app.include_router(tournaments.router)
app.include_router(auth.router)
app.include_router(websocket.router)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
if __name__ == "__main__":
import uvicorn
uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)
+121
View File
@@ -0,0 +1,121 @@
# backend/app/models.py
from datetime import datetime
from typing import Optional
from sqlalchemy import JSON, DateTime
from sqlalchemy import Enum as SqlEnum
from sqlalchemy import ForeignKey, Integer, String
from sqlalchemy.orm import Mapped, mapped_column, relationship
from .constants import (
TournamentTypes,
MatchSourceType,
MatchStatus,
WinnerSide,
BracketType,
)
from .database import Base
class Tournament(Base):
__tablename__ = "tournaments"
id: Mapped[str] = mapped_column(String(8), primary_key=True, index=True)
name: Mapped[str] = mapped_column(String, nullable=False)
code: Mapped[str] = mapped_column(String, nullable=True)
timestamp: Mapped[datetime] = mapped_column(DateTime, nullable=False)
duration: Mapped[int] = mapped_column(Integer, default=30)
type: Mapped[TournamentTypes] = mapped_column(SqlEnum(TournamentTypes))
teams: Mapped[list["Team"]] = relationship(
"Team", back_populates="tournament", cascade="all, delete-orphan"
)
courts: Mapped[list["Court"]] = relationship(
"Court", back_populates="tournament", cascade="all, delete-orphan"
)
matches: Mapped[list["Match"]] = relationship(
"Match", back_populates="tournament", cascade="all, delete-orphan"
)
# --- ADD THESE PROPERTIES ---
@property
def team_count(self) -> int:
return len(self.teams)
@property
def court_count(self) -> int:
return len(self.courts)
class Team(Base):
__tablename__ = "teams"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
name: Mapped[str] = mapped_column(String, nullable=False)
tournament_id: Mapped[str] = mapped_column(ForeignKey("tournaments.id"))
tournament: Mapped["Tournament"] = relationship(back_populates="teams")
class Court(Base):
__tablename__ = "courts"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
name: Mapped[str] = mapped_column(String, nullable=False)
tournament_id: Mapped[str] = mapped_column(ForeignKey("tournaments.id"))
tournament: Mapped["Tournament"] = relationship(back_populates="courts")
class Match(Base):
__tablename__ = "matches"
id: Mapped[str] = mapped_column(String, primary_key=True)
tournament_id: Mapped[str] = mapped_column(
ForeignKey("tournaments.id"), primary_key=True
)
# --- Structural Info ---
bracket: Mapped[BracketType] = mapped_column(
SqlEnum(BracketType, native_enum=False)
)
round: Mapped[int] = mapped_column(Integer)
number: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
# --- Scheduling ---
timestamp: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)
court_name: Mapped[Optional[str]] = mapped_column(String, nullable=True)
# --- Player Info ---
p1_name: Mapped[Optional[str]] = mapped_column(String, nullable=True)
p2_name: Mapped[Optional[str]] = mapped_column(String, nullable=True)
winner: Mapped[Optional[str]] = mapped_column(String, nullable=True)
status: Mapped[MatchStatus] = mapped_column(
SqlEnum(MatchStatus, native_enum=False), default=MatchStatus.PENDING
)
sets: Mapped[list[dict]] = mapped_column(JSON, default=list)
previous_match_p1_id: Mapped[Optional[str]] = mapped_column(String, nullable=True)
previous_match_p2_id: Mapped[Optional[str]] = mapped_column(String, nullable=True)
source_p1_type: Mapped[Optional[MatchSourceType]] = mapped_column(
SqlEnum(MatchSourceType, native_enum=False), nullable=True
)
source_p2_type: Mapped[Optional[MatchSourceType]] = mapped_column(
SqlEnum(MatchSourceType, native_enum=False), nullable=True
)
winner_next_match_id: Mapped[Optional[str]] = mapped_column(String, nullable=True)
loser_next_match_id: Mapped[Optional[str]] = mapped_column(String, nullable=True)
tournament: Mapped["Tournament"] = relationship(back_populates="matches")
@property
def winner_side(self) -> WinnerSide:
if not self.winner:
return WinnerSide.NONE
if self.winner == self.p1_name:
return WinnerSide.P1
if self.winner == self.p2_name:
return WinnerSide.P2
return WinnerSide.NONE
+1
View File
@@ -0,0 +1 @@
# backend/app/routes/__init__.py
+48
View File
@@ -0,0 +1,48 @@
# backend/app/routes/auth.py
from datetime import timedelta
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from ..schemas import Token
from ..core.auth import create_access_token, get_current_user
from ..core.config import (
ACCESS_TOKEN_EXPIRE_MINUTES,
ADMIN_HASH,
ADMIN_USER,
verify_password,
)
router = APIRouter(prefix="/auth", tags=["Auth"])
@router.post("/token", response_model=Token)
async def login_for_access_token(
form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
):
if form_data.username != ADMIN_USER:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
if not verify_password(form_data.password, ADMIN_HASH):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": form_data.username}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
@router.get("/check")
async def check_auth(user: str = Depends(get_current_user)):
return {"is_admin": True, "user": user}
@@ -0,0 +1,6 @@
# backend/app/routes/tournaments/__init__.py
from fastapi import APIRouter
router = APIRouter(prefix="/tournaments", tags=["Tournaments"])
from . import report, courts, matches, teams, tournaments
+61
View File
@@ -0,0 +1,61 @@
# backend/app/routes/tournaments/courts.py
from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session
from ... import crud, schemas
from ...constants import SUCCESS
from ...core.websocket_manager import send_ws_update
from ...core.auth import get_current_user
from ...database import get_db
from . import router
@router.get("/{id}/courts", response_model=list[schemas.CourtSchema])
def get_courts(id: str, db: Session = Depends(get_db)):
return crud.get_courts(db, id)
@router.post("/{id}/courts", response_model=schemas.CourtSchema)
async def create_court(
id: str,
court: schemas.CourtCreate,
db: Session = Depends(get_db),
user: str = Depends(get_current_user),
):
new_court = crud.create_court(db, id, court)
if not new_court:
raise HTTPException(404, "Tournament not found")
await send_ws_update(id)
return new_court
@router.patch("/{id}/courts", response_model=schemas.TournamentDetail)
async def update_courts(
id: str,
courts: list[str],
db: Session = Depends(get_db),
user: str = Depends(get_current_user),
):
t = crud.update_tournament_courts(db, id, courts)
if not t:
raise HTTPException(404, "Not found")
await send_ws_update(id)
return t
@router.delete("/{id}/courts/{court_id}")
async def delete_court(
id: str,
court_id: int,
db: Session = Depends(get_db),
user: str = Depends(get_current_user),
):
success = crud.delete_court(db, id, court_id)
if not success:
raise HTTPException(404, "Court or Tournament not found")
await send_ws_update(id)
return SUCCESS
+24
View File
@@ -0,0 +1,24 @@
# backend/app/routes/tournaments/matches.py
from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session
from ... import crud, schemas
from ...database import get_db
from . import router
@router.get("/{id}/matches", response_model=list[schemas.MatchOut])
def get_tournament_matches(id: str, db: Session = Depends(get_db)):
matches = crud.get_tournament_matches(db, id)
return matches
@router.get("/{id}/matches/{match_id}", response_model=schemas.MatchOut)
def get_match_details(id: str, match_id: str, db: Session = Depends(get_db)):
match = crud.get_match(db, id, match_id)
if not match:
raise HTTPException(404, "Match not found")
return match
+145
View File
@@ -0,0 +1,145 @@
# backend/app/routes/tournaments/report.py
from typing import List, Optional
from fastapi import Depends, HTTPException, Query
from sqlalchemy.orm import Session
from sqlalchemy.orm.attributes import flag_modified
from ... import crud, logic, models, schemas
from ...constants import SUCCESS, MatchStatus
from ...core.auth import get_optional_user
from ...core.websocket_manager import send_ws_update
from ...database import get_db
from . import router
# --- Helper: Centralize Auth Logic ---
def _check_auth(t: models.Tournament, user: Optional[str], code: Optional[str]):
is_admin = user is not None
code_matches = code is not None and str(code).strip() == str(t.code).strip()
if not is_admin and not code_matches:
raise HTTPException(403, "Invalid tournament code or admin privileges required")
@router.post("/{id}/matches/{match_id}/score")
async def report_score(
id: str,
match_id: str,
report: schemas.ScoreReport,
db: Session = Depends(get_db),
user: Optional[str] = Depends(get_optional_user),
):
t = crud.get_tournament(db, id)
if not t:
raise HTTPException(404, "Tournament not found")
_check_auth(t, user, report.code)
match = crud.get_match(db, id, match_id)
if not match:
raise HTTPException(404, "Match not found")
if not report.sets:
raise HTTPException(400, "No sets submitted")
_apply_score(match, report.sets)
flag_modified(match, "sets")
logic.refresh_bracket(t)
logic.update_schedule(t)
db.commit()
await send_ws_update(id)
return SUCCESS
@router.patch("/{id}/matches/{match_id}/score")
async def edit_score(
id: str,
match_id: str,
report: schemas.ScoreReport,
db: Session = Depends(get_db),
user: Optional[str] = Depends(get_optional_user),
):
"""
Allows correcting a score without resetting the match status logic entirely,
or just re-applying the new sets.
"""
t = crud.get_tournament(db, id)
if not t:
raise HTTPException(404, "Tournament not found")
_check_auth(t, user, report.code)
match = crud.get_match(db, id, match_id)
if not match:
raise HTTPException(404, "Match not found")
if report.sets:
_apply_score(match, report.sets)
flag_modified(match, "sets")
logic.refresh_bracket(t)
logic.update_schedule(t)
db.commit()
await send_ws_update(id)
return SUCCESS
@router.delete("/{id}/matches/{match_id}/score")
async def clear_score(
id: str,
match_id: str,
code: Optional[str] = Query(None),
db: Session = Depends(get_db),
user: Optional[str] = Depends(get_optional_user),
):
t = crud.get_tournament(db, id)
if not t:
raise HTTPException(404, "Tournament not found")
_check_auth(t, user, code)
match = crud.get_match(db, id, match_id)
if not match:
raise HTTPException(404, "Match not found")
match.winner = None
match.status = MatchStatus.PENDING.value
match.sets = []
flag_modified(match, "sets")
logic.refresh_bracket(t)
logic.update_schedule(t)
db.commit()
await send_ws_update(id)
return SUCCESS
def _apply_score(match: models.Match, sets: List[schemas.SetScore]):
"""
Calculates winner based on sets and updates the match object.
Does NOT commit to DB.
"""
p1_wins = sum(1 for s in sets if s.p1 > s.p2)
p2_wins = sum(1 for s in sets if s.p2 > s.p1)
if p1_wins > p2_wins:
match.winner = match.p1_name
elif p2_wins > p1_wins:
match.winner = match.p2_name
else:
p1_points = sum(s.p1 for s in sets)
p2_points = sum(s.p2 for s in sets)
if p1_points == p2_points:
raise HTTPException(400, "Absolute tie: Sets and Points are equal.")
match.winner = match.p1_name if p1_points > p2_points else match.p2_name
match.status = MatchStatus.FINISHED.value
match.sets = [s.model_dump() for s in sets]
+61
View File
@@ -0,0 +1,61 @@
# backend/app/routes/tournaments/teams.py
from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session
from ... import crud, schemas
from ...constants import SUCCESS
from ...core.websocket_manager import send_ws_update
from ...core.auth import get_current_user
from ...database import get_db
from . import router
@router.get("/{id}/teams", response_model=list[schemas.TeamSchema])
def get_teams(id: str, db: Session = Depends(get_db)):
return crud.get_teams(db, id)
@router.post("/{id}/teams", response_model=schemas.TeamSchema)
async def create_team(
id: str,
team: schemas.TeamCreate,
db: Session = Depends(get_db),
user: str = Depends(get_current_user),
):
new_team = crud.create_team(db, id, team)
if not new_team:
raise HTTPException(404, "Tournament not found")
await send_ws_update(id)
return new_team
@router.patch("/{id}/teams", response_model=list[schemas.TeamSchema])
async def update_teams(
id: str,
teams: list[str],
db: Session = Depends(get_db),
user: str = Depends(get_current_user),
):
t = crud.update_tournament_teams(db, id, teams)
if not t:
raise HTTPException(404, "Not found")
await send_ws_update(id)
return t.teams
@router.delete("/{id}/teams/{team_id}")
async def delete_team(
id: str,
team_id: int,
db: Session = Depends(get_db),
user: str = Depends(get_current_user),
):
success = crud.delete_team(db, id, team_id)
if not success:
raise HTTPException(404, "Team or Tournament not found")
await send_ws_update(id)
return SUCCESS
@@ -0,0 +1,62 @@
# backend/app/routes/tournaments/tournaments.py
from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session
from ... import crud, schemas
from ...constants import SUCCESS
from ...core.websocket_manager import send_ws_update
from ...database import get_db
from ...core.auth import get_current_user
from . import router
@router.post("", response_model=schemas.TournamentOut)
async def create_tournament(
data: schemas.TournamentCreate,
db: Session = Depends(get_db),
user: str = Depends(get_current_user),
):
new_t = crud.create_tournament(db, data)
await send_ws_update(new_t.id)
return new_t
@router.get("", response_model=list[schemas.TournamentOut])
def list_tournaments(db: Session = Depends(get_db)):
return crud.get_tournaments(db)
@router.get("/{id}", response_model=schemas.TournamentDetail)
def get_tournament(id: str, db: Session = Depends(get_db)):
t = crud.get_tournament(db, id)
if not t:
raise HTTPException(404, "Tournament not found")
return t
@router.patch("/{id}", response_model=schemas.TournamentUpdateResponse)
async def update_settings(
id: str,
data: schemas.TournamentUpdate,
db: Session = Depends(get_db),
user: str = Depends(get_current_user),
):
t = crud.update_tournament_details(db, id, data)
if not t:
raise HTTPException(404, "Not found")
await send_ws_update(id)
return t
@router.delete("/{id}")
async def delete_tournament(
id: str, db: Session = Depends(get_db), user: str = Depends(get_current_user)
):
success = crud.delete_tournament(db, id)
if not success:
raise HTTPException(404, "Tournament not found")
await send_ws_update(id)
return SUCCESS
+16
View File
@@ -0,0 +1,16 @@
# backend/app/routes/websocket.py
from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from ..core.websocket_manager import manager
router = APIRouter(prefix="/ws", tags=["Websocket"])
@router.websocket("/")
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
await websocket.receive_text()
except WebSocketDisconnect:
manager.disconnect(websocket)
+139
View File
@@ -0,0 +1,139 @@
# backend/app/schemas.py
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
from .constants import (
BracketType,
MatchSourceType,
MatchStatus,
TournamentTypes,
WinnerSide,
)
class TeamSchema(BaseModel):
id: int
name: str
model_config = ConfigDict(from_attributes=True)
class CourtSchema(BaseModel):
id: int
name: str
model_config = ConfigDict(from_attributes=True)
class TeamCreate(BaseModel):
name: str
class CourtCreate(BaseModel):
name: str
class SetScore(BaseModel):
p1: int
p2: int
class ScoreReport(BaseModel):
id: str
code: str | None = None
sets: list[SetScore]
class MatchOut(BaseModel):
id: str
number: int | None = None
timestamp: datetime | None = None
court: str | None = Field(
default=None, validation_alias="court_name", serialization_alias="court"
)
bracket: BracketType
round: int
p1: str | None = Field(
default=None, validation_alias="p1_name", serialization_alias="p1"
)
p2: str | None = Field(
default=None, validation_alias="p2_name", serialization_alias="p2"
)
status: MatchStatus
sets: list[SetScore] = []
previous_match_p1_id: str | None = Field(
default=None, serialization_alias="source_p1"
)
previous_match_p2_id: str | None = Field(
default=None, serialization_alias="source_p2"
)
source_p1_type: MatchSourceType | None = None
source_p2_type: MatchSourceType | None = None
winner_next_match_id: str | None = Field(
default=None, serialization_alias="next_win"
)
loser_next_match_id: str | None = Field(
default=None, serialization_alias="next_loss"
)
winner_side: WinnerSide = WinnerSide.NONE
model_config = ConfigDict(from_attributes=True, populate_by_name=True)
class TournamentCreate(BaseModel):
name: str
code: str
type: TournamentTypes
timestamp: datetime
duration: int
teams: list[str]
courts: list[str]
class TournamentOut(BaseModel):
id: str
name: str
timestamp: datetime
type: TournamentTypes
team_count: int
court_count: int
model_config = ConfigDict(from_attributes=True)
class TournamentUpdate(BaseModel):
name: str | None = None
code: str | None = None
timestamp: datetime | None = None
duration: int | None = None
type: TournamentTypes | None = None
class TournamentDetail(BaseModel):
id: str
name: str
code: str
timestamp: datetime
type: TournamentTypes
teams: list[TeamSchema]
courts: list[CourtSchema]
matches: list[MatchOut]
model_config = ConfigDict(from_attributes=True)
class TournamentUpdateResponse(TournamentDetail):
code: str
class Token(BaseModel):
access_token: str
token_type: str
Binary file not shown.
+1
View File
@@ -0,0 +1 @@
# backend\tests\__init__.py
+97
View File
@@ -0,0 +1,97 @@
# backend/tests/conftest.py
from typing import AsyncGenerator, Generator
import pytest
from app.core.auth import get_current_user, get_optional_user
from app.database import Base, get_db
# Import your app and models
from app.main import app
from fastapi import Request
from httpx import ASGITransport, AsyncClient
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool
# --- DATABASE SETUP ---
# Use in-memory SQLite.
# StaticPool is CRITICAL for in-memory SQLite with async tests to share connection.
SQLALCHEMY_DATABASE_URL = "sqlite:///"
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
@pytest.fixture(scope="session")
def prepare_db():
Base.metadata.create_all(bind=engine)
yield
Base.metadata.drop_all(bind=engine)
@pytest.fixture(scope="function")
def db(prepare_db) -> Generator[Session, None, None]:
connection = engine.connect()
transaction = connection.begin()
session = TestingSessionLocal(bind=connection)
yield session
session.close()
transaction.rollback()
connection.close()
@pytest.fixture(scope="function")
async def client(db: Session) -> AsyncGenerator[AsyncClient, None]:
def override_get_db():
try:
yield db
finally:
pass
# Strict Auth: Always requires a token (simulated by header presence)
def override_get_current_user(request: Request):
if "Authorization" not in request.headers:
# Let FastAPI raise the 401 naturally if header is missing
raise pytest.skip("Auth header missing in strict auth test")
return "test_admin"
# Optional Auth: Returns Admin IF header exists, else None
def override_get_optional_user(request: Request):
if "Authorization" in request.headers:
return "test_admin"
return None
app.dependency_overrides[get_db] = override_get_db
app.dependency_overrides[get_current_user] = override_get_current_user
app.dependency_overrides[get_optional_user] = override_get_optional_user
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as c:
yield c
app.dependency_overrides.clear()
# --- HELPER FIXTURES ---
@pytest.fixture
def auth_headers(client):
return {"Authorization": "Bearer test_token"}
@pytest.fixture
def valid_tournament_payload():
return {
"name": "Test Tournament",
"code": "1234",
"type": "Double",
"timestamp": "2024-01-01T10:00:00",
"duration": 15,
"teams": ["Team A", "Team B", "Team C", "Team D"],
"courts": ["Court 1", "Court 2"],
}
+94
View File
@@ -0,0 +1,94 @@
# backend/tests/test_scoring.py
import pytest
from httpx import AsyncClient
pytestmark = pytest.mark.anyio
async def test_scoring_flow(
client: AsyncClient, auth_headers, valid_tournament_payload
):
# 1. Create Tournament
res = await client.post(
"/tournaments", json=valid_tournament_payload, headers=auth_headers
)
t_id = res.json()["id"]
t_code = valid_tournament_payload["code"]
# 2. Get Matches to find a Round 1 match
matches_res = await client.get(f"/tournaments/{t_id}/matches")
matches = matches_res.json()
# Find a match that has real players (not BYE)
# In double elim, Round 1 matches usually have seeds.
target_match = next(m for m in matches if m["p1"] and m["p2"])
match_id = target_match["id"]
next_match_id = target_match["next_win"] # Note: using alias from schema
# 3. Report Score WITHOUT Auth Header (Public user with Code)
score_payload = {
"id": match_id,
"code": t_code,
"sets": [{"p1": 21, "p2": 19}, {"p1": 21, "p2": 15}], # P1 Wins
}
report_res = await client.post(
f"/tournaments/{t_id}/matches/{match_id}/score", json=score_payload
)
assert report_res.status_code == 200
# 4. Verify Winner Advanced
# Fetch the *Next* match
next_match_res = await client.get(f"/tournaments/{t_id}/matches/{next_match_id}")
next_match = next_match_res.json()
# Assert P1 from previous match is now in the next match
# Note: We check if the name matches the winner
winner_name = target_match["p1"]
assert (next_match["p1"] == winner_name) or (next_match["p2"] == winner_name)
# 5. Test Invalid Code
bad_payload = score_payload.copy()
bad_payload["code"] = "WRONG"
bad_res = await client.post(
f"/tournaments/{t_id}/matches/{match_id}/score", json=bad_payload
)
assert bad_res.status_code == 403
async def test_clear_score(client: AsyncClient, auth_headers, valid_tournament_payload):
# Setup: Create & Score
res = await client.post(
"/tournaments", json=valid_tournament_payload, headers=auth_headers
)
t_id = res.json()["id"]
matches = (await client.get(f"/tournaments/{t_id}/matches")).json()
target = next(m for m in matches if m["p1"] and m["p2"])
score_payload = {"id": target["id"], "code": "1234", "sets": [{"p1": 25, "p2": 0}]}
await client.post(
f"/tournaments/{t_id}/matches/{target['id']}/score", json=score_payload
)
# Verify Finished
check_res = await client.get(f"/tournaments/{t_id}/matches/{target['id']}")
assert check_res.json()["status"] == "Finished"
# Action: Clear Score
clear_res = await client.delete(
f"/tournaments/{t_id}/matches/{target['id']}/score", headers=auth_headers
)
assert clear_res.status_code == 200
# Verify Reset
final_res = await client.get(f"/tournaments/{t_id}/matches/{target['id']}")
data = final_res.json()
# 1. We already fixed this to expect 'Scheduled'
assert data["status"] == "Scheduled"
# 2. FIX: Check 'winner_side' instead of 'winner'
# Use the Enum value "none"
assert data["winner_side"] == "none"
assert len(data["sets"]) == 0
+66
View File
@@ -0,0 +1,66 @@
# backend/tests/test_structure.py
import pytest
from httpx import AsyncClient
pytestmark = pytest.mark.anyio
async def test_manage_teams(
client: AsyncClient, auth_headers, valid_tournament_payload
):
# Setup
res = await client.post(
"/tournaments", json=valid_tournament_payload, headers=auth_headers
)
t_id = res.json()["id"]
# 1. Add a Team via POST
new_team = {"name": "Team E"}
post_res = await client.post(
f"/tournaments/{t_id}/teams", json=new_team, headers=auth_headers
)
assert post_res.status_code == 200
assert post_res.json()["name"] == "Team E"
# 2. Verify Bracket Regenerated (Match count should likely change or re-seed)
matches_res = await client.get(f"/tournaments/{t_id}/matches")
# With 4 teams -> ~6 matches. With 5 teams -> ~8-10 matches in Double Elim.
assert len(matches_res.json()) > 0
# 3. Bulk Update via PATCH (Replace all teams)
new_team_list = ["Team X", "Team Y"]
patch_res = await client.patch(
f"/tournaments/{t_id}/teams", json=new_team_list, headers=auth_headers
)
assert patch_res.status_code == 200
data = patch_res.json()
assert len(data) == 2
assert data[0]["name"] in ["Team X", "Team Y"]
async def test_manage_courts(
client: AsyncClient, auth_headers, valid_tournament_payload
):
res = await client.post(
"/tournaments", json=valid_tournament_payload, headers=auth_headers
)
t_id = res.json()["id"]
# Get initial courts
courts_res = await client.get(f"/tournaments/{t_id}/courts")
initial_courts = courts_res.json()
assert len(initial_courts) == 2
# Delete a court
court_id = initial_courts[0]["id"]
del_res = await client.delete(
f"/tournaments/{t_id}/courts/{court_id}", headers=auth_headers
)
assert del_res.status_code == 200
# Create a court
create_res = await client.post(
f"/tournaments/{t_id}/courts", json={"name": "New Court"}, headers=auth_headers
)
assert create_res.status_code == 200
assert create_res.json()["name"] == "New Court"
+87
View File
@@ -0,0 +1,87 @@
# backend/tests/test_tournaments.py
import pytest
from httpx import AsyncClient
# Mark all tests in this file as async
pytestmark = pytest.mark.anyio
async def test_create_tournament(
client: AsyncClient, auth_headers, valid_tournament_payload
):
response = await client.post(
"/tournaments", json=valid_tournament_payload, headers=auth_headers
)
assert response.status_code == 200
data = response.json()
assert data["name"] == "Test Tournament"
assert data["team_count"] == 4
assert data["court_count"] == 2
assert "id" in data
async def test_list_tournaments(
client: AsyncClient, auth_headers, valid_tournament_payload
):
# Create one first
await client.post(
"/tournaments", json=valid_tournament_payload, headers=auth_headers
)
response = await client.get("/tournaments")
assert response.status_code == 200
data = response.json()
assert len(data) >= 1
assert data[0]["name"] == "Test Tournament"
async def test_get_tournament_detail(
client: AsyncClient, auth_headers, valid_tournament_payload
):
create_res = await client.post(
"/tournaments", json=valid_tournament_payload, headers=auth_headers
)
t_id = create_res.json()["id"]
response = await client.get(f"/tournaments/{t_id}")
assert response.status_code == 200
data = response.json()
# Check deeply nested fields
assert len(data["matches"]) > 0 # Logic should have generated matches
assert len(data["teams"]) == 4
async def test_update_settings(
client: AsyncClient, auth_headers, valid_tournament_payload
):
create_res = await client.post(
"/tournaments", json=valid_tournament_payload, headers=auth_headers
)
t_id = create_res.json()["id"]
update_payload = {"name": "Updated Name", "code": "9999"}
response = await client.patch(
f"/tournaments/{t_id}", json=update_payload, headers=auth_headers
)
# This will now succeed because we updated the response_model!
assert response.status_code == 200
data = response.json()
assert data["name"] == "Updated Name"
assert data["code"] == "9999"
async def test_delete_tournament(
client: AsyncClient, auth_headers, valid_tournament_payload
):
create_res = await client.post(
"/tournaments", json=valid_tournament_payload, headers=auth_headers
)
t_id = create_res.json()["id"]
del_res = await client.delete(f"/tournaments/{t_id}", headers=auth_headers)
assert del_res.status_code == 200
# Verify it's gone
get_res = await client.get(f"/tournaments/{t_id}")
assert get_res.status_code == 404