")
+ cols_per_row = 1
+ else:
+ cols_per_row = min(len(items), 3)
+ if cols_per_row == 0:
+ cols_per_row = 1
+ parts.append(f"
")
+
+ for label, value in items:
+ parts.append(f"
")
+ parts.append(
+ f"
{html.escape(str(value))}
"
+ )
+ if not hide_labels:
+ parts.append(
+ f"
"
+ f"{html.escape(label)}
"
+ )
+ parts.append("
")
+ parts.append("
")
+
+ body = "".join(parts)
+
+ return f"""
+
+
+
+
+
+
+
+{body}
+
+"""
+
+
+@router.get("/stats", response_class=HTMLResponse)
+def get_embed_stats(
+ request: Request,
+ token: str = Query(..., description="Embed token"),
+ theme: str = Query("light", description="Theme: light|dark"),
+ accent: str = Query("#3b82f6", description="Accent hex color"),
+ radius: str = Query("md", description="Border radius: none|sm|md|lg|xl"),
+ density: str = Query("normal", description="Density: compact|normal|comfortable"),
+ hide_labels: bool = Query(False, description="Hide text labels"),
+ show: Optional[str] = Query(None, description="Comma-separated stat keys to display"),
+ lang: str = Query("en", description="HTML lang attribute"),
+ font_scale: float = Query(1.0, ge=0.5, le=3.0, description="Font size multiplier"),
+ layout: str = Query("grid", description="Layout: grid|list"),
+ session: Session = Depends(get_session),
+):
+ user = _verify_embed_token(token, session, request)
+
+ books = list(
+ session.exec(
+ select(Book).where(Book.user_id == user.id)
+ ).all()
+ )
+
+ stats = _compute_stats(books)
+
+ show_set: Optional[set[str]] = None
+ if show:
+ keys = {k.strip() for k in show.split(",") if k.strip()}
+ invalid = keys - VALID_STAT_KEYS
+ if invalid:
+ raise HTTPException(
+ status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+ detail=f"Invalid stat keys: {', '.join(sorted(invalid))}. Valid: {', '.join(sorted(VALID_STAT_KEYS))}",
+ )
+ show_set = keys if keys else None
+
+ if layout not in LAYOUT_MODES:
+ raise HTTPException(
+ status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+ detail=f"Invalid layout '{layout}'. Valid: {', '.join(sorted(LAYOUT_MODES))}",
+ )
+
+ html_content = _render_stats_html(
+ stats, theme, accent, radius, density, hide_labels, lang, font_scale, layout, show_set,
+ )
+
+ return Response(
+ content=html_content,
+ media_type="text/html; charset=utf-8",
+ headers={
+ "Cache-Control": "private, max-age=60",
+ "X-Content-Type-Options": "nosniff",
+ "Referrer-Policy": "no-referrer",
+ "Content-Security-Policy": "default-src 'none'; style-src 'unsafe-inline'; frame-ancestors *",
+ },
+ )
diff --git a/backend/app/routers/profile.py b/backend/app/routers/profile.py
index 256a66aa..fa6c169b 100644
--- a/backend/app/routers/profile.py
+++ b/backend/app/routers/profile.py
@@ -9,14 +9,17 @@
clear_browser_session,
ensure_password_complexity,
generate_api_key,
+ generate_embed_token,
get_api_key_prefix,
+ get_embed_token_prefix,
get_password_hash,
hash_api_key,
+ hash_embed_token,
require_user,
)
from app.config import settings as app_settings
from app.database import get_session
-from app.models import ApiKey, User, UserSettings
+from app.models import ApiKey, EmbedToken, User, UserSettings
from app.schemas import (
ApiKeyCreate,
ApiKeyCreateResponse,
@@ -24,6 +27,10 @@
ConfirmationPhrase,
DataResetDeleted,
DataResetResponse,
+ EmbedTokenCreate,
+ EmbedTokenCreateResponse,
+ EmbedTokenRead,
+ EmbedTokenUpdate,
ProfileUpdate,
UserRead,
UserSettingsRead,
@@ -94,7 +101,6 @@ def get_settings(
user_id=settings.user_id,
language=settings.language,
timezone=settings.timezone,
- quote_service_enabled=app_settings.dashboard_quote_enabled,
theme=settings.theme,
custom_theme=settings.custom_theme,
)
@@ -123,7 +129,6 @@ def update_settings(
user_id=settings.user_id,
language=settings.language,
timezone=settings.timezone,
- quote_service_enabled=app_settings.dashboard_quote_enabled,
theme=settings.theme,
custom_theme=settings.custom_theme,
)
@@ -200,7 +205,7 @@ def list_api_keys(
select(ApiKey).where(
ApiKey.user_id == current_user.id,
ApiKey.revoked_at.is_(None),
- )
+ ).order_by(ApiKey.created_at.desc())
).all()
return [ApiKeyRead.model_validate(k) for k in keys]
@@ -239,3 +244,119 @@ def delete_api_key(
key.revoked_at = utcnow()
session.add(key)
session.commit()
+
+
+@router.get("/embed-tokens", response_model=list[EmbedTokenRead])
+def list_embed_tokens(
+ current_user: User = Depends(require_user),
+ session: Session = Depends(get_session),
+) -> list[EmbedTokenRead]:
+ """List non-revoked embed tokens for the current user."""
+ tokens = session.exec(
+ select(EmbedToken).where(
+ EmbedToken.user_id == current_user.id,
+ EmbedToken.revoked_at.is_(None),
+ ).order_by(EmbedToken.created_at.desc())
+ ).all()
+ return [EmbedTokenRead.model_validate(t) for t in tokens]
+
+
+@router.post("/embed-tokens", response_model=EmbedTokenCreateResponse, status_code=201)
+def create_embed_token(
+ body: EmbedTokenCreate,
+ current_user: User = Depends(require_user),
+ session: Session = Depends(get_session),
+) -> EmbedTokenCreateResponse:
+ """Create a new embed token for the current user."""
+ plain_token = generate_embed_token()
+ token = EmbedToken(
+ user_id=current_user.id,
+ name=body.name,
+ token_prefix=get_embed_token_prefix(plain_token),
+ token_hash=hash_embed_token(plain_token),
+ allowed_origins=body.allowed_origins or None,
+ expires_at=body.expires_at,
+ )
+ session.add(token)
+ session.commit()
+ session.refresh(token)
+ return EmbedTokenCreateResponse(
+ token=plain_token,
+ embed_token=EmbedTokenRead.model_validate(token),
+ )
+
+
+@router.patch("/embed-tokens/{token_id}", response_model=EmbedTokenRead)
+def update_embed_token(
+ token_id: int,
+ body: EmbedTokenUpdate,
+ current_user: User = Depends(require_user),
+ session: Session = Depends(get_session),
+) -> EmbedTokenRead:
+ """Update name, allowed_origins, or expires_at for an embed token."""
+ token = session.get(EmbedToken, token_id)
+ if not token or token.user_id != current_user.id or token.revoked_at is not None:
+ raise HTTPException(status_code=404, detail="Embed token not found")
+
+ update_data = body.model_dump(exclude_unset=True)
+ token.sqlmodel_update(update_data)
+ session.add(token)
+ session.commit()
+ session.refresh(token)
+ return EmbedTokenRead.model_validate(token)
+
+
+@router.post("/embed-tokens/{token_id}/rotate", response_model=EmbedTokenCreateResponse)
+def rotate_embed_token(
+ token_id: int,
+ current_user: User = Depends(require_user),
+ session: Session = Depends(get_session),
+) -> EmbedTokenCreateResponse:
+ """Revoke the current embed token and create a new one with the same settings."""
+ token = session.get(EmbedToken, token_id)
+ if not token or token.user_id != current_user.id or token.revoked_at is not None:
+ raise HTTPException(status_code=404, detail="Embed token not found")
+
+ now = utcnow()
+ if token.expires_at is not None and token.expires_at < now:
+ raise HTTPException(
+ status_code=409,
+ detail="Expired embed tokens cannot be rotated",
+ )
+
+ token.revoked_at = now
+ session.add(token)
+
+ plain_token = generate_embed_token()
+ new_token = EmbedToken(
+ user_id=current_user.id,
+ name=token.name,
+ token_prefix=get_embed_token_prefix(plain_token),
+ token_hash=hash_embed_token(plain_token),
+ scopes=token.scopes,
+ allowed_origins=token.allowed_origins,
+ expires_at=token.expires_at,
+ )
+ session.add(new_token)
+ session.commit()
+ session.refresh(new_token)
+ return EmbedTokenCreateResponse(
+ token=plain_token,
+ embed_token=EmbedTokenRead.model_validate(new_token),
+ )
+
+
+@router.delete("/embed-tokens/{token_id}", status_code=204)
+def delete_embed_token(
+ token_id: int,
+ current_user: User = Depends(require_user),
+ session: Session = Depends(get_session),
+) -> None:
+ """Revoke an embed token by ID."""
+ token = session.get(EmbedToken, token_id)
+ if not token or token.user_id != current_user.id:
+ raise HTTPException(status_code=404, detail="Embed token not found")
+
+ token.revoked_at = utcnow()
+ session.add(token)
+ session.commit()
diff --git a/backend/app/schemas.py b/backend/app/schemas.py
index 80810a62..96c3b018 100644
--- a/backend/app/schemas.py
+++ b/backend/app/schemas.py
@@ -324,7 +324,6 @@ class UserSettingsRead(SQLModel):
user_id: int
language: str
timezone: str
- quote_service_enabled: bool
theme: str
custom_theme: Optional[str] = None
@@ -419,6 +418,13 @@ class OidcConfigRead(SQLModel):
provider_name: Optional[str] = None
+class AppConfigRead(SQLModel):
+ """Application-level feature flags."""
+ embed_enabled: bool
+ dashboard_quote_enabled: bool
+ thalia_cover_search_enabled: bool
+
+
class OidcLinkRead(SQLModel):
"""OIDC link status."""
linked: bool
@@ -599,6 +605,38 @@ class DataImportPreviewResponse(SQLModel):
errors: list[str] = []
+class EmbedTokenCreate(SQLModel):
+ """Embed token creation request."""
+ name: str = Field(max_length=255)
+ allowed_origins: Optional[str] = None
+ expires_at: Optional[datetime] = None
+
+
+class EmbedTokenUpdate(SQLModel):
+ """Embed token update request."""
+ name: Optional[str] = None
+ allowed_origins: Optional[str] = None
+ expires_at: Optional[datetime] = None
+
+
+class EmbedTokenRead(SQLModel):
+ """Embed token read response (without the raw token value)."""
+ id: int
+ name: str
+ token_prefix: str
+ scopes: str
+ allowed_origins: Optional[str]
+ expires_at: Optional[datetime]
+ last_used_at: Optional[datetime]
+ created_at: datetime
+
+
+class EmbedTokenCreateResponse(SQLModel):
+ """Embed token creation response containing the raw token (shown once)."""
+ token: str
+ embed_token: EmbedTokenRead
+
+
class DataImportExecuteResult(SQLModel):
"""Import execution result summary."""
imported: int
diff --git a/backend/scripts/gen_db_docs.py b/backend/scripts/gen_db_docs.py
new file mode 100644
index 00000000..de62c19b
--- /dev/null
+++ b/backend/scripts/gen_db_docs.py
@@ -0,0 +1,225 @@
+#!/usr/bin/env python3
+"""Generate docs/guide/database-layout.md from SQLModel metadata.
+
+Usage:
+ uv run --directory backend python scripts/gen_db_docs.py
+"""
+
+import sys
+from pathlib import Path
+from datetime import datetime
+from enum import Enum as PyEnum
+
+BACKEND_DIR = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(BACKEND_DIR))
+
+import sqlalchemy as sa
+from sqlalchemy import TypeDecorator
+from sqlmodel import SQLModel
+from app import models # noqa: E402
+from app.models import UtcDateTime
+
+OUTPUT = Path(BACKEND_DIR).parent / "docs" / "guide" / "database-layout.md"
+NOW = datetime.now().strftime("%Y-%m-%d")
+
+
+def _model_for_table(name: str):
+ for obj in models.__dict__.values():
+ if isinstance(obj, type) and issubclass(obj, SQLModel) and hasattr(obj, '__tablename__'):
+ if obj.__tablename__ == name:
+ return obj
+ return None
+
+
+def _resolve_type(col: sa.Column) -> str:
+ t = col.type
+ if isinstance(t, UtcDateTime):
+ return "DATETIME"
+ if isinstance(t, sa.Enum):
+ return "VARCHAR"
+ return str(t).upper()
+
+
+def _resolve_constraints(col: sa.Column) -> str:
+ parts = []
+ if col.primary_key:
+ parts.append("PK")
+ for fk in col.foreign_keys:
+ parts.append(f"FK → {fk.column.table.name}.{fk.column.name}")
+ if col.unique and not col.primary_key:
+ parts.append("UNIQUE")
+ if not col.nullable and not col.primary_key:
+ parts.append("NOT NULL")
+ if col.index and not col.primary_key and not col.unique:
+ parts.append("INDEX")
+ return ", ".join(parts)
+
+
+def _field_metadata(col: sa.Column):
+ model = _model_for_table(col.table.name)
+ if model and col.name in (model.model_fields or {}):
+ return model.model_fields[col.name]
+ return None
+
+
+def _resolve_notes(col: sa.Column) -> str:
+ note_parts = []
+
+ if isinstance(col.type, UtcDateTime):
+ note_parts.append("UTC")
+
+ single_pk = len(col.table.primary_key.columns) == 1
+ if col.primary_key and single_pk and isinstance(col.type, sa.Integer):
+ note_parts.append("Auto-increment")
+
+ # default value
+ if col.default is not None:
+ arg = getattr(col.default, "arg", None)
+ if arg is not None and not callable(arg):
+ if isinstance(arg, PyEnum):
+ note_parts.append(f"default `{arg.value}`")
+ elif isinstance(arg, str):
+ note_parts.append(f"default `{arg}`")
+ else:
+ note_parts.append(f"default {arg}")
+
+ # ge / le from pydantic field metadata
+ field = _field_metadata(col)
+ if field:
+ for item in field.metadata:
+ cls_name = type(item).__name__
+ if cls_name == "Ge":
+ note_parts.append(f"≥ {item.ge}")
+ elif cls_name == "Le":
+ note_parts.append(f"≤ {item.le}")
+
+ return "; ".join(note_parts) if note_parts else ""
+
+
+def _cardinality(col: sa.Column) -> str:
+ if col.unique:
+ return "o|" if col.nullable else "||"
+ return "o{" if col.nullable else "|{"
+
+
+def _generate_er_diagram(tables) -> str:
+ lines = ["```mermaid", "erDiagram", ""]
+ rels = []
+ seen = set()
+ for table in tables:
+ for col in table.columns:
+ for fk in col.foreign_keys:
+ parent = fk.column.table.name
+ child = table.name
+ key = (parent, child)
+ if key not in seen:
+ seen.add(key)
+ rels.append((parent, child, col))
+ card_map = {"||": "1:1", "|{": "1:N", "o|": "0..1", "o{": "0..N"}
+ for parent, child, col in rels:
+ card = _cardinality(col)
+ label = card_map.get(card, card)
+ lines.append(f' {parent} ||--{card} {child} : "{label}"')
+ lines.append("")
+ for table in tables:
+ lines.append(f" {table.name} {{")
+ for col in table.columns:
+ typ = _resolve_type(col).lower()
+ name = col.name
+ key_flag = "PK" if col.primary_key else ("UK" if col.unique else "")
+ parts = [typ, name]
+ if key_flag:
+ parts.append(key_flag)
+ lines.append(f" {' '.join(parts)}")
+ lines.append(" }")
+ lines.append("")
+ lines.append("```")
+ return "\n".join(lines)
+
+
+def _generate_table_docs(tables) -> str:
+ lines = ["## Tables", ""]
+ for table in tables:
+ model = _model_for_table(table.name)
+ doc = (model.__doc__ or "").strip() if model else ""
+
+ lines.append(f"### `{table.name}`")
+ lines.append("")
+ if doc:
+ lines.append(doc)
+ lines.append("")
+
+ lines.append("| Column | Type | Constraints | Notes |")
+ lines.append("|--------|------|-------------|-------|")
+ for col in table.columns:
+ typ = _resolve_type(col)
+ cnst = _resolve_constraints(col)
+ notes = _resolve_notes(col)
+ lines.append(f"| `{col.name}` | `{typ}` | {cnst} | {notes} |")
+
+ ucs = [uc for uc in table.constraints
+ if isinstance(uc, sa.UniqueConstraint)
+ and not any(c.unique for c in uc.columns)]
+ if ucs:
+ lines.append("")
+ for uc in ucs:
+ col_names = ", ".join(c.name for c in uc.columns)
+ lines.append(f"**Unique constraint:** `({col_names})` — {uc.name}")
+
+ lines.append("")
+ return "\n".join(lines)
+
+
+def _generate_enum_docs() -> str:
+ lines = ["## Enums", ""]
+ for name in dir(models):
+ cls = getattr(models, name)
+ if isinstance(cls, type) and issubclass(cls, PyEnum) and cls.__module__ == models.__name__:
+ lines.append(f"### `{name}`")
+ lines.append("")
+ lines.append("| Value | Meaning |")
+ lines.append("|-------|---------|")
+ for member in cls:
+ lines.append(f"| `{member.value}` | {member.value.replace('_', ' ').title()} |")
+ lines.append("")
+ return "\n".join(lines)
+
+
+def main():
+ tables = list(SQLModel.metadata.sorted_tables)
+
+ content = f"""# Database Layout
+
+> _Auto-generated from SQLModel metadata on {NOW}._
+
+This page documents the LibrisLog database schema. It is intended for
+developers who need to understand the data model, write queries, or extend
+the application.
+
+{_generate_er_diagram(tables)}
+
+{_generate_table_docs(tables)}
+
+{_generate_enum_docs()}
+
+## Conventions
+
+- **Timestamps** are stored as UTC via the `UtcDateTime` type decorator.
+ Values are stored as naive UTC in SQLite and returned as timezone-aware
+ `datetime` objects by the application.
+- **Soft deletes** — `ApiKey` and `EmbedToken` use a `revoked_at` timestamp
+ instead of `DELETE`. Revoked entries are excluded from all queries.
+- **Foreign keys** — all user-owned tables reference `user.id` via foreign
+ key constraints. Cascading behavior is handled in application code (not
+ at the database level).
+- **Unique constraints** — compound constraints like `(user_id, isbn)` on
+ `book` and `(user_id, name)` on `tag` enforce per-user uniqueness without
+ restricting other users.
+"""
+ OUTPUT.parent.mkdir(parents=True, exist_ok=True)
+ OUTPUT.write_text(content)
+ print(f"✓ Wrote {OUTPUT}")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/backend/tests/test_embed.py b/backend/tests/test_embed.py
new file mode 100644
index 00000000..2b0e45b5
--- /dev/null
+++ b/backend/tests/test_embed.py
@@ -0,0 +1,359 @@
+"""Tests for embed token lifecycle and the embed HTML widget endpoint."""
+
+import re
+from collections.abc import Callable
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlmodel import Session, select
+
+from app.auth import generate_embed_token, hash_embed_token
+from app.models import Book, EmbedToken, UserRole
+from app.time_utils import utcnow
+
+# ── Helpers ──────────────────────────────────────────────────────────────
+
+
+def _create_token(
+ session: Session,
+ user_id: int,
+ *,
+ name: str = "Test Token",
+ allowed_origins: str | None = None,
+ expires_at: datetime | None = None,
+) -> str:
+ plain = generate_embed_token()
+ token = EmbedToken(
+ user_id=user_id,
+ name=name,
+ token_prefix=plain[:12],
+ token_hash=hash_embed_token(plain),
+ allowed_origins=allowed_origins,
+ expires_at=expires_at,
+ )
+ session.add(token)
+ session.commit()
+ return plain
+
+
+# ── Profile CRUD Tests ───────────────────────────────────────────────────
+
+
+class TestListEmbedTokens:
+ def test_lists_own_tokens(self, client: TestClient, session: Session) -> None:
+ plain = _create_token(session, user_id=1)
+ resp = client.get("/api/profile/embed-tokens")
+ assert resp.status_code == 200
+ data = resp.json()
+ assert len(data) >= 1
+ assert data[0]["token_prefix"] == plain[:12]
+
+ def test_does_not_list_revoked_tokens(
+ self, client: TestClient, session: Session, create_user_with_key: Callable[..., Any]
+ ) -> None:
+ user, key = create_user_with_key(email="embed_list@example.com")
+ client.headers["X-API-Key"] = key
+ plain = _create_token(session, user.id)
+ # Revoke it
+ token = session.exec(
+ select(EmbedToken).where(EmbedToken.token_hash == hash_embed_token(plain))
+ ).first()
+ assert token is not None
+ token.revoked_at = utcnow()
+ session.add(token)
+ session.commit()
+ resp = client.get("/api/profile/embed-tokens")
+ assert resp.status_code == 200
+ assert all(t["token_prefix"] != plain[:12] for t in resp.json())
+
+ def test_other_user_cannot_see_tokens(
+ self, client: TestClient, session: Session, create_user_with_key: Callable[..., Any]
+ ) -> None:
+ user, key = create_user_with_key(email="other_embed@example.com")
+ client.headers["X-API-Key"] = key
+ _create_token(session, user_id=user.id, name="Secret")
+ resp = client.get("/api/profile/embed-tokens")
+ assert resp.status_code == 200
+ assert len(resp.json()) == 1 # only own tokens
+
+
+class TestCreateEmbedToken:
+ def test_creates_token(self, client: TestClient) -> None:
+ resp = client.post(
+ "/api/profile/embed-tokens",
+ json={"name": "Homarr Widget"},
+ )
+ assert resp.status_code == 201
+ data = resp.json()
+ assert data["token"].startswith("le_")
+ assert data["embed_token"]["name"] == "Homarr Widget"
+ assert data["embed_token"]["token_prefix"] == data["token"][:12]
+
+ def test_with_allowed_origins(self, client: TestClient) -> None:
+ resp = client.post(
+ "/api/profile/embed-tokens",
+ json={
+ "name": "With Origins",
+ "allowed_origins": "https://homarr.local,https://dashy.local",
+ },
+ )
+ assert resp.status_code == 201
+ data = resp.json()
+ assert data["embed_token"]["allowed_origins"] == "https://homarr.local,https://dashy.local"
+
+ def test_with_expiry(self, client: TestClient) -> None:
+ expires = (datetime.now(timezone.utc) + timedelta(days=90)).isoformat()
+ resp = client.post(
+ "/api/profile/embed-tokens",
+ json={"name": "Expiring", "expires_at": expires},
+ )
+ assert resp.status_code == 201
+
+
+class TestUpdateEmbedToken:
+ def test_updates_name_and_origins(self, client: TestClient, session: Session) -> None:
+ plain = _create_token(session, user_id=1)
+ token = session.exec(
+ select(EmbedToken).where(EmbedToken.token_hash == hash_embed_token(plain))
+ ).first()
+ assert token is not None
+ resp = client.patch(
+ f"/api/profile/embed-tokens/{token.id}",
+ json={"name": "Renamed", "allowed_origins": "https://example.com"},
+ )
+ assert resp.status_code == 200
+ assert resp.json()["name"] == "Renamed"
+ assert resp.json()["allowed_origins"] == "https://example.com"
+
+ def test_not_found_for_other_user(
+ self, client: TestClient, session: Session, create_user_with_key: Callable[..., Any]
+ ) -> None:
+ user, key = create_user_with_key(email="other_update@example.com")
+ client.headers["X-API-Key"] = key
+ resp = client.patch("/api/profile/embed-tokens/99999", json={"name": "Nope"})
+ assert resp.status_code == 404
+
+
+class TestRotateEmbedToken:
+ def test_rotate_returns_new_token(self, client: TestClient, session: Session) -> None:
+ plain = _create_token(session, user_id=1, name="Rotate Me")
+ token = session.exec(
+ select(EmbedToken).where(EmbedToken.token_hash == hash_embed_token(plain))
+ ).first()
+ assert token is not None
+ old_id = token.id
+ resp = client.post(f"/api/profile/embed-tokens/{old_id}/rotate")
+ assert resp.status_code == 200
+ data = resp.json()
+ assert data["token"].startswith("le_")
+ assert data["token"] != plain
+ assert data["embed_token"]["name"] == "Rotate Me"
+ assert data["embed_token"]["id"] != old_id
+ # Old token should be revoked
+ session.expire_all()
+ old = session.get(EmbedToken, old_id)
+ assert old is not None
+ assert old.revoked_at is not None
+
+ def test_cannot_rotate_expired_token(self, client: TestClient, session: Session) -> None:
+ plain = _create_token(
+ session,
+ user_id=1,
+ name="Expired",
+ expires_at=datetime.now(timezone.utc) - timedelta(hours=1),
+ )
+ token = session.exec(
+ select(EmbedToken).where(EmbedToken.token_hash == hash_embed_token(plain))
+ ).first()
+ assert token is not None
+
+ resp = client.post(f"/api/profile/embed-tokens/{token.id}/rotate")
+ assert resp.status_code == 409
+ assert resp.json()["detail"] == "Expired embed tokens cannot be rotated"
+
+
+class TestDeleteEmbedToken:
+ def test_revokes_token(self, client: TestClient, session: Session) -> None:
+ plain = _create_token(session, user_id=1)
+ token = session.exec(
+ select(EmbedToken).where(EmbedToken.token_hash == hash_embed_token(plain))
+ ).first()
+ assert token is not None
+ resp = client.delete(f"/api/profile/embed-tokens/{token.id}")
+ assert resp.status_code == 204
+ session.expire_all()
+ deleted = session.get(EmbedToken, token.id)
+ assert deleted is not None
+ assert deleted.revoked_at is not None
+
+ def test_not_found(self, client: TestClient) -> None:
+ resp = client.delete("/api/profile/embed-tokens/99999")
+ assert resp.status_code == 404
+
+
+# ── Embed Widget Endpoint Tests ──────────────────────────────────────────
+
+
+class TestEmbedStatsEndpoint:
+ def test_returns_html_with_valid_token(self, client: TestClient, session: Session) -> None:
+ plain = _create_token(session, user_id=1)
+ resp = client.get(f"/embed/v1/stats?token={plain}")
+ assert resp.status_code == 200
+ assert resp.headers["content-type"].startswith("text/html")
+ assert "" in resp.text
+ assert "Books" in resp.text
+
+ def test_401_for_invalid_token(self, client: TestClient) -> None:
+ resp = client.get("/embed/v1/stats?token=le_invalid")
+ assert resp.status_code == 401
+
+ def test_401_for_revoked_token(self, client: TestClient, session: Session) -> None:
+ plain = _create_token(session, user_id=1)
+ token = session.exec(
+ select(EmbedToken).where(EmbedToken.token_hash == hash_embed_token(plain))
+ ).first()
+ assert token is not None
+ token.revoked_at = utcnow()
+ session.add(token)
+ session.commit()
+ resp = client.get(f"/embed/v1/stats?token={plain}")
+ assert resp.status_code == 401
+
+ def test_401_for_expired_token(self, client: TestClient, session: Session) -> None:
+ plain = _create_token(
+ session, user_id=1,
+ expires_at=datetime.now(timezone.utc) - timedelta(hours=1),
+ )
+ resp = client.get(f"/embed/v1/stats?token={plain}")
+ assert resp.status_code == 401
+
+ def test_shows_stats_for_user(self, client: TestClient, session: Session) -> None:
+ for i in range(3):
+ session.add(Book(title=f"Book {i}", author="Author", page_count=200, user_id=1, reading_status="read"))
+ session.commit()
+
+ plain = _create_token(session, user_id=1)
+ resp = client.get(f"/embed/v1/stats?token={plain}")
+ assert resp.status_code == 200
+ assert "3" in resp.text
+
+ def test_user_isolation(self, client: TestClient, session: Session) -> None:
+ for i in range(5):
+ session.add(Book(title=f"U1 Book {i}", author="Author", page_count=100, user_id=1, reading_status="want_to_read"))
+ session.commit()
+
+ from app.auth import generate_api_key, get_api_key_prefix, get_password_hash, hash_api_key, encrypt_api_key
+ from app.models import ApiKey, User, UserSettings
+
+ user2 = User(firstname="User", lastname="Two", email="u2@example.com",
+ role=UserRole.user, hashed_password=get_password_hash("secret"))
+ session.add(user2)
+ session.commit()
+ session.refresh(user2)
+ session.add(UserSettings(user_id=user2.id, language="en"))
+ key2 = generate_api_key()
+ session.add(ApiKey(user_id=user2.id, key_prefix=get_api_key_prefix(key2),
+ key_hash=hash_api_key(key2), key_encrypted=encrypt_api_key(key2),
+ description="Key"))
+ session.commit()
+ plain = _create_token(session, user2.id)
+ resp = client.get(f"/embed/v1/stats?token={plain}")
+ assert resp.status_code == 200
+ text = resp.text
+ books_match = re.search(r'
]*>\s*0\s*