Merge branch 'feature/remove-auth' into master

This commit is contained in:
2026-03-13 12:40:58 -05:00
44 changed files with 523 additions and 1024 deletions

View File

@@ -1,10 +1,6 @@
# SneakySwole Environment Configuration
# Copy this file to .env and fill in real values.
# Admin credentials (used to create admin user on first run)
ADMIN_USERNAME=admin
ADMIN_PASSWORD=changeme
# App settings
APP_ENV=development
APP_HOST=0.0.0.0

View File

@@ -1,9 +1,6 @@
# SneakySwole Production Environment
# Copy to .env on your production server and fill in real values.
ADMIN_USERNAME=admin
ADMIN_PASSWORD=
APP_ENV=production
APP_HOST=0.0.0.0
APP_PORT=8000

View File

@@ -0,0 +1,37 @@
"""remove auth columns from users
Revision ID: a1b2c3d4e5f6
Revises: 1855836abf6c
Create Date: 2026-03-13
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
import sqlmodel
# revision identifiers, used by Alembic.
revision: str = 'a1b2c3d4e5f6'
down_revision: Union[str, Sequence[str], None] = '1855836abf6c'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Drop password_hash and is_admin columns, delete admin users."""
# Delete admin users before dropping the column
op.execute("DELETE FROM users WHERE is_admin = 1")
# SQLite requires table recreation for column drops
with op.batch_alter_table('users', schema=None) as batch_op:
batch_op.drop_column('password_hash')
batch_op.drop_column('is_admin')
def downgrade() -> None:
"""Re-add password_hash and is_admin columns."""
with op.batch_alter_table('users', schema=None) as batch_op:
batch_op.add_column(sa.Column('password_hash', sa.String(), nullable=False, server_default=''))
batch_op.add_column(sa.Column('is_admin', sa.Boolean(), nullable=False, server_default='0'))

View File

@@ -20,8 +20,6 @@ class Settings(BaseSettings):
"""Typed application settings loaded from environment variables.
Attributes:
admin_username: Admin login username (from ADMIN_USERNAME env var).
admin_password: Admin login password (from ADMIN_PASSWORD env var).
app_env: Runtime environment — 'development' or 'production'.
app_host: Host address to bind the server to.
app_port: Port number for the server.
@@ -29,8 +27,6 @@ class Settings(BaseSettings):
database_url: SQLite connection string.
"""
admin_username: str
admin_password: str
app_env: str = "development"
app_host: str = "0.0.0.0"
app_port: int = 8000
@@ -40,6 +36,7 @@ class Settings(BaseSettings):
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
extra="ignore",
)

View File

