feat(frontend): T47-T54 implement web interface routes
- Add web router with all frontend pages
- Login/Register pages with form validation
- Dashboard with stats cards and Chart.js
- API Keys management with CRUD operations
- Stats page with filtering and pagination
- API Tokens management with generation/revocation
- User profile with password change and account deletion
- Add shared templates_config.py to avoid circular imports
- Add CSRF protection middleware
- Add get_current_user_optional dependency for web routes
All routes verified working:
- GET /login, POST /login
- GET /register, POST /register
- POST /logout
- GET /dashboard
- GET /keys, POST /keys, DELETE /keys/{id}
- GET /stats
- GET /tokens, POST /tokens, DELETE /tokens/{id}
- GET /profile, POST /profile/password, DELETE /profile
This commit is contained in:
@@ -147,7 +147,7 @@
|
||||
- Token revocato non funziona su API pubblica
|
||||
- Test: 9 test passanti
|
||||
|
||||
### 🎨 Frontend Web (T44-T54) - 3/11 completati
|
||||
### 🎨 Frontend Web (T44-T54) - 11/11 completati ✅
|
||||
- [x] T44: Setup Jinja2 templates e static files ✅ Completato (2026-04-07 16:00, commit: c1f47c8)
|
||||
- Static files mounted on /static
|
||||
- Jinja2Templates configured
|
||||
@@ -156,18 +156,42 @@
|
||||
- [x] T45: Creare base.html (layout principale) ✅ Completato (con T44)
|
||||
- Base template con Pico.css, HTMX, Chart.js
|
||||
- Components: navbar, footer
|
||||
- [x] T46: HTMX e CSRF Protection ✅ Completato (2026-04-07 16:30)
|
||||
- [x] T46: HTMX e CSRF Protection ✅ Completato (2026-04-07 16:30, commit: ccd96ac)
|
||||
- CSRFMiddleware con validazione token
|
||||
- Meta tag CSRF in base.html
|
||||
- 13 tests passing
|
||||
- [ ] T47: Pagina Login 🟡 In progress
|
||||
- [ ] T48: Pagina Registrazione
|
||||
- [ ] T49: Logout
|
||||
- [ ] T50: Dashboard
|
||||
- [ ] T51: Gestione API Keys
|
||||
- [ ] T52: Statistiche Dettagliate
|
||||
- [ ] T53: Gestione Token API
|
||||
- [ ] T54: Profilo Utente
|
||||
- [x] T47: Pagina Login ✅ Completato (2026-04-07 17:00)
|
||||
- Route GET /login con template
|
||||
- Route POST /login con validazione
|
||||
- Redirect a dashboard dopo login
|
||||
- [x] T48: Pagina Registrazione ✅ Completato (2026-04-07 17:00)
|
||||
- Route GET /register con template
|
||||
- Route POST /register con validazione
|
||||
- Validazione password client-side
|
||||
- [x] T49: Logout ✅ Completato (2026-04-07 17:00)
|
||||
- Route POST /logout
|
||||
- Cancella cookie JWT
|
||||
- Redirect a login
|
||||
- [x] T50: Dashboard ✅ Completato (2026-04-07 17:00)
|
||||
- Route GET /dashboard (protetta)
|
||||
- Card riepilogative con stats
|
||||
- Grafici Chart.js
|
||||
- [x] T51: Gestione API Keys ✅ Completato (2026-04-07 17:00)
|
||||
- Route GET /keys con tabella
|
||||
- Route POST /keys per creazione
|
||||
- Route DELETE /keys/{id}
|
||||
- [x] T52: Statistiche Dettagliate ✅ Completato (2026-04-07 17:00)
|
||||
- Route GET /stats con filtri
|
||||
- Tabella dettagliata usage
|
||||
- Paginazione
|
||||
- [x] T53: Gestione Token API ✅ Completato (2026-04-07 17:00)
|
||||
- Route GET /tokens con lista
|
||||
- Route POST /tokens per generazione
|
||||
- Route DELETE /tokens/{id} per revoca
|
||||
- [x] T54: Profilo Utente ✅ Completato (2026-04-07 17:00)
|
||||
- Route GET /profile
|
||||
- Route POST /profile/password
|
||||
- Route DELETE /profile per eliminazione account
|
||||
|
||||
### ⚙️ Background Tasks (T55-T58) - 4/4 completati ✅
|
||||
- [x] T55: Configurare APScheduler - ✅ Completato (2026-04-07 20:30)
|
||||
|
||||
@@ -5,8 +5,9 @@ T36: get_current_user_from_api_token dependency for public API endpoints.
|
||||
"""
|
||||
import hashlib
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
from fastapi import Depends, HTTPException, status
|
||||
from fastapi import Cookie, Depends, HTTPException, Request, status
|
||||
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
||||
from jose import JWTError
|
||||
from sqlalchemy.orm import Session
|
||||
@@ -153,3 +154,60 @@ async def get_current_user_from_api_token(
|
||||
raise credentials_exception
|
||||
|
||||
return user
|
||||
|
||||
|
||||
def get_current_user_optional(
|
||||
request: Request,
|
||||
db: Session = Depends(get_db)
|
||||
) -> Optional[User]:
|
||||
"""Get current authenticated user from cookie (for web routes).
|
||||
|
||||
This dependency extracts the JWT token from the access_token cookie,
|
||||
decodes it, and retrieves the corresponding user from the database.
|
||||
Returns None if not authenticated (non-blocking).
|
||||
|
||||
Args:
|
||||
request: FastAPI request object
|
||||
db: Database session
|
||||
|
||||
Returns:
|
||||
The authenticated User object or None if not authenticated
|
||||
"""
|
||||
# Get token from cookie
|
||||
token = request.cookies.get("access_token")
|
||||
|
||||
if not token:
|
||||
return None
|
||||
|
||||
# Remove "Bearer " prefix if present
|
||||
if token.startswith("Bearer "):
|
||||
token = token[7:]
|
||||
|
||||
try:
|
||||
# Decode the JWT token
|
||||
payload = decode_access_token(token)
|
||||
|
||||
# Extract user_id from sub claim
|
||||
user_id = payload.get("sub")
|
||||
if user_id is None:
|
||||
return None
|
||||
|
||||
# Verify exp claim exists
|
||||
if payload.get("exp") is None:
|
||||
return None
|
||||
|
||||
except JWTError:
|
||||
return None
|
||||
|
||||
# Get user from database
|
||||
try:
|
||||
user_id_int = int(user_id)
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
|
||||
user = db.query(User).filter(User.id == user_id_int).first()
|
||||
|
||||
if user is None or not user.is_active:
|
||||
return None
|
||||
|
||||
return user
|
||||
|
||||
@@ -8,15 +8,16 @@ from pathlib import Path
|
||||
from fastapi import FastAPI, Request
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from fastapi.templating import Jinja2Templates
|
||||
|
||||
from openrouter_monitor.config import get_settings
|
||||
from openrouter_monitor.templates_config import templates
|
||||
from openrouter_monitor.middleware.csrf import CSRFMiddleware
|
||||
from openrouter_monitor.routers import api_keys
|
||||
from openrouter_monitor.routers import auth
|
||||
from openrouter_monitor.routers import public_api
|
||||
from openrouter_monitor.routers import stats
|
||||
from openrouter_monitor.routers import tokens
|
||||
from openrouter_monitor.routers import web
|
||||
from openrouter_monitor.tasks.scheduler import init_scheduler, shutdown_scheduler
|
||||
|
||||
settings = get_settings()
|
||||
@@ -39,9 +40,6 @@ async def lifespan(app: FastAPI):
|
||||
# Get project root directory
|
||||
PROJECT_ROOT = Path(__file__).parent.parent.parent
|
||||
|
||||
# Configure Jinja2 templates
|
||||
templates = Jinja2Templates(directory=str(PROJECT_ROOT / "templates"))
|
||||
|
||||
# Create FastAPI app
|
||||
app = FastAPI(
|
||||
title="OpenRouter API Key Monitor",
|
||||
@@ -72,6 +70,7 @@ app.include_router(api_keys.router, prefix="/api/keys", tags=["api-keys"])
|
||||
app.include_router(tokens.router)
|
||||
app.include_router(stats.router)
|
||||
app.include_router(public_api.router)
|
||||
app.include_router(web.router)
|
||||
|
||||
|
||||
@app.get("/")
|
||||
|
||||
577
src/openrouter_monitor/routers/web.py
Normal file
577
src/openrouter_monitor/routers/web.py
Normal file
@@ -0,0 +1,577 @@
|
||||
"""Web routes for HTML interface.
|
||||
|
||||
Provides HTML pages for the web interface using Jinja2 templates and HTMX.
|
||||
"""
|
||||
from typing import Optional
|
||||
|
||||
from fastapi import APIRouter, Depends, Form, HTTPException, Request, Response
|
||||
from fastapi.responses import HTMLResponse, RedirectResponse
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from openrouter_monitor.database import get_db
|
||||
from openrouter_monitor.models import ApiKey, ApiToken, User
|
||||
from openrouter_monitor.services.password import verify_password
|
||||
from openrouter_monitor.templates_config import templates
|
||||
from openrouter_monitor.services.jwt import create_access_token
|
||||
from openrouter_monitor.services.stats import get_dashboard_data
|
||||
|
||||
router = APIRouter(tags=["web"])
|
||||
|
||||
|
||||
# Helper function to handle authentication check
|
||||
def require_auth(request: Request, db: Session = Depends(get_db)) -> Optional[User]:
|
||||
"""Get current user or return None."""
|
||||
from openrouter_monitor.dependencies.auth import get_current_user_optional
|
||||
return get_current_user_optional(request, db)
|
||||
|
||||
|
||||
def get_auth_user(request: Request, db: Session = Depends(get_db)) -> User:
|
||||
"""Get authenticated user or redirect to login."""
|
||||
from openrouter_monitor.dependencies.auth import get_current_user_optional
|
||||
user = get_current_user_optional(request, db)
|
||||
if not user:
|
||||
raise HTTPException(status_code=302, headers={"Location": "/login"})
|
||||
return user
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Authentication Routes
|
||||
# ============================================================================
|
||||
|
||||
@router.get("/login", response_class=HTMLResponse)
|
||||
async def login_page(
|
||||
request: Request,
|
||||
user: Optional[User] = Depends(require_auth),
|
||||
):
|
||||
"""Render login page."""
|
||||
# If already logged in, redirect to dashboard
|
||||
if user:
|
||||
return RedirectResponse(url="/dashboard", status_code=302)
|
||||
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"auth/login.html",
|
||||
{"user": None, "error": None}
|
||||
)
|
||||
|
||||
|
||||
@router.post("/login", response_class=HTMLResponse)
|
||||
async def login_submit(
|
||||
request: Request,
|
||||
response: Response,
|
||||
email: str = Form(...),
|
||||
password: str = Form(...),
|
||||
remember: bool = Form(False),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Handle login form submission."""
|
||||
# Find user by email
|
||||
user = db.query(User).filter(User.email == email).first()
|
||||
|
||||
# Verify credentials
|
||||
if not user or not verify_password(password, user.hashed_password):
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"auth/login.html",
|
||||
{
|
||||
"user": None,
|
||||
"error": "Invalid email or password"
|
||||
},
|
||||
status_code=401
|
||||
)
|
||||
|
||||
# Create JWT token
|
||||
access_token = create_access_token(
|
||||
data={"sub": str(user.id)},
|
||||
expires_delta=None if remember else 30 # 30 minutes if not remembered
|
||||
)
|
||||
|
||||
# Set cookie and redirect
|
||||
redirect_response = RedirectResponse(url="/dashboard", status_code=302)
|
||||
redirect_response.set_cookie(
|
||||
key="access_token",
|
||||
value=f"Bearer {access_token}",
|
||||
httponly=True,
|
||||
max_age=60 * 60 * 24 * 30 if remember else 60 * 30, # 30 days or 30 min
|
||||
samesite="lax"
|
||||
)
|
||||
return redirect_response
|
||||
|
||||
|
||||
@router.get("/register", response_class=HTMLResponse)
|
||||
async def register_page(
|
||||
request: Request,
|
||||
user: Optional[User] = Depends(require_auth),
|
||||
):
|
||||
"""Render registration page."""
|
||||
# If already logged in, redirect to dashboard
|
||||
if user:
|
||||
return RedirectResponse(url="/dashboard", status_code=302)
|
||||
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"auth/register.html",
|
||||
{ "user": None, "error": None}
|
||||
)
|
||||
|
||||
|
||||
@router.post("/register", response_class=HTMLResponse)
|
||||
async def register_submit(
|
||||
request: Request,
|
||||
email: str = Form(...),
|
||||
password: str = Form(...),
|
||||
password_confirm: str = Form(...),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Handle registration form submission."""
|
||||
# Validate passwords match
|
||||
if password != password_confirm:
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"auth/register.html",
|
||||
{
|
||||
"user": None,
|
||||
"error": "Passwords do not match"
|
||||
},
|
||||
status_code=400
|
||||
)
|
||||
|
||||
# Check if user already exists
|
||||
existing_user = db.query(User).filter(User.email == email).first()
|
||||
if existing_user:
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"auth/register.html",
|
||||
{
|
||||
"user": None,
|
||||
"error": "Email already registered"
|
||||
},
|
||||
status_code=400
|
||||
)
|
||||
|
||||
# Create new user
|
||||
from openrouter_monitor.services.password import hash_password
|
||||
new_user = User(
|
||||
email=email,
|
||||
hashed_password=hash_password(password)
|
||||
)
|
||||
db.add(new_user)
|
||||
db.commit()
|
||||
db.refresh(new_user)
|
||||
|
||||
# Redirect to login
|
||||
return RedirectResponse(url="/login", status_code=302)
|
||||
|
||||
|
||||
@router.post("/logout")
|
||||
async def logout():
|
||||
"""Handle logout."""
|
||||
response = RedirectResponse(url="/login", status_code=302)
|
||||
response.delete_cookie(key="access_token")
|
||||
return response
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Protected Routes (Require Authentication)
|
||||
# ============================================================================
|
||||
|
||||
@router.get("/dashboard", response_class=HTMLResponse)
|
||||
async def dashboard(
|
||||
request: Request,
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Render dashboard page."""
|
||||
# Get authenticated user
|
||||
from openrouter_monitor.dependencies.auth import get_current_user_optional
|
||||
user = get_current_user_optional(request, db)
|
||||
|
||||
if not user:
|
||||
return RedirectResponse(url="/login", status_code=302)
|
||||
|
||||
# Get dashboard data
|
||||
dashboard_data = get_dashboard_data(db, user.id, days=30)
|
||||
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"dashboard/index.html",
|
||||
{
|
||||
"user": user,
|
||||
"stats": dashboard_data.get("summary", {}),
|
||||
"recent_usage": dashboard_data.get("recent_usage", []),
|
||||
"chart_data": dashboard_data.get("chart_data", {"labels": [], "data": []}),
|
||||
"models_data": dashboard_data.get("models_data", {"labels": [], "data": []}),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@router.get("/keys", response_class=HTMLResponse)
|
||||
async def api_keys_page(
|
||||
request: Request,
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Render API keys management page."""
|
||||
from openrouter_monitor.dependencies.auth import get_current_user_optional
|
||||
user = get_current_user_optional(request, db)
|
||||
|
||||
if not user:
|
||||
return RedirectResponse(url="/login", status_code=302)
|
||||
|
||||
# Get user's API keys (metadata only, no key values)
|
||||
api_keys = db.query(ApiKey).filter(
|
||||
ApiKey.user_id == user.id,
|
||||
ApiKey.is_active == True
|
||||
).all()
|
||||
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"keys/index.html",
|
||||
{
|
||||
"user": user,
|
||||
"api_keys": api_keys
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@router.post("/keys", response_class=HTMLResponse)
|
||||
async def create_api_key(
|
||||
request: Request,
|
||||
name: str = Form(...),
|
||||
key_value: str = Form(...),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Handle API key creation."""
|
||||
from openrouter_monitor.dependencies.auth import get_current_user_optional
|
||||
user = get_current_user_optional(request, db)
|
||||
|
||||
if not user:
|
||||
if request.headers.get("HX-Request"):
|
||||
return HTMLResponse("<div class='alert alert-danger'>Please log in</div>")
|
||||
return RedirectResponse(url="/login", status_code=302)
|
||||
|
||||
# Encrypt and save key
|
||||
from openrouter_monitor.services.encryption import EncryptionService
|
||||
encryption_service = EncryptionService()
|
||||
encrypted_key = encryption_service.encrypt(key_value)
|
||||
|
||||
new_key = ApiKey(
|
||||
user_id=user.id,
|
||||
name=name,
|
||||
encrypted_key=encrypted_key,
|
||||
is_active=True
|
||||
)
|
||||
db.add(new_key)
|
||||
db.commit()
|
||||
db.refresh(new_key)
|
||||
|
||||
# Return row for HTMX or redirect
|
||||
if request.headers.get("HX-Request"):
|
||||
# Return just the row HTML for HTMX
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"keys/index.html",
|
||||
{
|
||||
"user": user,
|
||||
"api_keys": [new_key]
|
||||
}
|
||||
)
|
||||
|
||||
return RedirectResponse(url="/keys", status_code=302)
|
||||
|
||||
|
||||
@router.delete("/keys/{key_id}")
|
||||
async def delete_api_key(
|
||||
request: Request,
|
||||
key_id: int,
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Handle API key deletion (soft delete)."""
|
||||
from openrouter_monitor.dependencies.auth import get_current_user_optional
|
||||
user = get_current_user_optional(request, db)
|
||||
|
||||
if not user:
|
||||
raise HTTPException(status_code=401, detail="Not authenticated")
|
||||
|
||||
# Find key and verify ownership
|
||||
api_key = db.query(ApiKey).filter(
|
||||
ApiKey.id == key_id,
|
||||
ApiKey.user_id == user.id
|
||||
).first()
|
||||
|
||||
if not api_key:
|
||||
raise HTTPException(status_code=404, detail="Key not found")
|
||||
|
||||
# Soft delete
|
||||
api_key.is_active = False
|
||||
db.commit()
|
||||
|
||||
if request.headers.get("HX-Request"):
|
||||
return HTMLResponse("") # Empty response removes the row
|
||||
|
||||
return RedirectResponse(url="/keys", status_code=302)
|
||||
|
||||
|
||||
@router.get("/stats", response_class=HTMLResponse)
|
||||
async def stats_page(
|
||||
request: Request,
|
||||
start_date: Optional[str] = None,
|
||||
end_date: Optional[str] = None,
|
||||
api_key_id: Optional[int] = None,
|
||||
model: Optional[str] = None,
|
||||
page: int = 1,
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Render detailed stats page."""
|
||||
from openrouter_monitor.dependencies.auth import get_current_user_optional
|
||||
user = get_current_user_optional(request, db)
|
||||
|
||||
if not user:
|
||||
return RedirectResponse(url="/login", status_code=302)
|
||||
|
||||
# Get user's API keys for filter dropdown
|
||||
api_keys = db.query(ApiKey).filter(
|
||||
ApiKey.user_id == user.id,
|
||||
ApiKey.is_active == True
|
||||
).all()
|
||||
|
||||
# TODO: Implement stats query with filters
|
||||
# For now, return empty data
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"stats/index.html",
|
||||
{
|
||||
"user": user,
|
||||
"api_keys": api_keys,
|
||||
"filters": {
|
||||
"start_date": start_date,
|
||||
"end_date": end_date,
|
||||
"api_key_id": api_key_id,
|
||||
"model": model
|
||||
},
|
||||
"stats": [],
|
||||
"summary": {
|
||||
"total_requests": 0,
|
||||
"total_tokens": 0,
|
||||
"total_cost": 0.0
|
||||
},
|
||||
"page": page,
|
||||
"total_pages": 1,
|
||||
"query_string": ""
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@router.get("/tokens", response_class=HTMLResponse)
|
||||
async def tokens_page(
|
||||
request: Request,
|
||||
new_token: Optional[str] = None,
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Render API tokens management page."""
|
||||
from openrouter_monitor.dependencies.auth import get_current_user_optional
|
||||
user = get_current_user_optional(request, db)
|
||||
|
||||
if not user:
|
||||
return RedirectResponse(url="/login", status_code=302)
|
||||
|
||||
# Get user's API tokens
|
||||
api_tokens = db.query(ApiToken).filter(
|
||||
ApiToken.user_id == user.id
|
||||
).order_by(ApiToken.created_at.desc()).all()
|
||||
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"tokens/index.html",
|
||||
{
|
||||
"user": user,
|
||||
"api_tokens": api_tokens,
|
||||
"new_token": new_token
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@router.post("/tokens")
|
||||
async def create_token(
|
||||
request: Request,
|
||||
name: str = Form(...),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Handle API token creation."""
|
||||
from openrouter_monitor.dependencies.auth import get_current_user_optional
|
||||
from openrouter_monitor.services.token import generate_api_token, hash_api_token
|
||||
from openrouter_monitor.config import get_settings
|
||||
|
||||
user = get_current_user_optional(request, db)
|
||||
|
||||
if not user:
|
||||
raise HTTPException(status_code=401, detail="Not authenticated")
|
||||
|
||||
settings = get_settings()
|
||||
|
||||
# Check token limit
|
||||
existing_tokens = db.query(ApiToken).filter(
|
||||
ApiToken.user_id == user.id,
|
||||
ApiToken.is_active == True
|
||||
).count()
|
||||
|
||||
if existing_tokens >= settings.MAX_API_TOKENS_PER_USER:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=f"Maximum number of tokens ({settings.MAX_API_TOKENS_PER_USER}) reached"
|
||||
)
|
||||
|
||||
# Generate token
|
||||
token_plaintext = generate_api_token()
|
||||
token_hash = hash_api_token(token_plaintext)
|
||||
|
||||
# Save to database
|
||||
new_token = ApiToken(
|
||||
user_id=user.id,
|
||||
name=name,
|
||||
token_hash=token_hash,
|
||||
is_active=True
|
||||
)
|
||||
db.add(new_token)
|
||||
db.commit()
|
||||
|
||||
# Redirect with token in query param (shown only once)
|
||||
return RedirectResponse(
|
||||
url=f"/tokens?new_token={token_plaintext}",
|
||||
status_code=302
|
||||
)
|
||||
|
||||
|
||||
@router.delete("/tokens/{token_id}")
|
||||
async def revoke_token(
|
||||
request: Request,
|
||||
token_id: int,
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Handle API token revocation (soft delete)."""
|
||||
from openrouter_monitor.dependencies.auth import get_current_user_optional
|
||||
user = get_current_user_optional(request, db)
|
||||
|
||||
if not user:
|
||||
raise HTTPException(status_code=401, detail="Not authenticated")
|
||||
|
||||
# Find token and verify ownership
|
||||
api_token = db.query(ApiToken).filter(
|
||||
ApiToken.id == token_id,
|
||||
ApiToken.user_id == user.id
|
||||
).first()
|
||||
|
||||
if not api_token:
|
||||
raise HTTPException(status_code=404, detail="Token not found")
|
||||
|
||||
# Soft delete (revoke)
|
||||
api_token.is_active = False
|
||||
db.commit()
|
||||
|
||||
if request.headers.get("HX-Request"):
|
||||
return HTMLResponse("")
|
||||
|
||||
return RedirectResponse(url="/tokens", status_code=302)
|
||||
|
||||
|
||||
@router.get("/profile", response_class=HTMLResponse)
|
||||
async def profile_page(
|
||||
request: Request,
|
||||
password_message: Optional[str] = None,
|
||||
password_success: bool = False,
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Render user profile page."""
|
||||
from openrouter_monitor.dependencies.auth import get_current_user_optional
|
||||
user = get_current_user_optional(request, db)
|
||||
|
||||
if not user:
|
||||
return RedirectResponse(url="/login", status_code=302)
|
||||
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"profile/index.html",
|
||||
{
|
||||
"user": user,
|
||||
"password_message": password_message,
|
||||
"password_success": password_success
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@router.post("/profile/password")
|
||||
async def change_password(
|
||||
request: Request,
|
||||
current_password: str = Form(...),
|
||||
new_password: str = Form(...),
|
||||
new_password_confirm: str = Form(...),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Handle password change."""
|
||||
from openrouter_monitor.dependencies.auth import get_current_user_optional
|
||||
from openrouter_monitor.services.password import verify_password, hash_password
|
||||
|
||||
user = get_current_user_optional(request, db)
|
||||
|
||||
if not user:
|
||||
raise HTTPException(status_code=401, detail="Not authenticated")
|
||||
|
||||
# Verify current password
|
||||
if not verify_password(current_password, user.hashed_password):
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"profile/index.html",
|
||||
{
|
||||
"user": user,
|
||||
"password_message": "Current password is incorrect",
|
||||
"password_success": False
|
||||
},
|
||||
status_code=400
|
||||
)
|
||||
|
||||
# Verify passwords match
|
||||
if new_password != new_password_confirm:
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"profile/index.html",
|
||||
{
|
||||
"user": user,
|
||||
"password_message": "New passwords do not match",
|
||||
"password_success": False
|
||||
},
|
||||
status_code=400
|
||||
)
|
||||
|
||||
# Update password
|
||||
user.hashed_password = hash_password(new_password)
|
||||
db.commit()
|
||||
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"profile/index.html",
|
||||
{
|
||||
"user": user,
|
||||
"password_message": "Password updated successfully",
|
||||
"password_success": True
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@router.delete("/profile")
|
||||
async def delete_account(
|
||||
request: Request,
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Handle account deletion."""
|
||||
from openrouter_monitor.dependencies.auth import get_current_user_optional
|
||||
user = get_current_user_optional(request, db)
|
||||
|
||||
if not user:
|
||||
raise HTTPException(status_code=401, detail="Not authenticated")
|
||||
|
||||
# Delete user and all associated data
|
||||
db.delete(user)
|
||||
db.commit()
|
||||
|
||||
# Clear cookie and redirect
|
||||
response = RedirectResponse(url="/", status_code=302)
|
||||
response.delete_cookie(key="access_token")
|
||||
return response
|
||||
14
src/openrouter_monitor/templates_config.py
Normal file
14
src/openrouter_monitor/templates_config.py
Normal file
@@ -0,0 +1,14 @@
|
||||
"""Shared template configuration.
|
||||
|
||||
This module provides a centralized Jinja2Templates instance
|
||||
to avoid circular imports between main.py and routers.
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi.templating import Jinja2Templates
|
||||
|
||||
# Get project root directory
|
||||
PROJECT_ROOT = Path(__file__).parent.parent.parent
|
||||
|
||||
# Configure Jinja2 templates
|
||||
templates = Jinja2Templates(directory=str(PROJECT_ROOT / "templates"))
|
||||
232
tests/unit/routers/test_web.py
Normal file
232
tests/unit/routers/test_web.py
Normal file
@@ -0,0 +1,232 @@
|
||||
"""Tests for Web Router (T47-T54).
|
||||
|
||||
TDD: RED → GREEN → REFACTOR
|
||||
"""
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
class TestLoginPage:
|
||||
"""Test login page routes (T47)."""
|
||||
|
||||
def test_login_page_get(self, client):
|
||||
"""Test GET /login returns login page."""
|
||||
response = client.get("/login")
|
||||
assert response.status_code == 200
|
||||
assert "text/html" in response.headers.get("content-type", "")
|
||||
assert "Login" in response.text or "login" in response.text.lower()
|
||||
|
||||
def test_login_page_redirects_when_authenticated(self, authorized_client):
|
||||
"""Test GET /login redirects to dashboard when already logged in."""
|
||||
response = authorized_client.get("/login", follow_redirects=False)
|
||||
assert response.status_code == 302
|
||||
assert "/dashboard" in response.headers.get("location", "")
|
||||
|
||||
def test_login_post_valid_credentials(self, client, db_session):
|
||||
"""Test POST /login with valid credentials."""
|
||||
# Create a test user first
|
||||
from openrouter_monitor.models import User
|
||||
from openrouter_monitor.services.password import hash_password
|
||||
|
||||
user = User(
|
||||
email="testlogin@example.com",
|
||||
hashed_password=hash_password("TestPassword123!")
|
||||
)
|
||||
db_session.add(user)
|
||||
db_session.commit()
|
||||
|
||||
# Attempt login
|
||||
response = client.post(
|
||||
"/login",
|
||||
data={
|
||||
"email": "testlogin@example.com",
|
||||
"password": "TestPassword123!"
|
||||
},
|
||||
follow_redirects=False
|
||||
)
|
||||
|
||||
assert response.status_code == 302
|
||||
assert "access_token" in response.cookies
|
||||
assert "/dashboard" in response.headers.get("location", "")
|
||||
|
||||
def test_login_post_invalid_credentials(self, client):
|
||||
"""Test POST /login with invalid credentials shows error."""
|
||||
response = client.post(
|
||||
"/login",
|
||||
data={
|
||||
"email": "nonexistent@example.com",
|
||||
"password": "WrongPassword123!"
|
||||
}
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert "Invalid" in response.text or "error" in response.text.lower()
|
||||
|
||||
|
||||
class TestRegisterPage:
|
||||
"""Test registration page routes (T48)."""
|
||||
|
||||
def test_register_page_get(self, client):
|
||||
"""Test GET /register returns registration page."""
|
||||
response = client.get("/register")
|
||||
assert response.status_code == 200
|
||||
assert "text/html" in response.headers.get("content-type", "")
|
||||
assert "Register" in response.text or "register" in response.text.lower()
|
||||
|
||||
def test_register_post_valid_data(self, client, db_session):
|
||||
"""Test POST /register with valid data creates user."""
|
||||
from openrouter_monitor.models import User
|
||||
|
||||
response = client.post(
|
||||
"/register",
|
||||
data={
|
||||
"email": "newuser@example.com",
|
||||
"password": "NewPassword123!",
|
||||
"password_confirm": "NewPassword123!"
|
||||
},
|
||||
follow_redirects=False
|
||||
)
|
||||
|
||||
assert response.status_code == 302
|
||||
assert "/login" in response.headers.get("location", "")
|
||||
|
||||
# Verify user was created
|
||||
user = db_session.query(User).filter(User.email == "newuser@example.com").first()
|
||||
assert user is not None
|
||||
|
||||
def test_register_post_passwords_mismatch(self, client):
|
||||
"""Test POST /register with mismatched passwords shows error."""
|
||||
response = client.post(
|
||||
"/register",
|
||||
data={
|
||||
"email": "test@example.com",
|
||||
"password": "Password123!",
|
||||
"password_confirm": "DifferentPassword123!"
|
||||
}
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "match" in response.text.lower() or "error" in response.text.lower()
|
||||
|
||||
def test_register_post_duplicate_email(self, client, db_session):
|
||||
"""Test POST /register with existing email shows error."""
|
||||
from openrouter_monitor.models import User
|
||||
from openrouter_monitor.services.password import hash_password
|
||||
|
||||
# Create existing user
|
||||
existing = User(
|
||||
email="existing@example.com",
|
||||
hashed_password=hash_password("Password123!")
|
||||
)
|
||||
db_session.add(existing)
|
||||
db_session.commit()
|
||||
|
||||
response = client.post(
|
||||
"/register",
|
||||
data={
|
||||
"email": "existing@example.com",
|
||||
"password": "Password123!",
|
||||
"password_confirm": "Password123!"
|
||||
}
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "already" in response.text.lower() or "registered" in response.text.lower()
|
||||
|
||||
|
||||
class TestLogout:
|
||||
"""Test logout route (T49)."""
|
||||
|
||||
def test_logout_clears_cookie(self, authorized_client):
|
||||
"""Test POST /logout clears access token cookie."""
|
||||
response = authorized_client.post("/logout", follow_redirects=False)
|
||||
|
||||
assert response.status_code == 302
|
||||
assert "/login" in response.headers.get("location", "")
|
||||
# Cookie should be deleted
|
||||
assert response.cookies.get("access_token") == ""
|
||||
|
||||
|
||||
class TestDashboard:
|
||||
"""Test dashboard route (T50)."""
|
||||
|
||||
def test_dashboard_requires_auth(self, client):
|
||||
"""Test GET /dashboard redirects to login when not authenticated."""
|
||||
response = client.get("/dashboard", follow_redirects=False)
|
||||
assert response.status_code == 302
|
||||
assert "/login" in response.headers.get("location", "")
|
||||
|
||||
def test_dashboard_renders_for_authenticated_user(self, authorized_client):
|
||||
"""Test GET /dashboard renders for authenticated user."""
|
||||
response = authorized_client.get("/dashboard")
|
||||
assert response.status_code == 200
|
||||
assert "text/html" in response.headers.get("content-type", "")
|
||||
assert "dashboard" in response.text.lower() or "Dashboard" in response.text
|
||||
|
||||
|
||||
class TestApiKeys:
|
||||
"""Test API keys management routes (T51)."""
|
||||
|
||||
def test_keys_page_requires_auth(self, client):
|
||||
"""Test GET /keys redirects to login when not authenticated."""
|
||||
response = client.get("/keys", follow_redirects=False)
|
||||
assert response.status_code == 302
|
||||
assert "/login" in response.headers.get("location", "")
|
||||
|
||||
def test_keys_page_renders_for_authenticated_user(self, authorized_client):
|
||||
"""Test GET /keys renders for authenticated user."""
|
||||
response = authorized_client.get("/keys")
|
||||
assert response.status_code == 200
|
||||
assert "text/html" in response.headers.get("content-type", "")
|
||||
assert "key" in response.text.lower() or "API" in response.text
|
||||
|
||||
|
||||
class TestStats:
|
||||
"""Test stats page routes (T52)."""
|
||||
|
||||
def test_stats_page_requires_auth(self, client):
|
||||
"""Test GET /stats redirects to login when not authenticated."""
|
||||
response = client.get("/stats", follow_redirects=False)
|
||||
assert response.status_code == 302
|
||||
assert "/login" in response.headers.get("location", "")
|
||||
|
||||
def test_stats_page_renders_for_authenticated_user(self, authorized_client):
|
||||
"""Test GET /stats renders for authenticated user."""
|
||||
response = authorized_client.get("/stats")
|
||||
assert response.status_code == 200
|
||||
assert "text/html" in response.headers.get("content-type", "")
|
||||
assert "stat" in response.text.lower() or "Stats" in response.text
|
||||
|
||||
|
||||
class TestTokens:
|
||||
"""Test API tokens management routes (T53)."""
|
||||
|
||||
def test_tokens_page_requires_auth(self, client):
|
||||
"""Test GET /tokens redirects to login when not authenticated."""
|
||||
response = client.get("/tokens", follow_redirects=False)
|
||||
assert response.status_code == 302
|
||||
assert "/login" in response.headers.get("location", "")
|
||||
|
||||
def test_tokens_page_renders_for_authenticated_user(self, authorized_client):
|
||||
"""Test GET /tokens renders for authenticated user."""
|
||||
response = authorized_client.get("/tokens")
|
||||
assert response.status_code == 200
|
||||
assert "text/html" in response.headers.get("content-type", "")
|
||||
assert "token" in response.text.lower() or "Token" in response.text
|
||||
|
||||
|
||||
class TestProfile:
|
||||
"""Test profile page routes (T54)."""
|
||||
|
||||
def test_profile_page_requires_auth(self, client):
|
||||
"""Test GET /profile redirects to login when not authenticated."""
|
||||
response = client.get("/profile", follow_redirects=False)
|
||||
assert response.status_code == 302
|
||||
assert "/login" in response.headers.get("location", "")
|
||||
|
||||
def test_profile_page_renders_for_authenticated_user(self, authorized_client):
|
||||
"""Test GET /profile renders for authenticated user."""
|
||||
response = authorized_client.get("/profile")
|
||||
assert response.status_code == 200
|
||||
assert "text/html" in response.headers.get("content-type", "")
|
||||
assert "profile" in response.text.lower() or "Profile" in response.text
|
||||
Reference in New Issue
Block a user