Compare commits

..
8 Commits
26 changed files with 1268 additions and 47 deletions

No files matched your search

+3 -1
View File
@@ -11,7 +11,9 @@
"Bash(DATABASE_URL=\"sqlite+aiosqlite:////tmp/claude-1000/-home-iroddis-dev-armament-fitguide/41e5e14d-88fe-45e7-ba3a-be5a2cbde2cd/scratchpad/mig_test.db\" uv run alembic upgrade head)",
"Bash(sqlite3 /tmp/claude-1000/-home-iroddis-dev-armament-fitguide/41e5e14d-88fe-45e7-ba3a-be5a2cbde2cd/scratchpad/mig_test.db '.schema users')",
"Bash(docker build *)",
"Bash(git status *)"
"Bash(git status *)",
"Bash(git -C /home/iroddis/dev/armament/fitguide status --short)",
"Bash(git -C /home/iroddis/dev/armament/fitguide log --oneline -10)"
]
}
}
+94 -13
View File
@@ -1,26 +1,41 @@
# CLAUDE.md
Guidance for working in this repo. Read alongside `README.md` (user-facing setup).
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
Guidance for working in this repo. Read alongside `README.md` (user-facing setup
for the original FastAPI skeleton this app was built from).
## What this is
A FastAPI skeleton serving **both** a JSON API (`/api/*`) and server-rendered
HTML via **Jinja2 + HTMX** (`/`, `/admin/*`). Async SQLAlchemy 2.0 ORM, JWT auth
with refresh-token revocation, and RBAC (users → roles → permissions).
**Fitguide**: a parts-compatibility guide for Armament scopes and accessories,
built on a FastAPI skeleton that serves **both** a JSON API (`/api/*`) and
server-rendered HTML via **Jinja2 + HTMX** (`/`, `/scopes`, `/admin/*`). Async
SQLAlchemy 2.0 ORM, JWT auth with refresh-token revocation (plus optional LDAP
direct-bind login), and RBAC (users → roles → permissions). The domain layer on
top of the skeleton tracks parts, their attachment points, and which parts fit
together — see "Domain model" below.
## Commands
```bash
uv sync --extra dev # install (incl. dev tools)
uv run alembic upgrade head # apply migrations
uv run python -m scripts.seed # seed perms/roles/superuser (idempotent)
uv run uvicorn app.main:app --reload # dev server on :8000
uv run pytest # tests (in-memory SQLite)
uv run pytest tests/test_foo.py::test_bar # single test
uv run ruff check . [--fix] # lint
uv run alembic upgrade head # apply migrations
uv run black . # format (ruff does not format here)
uv run alembic revision --autogenerate -m "msg" # new migration after model changes
docker compose up --build # app + Postgres, auto-seeded
```
`Makefile` wraps the common ones (`make sync|seed|dev|test|fmt|migrate`).
`make clean` deletes the local `app.db`, re-migrates, and reseeds — the quick
way to get a clean dev database. `make seed` runs `alembic upgrade head` before
seeding, i.e. day-to-day dev relies on migrations, not just
`AUTO_CREATE_TABLES` (see "Database").
## Architecture / layering
Request → router (`app/api/` JSON or `app/web/` HTML) → `app/crud/` (data access)
@@ -28,40 +43,106 @@ Request → router (`app/api/` JSON or `app/web/` HTML) → `app/crud/` (data ac
token logic in `app/services/auth.py`. Keep route handlers thin; put DB logic in
`crud/`, not in routers.
`app/api/` is one router module per resource, all mounted under `/api` by
`app/api/router.py`. `app/web/` is organized differently: `pages.py` (dashboard/
login), `scopes.py` (the public-facing `/scopes` fit guide), and **`admin.py`**
— a single ~3000-line router covering every admin screen (users, roles, part
taxonomy tables, imported parts, augmented parts + detail + nested attachment
points/fits, audit log, etc.), organized into `# --- Section ---` banner
comments per resource rather than split into files. When adding an admin screen,
follow the existing banner-section convention there rather than starting a new
web router module.
## Domain model
The business domain, independent of the auth/RBAC skeleton:
- **`AugmentedPart`** (`augmented_parts`) is the generic "part" entity — a
scope, flip cover, ARD, polarizer, adapter ring, etc. — distinguished by
`PartType`/`SubPartType`, with `PartColour`, `Manufacturer`, and an optional
`ScopeModel`. "Scope" is just one `PartType` value (see `SCOPE_TYPE_NAME` in
`app/services/scope_guide.py`), not a separate table.
- **`ScopeInfo`** is a 1:1 extension of `AugmentedPart` (magnification,
objective thread, model suffix) populated only for scope-type parts.
- **`AttachmentPoint`** belongs to one `AugmentedPart`, is named via
`AttachmentPointName` (e.g. "Ocular", "Objective"), and connects via
*exactly one* of `IsoThread` / `UncThread` / `ClampConnection` /
`StaticConnection` — that "exactly one" invariant is enforced in the app
layer (CRUD/API validation), not by a DB constraint.
- **`Fit`** (`app/models/fit.py`) pairs two `AttachmentPoint`s (canonical
ordering enforced in `app/crud/fit.py` so a pair only ever has one row),
labeled `possible` / `exact` / `none`, with threaded `FitNote`s.
`app/services/fit_matching.py` computes `possible` fits automatically
(matching ISO/UNC thread specs exactly, or a static connection's dimension
falling inside a clamp connection's range); `exact`/`none` are user-asserted
and never touched by the algorithm.
- **`ImportedPart`** (`imported_parts`, PK is the string `InventoryID` — not an
int id) holds raw rows from an external inventory feed, fetched via
`app/services/odata.py` (OData v4, paginated). `app/services/
augmented_part_sync.py` reconciles these into `AugmentedPart` +
`AttachmentPoint` rows: infers `PartType` by substring match on the
description, parses `TBXThreadAttribute` against ISO/UNC regexes (falling
back to an OD-derived `StaticConnection`), and prunes `AugmentedPart`s whose
backing `ImportedPart`s have disappeared.
- **`app/services/scope_import.py`** is a separate, manually-triggered bulk
importer: reads an uploaded `.xlsx` of third-party scopes, upserting by
`product_number` (→ `AugmentedPart.sku`) and creating `ScopeModel`,
`ScopeInfo`, up to three `AttachmentPoint`s, and up to two `Fit`s per row
(see the module docstring for the full column mapping).
- **`app/services/scope_guide.py`** assembles the `/scopes` page: one row per
Scope-type `AugmentedPart`, one column per other `PartType`, each cell
listing parts of that type with a recorded `Fit`.
## Conventions that matter
- **Everything async.** Engine, sessions, CRUD, routes, tests. Use `await`; never
add a sync DB call. `get_db` (in `app/core/database.py`) yields the request session.
- **Relationships use `lazy="selectin"`** (`User.roles`, `Role.permissions`) so
attributes are safe to read after the session closes — required under async.
Preserve this when adding relationships that templates/schemas read.
- **Relationships use `lazy="selectin"`** (`User.roles`, `Role.permissions`,
`AttachmentPoint.iso_thread`, etc.) so attributes are safe to read after the
session closes — required under async. Preserve this when adding relationships
that templates/schemas read.
- **Config** is `app/core/config.py` (`pydantic-settings`). Add new settings there;
they're auto-read from env vars (uppercased field name) or `.env`. Env wins.
- **RBAC**: guard API routes with `Depends(require_permission("resource:action"))`;
guard web routes with `web_require(...)` in `app/web/admin.py`. Superusers bypass
all checks (`User.is_superuser` / `User.has_permission`). Add new permission
strings to `scripts/seed.py`'s `PERMISSIONS`/`ROLES`.
- **HTMX UI-level write gating**: on top of route-level RBAC, pages that mix
read-only and read-write viewers (e.g. `augmented_part_detail.html`'s Fits and
Attachment Points sections) compute a `can_write_<thing>` bool from the
current user's permissions and wrap add/edit/delete controls in
`{% if can_write_<thing> %}` in the template, so read-only users see data but
not controls they'd get a 403 on. The route-level `web_require(...)` check
remains the real security boundary; the template flag is only for UX.
- **Auth, two transports**: API uses `Authorization: Bearer`; the HTMX UI uses an
httpOnly cookie (`settings.cookie_name`). `app/deps.py` reads both. Web auth
failures raise `NotAuthenticated`, handled in `main.py` as a redirect to `/login`.
failures raise `NotAuthenticated`, handled in `main.py` as a redirect to `/login`
(preserving the original path as `?next=`).
- **Refresh tokens** are stored server-side (hash + `jti` + `revoked`) in the
`refresh_tokens` table. `services/auth.rotate_refresh_token` revokes-then-reissues.
Note: JWT payload `exp` is a tz-aware `datetime` (PyJWT only epoch-encodes inside
the token) — pass it straight to the model, don't `fromtimestamp` it.
- **LDAP login** (`app/services/ldap.py`) is optional (`settings.ldap_enabled`),
direct-bind, and just-in-time provisions a local `User` on first successful
login. `ldap3` is sync, so its calls run via `run_in_threadpool`.
## Database
One `DATABASE_URL` drives SQLite (`sqlite+aiosqlite://`, the default) or Postgres
(`postgresql+asyncpg://`). `AUTO_CREATE_TABLES=true` calls `Base.metadata.create_all`
on startup for dev convenience; production should use Alembic instead. `init_db()`
imports `app.models` inside the function to avoid a circular import.
on startup for dev convenience, but the day-to-day `make seed`/`make clean` flow
runs `alembic upgrade head` first — treat Alembic migrations as the source of
truth and keep them in sync with model changes. `init_db()` imports `app.models`
inside the function to avoid a circular import.
## Tests
`tests/conftest.py` builds an in-memory SQLite engine with `StaticPool`, creates
tables, and overrides `get_db`. Fixtures: `client` (httpx ASGI), `seeded`
(perms+roles), `make_user`. `asyncio_mode = "auto"` — write `async def test_*`
without decorators.
(perms+roles from `scripts.seed.PERMISSIONS`/`ROLES`), `make_user`, and
`db_sessionmaker` (used directly in tests to seed domain rows — models, fits,
attachment points — beyond what `seeded`/`make_user` cover). `asyncio_mode =
"auto"` — write `async def test_*` without decorators.
## When extending
+10
View File
@@ -31,6 +31,16 @@ async def list_all(
return list(result.scalars().all())
async def get_many(db: AsyncSession, ids: list[int]) -> list[AugmentedPart]:
"""Fetch augmented parts by id in one query (selectin relationships loaded)."""
if not ids:
return []
result = await db.execute(
select(AugmentedPart).where(AugmentedPart.id.in_(ids))
)
return list(result.scalars().all())
async def list_filtered(
db: AsyncSession,
description: str = "",
+23
View File
@@ -52,6 +52,29 @@ async def list_for_part(db: AsyncSession, augmented_part_id: int) -> list[Fit]:
return list(result.scalars().all())
async def list_for_parts(
db: AsyncSession,
part_ids: list[int],
exclude_label: FitLabel | None = None,
) -> list[Fit]:
"""All fits touching any of the given augmented parts, in one query.
Pass ``exclude_label`` to drop, e.g., confirmed non-fits (``FitLabel.NONE``).
"""
if not part_ids:
return []
q = select(Fit).where(
or_(
Fit.augmented_part_a_id.in_(part_ids),
Fit.augmented_part_b_id.in_(part_ids),
)
)
if exclude_label is not None:
q = q.where(Fit.label != exclude_label)
result = await db.execute(q.order_by(Fit.id))
return list(result.scalars().all())
async def list_all(
db: AsyncSession,
skip: int = 0,
+14
View File
@@ -17,6 +17,20 @@ async def get_by_augmented_part(
return result.scalar_one_or_none()
async def list_by_augmented_parts(
db: AsyncSession, augmented_part_ids: list[int]
) -> list[ScopeInfo]:
"""ScopeInfo rows for the given augmented parts, fetched in one query."""
if not augmented_part_ids:
return []
result = await db.execute(
select(ScopeInfo).where(
ScopeInfo.augmented_part_id.in_(augmented_part_ids)
)
)
return list(result.scalars().all())
async def create(
db: AsyncSession,
*,
+12 -1
View File
@@ -1,5 +1,6 @@
from contextlib import asynccontextmanager
from pathlib import Path
from urllib.parse import quote
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
@@ -12,6 +13,7 @@ from app.core.database import init_db
from app.deps import NotAuthenticated
from app.web.admin import router as admin_router
from app.web.pages import router as pages_router
from app.web.scopes import router as scopes_router
STATIC_DIR = Path(__file__).resolve().parent / "static"
@@ -39,11 +41,20 @@ def create_app() -> FastAPI:
app.include_router(api_router, prefix="/api")
app.include_router(pages_router)
app.include_router(scopes_router)
app.include_router(admin_router)
@app.exception_handler(NotAuthenticated)
async def _redirect_to_login(request: Request, exc: NotAuthenticated):
return RedirectResponse(url="/login", status_code=302)
# Preserve the originally requested URL so we can return there after
# login. Only meaningful for GET navigations (the ones users deep-link).
login_url = "/login"
if request.method == "GET" and request.url.path not in ("/", "/login"):
target = request.url.path
if request.url.query:
target += "?" + request.url.query
login_url = f"/login?next={quote(target, safe='')}"
return RedirectResponse(url=login_url, status_code=302)
@app.get("/healthz", tags=["health"])
async def healthz():
+32 -12
View File
@@ -55,7 +55,7 @@ MM_PER_INCH = 25.4
# M50x.75 - No leading digit on floats
# Float pattern: \d*\.?\d+
_ISO_THREAD_RE = re.compile(
r"^M\s*(?P<diameter>\d*\.?\d+)\s*[xX-]\s*P?(?P<pitch>\d*\.?\d+)(?P<extra>.*)$"
r"M\s*(?P<diameter>\d*\.?\d+)\s*[xX-]\s*P?(?P<pitch>\d*\.?\d+)(?P<extra>.*)"
)
# e.g.
@@ -65,8 +65,8 @@ _ISO_THREAD_RE = re.compile(
#
# Leading digit / period, one or more digits, a dash, then
_UNC_THREAD_RE = re.compile(
r"^(?P<diameter>\d*\.?\d+)\s*-\s*(?P<tpi>\d*\.?\d+)"
r"\s*(?P<series>UNEF|UNC|UNF|UNS|UN)?$",
r"(?P<diameter>\d*\.?\d+)\s*-\s*(?P<tpi>\d*\.?\d+)"
r"\s*(?P<series>UNEF|UNC|UNF|UNS|UN)?",
re.IGNORECASE,
)
@@ -111,7 +111,9 @@ async def _get_or_create_attachment_point_name(db: AsyncSession, name: str):
return apn
async def _add_attachment_point(db: AsyncSession, ap: AugmentedPart, ip: ImportedPart) -> None:
async def _add_attachment_point(
db: AsyncSession, ap: AugmentedPart, ip: ImportedPart
) -> None:
"""Infer and create an attachment point for a newly-synced AugmentedPart.
Preference order: TBXThreadAttribute (ISO/UNC thread) then ProductODmm/in
@@ -119,7 +121,11 @@ async def _add_attachment_point(db: AsyncSession, ap: AugmentedPart, ip: Importe
attribute is present but unparseable.
"""
thread_value = (ip.TBXThreadAttribute or "").strip()
parsed = _parse_thread_attribute(thread_value) if thread_value else None
parsed = None
if thread_value:
parsed = _parse_thread_attribute(thread_value)
else:
parsed = _parse_thread_attribute(ip.Description or "")
if parsed is not None:
kind, fields = parsed
@@ -127,7 +133,9 @@ async def _add_attachment_point(db: AsyncSession, ap: AugmentedPart, ip: Importe
if kind == "iso":
depth_mm = _parse_float(ip.ProductMaxThreadDepthmm)
thread = await iso_thread_crud.get_by_values(db, depth_mm=depth_mm, **fields)
thread = await iso_thread_crud.get_by_values(
db, depth_mm=depth_mm, **fields
)
if thread is None:
thread = await iso_thread_crud.create(db, depth_mm=depth_mm, **fields)
await attachment_point_crud.create(
@@ -138,7 +146,9 @@ async def _add_attachment_point(db: AsyncSession, ap: AugmentedPart, ip: Importe
)
elif kind == "unc":
depth_in = _parse_float(ip.ProductMaxThreadDepthin)
thread = await unc_thread_crud.get_by_values(db, depth_in=depth_in, **fields)
thread = await unc_thread_crud.get_by_values(
db, depth_in=depth_in, **fields
)
if thread is None:
thread = await unc_thread_crud.create(db, depth_in=depth_in, **fields)
await attachment_point_crud.create(
@@ -236,7 +246,9 @@ async def sync(db: AsyncSession) -> AugmentedPartSyncResult:
# ------------------------------------------------------------------ phase 1
# IDs referenced in existing associations
result = await db.execute(select(augmented_part_imported_parts.c.imported_part_id).distinct())
result = await db.execute(
select(augmented_part_imported_parts.c.imported_part_id).distinct()
)
all_linked_ids: set[str] = {row[0] for row in result.all()}
# IDs that still exist in imported_parts
@@ -250,7 +262,9 @@ async def sync(db: AsyncSession) -> AugmentedPartSyncResult:
# Which AugmentedParts are affected?
result = await db.execute(
select(augmented_part_imported_parts.c.augmented_part_id).where(
augmented_part_imported_parts.c.imported_part_id.in_(orphaned_import_ids)
augmented_part_imported_parts.c.imported_part_id.in_(
orphaned_import_ids
)
)
)
orphaned_ap_ids: set[int] = {row[0] for row in result.all()}
@@ -258,7 +272,9 @@ async def sync(db: AsyncSession) -> AugmentedPartSyncResult:
# Remove the stale association rows
await db.execute(
augmented_part_imported_parts.delete().where(
augmented_part_imported_parts.c.imported_part_id.in_(orphaned_import_ids)
augmented_part_imported_parts.c.imported_part_id.in_(
orphaned_import_ids
)
)
)
await db.flush()
@@ -281,7 +297,9 @@ async def sync(db: AsyncSession) -> AugmentedPartSyncResult:
# ------------------------------------------------------------------ phase 2
# Re-query linked IDs after phase 1 cleanup
result = await db.execute(select(augmented_part_imported_parts.c.imported_part_id).distinct())
result = await db.execute(
select(augmented_part_imported_parts.c.imported_part_id).distinct()
)
now_linked_ids: set[str] = {row[0] for row in result.all()}
unlinked_ids = current_ids - now_linked_ids
@@ -289,7 +307,9 @@ async def sync(db: AsyncSession) -> AugmentedPartSyncResult:
skipped = 0
if not unlinked_ids:
return AugmentedPartSyncResult(created=created, deleted=deleted, skipped=skipped)
return AugmentedPartSyncResult(
created=created, deleted=deleted, skipped=skipped
)
result = await db.execute(
select(ImportedPart).where(ImportedPart.InventoryID.in_(unlinked_ids))
+137
View File
@@ -0,0 +1,137 @@
"""Assemble the scope-centric fit guide.
Builds the data for the top-level ``/scopes`` page: one row per augmented part
of part type "Scope", one column per non-Scope part type, where each cell lists
the parts of that type that fit the row's scope.
"""
from sqlalchemy.ext.asyncio import AsyncSession
from app.crud import augmented_part as augmented_part_crud
from app.crud import fit as fit_crud
from app.crud import part_type as part_type_crud
from app.crud import scope_info as scope_info_crud
from app.models.fit import FitLabel
SCOPE_TYPE_NAME = "Scope"
# "exact" outranks "possible" when the same part fits a scope via several fits.
_LABEL_RANK = {FitLabel.EXACT.value: 2, FitLabel.POSSIBLE.value: 1}
async def build(
db: AsyncSession,
*,
description: str = "",
manufacturer_id: int | None = None,
scope_model_id: int | None = None,
sub_part_type_id: int | None = None,
fits_part_type_id: int | None = None,
) -> dict:
"""Return the context for ``scopes.html``.
The scope rows are narrowed by the scope-attribute filters (``description``
matches the scope's description; the ``*_id`` filters match exactly).
``fits_part_type_id`` keeps only scopes that have at least one (non-``none``)
fit for that part type.
Shape::
{
"column_types": [PartType, ...], # non-Scope, ordered by name
"rows": [
{
"scope": AugmentedPart,
"info": ScopeInfo | None,
"cells": {part_type_id: [{"part": AugmentedPart, "label": str}]},
},
...,
],
}
"""
part_types = await part_type_crud.list_all(db, limit=1000)
scope_type = next(
(pt for pt in part_types if pt.name == SCOPE_TYPE_NAME), None
)
column_types = [
pt for pt in part_types if scope_type is None or pt.id != scope_type.id
]
if scope_type is None:
return {"column_types": column_types, "rows": []}
scopes = await augmented_part_crud.list_filtered(
db,
description=description,
manufacturer_id=manufacturer_id,
scope_model_id=scope_model_id,
sub_part_type_id=sub_part_type_id,
part_type_id=scope_type.id,
limit=1000,
)
scope_ids = {s.id for s in scopes}
if not scope_ids:
return {"column_types": column_types, "rows": []}
fits = await fit_crud.list_for_parts(
db, list(scope_ids), exclude_label=FitLabel.NONE
)
# Resolve the "other" (non-scope) part on each fit in a single batched query.
other_ids: set[int] = set()
for fit in fits:
scope_id, other_id = _scope_and_other(fit, scope_ids)
if scope_id is not None and other_id not in scope_ids:
other_ids.add(other_id)
other_parts = {
p.id: p for p in await augmented_part_crud.get_many(db, list(other_ids))
}
infos = {
info.augmented_part_id: info
for info in await scope_info_crud.list_by_augmented_parts(db, list(scope_ids))
}
# cell_map[scope_id][type_id][other_part_id] -> {"part", "label"}, dedup'd so a
# part appears once per cell, keeping its strongest fit label.
cell_map: dict[int, dict[int, dict[int, dict]]] = {
sid: {} for sid in scope_ids
}
for fit in fits:
scope_id, other_id = _scope_and_other(fit, scope_ids)
if scope_id is None:
continue
other = other_parts.get(other_id)
if other is None or other.part_type_id == scope_type.id:
continue # scope-to-scope fits have no column
by_type = cell_map[scope_id].setdefault(other.part_type_id, {})
existing = by_type.get(other.id)
label = fit.label.value
if existing is None or _LABEL_RANK[label] > _LABEL_RANK[existing["label"]]:
by_type[other.id] = {"part": other, "label": label}
rows = []
for scope in scopes:
cells = {
type_id: sorted(
entries.values(), key=lambda e: e["part"].description.lower()
)
for type_id, entries in cell_map[scope.id].items()
}
if fits_part_type_id is not None and fits_part_type_id not in cells:
continue # scope has no fit for the required accessory type
rows.append(
{"scope": scope, "info": infos.get(scope.id), "cells": cells}
)
return {"column_types": column_types, "rows": rows}
def _scope_and_other(fit, scope_ids: set[int]) -> tuple[int | None, int]:
"""Return (scope_part_id, other_part_id) for a fit, or (None, _) if neither
side is a scope. When both sides are scopes, side A is treated as the scope."""
if fit.augmented_part_a_id in scope_ids:
return fit.augmented_part_a_id, fit.augmented_part_b_id
if fit.augmented_part_b_id in scope_ids:
return fit.augmented_part_b_id, fit.augmented_part_a_id
return None, fit.augmented_part_b_id
+10
View File
@@ -493,6 +493,16 @@ textarea:focus {
.fit-notes { margin-top: 0.4rem; padding-left: 0.25rem; }
/* Scope guide: stacked matching-part entries within a fit column cell. */
.fit-cell-item {
display: flex;
align-items: center;
gap: 0.4rem;
font-size: 0.85rem;
white-space: nowrap;
}
.fit-cell-item + .fit-cell-item { margin-top: 0.3rem; }
.fit-note { font-size: 0.85rem; margin-bottom: 0.3rem; }
.fit-note-meta { color: var(--muted); font-size: 0.75rem; }
.fit-note-body { margin: 0.1rem 0 0; }
+5
View File
@@ -5,6 +5,7 @@
{% set can_labels = su or "attachment_point_names:read" in p or "manufacturers:read" in p or "part_colours:read" in p or "part_types:read" in p or "scope_models:read" in p or "sub_part_types:read" in p %}
{% set can_database = su or "imported_parts:read" in p or "augmented_parts:read" in p or "attachment_points:read" in p %}
{% set can_parts = can_connections or can_labels or can_database %}
{% set can_scopes = su or ("augmented_parts:read" in p and "fits:read" in p) %}
{# ── SVG icon sprite ───────────────────────────────────────────────────────── #}
<svg xmlns="http://www.w3.org/2000/svg" style="display:none">
@@ -127,6 +128,10 @@
<nav>
<a href="/">{{ icon("home") }} Dashboard</a>
{% if can_scopes %}
<a href="/scopes">{{ icon("crosshair") }} Scopes</a>
{% endif %}
{% if can_admin %}
<div class="nav-group">
<span class="nav-label">{{ icon("shield") }} Admin ▾</span>
+71
View File
@@ -0,0 +1,71 @@
{%- macro mag(v) -%}{{ v | int if v == v | int else v | round(1) }}{%- endmacro -%}
{%- macro magnification(info) -%}
{%- if info and info.magnification_min is not none and info.magnification_max is not none -%}
{{ mag(info.magnification_min) }}–{{ mag(info.magnification_max) }}×
{%- elif info and info.magnification_min is not none -%}
{{ mag(info.magnification_min) }}×
{%- elif info and info.magnification_max is not none -%}
{{ mag(info.magnification_max) }}×
{%- else -%}
<span class="muted">—</span>
{%- endif -%}
{%- endmacro -%}
<div id="scopes-table-card" class="card" style="overflow-x:auto">
<table class="table">
<thead>
<tr>
<th>Scope</th>
<th>Manufacturer</th>
<th>Model</th>
<th>Type</th>
<th>Magnification</th>
<th>Objective thread</th>
{% for ct in column_types %}
<th>{{ ct.name }}</th>
{% endfor %}
</tr>
</thead>
<tbody>
{% for row in rows %}
{% set scope = row.scope %}
{% set info = row.info %}
<tr>
<td>
<strong><a href="/admin/augmented-parts/{{ scope.id }}">{{ scope.description }}</a></strong>
{% if scope.sku %}<br><span class="muted">{{ scope.sku }}</span>{% endif %}
</td>
<td>{{ scope.manufacturer.name if scope.manufacturer else "—" }}</td>
<td>{{ scope.scope_model.name if scope.scope_model else "—" }}</td>
<td>{{ scope.sub_part_type.name if scope.sub_part_type else "—" }}</td>
<td>{{ magnification(info) }}</td>
<td>
{% if info and info.objective_thread_pitch %}
{{ info.objective_thread_pitch }}
{% else %}<span class="muted">—</span>{% endif %}
</td>
{% for ct in column_types %}
{% set entries = row.cells.get(ct.id) %}
<td>
{% if entries %}
{% for e in entries %}
<div class="fit-cell-item">
<span class="fit-badge fit-badge-{{ e.label }}">{{ e.label | capitalize }}</span>
<a href="/admin/augmented-parts/{{ e.part.id }}">{{ e.part.description | truncate(48) }}</a>
</div>
{% endfor %}
{% else %}
<span class="muted">—</span>
{% endif %}
</td>
{% endfor %}
</tr>
{% else %}
<tr>
<td colspan="{{ 6 + column_types | length }}" class="muted">
No scopes match your search.
</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
@@ -22,6 +22,7 @@
<span class="ap-item-name">{{ apoint.attachment_point_name.name }}</span>
<span class="ap-item-type">{{ conn_type }}</span>
<code class="ap-item-conn">{{ conn_label }}</code>
{% if can_write_attachment_points %}
<div class="row-actions">
<button class="btn btn-ghost"
hx-get="/admin/augmented-parts/{{ apoint.augmented_part_id }}/attachment-points/{{ apoint.id }}/edit"
@@ -33,4 +34,5 @@
hx-swap="outerHTML"
hx-confirm="Delete this attachment point?">Delete</button>
</div>
{% endif %}
</div>
@@ -7,6 +7,7 @@
<p class="muted" style="font-size:0.85rem;margin:0 0 0.5rem">No attachment points yet.</p>
{% endfor %}
{% if can_write_attachment_points %}
<form hx-post="/admin/augmented-parts/{{ augmented_part_id }}/attachment-points"
hx-target="#ap-{{ augmented_part_id }}-attachment-points"
hx-swap="outerHTML"
@@ -35,4 +36,5 @@
<div id="ap-{{ augmented_part_id }}-add-connection-options"></div>
<button type="submit" class="btn btn-primary">Add</button>
</form>
{% endif %}
</div>
@@ -19,6 +19,7 @@
<span class="fit-item-arrow">↔</span>
<span class="fit-item-side">{{ other_part.description if other_part else "Unknown part" }} — <strong>{{ other_ap.attachment_point_name.name }}</strong> ({{ conn_desc(other_ap) }})</span>
</div>
{% if can_write %}
<div class="row-actions">
{% if fit.label.value != "exact" %}
<button class="btn btn-ghost"
@@ -44,6 +45,7 @@
hx-swap="outerHTML"
hx-confirm="Delete this fit?">Delete</button>
</div>
{% endif %}
<div class="fit-notes">
{% for note in fit.notes %}
@@ -52,11 +54,13 @@
<p class="fit-note-body">{{ note.body }}</p>
</div>
{% endfor %}
{% if can_write %}
<form hx-post="/admin/augmented-parts/{{ augmented_part_id }}/fits/{{ fit.id }}/notes"
hx-target="#fit-{{ fit.id }}" hx-swap="outerHTML"
class="inline-edit fit-note-form">
<input type="text" name="body" placeholder="Add a note…" required>
<button type="submit" class="btn btn-ghost">Add note</button>
</form>
{% endif %}
</div>
</div>
@@ -12,6 +12,7 @@
<p class="muted" style="font-size:0.85rem;margin:0 0 0.5rem">No fits recorded yet.</p>
{% endfor %}
{% if can_write %}
<div class="row-actions" style="margin:0.5rem 0">
<form hx-post="/admin/augmented-parts/{{ augmented_part_id }}/fits/recompute"
hx-target="#ap-{{ augmented_part_id }}-fits" hx-swap="outerHTML">
@@ -46,4 +47,5 @@
<div id="ap-{{ augmented_part_id }}-fit-other-points"></div>
<button type="submit" class="btn btn-primary">Add fit</button>
</form>
{% endif %}
</div>
+1 -1
View File
@@ -1,7 +1,7 @@
<tr id="augmented-part-{{ ap.id }}">
<td>{{ ap.id }}</td>
<td>{{ ap.sku or '—' }}</td>
<td>{{ ap.description | truncate(60) }}</td>
<td><a href="/admin/augmented-parts/{{ ap.id }}">{{ ap.description | truncate(60) }}</a></td>
<td>{{ ap.colour.name }}</td>
<td>{{ ap.manufacturer.name }}</td>
<td>{{ ap.scope_model.name if ap.scope_model else '—' }}</td>
@@ -0,0 +1,65 @@
{% extends "base.html" %}
{% block title %}{{ ap.description | truncate(40) }} · {{ app_name }}{% endblock %}
{% block body %}
{% include "_nav.html" %}
<main class="container">
<p><a href="/admin/augmented-parts" class="btn btn-ghost">← Augmented Parts</a></p>
<h1>{{ ap.description }}</h1>
<p class="muted">
<span class="tag">{{ ap.part_type.name }}</span>
{% if ap.sub_part_type %}<span class="tag">{{ ap.sub_part_type.name }}</span>{% endif %}
{% if ap.sku %}· SKU {{ ap.sku }}{% endif %}
</p>
<div class="grid">
<div class="card">
<h2>Details</h2>
<dl class="kv">
<dt>SKU</dt><dd>{{ ap.sku or "—" }}</dd>
<dt>Description</dt><dd>{{ ap.description }}</dd>
<dt>Colour</dt><dd>{{ ap.colour.name if ap.colour else "—" }}</dd>
<dt>Manufacturer</dt><dd>{{ ap.manufacturer.name if ap.manufacturer else "—" }}</dd>
<dt>Type</dt><dd>{{ ap.part_type.name }}</dd>
<dt>Sub type</dt><dd>{{ ap.sub_part_type.name if ap.sub_part_type else "—" }}</dd>
<dt>Scope model</dt>
<dd>
{% if ap.scope_model %}
{{ ap.scope_model.name }}
{% if ap.scope_model.url %} · <a href="{{ ap.scope_model.url }}" target="_blank" rel="noopener">link</a>{% endif %}
{% else %}—{% endif %}
</dd>
</dl>
</div>
{% if scope_info %}
<div class="card">
<h2>Scope information</h2>
<dl class="kv">
<dt>Magnification</dt>
<dd>
{% if scope_info.magnification_min is not none or scope_info.magnification_max is not none %}
{{ scope_info.magnification_min if scope_info.magnification_min is not none else "?" }}–{{ scope_info.magnification_max if scope_info.magnification_max is not none else "?" }}×
{% else %}—{% endif %}
</dd>
<dt>Model suffix</dt><dd>{{ scope_info.model_suffix or "—" }}</dd>
<dt>Objective thread</dt><dd>{{ scope_info.objective_thread_pitch or "—" }}</dd>
</dl>
</div>
{% endif %}
</div>
{% if can_read_attachment_points %}
<div class="card">
{% include "admin/_augmented_part_attachment_points.html" %}
</div>
{% endif %}
{% if can_read_fits %}
<div class="card">
{% include "admin/_augmented_part_fits.html" %}
</div>
{% endif %}
</main>
{% endblock %}
+3
View File
@@ -9,6 +9,7 @@
{% if ldap_enabled %}
<form method="post" action="/login/ldap">
<input type="hidden" name="next" value="{{ next_url or '' }}">
{% if ldap_error %}<div class="alert">{{ ldap_error }}</div>{% endif %}
<label>Username
<input type="text" name="username" value="{{ username or '' }}" required autofocus>
@@ -22,6 +23,7 @@
<details class="local-login-panel" {% if local_error %}open{% endif %}>
<summary>Use a local account instead</summary>
<form method="post" action="/login">
<input type="hidden" name="next" value="{{ next_url or '' }}">
{% if local_error %}<div class="alert">{{ local_error }}</div>{% endif %}
<label>Email
<input type="email" name="email" value="{{ email or '' }}" required>
@@ -36,6 +38,7 @@
{% else %}
<form method="post" action="/login">
<input type="hidden" name="next" value="{{ next_url or '' }}">
{% if local_error %}<div class="alert">{{ local_error }}</div>{% endif %}
<label>Email
<input type="email" name="email" value="{{ email or '' }}" required autofocus>
+49
View File
@@ -0,0 +1,49 @@
{% extends "base.html" %}
{% block title %}Scopes · {{ app_name }}{% endblock %}
{% block body %}
{% include "_nav.html" %}
<main class="container">
<h1>Scopes</h1>
<p class="muted">
Every scope on record, with the accessories that fit it. Each part-type
column lists matching parts; <span class="fit-badge fit-badge-exact">Exact</span>
is a confirmed fit, <span class="fit-badge fit-badge-possible">Possible</span> a candidate.
</p>
<div class="filter-bar">
<form hx-get="/scopes/filter"
hx-target="#scopes-table-card"
hx-trigger="input delay:300ms, change"
hx-swap="outerHTML">
<input type="text" name="description" placeholder="Search scopes…"
value="{{ filters.description or '' }}">
<select name="manufacturer_id">
<option value="">All manufacturers</option>
{% for m in manufacturers %}
<option value="{{ m.id }}" {{ "selected" if filters.manufacturer_id and filters.manufacturer_id|int == m.id }}>{{ m.name }}</option>
{% endfor %}
</select>
<select name="scope_model_id">
<option value="">All models</option>
{% for sm in scope_models %}
<option value="{{ sm.id }}" {{ "selected" if filters.scope_model_id and filters.scope_model_id|int == sm.id }}>{{ sm.name }}</option>
{% endfor %}
</select>
<select name="sub_part_type_id">
<option value="">All types</option>
{% for spt in sub_part_types %}
<option value="{{ spt.id }}" {{ "selected" if filters.sub_part_type_id and filters.sub_part_type_id|int == spt.id }}>{{ spt.name }}</option>
{% endfor %}
</select>
<select name="fits_part_type_id">
<option value="">Fits any accessory</option>
{% for ct in column_types %}
<option value="{{ ct.id }}" {{ "selected" if filters.fits_part_type_id and filters.fits_part_type_id|int == ct.id }}>Fits {{ ct.name }}</option>
{% endfor %}
</select>
</form>
</div>
{% include "_scopes_table_card.html" %}
</main>
{% endblock %}
+99 -14
View File
@@ -26,6 +26,7 @@ from app.crud import manufacturer as manufacturer_crud
from app.crud import part_colour as part_colour_crud
from app.crud import part_type as part_type_crud
from app.crud import role as role_crud
from app.crud import scope_info as scope_info_crud
from app.crud import scope_model as scope_model_crud
from app.crud import attachment_point as attachment_point_crud
from app.crud import attachment_point_name as apn_crud
@@ -1780,6 +1781,46 @@ async def filter_skipped_parts(
)
@router.get("/augmented-parts/{augmented_part_id}", response_class=HTMLResponse)
async def augmented_part_detail(
request: Request,
augmented_part_id: int,
db: AsyncSession = Depends(get_db),
user: User = Depends(web_require("augmented_parts:read")),
):
ap = await augmented_part_crud.get(db, augmented_part_id)
if ap is None:
raise HTTPException(status_code=404, detail="Augmented part not found")
can_read_fits = user.is_superuser or "fits:read" in user.permissions
can_read_attachment_points = (
user.is_superuser or "attachment_points:read" in user.permissions
)
ctx = {
"user": user,
"ap": ap,
"scope_info": await scope_info_crud.get_by_augmented_part(
db, augmented_part_id
),
"can_read_fits": can_read_fits,
"can_read_attachment_points": can_read_attachment_points,
}
if can_read_attachment_points:
ctx.update(
await _augmented_part_ap_ctx(
db,
augmented_part_id,
can_write=_can_write_attachment_points(user),
)
)
if can_read_fits:
ctx.update(
await _augmented_part_fits_ctx(
db, augmented_part_id, can_write=_can_write_fits(user)
)
)
return templates.TemplateResponse(request, "admin/augmented_part_detail.html", ctx)
@router.get("/augmented-parts/{augmented_part_id}/edit", response_class=HTMLResponse)
async def edit_augmented_part_row(
request: Request,
@@ -1877,7 +1918,13 @@ async def delete_augmented_part(
return Response(status_code=status.HTTP_200_OK)
async def _augmented_part_ap_ctx(db: AsyncSession, augmented_part_id: int) -> dict:
def _can_write_attachment_points(user: User) -> bool:
return user.is_superuser or "attachment_points:write" in user.permissions
async def _augmented_part_ap_ctx(
db: AsyncSession, augmented_part_id: int, *, can_write: bool = True
) -> dict:
return {
"augmented_part_id": augmented_part_id,
"attachment_points_list": await attachment_point_crud.list_for_augmented_part(
@@ -1888,6 +1935,7 @@ async def _augmented_part_ap_ctx(db: AsyncSession, augmented_part_id: int) -> di
"unc_threads": await unc_thread_crud.list_all(db),
"clamp_connections": await clamp_connection_crud.list_all(db),
"static_connections": await static_connection_crud.list_all(db),
"can_write_attachment_points": can_write,
}
@@ -1899,9 +1947,11 @@ async def augmented_part_attachment_points(
request: Request,
augmented_part_id: int,
db: AsyncSession = Depends(get_db),
_: User = Depends(web_require("attachment_points:read")),
user: User = Depends(web_require("attachment_points:read")),
):
ctx = await _augmented_part_ap_ctx(db, augmented_part_id)
ctx = await _augmented_part_ap_ctx(
db, augmented_part_id, can_write=_can_write_attachment_points(user)
)
return templates.TemplateResponse(
request, "admin/_augmented_part_attachment_points.html", ctx
)
@@ -1944,7 +1994,9 @@ async def create_augmented_part_attachment_point(
"AttachmentPoint", apoint.id, f"part {augmented_part_id}"
),
)
ctx = await _augmented_part_ap_ctx(db, augmented_part_id)
ctx = await _augmented_part_ap_ctx(
db, augmented_part_id, can_write=_can_write_attachment_points(actor)
)
return templates.TemplateResponse(
request, "admin/_augmented_part_attachment_points.html", ctx
)
@@ -1990,13 +2042,18 @@ async def augmented_part_attachment_point_row(
augmented_part_id: int,
attachment_point_id: int,
db: AsyncSession = Depends(get_db),
_: User = Depends(web_require("attachment_points:read")),
user: User = Depends(web_require("attachment_points:read")),
):
apoint = await attachment_point_crud.get(db, attachment_point_id)
if apoint is None:
raise HTTPException(status_code=404, detail="Attachment point not found")
return templates.TemplateResponse(
request, "admin/_augmented_part_ap_row.html", {"apoint": apoint}
request,
"admin/_augmented_part_ap_row.html",
{
"apoint": apoint,
"can_write_attachment_points": _can_write_attachment_points(user),
},
)
@@ -2043,7 +2100,12 @@ async def update_augmented_part_attachment_point(
),
)
return templates.TemplateResponse(
request, "admin/_augmented_part_ap_row.html", {"apoint": apoint}
request,
"admin/_augmented_part_ap_row.html",
{
"apoint": apoint,
"can_write_attachment_points": _can_write_attachment_points(actor),
},
)
@@ -2071,7 +2133,9 @@ async def delete_augmented_part_attachment_point(
"AttachmentPoint", attachment_point_id, f"part {augmented_part_id}"
),
)
ctx = await _augmented_part_ap_ctx(db, augmented_part_id)
ctx = await _augmented_part_ap_ctx(
db, augmented_part_id, can_write=_can_write_attachment_points(actor)
)
return templates.TemplateResponse(
request, "admin/_augmented_part_attachment_points.html", ctx
)
@@ -2102,10 +2166,15 @@ async def _fit_view(db: AsyncSession, fit, part_id: int) -> dict:
}
def _can_write_fits(user: User) -> bool:
return user.is_superuser or "fits:write" in user.permissions
async def _augmented_part_fits_ctx(
db: AsyncSession,
augmented_part_id: int,
*,
can_write: bool = True,
message: str | None = None,
error: str | None = None,
) -> dict:
@@ -2122,6 +2191,7 @@ async def _augmented_part_fits_ctx(
db, augmented_part_id
),
"other_parts": other_parts,
"can_write": can_write,
"message": message,
"error": error,
}
@@ -2132,9 +2202,11 @@ async def augmented_part_fits(
request: Request,
augmented_part_id: int,
db: AsyncSession = Depends(get_db),
_: User = Depends(web_require("fits:read")),
user: User = Depends(web_require("fits:read")),
):
ctx = await _augmented_part_fits_ctx(db, augmented_part_id)
ctx = await _augmented_part_fits_ctx(
db, augmented_part_id, can_write=_can_write_fits(user)
)
return templates.TemplateResponse(request, "admin/_augmented_part_fits.html", ctx)
@@ -2162,6 +2234,7 @@ async def recompute_augmented_part_fits(
ctx = await _augmented_part_fits_ctx(
db,
augmented_part_id,
can_write=_can_write_fits(actor),
message=f"{result.created} created, {result.removed} removed.",
)
return templates.TemplateResponse(request, "admin/_augmented_part_fits.html", ctx)
@@ -2223,7 +2296,9 @@ async def create_augmented_part_fit(
"Fit", fit.id, f"attachment points {ap_a.id}↔{ap_b.id}"
),
)
ctx = await _augmented_part_fits_ctx(db, augmented_part_id, error=error)
ctx = await _augmented_part_fits_ctx(
db, augmented_part_id, can_write=_can_write_fits(actor), error=error
)
return templates.TemplateResponse(request, "admin/_augmented_part_fits.html", ctx)
@@ -2255,7 +2330,11 @@ async def update_augmented_part_fit_label(
return templates.TemplateResponse(
request,
"admin/_augmented_part_fit_row.html",
{"augmented_part_id": augmented_part_id, **view},
{
"augmented_part_id": augmented_part_id,
"can_write": _can_write_fits(actor),
**view,
},
)
@@ -2281,7 +2360,11 @@ async def create_augmented_part_fit_note(
return templates.TemplateResponse(
request,
"admin/_augmented_part_fit_row.html",
{"augmented_part_id": augmented_part_id, **view},
{
"augmented_part_id": augmented_part_id,
"can_write": _can_write_fits(actor),
**view,
},
)
@@ -2309,7 +2392,9 @@ async def delete_augmented_part_fit(
resource_id=fit_id,
description=audit_crud.describe_delete("Fit", fit_id, label),
)
ctx = await _augmented_part_fits_ctx(db, augmented_part_id)
ctx = await _augmented_part_fits_ctx(
db, augmented_part_id, can_write=_can_write_fits(actor)
)
return templates.TemplateResponse(request, "admin/_augmented_part_fits.html", ctx)
+30 -5
View File
@@ -1,4 +1,4 @@
from fastapi import APIRouter, Depends, Form, HTTPException, Request, status
from fastapi import APIRouter, Depends, Form, HTTPException, Query, Request, status
from fastapi.responses import HTMLResponse, RedirectResponse, Response
from sqlalchemy.ext.asyncio import AsyncSession
@@ -14,6 +14,21 @@ from app.web.templates import templates
router = APIRouter(include_in_schema=False)
def _safe_next(next_url: str) -> str:
"""Return next_url only if it's a safe same-site path, else "/".
Guards against open-redirect: rejects absolute URLs and protocol-relative
("//host", "/\\host") targets, and never bounces back to the auth pages.
"""
if (
next_url.startswith("/")
and not next_url.startswith(("//", "/\\"))
and not next_url.startswith(("/login", "/logout"))
):
return next_url
return "/"
def _set_auth_cookie(response: Response, access_token: str) -> None:
response.set_cookie(
key=settings.cookie_name,
@@ -35,9 +50,11 @@ async def dashboard(
@router.get("/login", response_class=HTMLResponse)
async def login_page(request: Request):
async def login_page(request: Request, next_url: str = Query("", alias="next")):
return templates.TemplateResponse(
request, "login.html", {"ldap_enabled": settings.ldap_enabled}
request,
"login.html",
{"ldap_enabled": settings.ldap_enabled, "next_url": _safe_next(next_url)},
)
@@ -46,6 +63,7 @@ async def login_submit(
request: Request,
email: str = Form(...),
password: str = Form(...),
next_url: str = Form("", alias="next"),
db: AsyncSession = Depends(get_db),
):
user = await user_crud.authenticate(db, email, password)
@@ -57,11 +75,14 @@ async def login_submit(
"local_error": "Incorrect email or password",
"email": email,
"ldap_enabled": settings.ldap_enabled,
"next_url": _safe_next(next_url),
},
status_code=status.HTTP_401_UNAUTHORIZED,
)
access, _ = await issue_token_pair(db, user)
response = RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER)
response = RedirectResponse(
url=_safe_next(next_url), status_code=status.HTTP_303_SEE_OTHER
)
_set_auth_cookie(response, access)
return response
@@ -71,6 +92,7 @@ async def login_ldap_submit(
request: Request,
username: str = Form(...),
password: str = Form(...),
next_url: str = Form("", alias="next"),
db: AsyncSession = Depends(get_db),
):
if not settings.ldap_enabled:
@@ -84,11 +106,14 @@ async def login_ldap_submit(
"ldap_error": "Incorrect username or password",
"username": username,
"ldap_enabled": True,
"next_url": _safe_next(next_url),
},
status_code=status.HTTP_401_UNAUTHORIZED,
)
access, _ = await issue_token_pair(db, user)
response = RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER)
response = RedirectResponse(
url=_safe_next(next_url), status_code=status.HTTP_303_SEE_OTHER
)
_set_auth_cookie(response, access)
return response
+103
View File
@@ -0,0 +1,103 @@
"""Top-level, read-only scope fit guide at ``/scopes``.
The site's primary view: a table of scopes with their specs, plus one column per
non-Scope part type listing the accessories that fit each scope. Supports HTMX
search/filter over scope attributes and required accessory fits.
"""
from fastapi import APIRouter, Depends, Request
from fastapi.responses import HTMLResponse
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.crud import manufacturer as manufacturer_crud
from app.crud import part_type as part_type_crud
from app.crud import scope_model as scope_model_crud
from app.crud import sub_part_type as sub_part_type_crud
from app.models.user import User
from app.services import scope_guide as scope_guide_service
from app.web.admin import web_require
from app.web.templates import templates
router = APIRouter(include_in_schema=False)
def _to_int(value: str) -> int | None:
return int(value) if value.strip() else None
async def _build(db: AsyncSession, filters: dict) -> dict:
"""Run the guide query for the given raw (string) filter values."""
return await scope_guide_service.build(
db,
description=filters["description"],
manufacturer_id=_to_int(filters["manufacturer_id"]),
scope_model_id=_to_int(filters["scope_model_id"]),
sub_part_type_id=_to_int(filters["sub_part_type_id"]),
fits_part_type_id=_to_int(filters["fits_part_type_id"]),
)
async def _filter_options(db: AsyncSession) -> dict:
"""Dropdown option lists for the filter bar."""
scope_type = await part_type_crud.get_by_name(
db, scope_guide_service.SCOPE_TYPE_NAME
)
sub_part_types = [
spt
for spt in await sub_part_type_crud.list_all(db, limit=1000)
if scope_type is not None and spt.part_type_id == scope_type.id
]
return {
"manufacturers": await manufacturer_crud.list_all(db, limit=1000),
"scope_models": await scope_model_crud.list_all(db, limit=1000),
"sub_part_types": sub_part_types,
}
@router.get("/scopes", response_class=HTMLResponse)
async def scopes_page(
request: Request,
description: str = "",
manufacturer_id: str = "",
scope_model_id: str = "",
sub_part_type_id: str = "",
fits_part_type_id: str = "",
db: AsyncSession = Depends(get_db),
user: User = Depends(web_require("augmented_parts:read", "fits:read")),
):
filters = {
"description": description,
"manufacturer_id": manufacturer_id,
"scope_model_id": scope_model_id,
"sub_part_type_id": sub_part_type_id,
"fits_part_type_id": fits_part_type_id,
}
guide = await _build(db, filters)
return templates.TemplateResponse(
request,
"scopes.html",
{"user": user, "filters": filters, **guide, **(await _filter_options(db))},
)
@router.get("/scopes/filter", response_class=HTMLResponse)
async def scopes_filter(
request: Request,
description: str = "",
manufacturer_id: str = "",
scope_model_id: str = "",
sub_part_type_id: str = "",
fits_part_type_id: str = "",
db: AsyncSession = Depends(get_db),
_: User = Depends(web_require("augmented_parts:read", "fits:read")),
):
filters = {
"description": description,
"manufacturer_id": manufacturer_id,
"scope_model_id": scope_model_id,
"sub_part_type_id": sub_part_type_id,
"fits_part_type_id": fits_part_type_id,
}
guide = await _build(db, filters)
return templates.TemplateResponse(request, "_scopes_table_card.html", guide)
+182
View File
@@ -0,0 +1,182 @@
import pytest
from app.crud import role as role_crud
from app.models.attachment_point import AttachmentPoint
from app.models.attachment_point_name import AttachmentPointName
from app.models.augmented_part import AugmentedPart
from app.models.fit import Fit, FitLabel
from app.models.part_type import PartType
from app.models.scope_info import ScopeInfo
async def _seed_part_with_fit(db):
apn = AttachmentPointName(name="Objective")
scope_t = PartType(name="Scope")
cover_t = PartType(name="Flip Cover")
db.add_all([apn, scope_t, cover_t])
await db.flush()
scope = AugmentedPart(
description="Alpha 3-18x50",
sku="ALP-318",
colour_id=1,
manufacturer_id=1,
part_type_id=scope_t.id,
)
cover = AugmentedPart(
description="Flip cover 50mm",
colour_id=1,
manufacturer_id=1,
part_type_id=cover_t.id,
)
db.add_all([scope, cover])
await db.flush()
db.add(
ScopeInfo(
augmented_part_id=scope.id,
magnification_min=3,
magnification_max=18,
objective_thread_pitch="M52x0.75",
)
)
ap_s = AttachmentPoint(
augmented_part_id=scope.id, attachment_point_name_id=apn.id
)
ap_c = AttachmentPoint(
augmented_part_id=cover.id, attachment_point_name_id=apn.id
)
db.add_all([ap_s, ap_c])
await db.flush()
db.add(
Fit(
attachment_point_a_id=ap_s.id,
attachment_point_b_id=ap_c.id,
augmented_part_a_id=scope.id,
augmented_part_b_id=cover.id,
label=FitLabel.POSSIBLE,
)
)
await db.commit()
return {"scope_id": scope.id, "cover_id": cover.id}
async def _web_login(client, email, password="secret123"):
resp = await client.post("/login", data={"email": email, "password": password})
assert resp.status_code == 303
@pytest.mark.asyncio
async def test_detail_shows_attributes_and_fits_readonly(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
ids = await _seed_part_with_fit(db)
# The default "user" role has augmented_parts:read + fits:read, NOT fits:write.
await make_user("reader@example.com", role_ids=[seeded["roles"]["user"].id])
await _web_login(client, "reader@example.com")
resp = await client.get(f"/admin/augmented-parts/{ids['scope_id']}")
assert resp.status_code == 200
body = resp.text
# Attributes
assert "Alpha 3-18x50" in body
assert "ALP-318" in body
assert "M52x0.75" in body # scope info
assert "Objective" in body # attachment point
# Fits are shown (the other part in the fit)
assert "Flip cover 50mm" in body
# ...but write controls are hidden for a read-only user
assert "Add fit" not in body
assert "Mark Exact" not in body
assert "Recompute possible fits" not in body
# ...and attachment-point add/edit/delete controls are hidden too
assert 'name="attachment_point_name_id"' not in body
assert "Delete this attachment point?" not in body
@pytest.mark.asyncio
async def test_detail_write_controls_for_attachment_point_editor(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
ids = await _seed_part_with_fit(db)
perms = seeded["perms"]
editor = await role_crud.create(
db,
name="ap-editor",
permission_ids=[
perms["augmented_parts:read"].id,
perms["attachment_points:read"].id,
perms["attachment_points:write"].id,
],
)
editor_id = editor.id
await make_user("apeditor@example.com", role_ids=[editor_id])
await _web_login(client, "apeditor@example.com")
resp = await client.get(f"/admin/augmented-parts/{ids['scope_id']}")
assert resp.status_code == 200
body = resp.text
assert 'name="attachment_point_name_id"' in body
assert "Delete this attachment point?" in body
@pytest.mark.asyncio
async def test_detail_write_controls_for_fit_editor(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
ids = await _seed_part_with_fit(db)
perms = seeded["perms"]
editor = await role_crud.create(
db,
name="fit-editor",
permission_ids=[
perms["augmented_parts:read"].id,
perms["fits:read"].id,
perms["fits:write"].id,
],
)
editor_id = editor.id
await make_user("editor@example.com", role_ids=[editor_id])
await _web_login(client, "editor@example.com")
resp = await client.get(f"/admin/augmented-parts/{ids['scope_id']}")
assert resp.status_code == 200
body = resp.text
assert "Add fit" in body
assert "Mark Exact" in body
assert "Recompute possible fits" in body
@pytest.mark.asyncio
async def test_detail_404_for_missing_part(client, seeded, make_user):
await make_user("reader2@example.com", role_ids=[seeded["roles"]["user"].id])
await _web_login(client, "reader2@example.com")
resp = await client.get("/admin/augmented-parts/99999")
assert resp.status_code == 404
@pytest.mark.asyncio
async def test_detail_requires_read_permission(client, seeded, make_user, db_sessionmaker):
async with db_sessionmaker() as db:
ids = await _seed_part_with_fit(db)
await make_user("plain@example.com") # no roles, not superuser
await _web_login(client, "plain@example.com")
resp = await client.get(f"/admin/augmented-parts/{ids['scope_id']}")
assert resp.status_code == 403
@pytest.mark.asyncio
async def test_augmented_parts_list_links_to_detail(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
ids = await _seed_part_with_fit(db)
await make_user("reader3@example.com", role_ids=[seeded["roles"]["user"].id])
await _web_login(client, "reader3@example.com")
resp = await client.get("/admin/augmented-parts")
assert resp.status_code == 200
assert f'href="/admin/augmented-parts/{ids["scope_id"]}"' in resp.text
+73
View File
@@ -0,0 +1,73 @@
import pytest
@pytest.mark.asyncio
async def test_deep_link_redirects_to_login_with_next(client):
resp = await client.get("/scopes")
assert resp.status_code == 302
assert resp.headers["location"] == "/login?next=%2Fscopes"
@pytest.mark.asyncio
async def test_deep_link_preserves_query_string(client):
resp = await client.get("/scopes?description=alpha")
assert resp.status_code == 302
assert resp.headers["location"] == "/login?next=%2Fscopes%3Fdescription%3Dalpha"
@pytest.mark.asyncio
async def test_root_redirect_has_no_next(client):
resp = await client.get("/")
assert resp.status_code == 302
assert resp.headers["location"] == "/login"
@pytest.mark.asyncio
async def test_login_page_renders_next_field(client):
resp = await client.get("/login?next=/scopes")
assert resp.status_code == 200
assert 'name="next" value="/scopes"' in resp.text
@pytest.mark.asyncio
async def test_login_redirects_to_next_on_success(client, make_user):
await make_user("u@example.com")
resp = await client.post(
"/login",
data={"email": "u@example.com", "password": "secret123", "next": "/scopes"},
)
assert resp.status_code == 303
assert resp.headers["location"] == "/scopes"
@pytest.mark.asyncio
async def test_login_defaults_to_root_without_next(client, make_user):
await make_user("u2@example.com")
resp = await client.post(
"/login", data={"email": "u2@example.com", "password": "secret123"}
)
assert resp.status_code == 303
assert resp.headers["location"] == "/"
@pytest.mark.asyncio
async def test_login_rejects_unsafe_next(client, make_user):
await make_user("u3@example.com")
for bad in ["https://evil.com", "//evil.com", "/\\evil.com", "/login"]:
resp = await client.post(
"/login",
data={"email": "u3@example.com", "password": "secret123", "next": bad},
)
assert resp.status_code == 303
assert resp.headers["location"] == "/", bad
@pytest.mark.asyncio
async def test_failed_login_preserves_next(client, make_user):
await make_user("u4@example.com")
resp = await client.post(
"/login",
data={"email": "u4@example.com", "password": "wrong", "next": "/scopes"},
)
assert resp.status_code == 401
assert 'name="next" value="/scopes"' in resp.text
+242
View File
@@ -0,0 +1,242 @@
import pytest
from app.models.augmented_part import AugmentedPart
from app.models.fit import Fit, FitLabel
from app.models.part_type import PartType
from app.models.scope_info import ScopeInfo
from app.models.sub_part_type import SubPartType
from app.services import scope_guide
async def _seed_guide_data(db):
scope_t = PartType(name="Scope")
cover_t = PartType(name="Flip Cover")
ard_t = PartType(name="ARD")
db.add_all([scope_t, cover_t, ard_t])
await db.flush()
rifle = SubPartType(name="Riflescope", part_type_id=scope_t.id)
reddot = SubPartType(name="Red Dot", part_type_id=scope_t.id)
db.add_all([rifle, reddot])
await db.flush()
def part(desc, type_id, manufacturer_id=1, sub_part_type_id=None):
p = AugmentedPart(
description=desc,
colour_id=1,
manufacturer_id=manufacturer_id,
part_type_id=type_id,
sub_part_type_id=sub_part_type_id,
)
db.add(p)
return p
s1 = part("Alpha 3-18x50", scope_t.id, manufacturer_id=10, sub_part_type_id=rifle.id)
s2 = part("Zeta 1-6x24", scope_t.id, manufacturer_id=20, sub_part_type_id=reddot.id)
fc1 = part("Flip cover 50mm", cover_t.id)
a1 = part("ARD 50mm", ard_t.id)
a2 = part("ARD 56mm", ard_t.id)
await db.flush()
db.add(
ScopeInfo(
augmented_part_id=s1.id,
magnification_min=3,
magnification_max=18,
objective_thread_pitch="M52x0.75",
)
)
def fit(ap_a, ap_b, a, b, label):
db.add(
Fit(
attachment_point_a_id=ap_a,
attachment_point_b_id=ap_b,
augmented_part_a_id=a.id,
augmented_part_b_id=b.id,
label=label,
)
)
# S1 fits FC1 (exact) via two attachment points -> should dedup to one exact.
fit(1, 2, s1, fc1, FitLabel.POSSIBLE)
fit(3, 4, s1, fc1, FitLabel.EXACT)
# S1 fits A1 (possible); an explicit NONE to A2 must be excluded.
fit(5, 6, s1, a1, FitLabel.POSSIBLE)
fit(7, 8, s1, a2, FitLabel.NONE)
# S2 fits FC1 (possible); order reversed to exercise the b-side branch.
fit(9, 10, fc1, s2, FitLabel.POSSIBLE)
await db.commit()
return {
"scope": scope_t,
"cover": cover_t,
"ard": ard_t,
"s1": s1,
"s2": s2,
"rifle": rifle,
"reddot": reddot,
}
@pytest.mark.asyncio
async def test_build_columns_exclude_scope_type(db_sessionmaker):
async with db_sessionmaker() as db:
await _seed_guide_data(db)
guide = await scope_guide.build(db)
names = [ct.name for ct in guide["column_types"]]
assert names == ["ARD", "Flip Cover"] # sorted, Scope excluded
@pytest.mark.asyncio
async def test_build_rows_group_fits_by_type_and_dedup(db_sessionmaker):
async with db_sessionmaker() as db:
ids = await _seed_guide_data(db)
guide = await scope_guide.build(db)
cover_id, ard_id = ids["cover"].id, ids["ard"].id
rows = {r["scope"].description: r for r in guide["rows"]}
assert list(rows) == ["Alpha 3-18x50", "Zeta 1-6x24"] # ordered by description
s1 = rows["Alpha 3-18x50"]
# Two fits to the same cover collapse to one entry, keeping the exact label.
cover_cell = s1["cells"][cover_id]
assert [(e["part"].description, e["label"]) for e in cover_cell] == [
("Flip cover 50mm", "exact")
]
# Possible ARD present; the NONE-labelled ARD is excluded entirely.
ard_cell = s1["cells"][ard_id]
assert [(e["part"].description, e["label"]) for e in ard_cell] == [
("ARD 50mm", "possible")
]
assert s1["info"] is not None
assert s1["info"].magnification_max == 18
s2 = rows["Zeta 1-6x24"]
assert [e["part"].description for e in s2["cells"][cover_id]] == ["Flip cover 50mm"]
assert ard_id not in s2["cells"] # no ARD fits for S2
assert s2["info"] is None
@pytest.mark.asyncio
async def test_build_no_scope_type_returns_empty_rows(db_sessionmaker):
async with db_sessionmaker() as db:
db.add(PartType(name="Flip Cover"))
await db.commit()
guide = await scope_guide.build(db)
assert guide["rows"] == []
assert [ct.name for ct in guide["column_types"]] == ["Flip Cover"]
def _descriptions(guide):
return [r["scope"].description for r in guide["rows"]]
@pytest.mark.asyncio
async def test_build_filters_by_description(db_sessionmaker):
async with db_sessionmaker() as db:
await _seed_guide_data(db)
guide = await scope_guide.build(db, description="alpha")
assert _descriptions(guide) == ["Alpha 3-18x50"]
@pytest.mark.asyncio
async def test_build_filters_by_manufacturer(db_sessionmaker):
async with db_sessionmaker() as db:
await _seed_guide_data(db)
guide = await scope_guide.build(db, manufacturer_id=20)
assert _descriptions(guide) == ["Zeta 1-6x24"]
@pytest.mark.asyncio
async def test_build_filters_by_sub_part_type(db_sessionmaker):
async with db_sessionmaker() as db:
ids = await _seed_guide_data(db)
guide = await scope_guide.build(db, sub_part_type_id=ids["rifle"].id)
assert _descriptions(guide) == ["Alpha 3-18x50"]
@pytest.mark.asyncio
async def test_build_filters_by_required_fit_type(db_sessionmaker):
async with db_sessionmaker() as db:
ids = await _seed_guide_data(db)
# Only Alpha has an ARD fit; both scopes fit a flip cover.
ard_only = await scope_guide.build(db, fits_part_type_id=ids["ard"].id)
cover = await scope_guide.build(db, fits_part_type_id=ids["cover"].id)
assert _descriptions(ard_only) == ["Alpha 3-18x50"]
assert _descriptions(cover) == ["Alpha 3-18x50", "Zeta 1-6x24"]
async def _web_login(client, email, password="secret123"):
resp = await client.post(
"/login", data={"email": email, "password": password}
)
assert resp.status_code == 303
@pytest.mark.asyncio
async def test_scopes_page_renders_for_reader(client, seeded, make_user):
await make_user("reader@example.com", role_ids=[seeded["roles"]["user"].id])
await _web_login(client, "reader@example.com")
resp = await client.get("/scopes")
assert resp.status_code == 200
assert "Scopes" in resp.text
@pytest.mark.asyncio
async def test_scopes_page_renders_rows_and_fits(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
await _seed_guide_data(db)
await make_user("reader2@example.com", role_ids=[seeded["roles"]["user"].id])
await _web_login(client, "reader2@example.com")
resp = await client.get("/scopes")
assert resp.status_code == 200
body = resp.text
assert "Alpha 3-18x50" in body # scope row
assert "Flip cover 50mm" in body # a matching part in a fit column
assert "fit-badge-exact" in body # the exact fit badge
assert "M52x0.75" in body # scope info: objective thread
assert "3–18×" in body # magnification range, integral values un-padded
@pytest.mark.asyncio
async def test_scopes_filter_endpoint_narrows_rows(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
await _seed_guide_data(db)
await make_user("reader3@example.com", role_ids=[seeded["roles"]["user"].id])
await _web_login(client, "reader3@example.com")
resp = await client.get("/scopes/filter", params={"description": "Alpha"})
assert resp.status_code == 200
assert "Alpha 3-18x50" in resp.text
assert "Zeta 1-6x24" not in resp.text
@pytest.mark.asyncio
async def test_scopes_filter_endpoint_forbidden_without_permission(
client, seeded, make_user
):
await make_user("plain2@example.com") # no roles, not a superuser
await _web_login(client, "plain2@example.com")
resp = await client.get("/scopes/filter")
assert resp.status_code == 403
@pytest.mark.asyncio
async def test_scopes_page_forbidden_without_permission(client, seeded, make_user):
await make_user("plain@example.com") # no roles, not a superuser
await _web_login(client, "plain@example.com")
resp = await client.get("/scopes")
assert resp.status_code == 403
@pytest.mark.asyncio
async def test_scopes_page_redirects_when_anonymous(client):
resp = await client.get("/scopes")
assert resp.status_code == 302
# Deep link is preserved so the user lands back here after logging in.
assert resp.headers["location"] == "/login?next=%2Fscopes"