@@ -4,8 +4,6 @@ Creates and configures the FastAPI app with routes, templates,
static files, and structured logging.
"""
import os
import secrets
from pathlib import Path
import structlog
@@ -23,7 +21,6 @@ from app.models.user import User
from app.config import get_settings
from app.database import get_engine, get_db_session
from app.logging_config import setup_logging
from app.routes.auth import router as auth_router
from app.routes.exercises import router as exercises_router
from app.routes.history import router as history_router
from app.routes.health import router as health_router
@@ -34,9 +31,8 @@ from app.routes.workouts import router as workouts_router
from app.routes.dashboard import router as dashboard_router
from app.routes.schedule import router as schedule_router
from app.services.seed_service import SeedService
from app.services.auth_service import AuthService
from app.services.user_service import UserService
from app.utils.auth import SESSION_COOKIE_NAME, NotAuthenticatedError
from app.utils.auth import NoProfileSelectedError
logger = structlog.get_logger(__name__)
@@ -47,30 +43,21 @@ STATIC_DIR = _BASE_DIR / "static"
class NavContextMiddleware(BaseHTTPMiddleware):
"""Injects admin, profiles, and active_profile into request.state for templates."""
"""Injects profiles and active_profile into request.state for templates."""
async def dispatch(self, request: StarletteRequest, call_next: RequestResponseEndpoint) -> Response:
request.state.admin = None
request.state.profiles = []
request.state.active_profile = None
token = request.cookies.get(SESSION_COOKIE_NAME)
if token and hasattr(request.app.state, "engine"):
if hasattr(request.app.state, "engine"):
try:
with Session(request.app.state.engine) as session:
secret_key = getattr(request.app.state, "secret_key", "")
auth_service = AuthService(session, secret_key=secret_key)
user_id = auth_service.validate_session_token(token)
if user_id:
admin = session.get(User, user_id)
if admin and admin.is_admin:
request.state.admin = admin
user_service = UserService(session)
request.state.profiles = user_service.list_users(exclude_admin=True)
user_service = UserService(session)
request.state.profiles = user_service.list_users()
profile_id = request.cookies.get("active_profile_id")
if profile_id and profile_id.isdigit():
request.state.active_profile = user_service.get_user_by_id(int(profile_id))
profile_id = request.cookies.get("active_profile_id")
if profile_id and profile_id.isdigit():
request.state.active_profile = user_service.get_user_by_id(int(profile_id))
except Exception:
pass
@@ -92,10 +79,10 @@ def create_app() -> FastAPI:
version="0.1.0",
)
# Redirect unauthenticated requests to login
@app.exception_handler(NotAuthenticatedError)
async def _not_authenticated_handler(request, exc):
return RedirectResponse(url="/login", status_code=302)
# Redirect to home when no profile is selected
@app.exception_handler(NoProfileSelectedError)
async def _no_profile_handler(request, exc):
return RedirectResponse(url="/", status_code=302)
# Mount static files
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
@@ -104,14 +91,10 @@ def create_app() -> FastAPI:
templates = Jinja2Templates(directory=str(TEMPLATES_DIR))
app.state.templates = templates
# Secret key for session signing
app.state.secret_key = os.environ.get("SECRET_KEY", secrets.token_hex(32))
# Nav context middleware — injects admin/profiles/active_profile into request.state
# Nav context middleware — injects profiles/active_profile into request.state
app.add_middleware(NavContextMiddleware)
# Register route modules
app.include_router(auth_router)
app.include_router(exercises_router)
app.include_router(health_router)
app.include_router(history_router)

View File

@@ -1,6 +1,6 @@
"""User model for profile management.
Stores admin and regular user profiles with physical stats and goals.
Stores user profiles with physical stats and goals.
"""
from datetime import datetime
@@ -14,13 +14,11 @@ class User(SQLModel, table=True):
Attributes:
id: Primary key, auto-incremented.
username: Unique login identifier.
password_hash: bcrypt-hashed password (admin only initially).
username: Unique identifier.
display_name: Human-readable name shown in the UI.
height: User's height as a string (e.g., "6'0\"").
weight: User's weight as a string (e.g., "260 lbs").
goals: Free-text training goals.
is_admin: Whether this user has admin privileges.
created_at: Timestamp when the record was created.
updated_at: Timestamp of the last update.
"""
@@ -29,11 +27,9 @@ class User(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
username: str = Field(index=True, unique=True)
password_hash: str = Field(default="")
display_name: str = Field(default="")
height: Optional[str] = Field(default=None)
weight: Optional[str] = Field(default=None)
goals: Optional[str] = Field(default=None)
is_admin: bool = Field(default=False)
created_at: datetime = Field(default_factory=datetime.utcnow)
updated_at: datetime = Field(default_factory=datetime.utcnow)

View File

@@ -1,99 +0,0 @@
"""Authentication routes for admin login and logout.
Handles the login form, credential verification, session cookie
management, and logout.
"""
import structlog
from fastapi import APIRouter, Depends, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from sqlmodel import Session
from app.database import get_db_session
from app.services.auth_service import AuthService
from app.utils.auth import SESSION_COOKIE_NAME
logger = structlog.get_logger(__name__)
router = APIRouter(tags=["auth"])
@router.get("/login", response_class=HTMLResponse)
async def login_page(request: Request):
"""Render the login form.
Args:
request: The incoming HTTP request.
Returns:
Rendered login page HTML.
"""
templates = request.app.state.templates
return templates.TemplateResponse("pages/login.html", {
"request": request,
"error": None,
})
@router.post("/login")
async def login_submit(
request: Request,
session: Session = Depends(get_db_session),
):
"""Process login form submission.
Verifies credentials and sets a session cookie on success.
Re-renders the login page with an error on failure.
Args:
request: The incoming HTTP request.
session: Database session.
Returns:
Redirect to home on success, or login page with error.
"""
form = await request.form()
username = form.get("username", "")
password = form.get("password", "")
secret_key = request.app.state.secret_key
auth_service = AuthService(session, secret_key=secret_key)
user = auth_service.authenticate(username, password)
if user is None:
templates = request.app.state.templates
return templates.TemplateResponse("pages/login.html", {
"request": request,
"error": "Invalid username or password.",
}, status_code=200)
# Create session token and set cookie — httponly and samesite for security
token = auth_service.create_session_token(user_id=user.id)
response = RedirectResponse(url="/", status_code=303)
response.set_cookie(
key=SESSION_COOKIE_NAME,
value=token,
httponly=True,
samesite="lax",
max_age=86400, # 24 hours
)
logger.info("login_success", username=username)
return response
@router.get("/logout")
async def logout(request: Request):
"""Clear the session cookie and redirect to login.
Args:
request: The incoming HTTP request.
Returns:
Redirect to login page.
"""
response = RedirectResponse(url="/login", status_code=303)
response.delete_cookie(key=SESSION_COOKIE_NAME)
response.delete_cookie(key="active_profile_id")
logger.info("logout")
return response

View File

@@ -15,7 +15,7 @@ from app.models.user import User
from app.services.analytics_service import AnalyticsService
from app.services.exercise_service import ExerciseService
from app.services.progression_service import ProgressionService
from app.utils.auth import get_current_admin_user, get_active_profile_id
from app.utils.auth import require_active_profile
logger = structlog.get_logger(__name__)
@@ -26,25 +26,12 @@ router = APIRouter(prefix="/dashboard", tags=["dashboard"])
async def dashboard(
request: Request,
session: Session = Depends(get_db_session),
admin: User = Depends(get_current_admin_user),
profile: User = Depends(require_active_profile),
):
"""Render the progress dashboard for the active profile.
Shows: summary stats, volume by day chart, exercise progress links.
"""
active_profile_id = get_active_profile_id(request)
active_profile = (
session.get(User, active_profile_id)
if active_profile_id
else None
)
stats = {}
volume_data = {}
if active_profile_id:
analytics = AnalyticsService(session)
stats = analytics.get_user_stats(active_profile_id)
volume_data = analytics.get_volume_by_day(active_profile_id)
"""Render the progress dashboard for the active profile."""
analytics = AnalyticsService(session)
stats = analytics.get_user_stats(profile.id)
volume_data = analytics.get_volume_by_day(profile.id)
exercise_service = ExerciseService(session)
exercises = exercise_service.list_exercises()
@@ -55,8 +42,7 @@ async def dashboard(
"stats": stats,
"volume_data_json": json.dumps(volume_data),
"exercises": exercises,
"active_profile": active_profile,
"admin": admin,
"active_profile": profile,
})
@@ -65,31 +51,21 @@ async def exercise_progress(
exercise_id: int,
request: Request,
session: Session = Depends(get_db_session),
admin: User = Depends(get_current_admin_user),
profile: User = Depends(require_active_profile),
):
"""Render per-exercise progress page with charts and suggestions."""
active_profile_id = get_active_profile_id(request)
active_profile = (
session.get(User, active_profile_id)
if active_profile_id
else None
)
exercise_service = ExerciseService(session)
exercise = exercise_service.get_exercise_by_id(exercise_id)
progress_data = {}
suggestion = {}
if active_profile_id:
analytics = AnalyticsService(session)
progress_data = analytics.get_exercise_progress(
active_profile_id, exercise_id,
)
analytics = AnalyticsService(session)
progress_data = analytics.get_exercise_progress(
profile.id, exercise_id,
)
progression = ProgressionService(session)
suggestion = progression.get_suggestion(
active_profile_id, exercise_id,
)
progression = ProgressionService(session)
suggestion = progression.get_suggestion(
profile.id, exercise_id,
)
templates = request.app.state.templates
return templates.TemplateResponse("pages/exercise_progress.html", {
@@ -97,6 +73,5 @@ async def exercise_progress(
"exercise": exercise,
"progress_data_json": json.dumps(progress_data),
"suggestion": suggestion,
"active_profile": active_profile,
"admin": admin,
"active_profile": profile,
})

View File

@@ -1,6 +1,6 @@
"""Exercise browser routes with HTMX search/filter support.
All filtering is done via HTMX partial responses no JSON APIs.
All filtering is done via HTMX partial responses -- no JSON APIs.
"""
import structlog
@@ -11,7 +11,7 @@ from sqlmodel import Session
from app.database import get_db_session
from app.models.user import User
from app.services.exercise_service import ExerciseService
from app.utils.auth import get_current_admin_user
from app.utils.auth import require_active_profile
logger = structlog.get_logger(__name__)
@@ -22,18 +22,9 @@ router = APIRouter(prefix="/exercises", tags=["exercises"])
async def exercise_browser(
request: Request,
session: Session = Depends(get_db_session),
admin: User = Depends(get_current_admin_user),
profile: User = Depends(require_active_profile),
):
"""Render the exercise browser page with all exercises.
Args:
request: The incoming HTTP request.
session: Database session.
admin: The authenticated admin user.
Returns:
Rendered exercise browser page.
"""
"""Render the exercise browser page with all exercises."""
exercise_service = ExerciseService(session)
exercises = exercise_service.list_exercises()
workout_days = exercise_service.list_workout_days()
@@ -47,7 +38,6 @@ async def exercise_browser(
"exercises": exercises,
"workout_days": workout_days,
"muscle_groups": muscle_groups,
"admin": admin,
})
@@ -55,24 +45,11 @@ async def exercise_browser(
async def exercise_search(
request: Request,
session: Session = Depends(get_db_session),
admin: User = Depends(get_current_admin_user),
profile: User = Depends(require_active_profile),
workout_day: str = Query(default="", alias="workout_day"),
muscle_group: str = Query(default="", alias="muscle_group"),
):
"""Return filtered exercise list as an HTMX partial.
Called via hx-get from the exercise browser filter dropdowns.
Args:
request: The incoming HTTP request.
session: Database session.
admin: The authenticated admin user.
workout_day: Filter by workout day name.
muscle_group: Filter by muscle group.
Returns:
Rendered exercise list partial HTML.
"""
"""Return filtered exercise list as an HTMX partial."""
exercise_service = ExerciseService(session)
exercises = exercise_service.list_exercises(
workout_day=workout_day or None,

View File

@@ -13,7 +13,7 @@ from app.models.user import User
from app.services.exercise_service import ExerciseService
from app.services.log_service import LogService
from app.services.workout_session_service import WorkoutSessionService
from app.utils.auth import get_current_admin_user, get_active_profile_id
from app.utils.auth import require_active_profile
logger = structlog.get_logger(__name__)
@@ -24,31 +24,11 @@ router = APIRouter(prefix="/history", tags=["history"])
async def log_history(
request: Request,
session: Session = Depends(get_db_session),
admin: User = Depends(get_current_admin_user),
profile: User = Depends(require_active_profile),
):
"""Display log history for the active profile.
Shows a list of past workout sessions, most recent first.
Args:
request: The incoming HTTP request.
session: Database session.
admin: The authenticated admin user.
Returns:
Rendered log history page.
"""
active_profile_id = get_active_profile_id(request)
active_profile = (
session.get(User, active_profile_id)
if active_profile_id
else None
)
sessions_list = []
if active_profile_id:
ws_service = WorkoutSessionService(session)
sessions_list = ws_service.list_sessions(user_id=active_profile_id)
"""Display log history for the active profile."""
ws_service = WorkoutSessionService(session)
sessions_list = ws_service.list_sessions(user_id=profile.id)
# Resolve workout day names for display
exercise_service = ExerciseService(session)
@@ -59,8 +39,7 @@ async def log_history(
"request": request,
"sessions": sessions_list,
"days_by_id": days_by_id,
"active_profile": active_profile,
"admin": admin,
"active_profile": profile,
})
@@ -69,19 +48,9 @@ async def session_detail(
session_id: int,
request: Request,
session: Session = Depends(get_db_session),
admin: User = Depends(get_current_admin_user),
profile: User = Depends(require_active_profile),
):
"""Display detailed logs for a specific workout session.
Args:
session_id: The workout session ID.
request: The incoming HTTP request.
session: Database session.
admin: The authenticated admin user.
Returns:
Rendered session detail page.
"""
"""Display detailed logs for a specific workout session."""
ws_service = WorkoutSessionService(session)
ws = ws_service.get_session_by_id(session_id)
@@ -109,5 +78,4 @@ async def session_detail(
"logs_by_exercise": logs_by_exercise,
"exercises_by_id": exercises_by_id,
"days_by_id": days_by_id,
"admin": admin,
})

View File

@@ -15,7 +15,7 @@ from app.database import get_db_session
from app.models.user import User
from app.services.log_service import LogService
from app.services.workout_session_service import WorkoutSessionService
from app.utils.auth import get_current_admin_user, get_active_profile_id
from app.utils.auth import require_active_profile, get_active_profile_id
logger = structlog.get_logger(__name__)
@@ -26,20 +26,12 @@ router = APIRouter(prefix="/log", tags=["logging"])
async def log_set(
request: Request,
session: Session = Depends(get_db_session),
admin: User = Depends(get_current_admin_user),
profile: User = Depends(require_active_profile),
):
"""Log a single set for an exercise.
Creates the workout session if it doesn't exist yet (auto-create).
Returns the updated log entries partial for this exercise.
Args:
request: The incoming HTTP request.
session: Database session.
admin: The authenticated admin user.
Returns:
Rendered log entries partial for this exercise.
"""
form = await request.form()
exercise_id = int(form.get("exercise_id", 0))
@@ -49,18 +41,10 @@ async def log_set(
weight = form.get("weight", "")
felt_easy = form.get("felt_easy") == "on"
active_profile_id = get_active_profile_id(request)
if not active_profile_id:
templates = request.app.state.templates
return templates.TemplateResponse("partials/flash_message.html", {
"request": request,
"flash_error": "No profile selected. Switch profiles first.",
})
# Get or create today's session
ws_service = WorkoutSessionService(session)
ws = ws_service.get_or_create_session(
user_id=active_profile_id,
user_id=profile.id,
workout_day_id=workout_day_id,
session_date=date.today(),
)
@@ -96,19 +80,9 @@ async def edit_log(
log_id: int,
request: Request,
session: Session = Depends(get_db_session),
admin: User = Depends(get_current_admin_user),
profile: User = Depends(require_active_profile),
):
"""Edit an existing log entry.
Args:
log_id: The log entry ID.
request: The incoming HTTP request.
session: Database session.
admin: The authenticated admin user.
Returns:
Rendered updated log entry partial.
"""
"""Edit an existing log entry."""
form = await request.form()
log_service = LogService(session)
@@ -140,19 +114,9 @@ async def delete_log(
log_id: int,
request: Request,
session: Session = Depends(get_db_session),
admin: User = Depends(get_current_admin_user),
profile: User = Depends(require_active_profile),
):
"""Delete a log entry.
Args:
log_id: The log entry ID.
request: The incoming HTTP request.
session: Database session.
admin: The authenticated admin user.
Returns:
Rendered updated log entries partial.
"""
"""Delete a log entry."""
log_service = LogService(session)
log = log_service.get_log_by_id(log_id)

View File

@@ -3,21 +3,27 @@
Renders Jinja2 templates for user-facing pages.
"""
from fastapi import APIRouter, Request
from fastapi import APIRouter, Depends, Request
from fastapi.responses import HTMLResponse
from sqlmodel import Session
from app.database import get_db_session
from app.services.user_service import UserService
router = APIRouter(tags=["pages"])
@router.get("/")
async def home_page(request: Request) -> HTMLResponse:
"""Render the home page.
async def home_page(
request: Request,
session: Session = Depends(get_db_session),
) -> HTMLResponse:
"""Render the home page with profile picker or create-profile link."""
user_service = UserService(session)
profiles = user_service.list_users()
Args:
request: The incoming HTTP request.
Returns:
Rendered home page HTML.
"""
templates = request.app.state.templates
return templates.TemplateResponse(request, "pages/home.html")
return templates.TemplateResponse("pages/home.html", {
"request": request,
"profiles": profiles,
})

View File

@@ -1,6 +1,6 @@
"""Profile management routes.
Admin can view, create, edit user profiles and switch the active profile.
Users can view, create, edit profiles and switch the active profile.
"""
import structlog
@@ -11,7 +11,7 @@ from sqlmodel import Session
from app.database import get_db_session
from app.models.user import User
from app.services.user_service import UserService
from app.utils.auth import get_current_admin_user, get_active_profile_id
from app.utils.auth import require_active_profile, get_active_profile_id
logger = structlog.get_logger(__name__)
@@ -22,20 +22,11 @@ router = APIRouter(prefix="/profiles", tags=["profiles"])
async def list_profiles(
request: Request,
session: Session = Depends(get_db_session),
admin: User = Depends(get_current_admin_user),
profile: User = Depends(require_active_profile),
):
"""List all user profiles for the admin.
Args:
request: The incoming HTTP request.
session: Database session.
admin: The authenticated admin user.
Returns:
Rendered profile list page.
"""
"""List all user profiles."""
user_service = UserService(session)
profiles = user_service.list_users(exclude_admin=True)
profiles = user_service.list_users()
active_profile_id = get_active_profile_id(request)
templates = request.app.state.templates
@@ -43,7 +34,6 @@ async def list_profiles(
"request": request,
"profiles": profiles,
"active_profile_id": active_profile_id,
"admin": admin,
})
@@ -51,63 +41,101 @@ async def list_profiles(
async def switch_profile(
request: Request,
session: Session = Depends(get_db_session),
admin: User = Depends(get_current_admin_user),
):
"""Switch the active user profile.
Sets a cookie with the selected profile ID.
Args:
request: The incoming HTTP request.
session: Database session.
admin: The authenticated admin user.
Returns:
Redirect to the referring page with profile cookie set.
Sets a cookie with the selected profile ID (1 year expiry).
"""
form = await request.form()
profile_id = form.get("profile_id", "")
# Validate profile exists and is not an admin
user_service = UserService(session)
profile = user_service.get_user_by_id(int(profile_id)) if profile_id.isdigit() else None
referer = request.headers.get("referer", "/")
response = RedirectResponse(url=referer, status_code=303)
response = RedirectResponse(url="/workouts", status_code=303)
if profile and not profile.is_admin:
if profile:
response.set_cookie(
key="active_profile_id",
value=str(profile.id),
httponly=True,
samesite="lax",
max_age=86400,
max_age=31536000, # 1 year
)
logger.info("profile_switched", profile_id=profile.id, name=profile.display_name)
else:
logger.warning("profile_switch_failed", profile_id=profile_id)
response = RedirectResponse(url="/", status_code=303)
return response
@router.get("/create", response_class=HTMLResponse)
async def create_profile_form(request: Request):
"""Render the create-profile form."""
templates = request.app.state.templates
return templates.TemplateResponse("pages/profile_create.html", {
"request": request,
})
@router.post("/create")
async def create_profile(
request: Request,
session: Session = Depends(get_db_session),
):
"""Create a new profile, set cookie, redirect to workouts."""
form = await request.form()
display_name = form.get("display_name", "").strip()
if not display_name:
templates = request.app.state.templates
return templates.TemplateResponse("pages/profile_create.html", {
"request": request,
"error": "Display name is required.",
})
user_service = UserService(session)
# Generate username from display name
username = display_name.lower().replace(" ", "_")
# Ensure uniqueness
existing = user_service.get_user_by_username(username)
if existing:
templates = request.app.state.templates
return templates.TemplateResponse("pages/profile_create.html", {
"request": request,
"error": "A profile with that name already exists.",
})
profile = user_service.create_user(
username=username,
display_name=display_name,
height=form.get("height", "").strip() or None,
weight=form.get("weight", "").strip() or None,
goals=form.get("goals", "").strip() or None,
)
response = RedirectResponse(url="/workouts", status_code=303)
response.set_cookie(
key="active_profile_id",
value=str(profile.id),
httponly=True,
samesite="lax",
max_age=31536000, # 1 year
)
logger.info("profile_created", profile_id=profile.id, name=profile.display_name)
return response
@router.get("/{profile_id}/edit", response_class=HTMLResponse)
async def edit_profile_page(
profile_id: int,
request: Request,
session: Session = Depends(get_db_session),
admin: User = Depends(get_current_admin_user),
active_profile: User = Depends(require_active_profile),
):
"""Render the profile edit form.
Args:
profile_id: The profile ID to edit.
request: The incoming HTTP request.
session: Database session.
admin: The authenticated admin user.
Returns:
Rendered profile edit page.
"""
"""Render the profile edit form."""
user_service = UserService(session)
profile = user_service.get_user_by_id(profile_id)
@@ -115,7 +143,6 @@ async def edit_profile_page(
return templates.TemplateResponse("pages/profile_edit.html", {
"request": request,
"profile": profile,
"admin": admin,
})
@@ -124,19 +151,9 @@ async def update_profile(
profile_id: int,
request: Request,
session: Session = Depends(get_db_session),
admin: User = Depends(get_current_admin_user),
active_profile: User = Depends(require_active_profile),
):
"""Process profile edit form submission.
Args:
profile_id: The profile ID to update.
request: The incoming HTTP request.
session: Database session.
admin: The authenticated admin user.
Returns:
Redirect to profiles page.
"""
"""Process profile edit form submission."""
form = await request.form()
user_service = UserService(session)

View File

@@ -14,7 +14,7 @@ from app.database import get_db_session
from app.models.user import User
from app.services.exercise_service import ExerciseService
from app.services.workout_session_service import WorkoutSessionService
from app.utils.auth import get_current_admin_user, get_active_profile_id
from app.utils.auth import require_active_profile
logger = structlog.get_logger(__name__)
@@ -25,20 +25,13 @@ router = APIRouter(prefix="/schedule", tags=["schedule"])
async def schedule_view(
request: Request,
session: Session = Depends(get_db_session),
admin: User = Depends(get_current_admin_user),
profile: User = Depends(require_active_profile),
):
"""Render the 4-week schedule calendar.
Shows a 4-week grid where each training day is mapped to a
calendar date. Days with completed sessions are highlighted.
"""
active_profile_id = get_active_profile_id(request)
active_profile = (
session.get(User, active_profile_id)
if active_profile_id
else None
)
exercise_service = ExerciseService(session)
workout_days = exercise_service.list_workout_days()
@@ -50,12 +43,11 @@ async def schedule_view(
completed_dates = set()
# Get completed sessions for highlighting
if active_profile_id:
ws_service = WorkoutSessionService(session)
sessions_list = ws_service.list_sessions(
user_id=active_profile_id, limit=100,
)
completed_dates = {ws.date for ws in sessions_list}
ws_service = WorkoutSessionService(session)
sessions_list = ws_service.list_sessions(
user_id=profile.id, limit=100,
)
completed_dates = {ws.date for ws in sessions_list}
# 4 workout days per week, 4 weeks
for week_num in range(4):
@@ -81,6 +73,5 @@ async def schedule_view(
return templates.TemplateResponse("pages/schedule.html", {
"request": request,
"weeks": weeks,
"active_profile": active_profile,
"admin": admin,
"active_profile": profile,
})

View File

@@ -18,7 +18,7 @@ from app.services.exercise_service import ExerciseService
from app.services.log_service import LogService
from app.services.progression_service import ProgressionService
from app.services.workout_session_service import WorkoutSessionService
from app.utils.auth import get_current_admin_user, get_active_profile_id
from app.utils.auth import require_active_profile, get_active_profile_id
logger = structlog.get_logger(__name__)
@@ -29,18 +29,9 @@ router = APIRouter(prefix="/workouts", tags=["workouts"])
async def workout_days_list(
request: Request,
session: Session = Depends(get_db_session),
admin: User = Depends(get_current_admin_user),
profile: User = Depends(require_active_profile),
):
"""List all workout days as clickable cards.
Args:
request: The incoming HTTP request.
session: Database session.
admin: The authenticated admin user.
Returns:
Rendered workout days list page.
"""
"""List all workout days as clickable cards."""
exercise_service = ExerciseService(session)
days = exercise_service.list_workout_days()
@@ -48,7 +39,6 @@ async def workout_days_list(
return templates.TemplateResponse("pages/workout_days.html", {
"request": request,
"days": days,
"admin": admin,
})
@@ -57,19 +47,9 @@ async def workout_day_detail(
day_name: str,
request: Request,
session: Session = Depends(get_db_session),
admin: User = Depends(get_current_admin_user),
profile: User = Depends(require_active_profile),
):
"""Display a full workout day -- warmups + exercises with form cues.
Args:
day_name: The workout day name (e.g., "push", "pull").
request: The incoming HTTP request.
session: Database session.
admin: The authenticated admin user.
Returns:
Rendered workout day detail page.
"""
"""Display a full workout day -- warmups + exercises with form cues."""
exercise_service = ExerciseService(session)
# Normalize day name for DB lookup (e.g., "push" -> "Push", "full-body" -> "Full Body")
@@ -78,19 +58,15 @@ async def workout_day_detail(
warmups = exercise_service.list_warmups()
exercises = exercise_service.list_exercises(workout_day=day_display)
# Get active profile's programming if set
active_profile_id = get_active_profile_id(request)
# Get active profile's programming
active_profile_id = profile.id
programs = {}
active_profile = None
existing_logs = {}
if active_profile_id:
active_profile = session.get(User, active_profile_id)
if active_profile:
statement = select(UserExerciseProgram).where(
UserExerciseProgram.user_id == active_profile_id
)
for prog in session.exec(statement).all():
programs[prog.exercise_id] = prog
statement = select(UserExerciseProgram).where(
UserExerciseProgram.user_id == active_profile_id
)
for prog in session.exec(statement).all():
programs[prog.exercise_id] = prog
# Look up the workout day ID for logging forms
days = exercise_service.list_workout_days()
@@ -102,15 +78,14 @@ async def workout_day_detail(
# Get progression suggestions for each exercise
suggestions = {}
if active_profile_id:
progression = ProgressionService(session)
for exercise in exercises:
suggestions[exercise.id] = progression.get_suggestion(
active_profile_id, exercise.id,
)
progression = ProgressionService(session)
for exercise in exercises:
suggestions[exercise.id] = progression.get_suggestion(
active_profile_id, exercise.id,
)
# Load existing logs for today's session (if any)
if active_profile_id and workout_day_id:
if workout_day_id:
ws_service = WorkoutSessionService(session)
ws = ws_service.get_or_create_session(
user_id=active_profile_id,
@@ -129,9 +104,8 @@ async def workout_day_detail(
"warmups": warmups,
"exercises": exercises,
"programs": programs,
"active_profile": active_profile,
"active_profile": profile,
"existing_logs": existing_logs,
"suggestions": suggestions,
"workout_day_id": workout_day_id,
"admin": admin,
})

View File

@@ -1,100 +0,0 @@
"""Service layer for admin authentication and session management.
Handles password verification, session token creation/validation.
Uses bcrypt for password hashing and itsdangerous for session signing.
"""
from typing import Optional
import bcrypt
import structlog
from itsdangerous import BadSignature, URLSafeTimedSerializer
from sqlmodel import Session, select
from app.models.user import User
logger = structlog.get_logger(__name__)
# Session token max age: 24 hours
SESSION_MAX_AGE_SECONDS = 86400
class AuthService:
"""Handles admin authentication and session token management.
Args:
session: An active SQLModel Session.
secret_key: Secret key for signing session tokens.
"""
def __init__(self, session: Session, secret_key: str) -> None:
self._session = session
self._serializer = URLSafeTimedSerializer(secret_key)
def authenticate(self, username: str, password: str) -> Optional[User]:
"""Verify admin credentials and return the User if valid.
Only admin users can authenticate. Non-admin users are rejected.
Args:
username: The username to check.
password: The plaintext password to verify.
Returns:
The authenticated User, or None if credentials are invalid.
"""
statement = select(User).where(User.username == username)
user = self._session.exec(statement).first()
if user is None:
logger.warning("auth_failed", reason="user_not_found", username=username)
return None
if not user.is_admin:
logger.warning("auth_failed", reason="not_admin", username=username)
return None
if not user.password_hash:
logger.warning("auth_failed", reason="no_password_hash", username=username)
return None
# Verify password against bcrypt hash
if not bcrypt.checkpw(
password.encode("utf-8"),
user.password_hash.encode("utf-8"),
):
logger.warning("auth_failed", reason="wrong_password", username=username)
return None
logger.info("auth_success", username=username)
return user
def create_session_token(self, user_id: int) -> str:
"""Create a signed session token for the given user.
Args:
user_id: The authenticated user's ID.
Returns:
A signed, URL-safe token string.
"""
return self._serializer.dumps({"user_id": user_id})
def validate_session_token(self, token: str) -> Optional[int]:
"""Validate a session token and extract the user_id.
Args:
token: The session token to validate.
Returns:
The user_id if the token is valid, or None.
"""
try:
data = self._serializer.loads(token, max_age=SESSION_MAX_AGE_SECONDS)
return data.get("user_id")
except BadSignature:
logger.warning("session_invalid", reason="bad_signature")
return None
except Exception:
logger.warning("session_invalid", reason="unknown_error")
return None

View File

@@ -4,11 +4,9 @@ Reads config/exercises.yaml and config/user_programs.yaml to populate
the exercise library, warmups, workout days, and user programs.
"""
import os
from pathlib import Path
from typing import Optional
import bcrypt
import structlog
import yaml
from sqlmodel import Session, select
@@ -43,14 +41,7 @@ class SeedService:
self._config_dir = config_dir or Path(__file__).resolve().parent.parent.parent / "config"
def _load_yaml(self, filename: str) -> dict:
"""Load and parse a YAML file from the config directory.
Args:
filename: Name of the YAML file to load.
Returns:
Parsed YAML content as a dictionary.
"""
"""Load and parse a YAML file from the config directory."""
filepath = self._config_dir / filename
with open(filepath, "r") as f:
return yaml.safe_load(f)
@@ -111,48 +102,8 @@ class SeedService:
self._session.commit()
logger.info("seed_complete", table="warmups", count=len(warmups))
def seed_admin(self) -> None:
"""Create admin user from environment variables if not exists.
Reads ADMIN_USERNAME and ADMIN_PASSWORD from env,
hashes the password with bcrypt, and creates the admin user.
"""
admin_username = os.environ.get("ADMIN_USERNAME", "admin")
admin_password = os.environ.get("ADMIN_PASSWORD", "")
existing = self._session.exec(
select(User).where(User.username == admin_username)
).first()
if existing:
logger.info("seed_skipped", table="users", reason="admin already exists")
return
if not admin_password:
logger.warning("seed_skipped", table="users", reason="ADMIN_PASSWORD not set")
return
# Hash password with bcrypt
password_hash = bcrypt.hashpw(
admin_password.encode("utf-8"),
bcrypt.gensalt(),
).decode("utf-8")
admin = User(
username=admin_username,
password_hash=password_hash,
display_name="Admin",
is_admin=True,
)
self._session.add(admin)
self._session.commit()
logger.info("seed_complete", table="users", user="admin")
def seed_user_programs(self) -> None:
"""Seed user profiles and their exercise programs from user_programs.yaml.
Creates non-admin user profiles and links them to exercises
with week 1/4 rep and weight targets.
"""
"""Seed user profiles and their exercise programs from user_programs.yaml."""
data = self._load_yaml("user_programs.yaml")
programs = data.get("programs", [])
@@ -172,12 +123,10 @@ class SeedService:
else:
user = User(
username=username,
password_hash="", # Non-admin users don't log in initially
display_name=display_name,
height=profile.get("height", ""),
weight=profile.get("weight", ""),
goals=profile.get("goals", ""),
is_admin=False,
)
self._session.add(user)
self._session.commit()
@@ -219,15 +168,10 @@ class SeedService:
logger.info("seed_complete", table="user_exercise_programs", user=username)
def seed_all(self) -> None:
"""Run all seed operations in the correct order.
Order matters: workout_days and exercises must exist before
user_programs can reference them.
"""
"""Run all seed operations in the correct order."""
logger.info("seed_all_started")
self.seed_workout_days()
self.seed_exercises()
self.seed_warmups()
self.seed_admin()
self.seed_user_programs()
logger.info("seed_all_complete")

View File

@@ -27,77 +27,48 @@ class UserService:
def create_user(
self,
username: str,
password_hash: str,
display_name: str,
height: Optional[str] = None,
weight: Optional[str] = None,
goals: Optional[str] = None,
is_admin: bool = False,
) -> User:
"""Create a new user profile.
Args:
username: Unique login identifier.
password_hash: Pre-hashed password string.
username: Unique identifier.
display_name: Human-readable name.
height: User height as string.
weight: User weight as string.
goals: Free-text goals.
is_admin: Whether user has admin privileges.
Returns:
The newly created User record.
"""
user = User(
username=username,
password_hash=password_hash,
display_name=display_name,
height=height,
weight=weight,
goals=goals,
is_admin=is_admin,
)
self._session.add(user)
self._session.commit()
self._session.refresh(user)
logger.info("user_created", username=username, is_admin=is_admin)
logger.info("user_created", username=username)
return user
def get_user_by_id(self, user_id: int) -> Optional[User]:
"""Retrieve a user by primary key.
Args:
user_id: The user's ID.
Returns:
The User record, or None if not found.
"""
"""Retrieve a user by primary key."""
return self._session.get(User, user_id)
def get_user_by_username(self, username: str) -> Optional[User]:
"""Retrieve a user by username.
Args:
username: The username to look up.
Returns:
The User record, or None if not found.
"""
"""Retrieve a user by username."""
statement = select(User).where(User.username == username)
return self._session.exec(statement).first()
def list_users(self, exclude_admin: bool = False) -> list[User]:
"""List all user profiles.
Args:
exclude_admin: If True, omit admin users from the result.
Returns:
List of User records.
"""
def list_users(self) -> list[User]:
"""List all user profiles."""
statement = select(User)
if exclude_admin:
statement = statement.where(User.is_admin == False) # noqa: E712
return list(self._session.exec(statement).all())
def update_user(self, user_id: int, **kwargs) -> User:

View File

@@ -8,5 +8,33 @@
<p>Your open-source workout tracker</p>
</hgroup>
<p>Welcome to SneakySwole. Get started by logging in.</p>
{% if profiles %}
<h2>Select Profile</h2>
<div class="grid">
{% for profile in profiles %}
<article>
<header>
<strong>{{ profile.display_name }}</strong>
</header>
{% if profile.height or profile.weight %}
<p>
{% if profile.height %}Height: {{ profile.height }}{% endif %}
{% if profile.height and profile.weight %} &middot; {% endif %}
{% if profile.weight %}Weight: {{ profile.weight }}{% endif %}
</p>
{% endif %}
<footer>
<form method="POST" action="/profiles/switch">
<input type="hidden" name="profile_id" value="{{ profile.id }}">
<button type="submit" class="contrast">Select</button>
</form>
</footer>
</article>
{% endfor %}
</div>
{% else %}
<p>No profiles yet. Create one to get started.</p>
{% endif %}
<a href="/profiles/create" role="button" class="secondary">Create New Profile</a>
{% endblock %}

View File

@@ -1,27 +0,0 @@
{% extends "base.html" %}
{% block title %}SneakySwole — Login{% endblock %}
{% block content %}
<article>
<header>
<h2>Admin Login</h2>
</header>
{% if error %}
<div class="flash-error" role="alert">{{ error }}</div>
{% endif %}
<form method="POST" action="/login">
<label for="username">Username</label>
<input type="text" id="username" name="username"
placeholder="Enter username" required autofocus>
<label for="password">Password</label>
<input type="password" id="password" name="password"
placeholder="Enter password" required>
<button type="submit">Login</button>
</form>
</article>
{% endblock %}

View File

@@ -0,0 +1,46 @@
{% extends "base.html" %}
{% block title %}SneakySwole — Create Profile{% endblock %}
{% block content %}
<hgroup>
<h1>Create Profile</h1>
<p>Set up a new workout profile</p>
</hgroup>
{% if error %}
<article aria-label="Error" class="pico-background-red-500">
<p>{{ error }}</p>
</article>
{% endif %}
<form method="POST" action="/profiles/create">
<label for="display_name">
Display Name <span aria-hidden="true">*</span>
<input type="text" id="display_name" name="display_name" required
placeholder="e.g. Phillip" value="{{ request.query_params.get('display_name', '') }}">
</label>
<label for="height">
Height
<input type="text" id="height" name="height"
placeholder="e.g. 6'0&quot;">
</label>
<label for="weight">
Weight
<input type="text" id="weight" name="weight"
placeholder="e.g. 260 lbs">
</label>
<label for="goals">
Goals
<textarea id="goals" name="goals" rows="3"
placeholder="e.g. Build strength, improve mobility"></textarea>
</label>
<button type="submit">Create Profile</button>
</form>
<p><a href="/">&larr; Back to profiles</a></p>
{% endblock %}

View File

@@ -1,7 +1,5 @@
{% set admin = request.state.admin %}
{% set profiles = request.state.profiles %}
{% set active_profile = request.state.active_profile %}
{% if admin %}
<li>
<details class="dropdown">
<summary>
@@ -26,13 +24,10 @@
</ul>
</details>
</li>
<li><a href="/">Home</a></li>
<li><a href="/workouts">Workouts</a></li>
<li><a href="/schedule">Schedule</a></li>
<li><a href="/dashboard">Dashboard</a></li>
<li><a href="/history">History</a></li>
<li><a href="/exercises">Exercises</a></li>
<li><a href="/profiles">Profiles</a></li>
<li><a href="/logout">Logout</a></li>
{% else %}
<li><a href="/login">Login</a></li>
{% endif %}

View File

@@ -1,69 +1,27 @@
"""Authentication dependencies for FastAPI route protection.
"""Profile selection utilities for FastAPI route protection.
Provides dependency functions that verify the admin session cookie
and return the authenticated User, or redirect to /login.
Provides dependency functions that check the active_profile_id cookie
and return the selected User profile, or redirect to /.
"""
from typing import Optional
import structlog
from fastapi import Depends, Request
from fastapi.responses import RedirectResponse
from sqlmodel import Session
from app.database import get_db_session
from app.models.user import User
from app.services.auth_service import AuthService
logger = structlog.get_logger(__name__)
class NotAuthenticatedError(Exception):
"""Raised when a request lacks valid authentication."""
# Cookie name for the admin session
SESSION_COOKIE_NAME = "session"
def get_current_admin_user(request: Request, session: Session = Depends(get_db_session)) -> User:
"""FastAPI dependency that extracts and validates the admin session.
Reads the session cookie, validates the token, and returns the
authenticated admin User. Redirects to /login (303) if not authenticated.
Args:
request: The incoming HTTP request.
session: Database session (injected by FastAPI).
Returns:
The authenticated admin User.
Raises:
RedirectResponse: 303 redirect to /login if no valid admin session.
"""
token = request.cookies.get(SESSION_COOKIE_NAME)
if not token:
raise _login_redirect()
secret_key = getattr(request.app.state, "secret_key", "")
auth_service = AuthService(session, secret_key=secret_key)
user_id = auth_service.validate_session_token(token)
if user_id is None:
raise _login_redirect()
user = session.get(User, user_id)
if user is None or not user.is_admin:
raise _login_redirect()
return user
class NoProfileSelectedError(Exception):
"""Raised when a request lacks a valid profile selection."""
def get_active_profile_id(request: Request) -> Optional[int]:
"""Extract the active profile ID from the session cookie.
The admin selects which user profile to view/log as. This is stored
in a separate cookie called 'active_profile_id'.
"""Extract the active profile ID from the cookie.
Args:
request: The incoming HTTP request.
@@ -77,11 +35,28 @@ def get_active_profile_id(request: Request) -> Optional[int]:
return None
def _login_redirect():
"""Create a redirect exception to the login page.
def require_active_profile(request: Request, session: Session = Depends(get_db_session)) -> User:
"""FastAPI dependency that requires a valid profile selection.
Reads the active_profile_id cookie, loads the profile from DB,
and raises NoProfileSelectedError if missing or invalid.
Args:
request: The incoming HTTP request.
session: Database session (injected by FastAPI).
Returns:
A NotAuthenticatedError handled by a registered exception handler
in main.py that sends a 302 redirect to /login.
The selected User profile.
Raises:
NoProfileSelectedError: If no valid profile is selected.
"""
return NotAuthenticatedError()
profile_id = get_active_profile_id(request)
if profile_id is None:
raise NoProfileSelectedError()
user = session.get(User, profile_id)
if user is None:
raise NoProfileSelectedError()
return user

View File

@@ -3,8 +3,6 @@ uvicorn[standard]>=0.34.0,<1.0.0
jinja2>=3.1.0,<4.0.0
structlog>=24.0.0,<25.0.0
python-dotenv>=1.0.0,<2.0.0
bcrypt>=4.2.0,<5.0.0
itsdangerous>=2.2.0,<3.0.0
python-multipart>=0.0.20,<1.0.0
pyyaml>=6.0.0,<7.0.0
sqlmodel>=0.0.22,<1.0.0

View File

@@ -11,8 +11,6 @@ from fastapi.testclient import TestClient
def _set_test_env() -> None:
"""Ensure required env vars are set for all tests."""
with patch.dict(os.environ, {
"ADMIN_USERNAME": "testadmin",
"ADMIN_PASSWORD": "testpass123",
"APP_ENV": "development",
"DATABASE_URL": "sqlite:///data/test_sneakyswole.db",
}, clear=False):
@@ -26,3 +24,8 @@ def client() -> TestClient:
app = create_app()
return TestClient(app)
def set_profile_cookie(client: TestClient, profile_id: int) -> None:
"""Helper to set the active_profile_id cookie on a test client."""
client.cookies.set("active_profile_id", str(profile_id))

View File

@@ -21,7 +21,7 @@ class TestAnalyticsService:
SQLModel.metadata.create_all(engine)
session = Session(engine)
user = User(username="phil", password_hash="h", display_name="Phillip")
user = User(username="phil", display_name="Phillip")
day = WorkoutDay(name="Push", day_number=1, description="Push day")
exercise = Exercise(
name="DB Chest Press", muscle_group="Chest",

View File

@@ -1,56 +0,0 @@
"""Tests for the auth dependency (require_admin)."""
from unittest.mock import MagicMock
import pytest
from fastapi import HTTPException
from app.utils.auth import get_current_admin_user, get_active_profile_id
class TestAuthDependency:
"""Tests for the require_admin dependency."""
def test_redirects_when_no_session_cookie(self) -> None:
"""Should redirect to /login (303) when no session cookie is present."""
request = MagicMock()
request.cookies = {}
with pytest.raises(HTTPException) as exc_info:
get_current_admin_user(request=request, session=MagicMock())
assert exc_info.value.status_code == 303
def test_redirects_when_invalid_token(self) -> None:
"""Should redirect to /login (303) when session cookie has invalid token."""
request = MagicMock()
request.cookies = {"session": "invalid-token"}
request.app.state.secret_key = "test-secret"
mock_session = MagicMock()
mock_session.get.return_value = None
with pytest.raises(HTTPException) as exc_info:
get_current_admin_user(request=request, session=mock_session)
assert exc_info.value.status_code == 303
class TestGetActiveProfileId:
"""Tests for the get_active_profile_id dependency."""
def test_returns_profile_id_from_cookie(self) -> None:
"""Should return the integer profile ID from cookie."""
request = MagicMock()
request.cookies = {"active_profile_id": "5"}
assert get_active_profile_id(request) == 5
def test_returns_none_when_no_cookie(self) -> None:
"""Should return None when no active_profile_id cookie is set."""
request = MagicMock()
request.cookies = {}
assert get_active_profile_id(request) is None
def test_returns_none_for_non_numeric(self) -> None:
"""Should return None for non-numeric cookie values."""
request = MagicMock()
request.cookies = {"active_profile_id": "abc"}
assert get_active_profile_id(request) is None

View File

@@ -1,43 +0,0 @@
"""Tests for login/logout routes."""
from fastapi.testclient import TestClient
class TestLoginPage:
"""Tests for GET /login."""
def test_login_page_returns_200(self, client: TestClient) -> None:
"""GET /login should render the login form."""
response = client.get("/login")
assert response.status_code == 200
assert "Login" in response.text
def test_login_page_has_form(self, client: TestClient) -> None:
"""GET /login should contain a login form with username and password."""
response = client.get("/login")
assert 'name="username"' in response.text
assert 'name="password"' in response.text
class TestLoginPost:
"""Tests for POST /login."""
def test_login_invalid_credentials_shows_error(self, client: TestClient) -> None:
"""POST /login with wrong credentials should show error message."""
response = client.post(
"/login",
data={"username": "wrong", "password": "wrong"},
follow_redirects=False,
)
# Should re-render login page with error or redirect back
assert response.status_code in (200, 303)
class TestLogout:
"""Tests for GET /logout."""
def test_logout_redirects_to_login(self, client: TestClient) -> None:
"""GET /logout should clear session and redirect to login."""
response = client.get("/logout", follow_redirects=False)
assert response.status_code in (303, 307)
assert "/login" in response.headers.get("location", "")

View File

@@ -1,93 +0,0 @@
"""Tests for the AuthService class."""
import bcrypt
from sqlmodel import SQLModel, Session, create_engine
from app.models.user import User
from app.services.auth_service import AuthService
class TestAuthService:
"""Tests for admin authentication logic."""
def _setup(self):
"""Create an in-memory DB with an admin user."""
engine = create_engine("sqlite:///:memory:")
SQLModel.metadata.create_all(engine)
session = Session(engine)
# Create admin user with known password
pw_hash = bcrypt.hashpw(b"adminpass", bcrypt.gensalt()).decode("utf-8")
admin = User(
username="admin",
password_hash=pw_hash,
display_name="Admin",
is_admin=True,
)
session.add(admin)
session.commit()
service = AuthService(session, secret_key="test-secret-key")
return session, service
def test_authenticate_valid_credentials(self) -> None:
"""authenticate should return the User for valid credentials."""
session, service = self._setup()
user = service.authenticate("admin", "adminpass")
assert user is not None
assert user.username == "admin"
session.close()
def test_authenticate_wrong_password(self) -> None:
"""authenticate should return None for wrong password."""
session, service = self._setup()
user = service.authenticate("admin", "wrongpass")
assert user is None
session.close()
def test_authenticate_unknown_user(self) -> None:
"""authenticate should return None for unknown username."""
session, service = self._setup()
user = service.authenticate("nobody", "anything")
assert user is None
session.close()
def test_authenticate_non_admin(self) -> None:
"""authenticate should return None for non-admin users."""
session, service = self._setup()
pw_hash = bcrypt.hashpw(b"userpass", bcrypt.gensalt()).decode("utf-8")
non_admin = User(
username="regular",
password_hash=pw_hash,
display_name="Regular",
is_admin=False,
)
session.add(non_admin)
session.commit()
user = service.authenticate("regular", "userpass")
assert user is None
session.close()
def test_create_session_token(self) -> None:
"""create_session_token should return a non-empty signed string."""
session, service = self._setup()
token = service.create_session_token(user_id=1)
assert isinstance(token, str)
assert len(token) > 0
session.close()
def test_validate_session_token(self) -> None:
"""validate_session_token should return the user_id from a valid token."""
session, service = self._setup()
token = service.create_session_token(user_id=42)
user_id = service.validate_session_token(token)
assert user_id == 42
session.close()
def test_validate_session_token_invalid(self) -> None:
"""validate_session_token should return None for tampered tokens."""
session, service = self._setup()
user_id = service.validate_session_token("fake-token-value")
assert user_id is None
session.close()

View File

@@ -11,37 +11,17 @@ class TestSettings:
def test_settings_loads_defaults(self) -> None:
"""Settings should have sensible defaults for all fields."""
env = {
"ADMIN_USERNAME": "testadmin",
"ADMIN_PASSWORD": "testpass123",
}
# Remove DATABASE_URL so we test the actual default
env_clean = {k: v for k, v in os.environ.items() if k != "DATABASE_URL"}
env_clean.update(env)
with patch.dict(os.environ, env_clean, clear=True):
settings = Settings()
assert settings.admin_username == "testadmin"
assert settings.admin_password == "testpass123"
assert settings.app_env == "development"
assert settings.app_host == "0.0.0.0"
assert settings.app_port == 8000
assert settings.database_url == "sqlite:///data/sneakyswole.db"
def test_settings_requires_admin_username(self) -> None:
"""Settings should require ADMIN_USERNAME to be set."""
with patch.dict(os.environ, {"ADMIN_PASSWORD": "testpass"}, clear=True):
try:
Settings()
assert False, "Should have raised an error"
except Exception:
pass
def test_get_settings_returns_singleton(self) -> None:
"""get_settings should return the same instance on repeated calls."""
with patch.dict(os.environ, {
"ADMIN_USERNAME": "admin",
"ADMIN_PASSWORD": "pass",
}, clear=False):
s1 = get_settings()
s2 = get_settings()
assert s1 is s2
s1 = get_settings()
s2 = get_settings()
assert s1 is s2

View File

@@ -6,16 +6,18 @@ from fastapi.testclient import TestClient
class TestDashboard:
"""Tests for GET /dashboard."""
def test_dashboard_requires_auth(self, client: TestClient) -> None:
"""GET /dashboard should require admin login."""
def test_dashboard_requires_profile(self, client: TestClient) -> None:
"""GET /dashboard should redirect to / without profile cookie."""
response = client.get("/dashboard", follow_redirects=False)
assert response.status_code in (401, 303)
assert response.status_code == 302
assert response.headers["location"] == "/"
class TestExerciseProgress:
"""Tests for GET /dashboard/exercise/<id>."""
def test_exercise_progress_requires_auth(self, client: TestClient) -> None:
"""GET /dashboard/exercise/1 should require admin login."""
def test_exercise_progress_requires_profile(self, client: TestClient) -> None:
"""GET /dashboard/exercise/1 should redirect to / without profile cookie."""
response = client.get("/dashboard/exercise/1", follow_redirects=False)
assert response.status_code in (401, 303)
assert response.status_code == 302
assert response.headers["location"] == "/"

View File

@@ -6,19 +6,21 @@ from fastapi.testclient import TestClient
class TestExerciseBrowser:
"""Tests for GET /exercises."""
def test_exercise_browser_requires_auth(self, client: TestClient) -> None:
"""GET /exercises should require admin login."""
def test_exercise_browser_requires_profile(self, client: TestClient) -> None:
"""GET /exercises should redirect to / without profile cookie."""
response = client.get("/exercises", follow_redirects=False)
assert response.status_code in (401, 303)
assert response.status_code == 302
assert response.headers["location"] == "/"
class TestExerciseSearch:
"""Tests for HTMX exercise search."""
def test_exercise_search_requires_auth(self, client: TestClient) -> None:
"""GET /exercises/search should require admin login."""
def test_exercise_search_requires_profile(self, client: TestClient) -> None:
"""GET /exercises/search should redirect to / without profile cookie."""
response = client.get(
"/exercises/search?workout_day=Push",
follow_redirects=False,
)
assert response.status_code in (401, 303)
assert response.status_code == 302
assert response.headers["location"] == "/"

View File

@@ -6,16 +6,18 @@ from fastapi.testclient import TestClient
class TestLogHistory:
"""Tests for GET /history."""
def test_history_requires_auth(self, client: TestClient) -> None:
"""GET /history should require admin login."""
def test_history_requires_profile(self, client: TestClient) -> None:
"""GET /history should redirect to / without profile cookie."""
response = client.get("/history", follow_redirects=False)
assert response.status_code in (401, 303)
assert response.status_code == 302
assert response.headers["location"] == "/"
class TestSessionDetail:
"""Tests for GET /history/<session_id>."""
def test_session_detail_requires_auth(self, client: TestClient) -> None:
"""GET /history/1 should require admin login."""
def test_session_detail_requires_profile(self, client: TestClient) -> None:
"""GET /history/1 should redirect to / without profile cookie."""
response = client.get("/history/1", follow_redirects=False)
assert response.status_code in (401, 303)
assert response.status_code == 302
assert response.headers["location"] == "/"

View File

@@ -21,7 +21,7 @@ class TestLogService:
SQLModel.metadata.create_all(engine)
session = Session(engine)
user = User(username="phil", password_hash="h", display_name="Phillip")
user = User(username="phil", display_name="Phillip")
day = WorkoutDay(name="Push", day_number=1, description="Push day")
exercise = Exercise(
name="DB Chest Press", muscle_group="Chest",

View File

@@ -6,8 +6,8 @@ from fastapi.testclient import TestClient
class TestLogSet:
"""Tests for POST /log."""
def test_log_set_requires_auth(self, client: TestClient) -> None:
"""POST /log should require admin login."""
def test_log_set_requires_profile(self, client: TestClient) -> None:
"""POST /log should redirect to / without profile cookie."""
response = client.post(
"/log",
data={
@@ -19,26 +19,29 @@ class TestLogSet:
},
follow_redirects=False,
)
assert response.status_code in (401, 303)
assert response.status_code == 302
assert response.headers["location"] == "/"
class TestLogEdit:
"""Tests for POST /log/<id>/edit."""
def test_edit_log_requires_auth(self, client: TestClient) -> None:
"""POST /log/1/edit should require admin login."""
def test_edit_log_requires_profile(self, client: TestClient) -> None:
"""POST /log/1/edit should redirect to / without profile cookie."""
response = client.post(
"/log/1/edit",
data={"reps": "10", "weight": "35 lbs"},
follow_redirects=False,
)
assert response.status_code in (401, 303)
assert response.status_code == 302
assert response.headers["location"] == "/"
class TestLogDelete:
"""Tests for POST /log/<id>/delete."""
def test_delete_log_requires_auth(self, client: TestClient) -> None:
"""POST /log/1/delete should require admin login."""
def test_delete_log_requires_profile(self, client: TestClient) -> None:
"""POST /log/1/delete should redirect to / without profile cookie."""
response = client.post("/log/1/delete", follow_redirects=False)
assert response.status_code in (401, 303)
assert response.status_code == 302
assert response.headers["location"] == "/"

View File

@@ -28,12 +28,10 @@ class TestModels:
engine = self._create_engine()
user = User(
username="testuser",
password_hash="fakehash",
display_name="Test User",
display_name="Test User",
height="6'0\"",
weight="200 lbs",
goals="Get strong",
is_admin=False,
)
with Session(engine) as session:
session.add(user)
@@ -41,7 +39,6 @@ class TestModels:
session.refresh(user)
assert user.id is not None
assert user.username == "testuser"
assert user.is_admin is False
assert isinstance(user.created_at, datetime)
def test_exercise_model_roundtrip(self) -> None:
@@ -97,7 +94,7 @@ class TestModels:
"""UserExerciseProgram model should persist with FK references."""
engine = self._create_engine()
with Session(engine) as session:
user = User(username="u", password_hash="h", display_name="U")
user = User(username="u", display_name="U")
exercise = Exercise(
name="Test Ex", muscle_group="Test",
workout_day="Push", sets=3, tempo="3-1-2", form_cues="..."
@@ -126,7 +123,7 @@ class TestModels:
"""WorkoutSession model should persist correctly."""
engine = self._create_engine()
with Session(engine) as session:
user = User(username="u", password_hash="h", display_name="U")
user = User(username="u", display_name="U")
day = WorkoutDay(name="Push", day_number=1, description="Push day")
session.add(user)
session.add(day)
@@ -148,7 +145,7 @@ class TestModels:
"""WorkoutLog model should persist correctly."""
engine = self._create_engine()
with Session(engine) as session:
user = User(username="u", password_hash="h", display_name="U")
user = User(username="u", display_name="U")
day = WorkoutDay(name="Push", day_number=1, description="Push day")
exercise = Exercise(
name="Ex", muscle_group="Test",
@@ -187,7 +184,7 @@ class TestModels:
"""ProgressLog model should persist correctly."""
engine = self._create_engine()
with Session(engine) as session:
user = User(username="u", password_hash="h", display_name="U")
user = User(username="u", display_name="U")
exercise = Exercise(
name="Ex", muscle_group="Test",
workout_day="Push", sets=3, tempo="3-1-2", form_cues="..."

105
tests/test_profile_auth.py Normal file
View File

@@ -0,0 +1,105 @@
"""Tests for profile-based access control and landing page."""
from sqlmodel import SQLModel, Session, create_engine
from app.models.user import User
from app.services.user_service import UserService
from tests.conftest import set_profile_cookie
class TestProfileAuth:
"""Tests for the profile cookie-based access flow."""
def test_no_profile_redirects_to_home(self, client) -> None:
"""Routes requiring a profile should redirect to / when no cookie set."""
response = client.get("/workouts", follow_redirects=False)
assert response.status_code == 302
assert response.headers["location"] == "/"
def test_invalid_profile_redirects_to_home(self, client) -> None:
"""An invalid profile_id cookie should redirect to /."""
client.cookies.set("active_profile_id", "99999")
response = client.get("/workouts", follow_redirects=False)
assert response.status_code == 302
assert response.headers["location"] == "/"
def test_valid_profile_allows_access(self, client) -> None:
"""A valid profile cookie should allow access to protected routes."""
# The seeded profiles exist; find one
response = client.get("/")
assert response.status_code == 200
# Get the first profile ID from the seeded data
set_profile_cookie(client, 1)
response = client.get("/workouts")
assert response.status_code == 200
class TestLandingPage:
"""Tests for the home page profile picker."""
def test_home_page_shows_profiles(self, client) -> None:
"""GET / should show seeded profile names."""
response = client.get("/")
assert response.status_code == 200
assert "Select Profile" in response.text
def test_home_page_accessible_without_cookie(self, client) -> None:
"""GET / should work without any profile cookie."""
response = client.get("/")
assert response.status_code == 200
assert "SneakySwole" in response.text
class TestProfileCreation:
"""Tests for the profile creation flow."""
def test_create_profile_form_renders(self, client) -> None:
"""GET /profiles/create should render the form."""
response = client.get("/profiles/create")
assert response.status_code == 200
assert "Display Name" in response.text
def test_create_profile_sets_cookie(self, client) -> None:
"""POST /profiles/create should create profile and set cookie."""
response = client.post(
"/profiles/create",
data={"display_name": "NewUser", "height": "5'10\"", "weight": "180 lbs"},
follow_redirects=False,
)
assert response.status_code == 303
assert response.headers["location"] == "/workouts"
assert "active_profile_id" in response.cookies
def test_create_profile_requires_name(self, client) -> None:
"""POST /profiles/create with empty name should show error."""
response = client.post(
"/profiles/create",
data={"display_name": "", "height": "", "weight": ""},
)
assert response.status_code == 200
assert "required" in response.text.lower()
class TestProfileSwitch:
"""Tests for the profile switch flow."""
def test_switch_profile_sets_cookie(self, client) -> None:
"""POST /profiles/switch should set cookie and redirect."""
response = client.post(
"/profiles/switch",
data={"profile_id": "1"},
follow_redirects=False,
)
assert response.status_code == 303
assert "active_profile_id" in response.cookies
def test_switch_invalid_profile_redirects_home(self, client) -> None:
"""POST /profiles/switch with invalid ID should redirect to /."""
response = client.post(
"/profiles/switch",
data={"profile_id": "99999"},
follow_redirects=False,
)
assert response.status_code == 303
assert response.headers["location"] == "/"

View File

@@ -2,25 +2,34 @@
from fastapi.testclient import TestClient
from tests.conftest import set_profile_cookie
class TestProfileSwitcher:
"""Tests for POST /profiles/switch."""
def test_switch_profile_requires_auth(self, client: TestClient) -> None:
"""POST /profiles/switch should require admin login."""
def test_switch_profile_redirects_to_workouts(self, client: TestClient) -> None:
"""POST /profiles/switch should set cookie and redirect to /workouts."""
response = client.post(
"/profiles/switch",
data={"profile_id": "1"},
follow_redirects=False,
)
# Should redirect to login or return 401
assert response.status_code in (401, 303)
assert response.status_code == 303
assert response.headers["location"] == "/workouts"
class TestProfileList:
"""Tests for GET /profiles."""
def test_profiles_page_requires_auth(self, client: TestClient) -> None:
"""GET /profiles should require admin login."""
def test_profiles_page_requires_profile(self, client: TestClient) -> None:
"""GET /profiles should redirect to / without profile cookie."""
response = client.get("/profiles", follow_redirects=False)
assert response.status_code in (401, 303)
assert response.status_code == 302
assert response.headers["location"] == "/"
def test_profiles_page_with_profile(self, client: TestClient) -> None:
"""GET /profiles should succeed with a valid profile cookie."""
set_profile_cookie(client, 1)
response = client.get("/profiles")
assert response.status_code == 200

View File

@@ -23,7 +23,7 @@ class TestProgressionService:
SQLModel.metadata.create_all(engine)
session = Session(engine)
user = User(username="phil", password_hash="h", display_name="Phillip")
user = User(username="phil", display_name="Phillip")
day = WorkoutDay(name="Push", day_number=1, description="Push day")
exercise = Exercise(
name="DB Chest Press", muscle_group="Chest",

View File

@@ -6,7 +6,8 @@ from fastapi.testclient import TestClient
class TestSchedule:
"""Tests for GET /schedule."""
def test_schedule_requires_auth(self, client: TestClient) -> None:
"""GET /schedule should require admin login."""
def test_schedule_requires_profile(self, client: TestClient) -> None:
"""GET /schedule should redirect to / without profile cookie."""
response = client.get("/schedule", follow_redirects=False)
assert response.status_code in (401, 303)
assert response.status_code == 302
assert response.headers["location"] == "/"

View File

@@ -1,9 +1,7 @@
"""Tests for the SeedService class."""
from pathlib import Path
from unittest.mock import patch
import bcrypt
from sqlmodel import SQLModel, Session, create_engine, select
from app.models.user import User
@@ -53,20 +51,6 @@ class TestSeedService:
assert len(warmups) == 6
session.close()
def test_seed_admin_user(self) -> None:
"""seed_admin should create admin user with hashed password."""
session, service = self._setup()
with patch.dict("os.environ", {
"ADMIN_USERNAME": "admin",
"ADMIN_PASSWORD": "testpass",
}):
service.seed_admin()
admin = session.exec(select(User).where(User.is_admin == True)).first() # noqa: E712
assert admin is not None
assert admin.username == "admin"
assert bcrypt.checkpw(b"testpass", admin.password_hash.encode())
session.close()
def test_seed_user_programs(self) -> None:
"""seed_user_programs should create user profiles and link exercises."""
session, service = self._setup()
@@ -74,19 +58,15 @@ class TestSeedService:
service.seed_user_programs()
programs = session.exec(select(UserExerciseProgram)).all()
assert len(programs) > 0
users = session.exec(select(User).where(User.is_admin == False)).all() # noqa: E712
users = session.exec(select(User)).all()
assert len(users) == 2 # Phillip and Daughter
session.close()
def test_seed_is_idempotent(self) -> None:
"""Running seed_all twice should not create duplicate records."""
session, service = self._setup()
with patch.dict("os.environ", {
"ADMIN_USERNAME": "admin",
"ADMIN_PASSWORD": "testpass",
}):
service.seed_all()
service.seed_all()
service.seed_all()
service.seed_all()
users = session.exec(select(User)).all()
exercises = session.exec(select(Exercise)).all()
# Should not be doubled

View File

@@ -22,12 +22,10 @@ class TestUserService:
session, service = self._setup()
user = service.create_user(
username="phil",
password_hash="hashed",
display_name="Phillip",
height="6'0\"",
weight="260 lbs",
goals="Muscle build",
is_admin=False,
)
assert user.id is not None
assert user.username == "phil"
@@ -37,7 +35,7 @@ class TestUserService:
"""get_user_by_id should return the correct user."""
session, service = self._setup()
created = service.create_user(
username="test", password_hash="h", display_name="Test"
username="test", display_name="Test"
)
found = service.get_user_by_id(created.id)
assert found is not None
@@ -47,10 +45,10 @@ class TestUserService:
def test_get_user_by_username(self) -> None:
"""get_user_by_username should return the correct user."""
session, service = self._setup()
service.create_user(username="admin", password_hash="h", display_name="Admin")
found = service.get_user_by_username("admin")
service.create_user(username="phil", display_name="Phillip")
found = service.get_user_by_username("phil")
assert found is not None
assert found.display_name == "Admin"
assert found.display_name == "Phillip"
session.close()
def test_get_user_by_username_not_found(self) -> None:
@@ -63,26 +61,16 @@ class TestUserService:
def test_list_users(self) -> None:
"""list_users should return all users."""
session, service = self._setup()
service.create_user(username="a", password_hash="h", display_name="A")
service.create_user(username="b", password_hash="h", display_name="B")
service.create_user(username="a", display_name="A")
service.create_user(username="b", display_name="B")
users = service.list_users()
assert len(users) == 2
session.close()
def test_list_non_admin_users(self) -> None:
"""list_users with exclude_admin=True should skip admin users."""
session, service = self._setup()
service.create_user(username="admin", password_hash="h", display_name="Admin", is_admin=True)
service.create_user(username="user", password_hash="h", display_name="User", is_admin=False)
users = service.list_users(exclude_admin=True)
assert len(users) == 1
assert users[0].username == "user"
session.close()
def test_update_user(self) -> None:
"""update_user should modify the specified fields."""
session, service = self._setup()
user = service.create_user(username="u", password_hash="h", display_name="Old")
user = service.create_user(username="u", display_name="Old")
updated = service.update_user(user.id, display_name="New", weight="250 lbs")
assert updated.display_name == "New"
assert updated.weight == "250 lbs"

View File

@@ -2,16 +2,26 @@
from fastapi.testclient import TestClient
from tests.conftest import set_profile_cookie
class TestWorkoutDayViewer:
"""Tests for GET /workouts/<day_name>."""
def test_workout_day_requires_auth(self, client: TestClient) -> None:
"""GET /workouts/push should require admin login."""
def test_workout_day_requires_profile(self, client: TestClient) -> None:
"""GET /workouts/push should redirect to / without profile cookie."""
response = client.get("/workouts/push", follow_redirects=False)
assert response.status_code in (401, 303)
assert response.status_code == 302
assert response.headers["location"] == "/"
def test_workout_days_list_requires_auth(self, client: TestClient) -> None:
"""GET /workouts should require admin login."""
def test_workout_days_list_requires_profile(self, client: TestClient) -> None:
"""GET /workouts should redirect to / without profile cookie."""
response = client.get("/workouts", follow_redirects=False)
assert response.status_code in (401, 303)
assert response.status_code == 302
assert response.headers["location"] == "/"
def test_workout_days_list_with_profile(self, client: TestClient) -> None:
"""GET /workouts should succeed with a valid profile cookie."""
set_profile_cookie(client, 1)
response = client.get("/workouts")
assert response.status_code == 200

View File

@@ -19,7 +19,7 @@ class TestWorkoutSessionService:
SQLModel.metadata.create_all(engine)
session = Session(engine)
user = User(username="phil", password_hash="h", display_name="Phillip")
user = User(username="phil", display_name="Phillip")
day = WorkoutDay(name="Push", day_number=1, description="Push day")
session.add_all([user, day])
session.commit()