Compare commits

...
28 Commits
Author SHA1 Message Date
Ian Roddis 2318a6daeb Making scope filter by product id 2026-08-11 13:11:15 -03:00
Ian Roddis 6d7c5c5604 Adding pagination to scopes page 2026-08-11 11:33:56 -03:00
Ian Roddis 7a15d9f7c0 Adding a bulk delete option to the scopes page 2026-08-10 12:15:09 -03:00
Ian Roddis a9c4c16d4f Extending session login time 2026-08-10 12:09:51 -03:00
Ian Roddis 14aa2ac5d6 Adding delete option for scopes 2026-08-10 12:04:56 -03:00
Ian Roddis f4771e33d4 Fixing tag detection 2026-08-05 15:27:17 -04:00
Ian Roddis cf8d72f73e Making OData Refresh a background process 2026-08-05 16:26:52 -03:00
Ian Roddis 2e7976e382 Rejigging async jobs 2026-08-05 16:21:31 -03:00
Ian Roddis 2a0b8fcab2 Fixing errors and danger zone 2026-08-05 14:19:25 -03:00
Ian Roddis 392b79e801 Fixing presentation of objective thread 2026-08-05 14:15:10 -03:00
Ian Roddis 3ebe9a0894 Fixing imports 2026-08-05 13:59:57 -03:00
Ian Roddis 8d37b2ad67 Fixing column names, fixing model number 2026-08-05 13:35:09 -03:00
Ian Roddis 9ebd0f4498 Fixing automatic product_number assignment 2026-08-04 16:55:05 -03:00
Ian Roddis d54df47a2c Making objective_lens a thing 2026-08-04 16:48:12 -03:00
Ian Roddis 12e6984c63 Making objective_lens a thing 2026-08-04 16:48:00 -03:00
Ian Roddis fcb4dbaa62 Adding notes 2026-08-04 09:39:19 -03:00
Ian Roddis a617b935d3 Fixing formatting of magnification 2026-08-04 09:02:38 -03:00
Ian Roddis d108cb7149 Making long running operations asynchronous 2026-07-31 10:40:54 -03:00
Ian Roddis 373a2c6acf Adding favicon 2026-07-31 10:09:41 -03:00
Ian Roddis 2fb662080b Adding pervasive column sorting 2026-07-31 09:52:09 -03:00
Ian Roddis 34679d53e8 Adding AugmentedItem notes 2026-07-31 09:12:41 -03:00
Ian Roddis 30a13644ce Adding fit note at manual fit creation 2026-07-31 09:02:24 -03:00
Ian Roddis 23d07505bd Adding ability to autovivify an appropriate attachment point when manually entering a fit 2026-07-31 08:59:34 -03:00
Ian Roddis 92ca6b2a9f Adding fit type to create fit form 2026-07-31 08:56:39 -03:00
Ian Roddis b2d01f1ab0 Allowing attachment points to be pulled from descriptions on sync 2026-07-28 22:17:47 -03:00
Ian Roddis e24b8e5ea6 Updating CLAUDE.md 2026-07-28 22:08:13 -03:00
Ian Roddis aef60d77be Adding ability to edit attachment points on augmented item details 2026-07-28 22:04:40 -03:00
Ian Roddis 25fccd617f Redirecting deep links after login 2026-07-27 15:43:24 -03:00
88 changed files with 3283 additions and 460 deletions

No files matched your search

+27 -1
View File
@@ -13,7 +13,33 @@
"Bash(docker build *)",
"Bash(git status *)",
"Bash(git -C /home/iroddis/dev/armament/fitguide status --short)",
"Bash(git -C /home/iroddis/dev/armament/fitguide log --oneline -10)"
"Bash(git -C /home/iroddis/dev/armament/fitguide log --oneline -10)",
"Bash(node --check /Users/roddis/dev/armament/fitguide/app/static/js/sortable-tables.js)",
"Bash(sqlite3 /Users/roddis/dev/armament/fitguide/app.db \"select count\\(*\\) from users;\")",
"Bash(sqlite3 /Users/roddis/dev/armament/fitguide/app.db \"select count\\(*\\) from part_colours;\")",
"Bash(sqlite3 /Users/roddis/dev/armament/fitguide/app.db \"select count\\(*\\) from manufacturers;\")",
"Bash(curl -s -c /tmp/fg_cookies.txt -o /dev/null -w \"login:%{http_code}\\\\n\" -d \"email=admin@armament.com&password=changethis\" http://localhost:8000/login)",
"Bash(curl -s -b /tmp/fg_cookies.txt http://localhost:8000/admin/manufacturers)",
"Bash(curl -s -o /dev/null -w \"%{http_code} %{content_type}\\\\n\" http://localhost:8000/static/js/sortable-tables.js)",
"Bash(curl -s -b /tmp/fg_cookies.txt \"http://localhost:8000/admin/augmented-parts\")",
"Bash(curl -s -b /tmp/fg_cookies.txt \"http://localhost:8000/admin/augmented-parts/filter?sort=colour&dir=desc\")",
"Bash(curl -s -b /tmp/fg_cookies.txt \"http://localhost:8000/admin/roles\")",
"Bash(curl -s http://localhost:8000/scopes)",
"Bash(curl -s -o /dev/null -w \"%{http_code}\\\\n\" http://localhost:8000/scopes)",
"Bash(curl -s -b /tmp/fg_cookies.txt http://localhost:8000/scopes)",
"Bash(lsof -ti:8000 -sTCP:LISTEN)",
"Bash(xargs -r kill)",
"Bash(rm -f /tmp/fg_cookies.txt /private/tmp/claude-501/-Users-roddis-dev-armament-fitguide/7873fd9c-2f2b-4583-877e-72fc88718f24/scratchpad/smoke_sort.py)",
"Bash(curl -s -o /dev/null -w \"%{http_code}\\\\n\" http://localhost:8000/login)",
"Bash(curl -s -b /tmp/fg_cookies.txt http://localhost:8000__TRACKED_VAR__)",
"Bash(sqlite3 app.db \"PRAGMA foreign_keys=ON; SELECT scope_model_id FROM augmented_parts WHERE id=1862;\")",
"Bash(sqlite3 app.db \"SELECT id, name FROM scope_models WHERE name='Viper';\")",
"Bash(sqlite3 app.db \"SELECT COUNT\\(*\\) FROM augmented_parts WHERE scope_model_id=\\(SELECT id FROM scope_models WHERE name='Viper'\\);\")",
"Bash(sqlite3 app.db \"PRAGMA wal_checkpoint\\(FULL\\);\")",
"Bash(curl -s -o /dev/null -w \"%{http_code}\\\\n\" http://127.0.0.1:8000/login)",
"Bash(curl -s -o /dev/null -w \"%{http_code}\\\\n\" http://127.0.0.1:8000/admin/imported-parts)",
"Bash(rg -n \"hx-delete|router.delete\" app/web/admin.py app/templates/admin/*.html)",
"Bash(rg -n \"is_superuser\" tests/conftest.py tests/*.py)"
]
}
}
+3
View File
@@ -25,6 +25,9 @@ AUTO_CREATE_TABLES=true
COOKIE_NAME=access_token
# Set true when serving over HTTPS.
COOKIE_SECURE=false
# How long a browser login lasts before being bounced back to /login.
# Independent of ACCESS_TOKEN_EXPIRE_MINUTES, which governs JSON API tokens.
SESSION_TIMEOUT_HOURS=24
# --- LDAP (optional; direct-bind) ---
# When enabled, the login page and /api/auth/ldap-login accept a directory
+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 two `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
+1
View File
@@ -21,3 +21,4 @@
- Models
- Colours
- Part Types
- Notes
+1 -1
View File
@@ -1,4 +1,4 @@
VERSION := $(shell git describe --tags --always)
VERSION := $(shell git tag | tail -1)
IMAGE_NAME := fitguide
sync:
+2 -1
View File
@@ -149,8 +149,9 @@ Every setting is read from an environment variable first, then `.env`
|----------|---------|-------|
| `SECRET_KEY` | dev placeholder | **set a strong value in production** |
| `DATABASE_URL` | `sqlite+aiosqlite:///./app.db` | use `postgresql+asyncpg://...` for Postgres |
| `ACCESS_TOKEN_EXPIRE_MINUTES` | `15` | |
| `ACCESS_TOKEN_EXPIRE_MINUTES` | `15` | JSON API bearer token lifetime |
| `REFRESH_TOKEN_EXPIRE_DAYS` | `7` | |
| `SESSION_TIMEOUT_HOURS` | `24` | browser login (HTMX UI cookie) lifetime |
| `AUTO_CREATE_TABLES` | `true` | create tables on startup (dev); use Alembic in prod |
| `COOKIE_SECURE` | `false` | set `true` behind HTTPS |
| `FIRST_SUPERUSER_EMAIL` / `_PASSWORD` | admin@example.com / changethis | used by seed |
@@ -0,0 +1,44 @@
"""add jobs table
Revision ID: 04425ce3d4d5
Revises: a3f7c19b2d40
Create Date: 2026-08-05 15:21:09.527442
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = "04425ce3d4d5"
down_revision: Union[str, None] = "a3f7c19b2d40"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table(
"jobs",
sa.Column("id", sa.String(length=32), nullable=False),
sa.Column("kind", sa.String(length=64), nullable=False),
sa.Column("status", sa.String(length=16), nullable=False),
sa.Column("message", sa.Text(), nullable=False),
sa.Column("result", sa.JSON(), nullable=True),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
server_default=sa.text("(CURRENT_TIMESTAMP)"),
nullable=False,
),
sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
sa.PrimaryKeyConstraint("id"),
)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table("jobs")
# ### end Alembic commands ###
@@ -31,10 +31,16 @@ def upgrade() -> None:
nullable=False,
),
sa.Column(
"created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
"created_at",
sa.DateTime(timezone=True),
server_default=sa.func.now(),
nullable=False,
),
sa.Column(
"updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
"updated_at",
sa.DateTime(timezone=True),
server_default=sa.func.now(),
nullable=False,
),
sa.ForeignKeyConstraint(
["attachment_point_a_id"], ["attachment_points.id"], ondelete="CASCADE"
@@ -59,7 +65,10 @@ def upgrade() -> None:
sa.Column("username", sa.String(length=255), nullable=False),
sa.Column("body", sa.Text(), nullable=False),
sa.Column(
"created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
"created_at",
sa.DateTime(timezone=True),
server_default=sa.func.now(),
nullable=False,
),
sa.ForeignKeyConstraint(["fit_id"], ["fits.id"], ondelete="CASCADE"),
sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="SET NULL"),
@@ -0,0 +1,43 @@
"""add_augmented_part_notes_table
Revision ID: 8012c00b0c04
Revises: d1e2f3a4b5c6
Create Date: 2026-07-31 09:07:06.137175
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = "8012c00b0c04"
down_revision: Union[str, None] = "d1e2f3a4b5c6"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
"augmented_part_notes",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("augmented_part_id", sa.Integer(), nullable=False),
sa.Column("user_id", sa.Integer(), nullable=True),
sa.Column("username", sa.String(length=255), nullable=False),
sa.Column("body", sa.Text(), nullable=False),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
server_default=sa.func.now(),
nullable=False,
),
sa.ForeignKeyConstraint(
["augmented_part_id"], ["augmented_parts.id"], ondelete="CASCADE"
),
sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="SET NULL"),
sa.PrimaryKeyConstraint("id"),
)
def downgrade() -> None:
op.drop_table("augmented_part_notes")
@@ -0,0 +1,28 @@
"""add_objective_lens_diameter_to_scope_infos
Revision ID: a3f7c19b2d40
Revises: 8012c00b0c04
Create Date: 2026-08-04 00:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = "a3f7c19b2d40"
down_revision: Union[str, None] = "8012c00b0c04"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.add_column(
"scope_infos",
sa.Column("objective_lens_diameter_mm", sa.Float(), nullable=True),
)
def downgrade() -> None:
op.drop_column("scope_infos", "objective_lens_diameter_mm")
+22
View File
@@ -4,10 +4,13 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.crud import audit as audit_crud
from app.crud import augmented_part as augmented_part_crud
from app.crud import augmented_part_note as augmented_part_note_crud
from app.deps import require_permission
from app.models.user import User
from app.schemas.augmented_part import (
AugmentedPartCreate,
AugmentedPartNoteCreate,
AugmentedPartNoteRead,
AugmentedPartRead,
AugmentedPartSyncResult,
AugmentedPartUpdate,
@@ -132,3 +135,22 @@ async def delete_augmented_part(
"AugmentedPart", augmented_part_id, label
),
)
@router.post(
"/{augmented_part_id}/notes",
response_model=AugmentedPartNoteRead,
status_code=status.HTTP_201_CREATED,
)
async def create_augmented_part_note(
augmented_part_id: int,
payload: AugmentedPartNoteCreate,
db: AsyncSession = Depends(get_db),
actor: User = Depends(require_permission("augmented_parts:write")),
):
ap = await augmented_part_crud.get(db, augmented_part_id)
if ap is None:
raise HTTPException(status_code=404, detail="Augmented part not found")
return await augmented_part_note_crud.create(
db, augmented_part=ap, author=actor, body=payload.body
)
+1 -3
View File
@@ -67,9 +67,7 @@ async def create_fit(
if ap_a is None or ap_b is None:
raise HTTPException(status_code=404, detail="Attachment point not found")
if ap_a.augmented_part_id == ap_b.augmented_part_id:
raise HTTPException(
status_code=422, detail="a part cannot fit itself"
)
raise HTTPException(status_code=422, detail="a part cannot fit itself")
existing = await fit_crud.get_by_attachment_points(db, ap_a, ap_b)
if existing is not None:
raise HTTPException(status_code=409, detail="Fit already exists")
+5 -1
View File
@@ -33,9 +33,13 @@ class Settings(BaseSettings):
database_url: str = "sqlite+aiosqlite:///./app.db"
auto_create_tables: bool = True
# Web / cookies (the HTMX UI stores the access token in an httpOnly cookie)
# Web / cookies (the HTMX UI stores the access token in an httpOnly cookie).
# session_timeout_hours governs how long a browser login lasts before the
# user is bounced back to /login; it's independent of the JSON API's
# short-lived access_token_expire_minutes.
cookie_name: str = "access_token"
cookie_secure: bool = False
session_timeout_hours: int = 24
# LDAP (direct-bind). When enabled, the login page offers an LDAP option.
# At login the user's DN is built from ``ldap_bind_dn_template`` (with
+1
View File
@@ -35,6 +35,7 @@ if engine.sync_engine.dialect.name == "sqlite":
cursor.execute("PRAGMA busy_timeout=5000")
cursor.close()
AsyncSessionLocal = async_sessionmaker(
engine,
class_=AsyncSession,
+4 -2
View File
@@ -47,10 +47,12 @@ def _create_token(
return token, payload
def create_access_token(subject: str | int) -> str:
def create_access_token(
subject: str | int, expires_delta: timedelta | None = None
) -> str:
token, _ = _create_token(
subject,
timedelta(minutes=settings.access_token_expire_minutes),
expires_delta or timedelta(minutes=settings.access_token_expire_minutes),
ACCESS_TOKEN_TYPE,
)
return token
+19
View File
@@ -2,6 +2,7 @@ from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.attachment_point import AttachmentPoint
from app.models.attachment_point_name import AttachmentPointName
_RELATIONSHIPS = [
"attachment_point_name",
@@ -27,6 +28,24 @@ async def list_for_augmented_part(
return list(result.scalars().all())
async def list_by_augmented_parts_and_name(
db: AsyncSession, augmented_part_ids: list[int], name: str
) -> list[AttachmentPoint]:
"""AttachmentPoints named ``name`` for the given augmented parts, fetched in
one query."""
if not augmented_part_ids:
return []
result = await db.execute(
select(AttachmentPoint)
.join(AttachmentPoint.attachment_point_name)
.where(
AttachmentPoint.augmented_part_id.in_(augmented_part_ids),
AttachmentPointName.name == name,
)
)
return list(result.scalars().all())
async def list_all(
db: AsyncSession, skip: int = 0, limit: int = 100
) -> list[AttachmentPoint]:
+54 -8
View File
@@ -1,12 +1,50 @@
from sqlalchemy import distinct, select
from sqlalchemy import distinct, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.associations import augmented_part_imported_parts
from app.models.augmented_part import AugmentedPart
from app.models.imported_part import ImportedPart
from app.models.manufacturer import Manufacturer
from app.models.part_colour import PartColour
from app.models.part_type import PartType
from app.models.scope_model import ScopeModel
from app.models.sub_part_type import SubPartType
_RELATIONSHIPS = ["colour", "manufacturer", "scope_model", "part_type", "sub_part_type"]
# Sortable columns for the admin list, keyed by the query-string "sort" value.
# Columns backed by a related table's name require an outer join (the FK may be
# nullable, e.g. scope_model_id) so unrelated rows still appear, ordered last.
SORT_COLUMNS = {
"id": (AugmentedPart.id, None),
"sku": (AugmentedPart.sku, None),
"description": (AugmentedPart.description, None),
"colour": (PartColour.name, (PartColour, AugmentedPart.colour_id == PartColour.id)),
"manufacturer": (
Manufacturer.name,
(Manufacturer, AugmentedPart.manufacturer_id == Manufacturer.id),
),
"scope_model": (
ScopeModel.name,
(ScopeModel, AugmentedPart.scope_model_id == ScopeModel.id),
),
"part_type": (PartType.name, (PartType, AugmentedPart.part_type_id == PartType.id)),
"sub_part_type": (
SubPartType.name,
(SubPartType, AugmentedPart.sub_part_type_id == SubPartType.id),
),
}
def _apply_sort(q, sort: str, direction: str):
column, join = SORT_COLUMNS.get(sort, SORT_COLUMNS["description"])
if join is not None:
table, cond = join
q = q.outerjoin(table, cond)
order = column.desc() if direction == "desc" else column.asc()
# AugmentedPart.id as a tiebreaker keeps paginated ordering stable.
return q.order_by(order, AugmentedPart.id)
async def get(db: AsyncSession, augmented_part_id: int) -> AugmentedPart | None:
return await db.get(AugmentedPart, augmented_part_id)
@@ -35,9 +73,7 @@ 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))
)
result = await db.execute(select(AugmentedPart).where(AugmentedPart.id.in_(ids)))
return list(result.scalars().all())
@@ -49,12 +85,23 @@ async def list_filtered(
scope_model_id: int | None = None,
part_type_id: int | None = None,
sub_part_type_id: int | None = None,
sort: str = "description",
direction: str = "asc",
skip: int = 0,
limit: int = 100,
match_sku: bool = False,
) -> list[AugmentedPart]:
q = select(AugmentedPart)
if description.strip():
q = q.where(AugmentedPart.description.ilike(f"%{description.strip()}%"))
term = f"%{description.strip()}%"
if match_sku:
q = q.where(
or_(
AugmentedPart.description.ilike(term), AugmentedPart.sku.ilike(term)
)
)
else:
q = q.where(AugmentedPart.description.ilike(term))
if colour_id is not None:
q = q.where(AugmentedPart.colour_id == colour_id)
if manufacturer_id is not None:
@@ -65,9 +112,8 @@ async def list_filtered(
q = q.where(AugmentedPart.part_type_id == part_type_id)
if sub_part_type_id is not None:
q = q.where(AugmentedPart.sub_part_type_id == sub_part_type_id)
result = await db.execute(
q.order_by(AugmentedPart.description).offset(skip).limit(limit)
)
q = _apply_sort(q, sort, direction)
result = await db.execute(q.offset(skip).limit(limit))
return list(result.scalars().all())
+19
View File
@@ -0,0 +1,19 @@
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.augmented_part import AugmentedPart, AugmentedPartNote
from app.models.user import User
async def create(
db: AsyncSession, *, augmented_part: AugmentedPart, author: User, body: str
) -> AugmentedPartNote:
note = AugmentedPartNote(
augmented_part_id=augmented_part.id,
user_id=author.id,
username=author.full_name or author.email,
body=body,
)
db.add(note)
await db.commit()
await db.refresh(note)
return note
+40 -6
View File
@@ -16,14 +16,37 @@ _FIELDS = frozenset(
}
)
# Sortable columns for the admin list, keyed by the query-string "sort" value.
SORT_COLUMNS = {
"InventoryID": ImportedPart.InventoryID,
"Description": ImportedPart.Description,
"TBXColour": ImportedPart.TBXColour,
"ProductMaxThreadDepthmm": ImportedPart.ProductMaxThreadDepthmm,
"ProductMaxThreadDepthin": ImportedPart.ProductMaxThreadDepthin,
"ProductODmm": ImportedPart.ProductODmm,
"ProductODin": ImportedPart.ProductODin,
"TBXThreadAttribute": ImportedPart.TBXThreadAttribute,
}
def _apply_sort(q, sort: str, direction: str):
column = SORT_COLUMNS.get(sort, ImportedPart.InventoryID)
order = column.desc() if direction == "desc" else column.asc()
return q.order_by(order, ImportedPart.InventoryID)
async def get(db: AsyncSession, inventory_id: str) -> ImportedPart | None:
return await db.get(ImportedPart, inventory_id)
async def list_all(db: AsyncSession, skip: int = 0, limit: int = 500) -> list[ImportedPart]:
async def list_all(
db: AsyncSession, skip: int = 0, limit: int = 500
) -> list[ImportedPart]:
result = await db.execute(
select(ImportedPart).order_by(ImportedPart.InventoryID).offset(skip).limit(limit)
select(ImportedPart)
.order_by(ImportedPart.InventoryID)
.offset(skip)
.limit(limit)
)
return list(result.scalars().all())
@@ -31,13 +54,16 @@ async def list_all(db: AsyncSession, skip: int = 0, limit: int = 500) -> list[Im
async def list_filtered(
db: AsyncSession,
description: str = "",
sort: str = "InventoryID",
direction: str = "asc",
skip: int = 0,
limit: int = 500,
) -> list[ImportedPart]:
q = select(ImportedPart)
if description.strip():
q = q.where(ImportedPart.Description.ilike(f"%{description.strip()}%"))
result = await db.execute(q.order_by(ImportedPart.InventoryID).offset(skip).limit(limit))
q = _apply_sort(q, sort, direction)
result = await db.execute(q.offset(skip).limit(limit))
return list(result.scalars().all())
@@ -78,7 +104,11 @@ async def upsert_bulk(db: AsyncSession, records: list[dict]) -> tuple[int, int]:
return 0, 0
clean = [
{k: str(v).strip() if v is not None else None for k, v in r.items() if k in _FIELDS}
{
k: str(v).strip() if v is not None else None
for k, v in r.items()
if k in _FIELDS
}
for r in records
if r.get("InventoryID")
]
@@ -88,7 +118,9 @@ async def upsert_bulk(db: AsyncSession, records: list[dict]) -> tuple[int, int]:
incoming_ids = {r["InventoryID"] for r in clean}
result = await db.execute(
select(ImportedPart.InventoryID).where(ImportedPart.InventoryID.in_(incoming_ids))
select(ImportedPart.InventoryID).where(
ImportedPart.InventoryID.in_(incoming_ids)
)
)
existing_ids = {row[0] for row in result.all()}
@@ -103,7 +135,9 @@ async def upsert_bulk(db: AsyncSession, records: list[dict]) -> tuple[int, int]:
vals = {k: v for k, v in record.items() if k != "InventoryID"}
if vals:
await db.execute(
update(ImportedPart).where(ImportedPart.InventoryID == inv_id).values(vals)
update(ImportedPart)
.where(ImportedPart.InventoryID == inv_id)
.values(vals)
)
await db.commit()
+1 -3
View File
@@ -23,9 +23,7 @@ async def get_default(db: AsyncSession) -> Role | None:
async def set_default(db: AsyncSession, role: Role) -> Role:
"""Make ``role`` the sole default, clearing the flag on every other role."""
await db.execute(
sa_update(Role).where(Role.id != role.id).values(is_default=False)
)
await db.execute(sa_update(Role).where(Role.id != role.id).values(is_default=False))
role.is_default = True
await db.commit()
await db.refresh(role)
+3 -3
View File
@@ -24,9 +24,7 @@ async def list_by_augmented_parts(
if not augmented_part_ids:
return []
result = await db.execute(
select(ScopeInfo).where(
ScopeInfo.augmented_part_id.in_(augmented_part_ids)
)
select(ScopeInfo).where(ScopeInfo.augmented_part_id.in_(augmented_part_ids))
)
return list(result.scalars().all())
@@ -38,6 +36,7 @@ async def create(
magnification_min: float | None = None,
magnification_max: float | None = None,
model_suffix: str | None = None,
objective_lens_diameter_mm: float | None = None,
objective_thread_pitch: str | None = None,
) -> ScopeInfo:
info = ScopeInfo(
@@ -45,6 +44,7 @@ async def create(
magnification_min=magnification_min,
magnification_max=magnification_max,
model_suffix=model_suffix,
objective_lens_diameter_mm=objective_lens_diameter_mm,
objective_thread_pitch=objective_thread_pitch,
)
db.add(info)
+10 -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
@@ -45,7 +46,15 @@ def create_app() -> FastAPI:
@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():
+4 -1
View File
@@ -8,12 +8,13 @@ from app.models.associations import (
from app.models.attachment_point import AttachmentPoint
from app.models.attachment_point_name import AttachmentPointName
from app.models.audit_log import AuditLog
from app.models.augmented_part import AugmentedPart
from app.models.augmented_part import AugmentedPart, AugmentedPartNote
from app.models.clamp_connection import ClampConnection
from app.models.fit import Fit, FitLabel, FitNote
from app.models.imported_part import ImportedPart
from app.models.imported_part_config import ImportedPartConfig
from app.models.iso_thread import IsoThread
from app.models.job import Job
from app.models.manufacturer import Manufacturer
from app.models.part_colour import PartColour
from app.models.part_type import PartType
@@ -32,6 +33,7 @@ __all__ = [
"AttachmentPointName",
"AuditLog",
"AugmentedPart",
"AugmentedPartNote",
"ClampConnection",
"Fit",
"FitLabel",
@@ -39,6 +41,7 @@ __all__ = [
"ImportedPart",
"ImportedPartConfig",
"IsoThread",
"Job",
"Manufacturer",
"PartColour",
"PartType",
+30 -1
View File
@@ -1,4 +1,6 @@
from sqlalchemy import ForeignKey, String
from datetime import datetime
from sqlalchemy import DateTime, ForeignKey, String, Text, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.core.database import Base
@@ -45,3 +47,30 @@ class AugmentedPart(Base):
scope_model: Mapped[ScopeModel | None] = relationship(lazy="selectin")
part_type: Mapped[PartType] = relationship(lazy="selectin")
sub_part_type: Mapped[SubPartType | None] = relationship(lazy="selectin")
notes: Mapped[list["AugmentedPartNote"]] = relationship(
back_populates="augmented_part",
lazy="selectin",
order_by="AugmentedPartNote.created_at",
cascade="all, delete-orphan",
)
class AugmentedPartNote(Base):
__tablename__ = "augmented_part_notes"
id: Mapped[int] = mapped_column(primary_key=True)
augmented_part_id: Mapped[int] = mapped_column(
ForeignKey("augmented_parts.id", ondelete="CASCADE"), nullable=False
)
user_id: Mapped[int | None] = mapped_column(
ForeignKey("users.id", ondelete="SET NULL"), nullable=True
)
# Denormalized snapshot of who wrote the note (mirrors FitNote.username), so
# the note stays attributable even if the user is later deleted.
username: Mapped[str] = mapped_column(String(255), nullable=False)
body: Mapped[str] = mapped_column(Text, nullable=False)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now()
)
augmented_part: Mapped[AugmentedPart] = relationship(back_populates="notes")
+31
View File
@@ -0,0 +1,31 @@
from datetime import datetime
from sqlalchemy import JSON, DateTime, String, Text, func
from sqlalchemy.orm import Mapped, mapped_column
from app.core.database import Base
class Job(Base):
"""Tracks a long-running admin operation (sync, import, fit recompute).
Persisted (rather than kept in a module-level dict) so that any worker
process/replica handling a status-poll request can see progress recorded
by whichever process is actually running the job.
"""
__tablename__ = "jobs"
id: Mapped[str] = mapped_column(String(32), primary_key=True)
kind: Mapped[str] = mapped_column(String(64), nullable=False)
status: Mapped[str] = mapped_column(
String(16), nullable=False, default="running"
) # running | done | error
message: Mapped[str] = mapped_column(Text, nullable=False, default="Working…")
result: Mapped[dict | None] = mapped_column(JSON, nullable=True, default=None)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now()
)
finished_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True, default=None
)
+5
View File
@@ -27,6 +27,11 @@ class ScopeInfo(Base):
model_suffix: Mapped[str | None] = mapped_column(
String, nullable=True, default=None
)
# The lens diameter is a descriptive attribute (the "50" in "3-18x50"), not a
# connection dimension — it has no AttachmentPoint.
objective_lens_diameter_mm: Mapped[float | None] = mapped_column(
Float, nullable=True, default=None
)
# Kept as the raw string from the sheet (e.g. "0.75", "1/16") so no precision
# or notation is lost; the parsed thread lives on its own AttachmentPoint.
objective_thread_pitch: Mapped[str | None] = mapped_column(
+26 -1
View File
@@ -1,4 +1,6 @@
from pydantic import BaseModel, ConfigDict
from datetime import datetime
from pydantic import BaseModel, ConfigDict, field_validator
from app.schemas.manufacturer import ManufacturerRead
from app.schemas.part_colour import PartColourRead
@@ -7,6 +9,28 @@ from app.schemas.scope_model import ScopeModelRead
from app.schemas.sub_part_type import SubPartTypeRead
class AugmentedPartNoteCreate(BaseModel):
body: str
@field_validator("body")
@classmethod
def validate_body(cls, v: str) -> str:
if not v.strip():
raise ValueError("body must not be blank")
return v.strip()
class AugmentedPartNoteRead(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
augmented_part_id: int
user_id: int | None
username: str
body: str
created_at: datetime
class AugmentedPartBase(BaseModel):
sku: str | None = None
description: str
@@ -46,3 +70,4 @@ class AugmentedPartRead(AugmentedPartBase):
scope_model: ScopeModelRead | None
part_type: PartTypeRead
sub_part_type: SubPartTypeRead | None
notes: list[AugmentedPartNoteRead] = []
+1
View File
@@ -5,6 +5,7 @@ class ScopeInfoBase(BaseModel):
magnification_min: float | None = None
magnification_max: float | None = None
model_suffix: str | None = None
objective_lens_diameter_mm: float | None = None
objective_thread_pitch: str | None = None
+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))
+12 -3
View File
@@ -1,5 +1,7 @@
"""Token issuance / rotation shared by the JSON API and the HTMX UI."""
from datetime import timedelta
import jwt
from sqlalchemy.ext.asyncio import AsyncSession
@@ -13,9 +15,16 @@ from app.crud import token as token_crud
from app.models.user import User
async def issue_token_pair(db: AsyncSession, user: User) -> tuple[str, str]:
"""Create an access + refresh token and persist the refresh token for revocation."""
access = create_access_token(user.id)
async def issue_token_pair(
db: AsyncSession, user: User, access_expires_delta: timedelta | None = None
) -> tuple[str, str]:
"""Create an access + refresh token and persist the refresh token for revocation.
``access_expires_delta`` overrides the access token's default lifetime
(``settings.access_token_expire_minutes``) — used by the web login flow to
issue a longer-lived session cookie without affecting API bearer tokens.
"""
access = create_access_token(user.id, access_expires_delta)
refresh, payload = create_refresh_token(user.id)
await token_crud.store(
db,
+54
View File
@@ -0,0 +1,54 @@
"""Irreversible bulk-delete of all domain (parts/taxonomy/fit) data.
Leaves users, roles/permissions, refresh tokens, the audit log, and the
ImportedPartConfig (OData connection settings) untouched.
"""
from sqlalchemy import delete
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.associations import augmented_part_imported_parts
from app.models.attachment_point import AttachmentPoint
from app.models.attachment_point_name import AttachmentPointName
from app.models.augmented_part import AugmentedPart, AugmentedPartNote
from app.models.clamp_connection import ClampConnection
from app.models.fit import Fit, FitNote
from app.models.imported_part import ImportedPart
from app.models.iso_thread import IsoThread
from app.models.manufacturer import Manufacturer
from app.models.scope_info import ScopeInfo
from app.models.scope_model import ScopeModel
from app.models.static_connection import StaticConnection
from app.models.unc_thread import UncThread
# Children before the RESTRICT-protected reference tables they point to.
_TARGETS = [
("Fit notes", FitNote),
("Fits", Fit),
("Augmented part notes", AugmentedPartNote),
("Scope info", ScopeInfo),
("Attachment points", AttachmentPoint),
("Augmented part / imported part links", augmented_part_imported_parts),
("Augmented parts", AugmentedPart),
("Imported parts", ImportedPart),
("Scope models", ScopeModel),
("Attachment point names", AttachmentPointName),
("ISO threads", IsoThread),
("UNC threads", UncThread),
("Clamp connections", ClampConnection),
("Static connections", StaticConnection),
("Manufacturers", Manufacturer),
]
async def purge_domain_data(db: AsyncSession) -> dict[str, int]:
"""Delete every row from every domain table, in FK-safe order.
Returns a label -> rows-deleted map for reporting/audit purposes.
"""
counts: dict[str, int] = {}
for label, target in _TARGETS:
result = await db.execute(delete(target))
counts[label] = result.rowcount or 0
await db.commit()
return counts
+4 -4
View File
@@ -142,9 +142,7 @@ async def _reconcile(
for own_ap in own_points:
for other_ap in points_by_part.get(other_id, []):
if attachment_points_match(own_ap, other_ap):
candidate_pairs.add(
_canonical_pair(own_ap.id, other_ap.id)
)
candidate_pairs.add(_canonical_pair(own_ap.id, other_ap.id))
result = RecomputeResult()
@@ -171,7 +169,9 @@ async def _reconcile(
return result
async def recompute_for_part(db: AsyncSession, augmented_part_id: int) -> RecomputeResult:
async def recompute_for_part(
db: AsyncSession, augmented_part_id: int
) -> RecomputeResult:
"""Reconcile Possible fits involving this part against the current algorithm.
Creates Possible fits for newly-matching attachment point pairs, and removes
+96
View File
@@ -0,0 +1,96 @@
"""DB-backed tracking for long-running admin operations.
Jobs are persisted to the ``jobs`` table rather than kept in a module-level
dict, so that a status-poll request landing on a *different* worker
process/replica than the one running the job can still see its progress —
required once the app runs with more than one uvicorn worker or replica
(a module-level dict is process-local and would report "not found" for a job
started elsewhere, even while that job is still running and writing to the
shared database).
"""
from __future__ import annotations
import asyncio
import uuid
from collections.abc import Awaitable, Callable
from datetime import UTC, datetime, timedelta
from sqlalchemy import delete
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import AsyncSessionLocal
from app.models.job import Job
_MAX_FINISHED_AGE = timedelta(hours=1)
async def get(db: AsyncSession, job_id: str) -> Job | None:
return await db.get(Job, job_id)
async def start(kind: str, run: Callable[[str], Awaitable[None]]) -> Job:
"""Create a job row, launch ``run(job_id)`` in the background, and return
the job immediately.
``run`` is responsible for calling ``mark_done``/``mark_error`` to record
the outcome, including on its own errors. Any exception that escapes
anyway is caught here as a last resort so the job doesn't hang forever in
"running" state.
"""
async with AsyncSessionLocal() as db:
await _evict_stale(db)
job = Job(id=uuid.uuid4().hex, kind=kind, status="running", message="Working…")
db.add(job)
await db.commit()
async def _runner() -> None:
try:
await run(job.id)
except Exception as exc:
await mark_error(job.id, f"{kind} failed: {exc}")
asyncio.create_task(_runner())
return job
async def mark_done(job_id: str, message: str, result: dict | None = None) -> None:
await _update(job_id, status="done", message=message, result=result)
async def mark_error(job_id: str, message: str) -> None:
await _update(job_id, status="error", message=message)
async def set_message(job_id: str, message: str) -> None:
await _update(job_id, message=message)
async def _update(
job_id: str,
*,
status: str | None = None,
message: str | None = None,
result: dict | None = None,
) -> None:
async with AsyncSessionLocal() as db:
job = await db.get(Job, job_id)
if job is None:
return
if status is not None:
job.status = status
if status in ("done", "error"):
job.finished_at = datetime.now(UTC)
if message is not None:
job.message = message
if result is not None:
job.result = result
await db.commit()
async def _evict_stale(db: AsyncSession) -> None:
cutoff = datetime.now(UTC) - _MAX_FINISHED_AGE
await db.execute(
delete(Job).where(Job.finished_at.is_not(None), Job.finished_at < cutoff)
)
await db.commit()
+25 -10
View File
@@ -7,6 +7,7 @@ the parts of that type that fit the row's scope.
from sqlalchemy.ext.asyncio import AsyncSession
from app.crud import attachment_point as attachment_point_crud
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
@@ -14,6 +15,7 @@ from app.crud import scope_info as scope_info_crud
from app.models.fit import FitLabel
SCOPE_TYPE_NAME = "Scope"
OBJECTIVE_ATTACHMENT_POINT_NAME = "Objective"
# "exact" outranks "possible" when the same part fits a scope via several fits.
_LABEL_RANK = {FitLabel.EXACT.value: 2, FitLabel.POSSIBLE.value: 1}
@@ -31,9 +33,12 @@ async def build(
"""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.
matches either the scope's description or its SKU/product ID; the ``*_id``
filters match exactly). ``fits_part_type_id`` keeps only scopes that have
at least one (non-``none``)
fit for that part type. Callers that paginate should slice ``rows``
themselves — this always returns the full filtered set, since
``fits_part_type_id`` is a post-filter that can't be pushed into the DB query.
Shape::
@@ -43,6 +48,7 @@ async def build(
{
"scope": AugmentedPart,
"info": ScopeInfo | None,
"objective_attachment_point": AttachmentPoint | None,
"cells": {part_type_id: [{"part": AugmentedPart, "label": str}]},
},
...,
@@ -50,9 +56,7 @@ async def build(
}
"""
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
)
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
]
@@ -68,6 +72,7 @@ async def build(
sub_part_type_id=sub_part_type_id,
part_type_id=scope_type.id,
limit=1000,
match_sku=True,
)
scope_ids = {s.id for s in scopes}
if not scope_ids:
@@ -87,6 +92,13 @@ async def build(
p.id: p for p in await augmented_part_crud.get_many(db, list(other_ids))
}
objective_aps = {
ap.augmented_part_id: ap
for ap in await attachment_point_crud.list_by_augmented_parts_and_name(
db, list(scope_ids), OBJECTIVE_ATTACHMENT_POINT_NAME
)
}
infos = {
info.augmented_part_id: info
for info in await scope_info_crud.list_by_augmented_parts(db, list(scope_ids))
@@ -94,9 +106,7 @@ async def build(
# 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
}
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:
@@ -121,7 +131,12 @@ async def build(
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}
{
"scope": scope,
"info": infos.get(scope.id),
"objective_attachment_point": objective_aps.get(scope.id),
"cells": cells,
}
)
return {"column_types": column_types, "rows": rows}
+21 -44
View File
@@ -2,7 +2,7 @@
The spreadsheet describes parts of PartType "Scope". Each row maps onto an
AugmentedPart (keyed on ``product_number`` → ``sku``, so re-importing upserts),
a ScopeModel, a ScopeInfo record for the scope-specific fields, up to three
a ScopeModel, a ScopeInfo record for the scope-specific fields, up to two
attachment points, and up to two Fits against other parts.
Column → destination
@@ -14,9 +14,9 @@ model_suffix → ScopeInfo.model_suffix (+ description)
sub_part_type → SubPartType (get-or-create under "Scope")
product_number → AugmentedPart.sku (also the upsert / fit-lookup key)
magnification_min / _max → ScopeInfo
objective_lens_diameter_mm → ScopeInfo (descriptive only, no attachment point)
objective_thread_pitch → ScopeInfo
ocular_outer_diameter_mm → StaticConnection + AttachmentPoint "Ocular"
objective_lens_outside_diameter_mm → StaticConnection + AttachmentPoint "Objective"
objective_thread → ISO/UNC thread + AttachmentPoint "Objective"
ocular_fit → Fit: scope's Ocular AP ↔ referenced part's first AP
objective_fit → Fit: scope's Objective AP ↔ referenced part's first AP
@@ -68,7 +68,7 @@ EXPECTED_COLUMNS = [
"model",
"magnification_min",
"magnification_max",
"objective_lens_outside_diameter_mm",
"objective_lens_diameter_mm",
"model_suffix",
"sub_part_type",
"product_number",
@@ -140,15 +140,11 @@ def parse_workbook(data: bytes) -> list[dict]:
wb.close()
async def _static_ap(
db: AsyncSession, *, augmented_part_id: int, label: str, od_mm: float
):
async def _static_ap(db: AsyncSession, *, augmented_part_id: int, label: str, od_mm: float):
"""Create an attachment point carrying a StaticConnection for ``od_mm``."""
sc = await static_connection_crud.get_by_od_mm(db, od_mm)
if sc is None:
sc = await static_connection_crud.create(
db, od_mm=od_mm, od_in=od_mm / MM_PER_INCH
)
sc = await static_connection_crud.create(db, od_mm=od_mm, od_in=od_mm / MM_PER_INCH)
apn = await apn_crud.get_by_name(db, label)
if apn is None:
apn = await apn_crud.create(db, name=label)
@@ -160,9 +156,7 @@ async def _static_ap(
)
async def _thread_ap(
db: AsyncSession, *, augmented_part_id: int, label: str, spec: str
):
async def _thread_ap(db: AsyncSession, *, augmented_part_id: int, label: str, spec: str):
"""Create an attachment point for a parsed ISO/UNC thread, or None."""
parsed = _parse_thread_attribute(spec)
if parsed is None:
@@ -206,8 +200,7 @@ async def _link_fit(
"""Create a POSSIBLE fit from the scope's AP to the referenced part's first AP."""
if scope_ap is None:
messages.append(
f"{scope_role}_fit '{target_sku}' skipped: scope has no "
f"{scope_role} attachment point"
f"{scope_role}_fit '{target_sku}' skipped: scope has no {scope_role} attachment point"
)
return
target = await augmented_part_crud.get_by_sku(db, target_sku)
@@ -220,16 +213,13 @@ async def _link_fit(
target_aps = await attachment_point_crud.list_for_augmented_part(db, target.id)
if not target_aps:
messages.append(
f"{scope_role}_fit '{target_sku}' skipped: referenced part has no "
"attachment points"
f"{scope_role}_fit '{target_sku}' skipped: referenced part has no attachment points"
)
return
target_ap = target_aps[0]
existing = await fit_crud.get_by_attachment_points(db, scope_ap, target_ap)
if existing is None:
await fit_crud.create(
db, ap_a=scope_ap, ap_b=target_ap, label=FitLabel.POSSIBLE
)
await fit_crud.create(db, ap_a=scope_ap, ap_b=target_ap, label=FitLabel.POSSIBLE)
async def import_scopes(db: AsyncSession, rows: list[dict]) -> ScopeImportResult:
@@ -250,7 +240,9 @@ async def import_scopes(db: AsyncSession, rows: list[dict]) -> ScopeImportResult
for offset, row in enumerate(rows):
# +2: 1-based, and the header occupies row 1.
row_num = offset + 2
product_number = _text(row.get("product_number")) or _text(row.get("model"))
model_suffix = _text(row.get("model_suffix"))
default_product_number = f"{_text(row.get('model'))} {row.get('magnification_min')}-{row.get('magnification_max')}x{row.get('objective_lens_diameter_mm')} {row.get('model_suffix')}".strip()
product_number = _text(row.get("product_number")) or default_product_number
result = ScopeImportRowResult(
row=row_num, product_number=product_number, status="skipped", messages=[]
)
@@ -271,9 +263,7 @@ async def import_scopes(db: AsyncSession, rows: list[dict]) -> ScopeImportResult
manufacturer = await manufacturer_crud.get_by_name(db, manufacturer_name)
if manufacturer is None:
manufacturer = await manufacturer_crud.create(
db, name=manufacturer_name, url=""
)
manufacturer = await manufacturer_crud.create(db, name=manufacturer_name, url="")
model_name = _text(row.get("model"))
url = _text(row.get("url"))
@@ -330,14 +320,13 @@ async def import_scopes(db: AsyncSession, rows: list[dict]) -> ScopeImportResult
magnification_min=_num(row.get("magnification_min")),
magnification_max=_num(row.get("magnification_max")),
model_suffix=model_suffix,
objective_lens_diameter_mm=_num(row.get("objective_lens_diameter_mm")),
objective_thread_pitch=_text(row.get("objective_thread_pitch")),
)
# Rebuild attachment points from scratch so re-imports stay idempotent
# (cascades delete any fits that referenced the old points).
for old_ap in await attachment_point_crud.list_for_augmented_part(
db, part.id
):
for old_ap in await attachment_point_crud.list_for_augmented_part(db, part.id):
await attachment_point_crud.delete(db, old_ap)
ocular_ap = None
@@ -350,34 +339,22 @@ async def import_scopes(db: AsyncSession, rows: list[dict]) -> ScopeImportResult
od_mm=ocular_od,
)
objective_static_ap = None
objective_od = _num(row.get("objective_lens_outside_diameter_mm"))
if objective_od is not None:
objective_static_ap = await _static_ap(
db,
augmented_part_id=part.id,
label=OBJECTIVE_LABEL,
od_mm=objective_od,
)
# The objective attachment point comes from the thread alone — the
# lens diameter is a descriptive attribute, not a connection.
objective_thread_spec = _text(row.get("objective_thread"))
objective_thread_ap = None
objective_ap = None
if objective_thread_spec:
objective_thread_ap = await _thread_ap(
objective_ap = await _thread_ap(
db,
augmented_part_id=part.id,
label=OBJECTIVE_LABEL,
spec=objective_thread_spec,
)
if objective_thread_ap is None:
if objective_ap is None:
result.messages.append(
f"could not parse objective_thread "
f"'{objective_thread_spec}'"
f"could not parse objective_thread '{objective_thread_spec}'"
)
# For objective_fit prefer the lens-OD point, falling back to the thread.
objective_ap = objective_static_ap or objective_thread_ap
staged.append((row, ocular_ap, objective_ap, part.id, result))
row_results.append(result)
except Exception as exc: # noqa: BLE001 - report, don't abort the batch
+36
View File
@@ -247,6 +247,37 @@ h2 {
gap: 0.4rem;
}
/* Sortable table headers */
table.sortable thead th:not([data-sort-off]) {
cursor: pointer;
user-select: none;
}
table.sortable thead th:not([data-sort-off]):hover {
color: var(--text);
}
table.sortable thead th[aria-sort]::after {
margin-left: 0.3rem;
}
table.sortable thead th[aria-sort="ascending"]::after {
content: "\25B2";
}
table.sortable thead th[aria-sort="descending"]::after {
content: "\25BC";
}
.sort-link {
color: inherit;
text-decoration: none;
}
.sort-link:hover {
text-decoration: underline;
}
/* Tags / definition lists */
.tag {
display: inline-block;
@@ -536,6 +567,11 @@ textarea:focus {
margin-bottom: 0.75rem;
}
/* Bulk actions (e.g. delete-selected above a checkbox table) */
.bulk-actions {
margin-bottom: 0.75rem;
}
.filter-bar form {
display: flex;
flex-wrap: wrap;
Binary file not shown.

After

Width:  |  Height:  |  Size: 27 KiB

+77
View File
@@ -0,0 +1,77 @@
(function () {
"use strict";
function getCellText(row, index) {
var cell = row.children[index];
return cell ? cell.textContent.trim() : "";
}
function isEmptyValue(text) {
return text === "" || text === "—";
}
function isPureNumber(text) {
return /^-?\d+(\.\d+)?$/.test(text);
}
function columnIsNumeric(rows, index) {
var sawValue = false;
for (var i = 0; i < rows.length; i++) {
var text = getCellText(rows[i], index);
if (isEmptyValue(text)) continue;
sawValue = true;
if (!isPureNumber(text)) return false;
}
return sawValue;
}
function compareRows(a, b, index, numeric, dir) {
var textA = getCellText(a, index);
var textB = getCellText(b, index);
var emptyA = isEmptyValue(textA);
var emptyB = isEmptyValue(textB);
if (emptyA && emptyB) return 0;
if (emptyA) return 1;
if (emptyB) return -1;
var result = numeric
? parseFloat(textA) - parseFloat(textB)
: textA.localeCompare(textB, undefined, { numeric: true, sensitivity: "base" });
return dir === "desc" ? -result : result;
}
function sortTable(table, th) {
var thead = table.tHead;
var tbody = table.tBodies[0];
if (!thead || !tbody) return;
var headerRow = thead.rows[0];
var index = Array.prototype.indexOf.call(headerRow.children, th);
if (index === -1) return;
var rows = Array.prototype.slice.call(tbody.rows);
if (rows.length < 2) return;
var nextDir = th.getAttribute("data-sort-dir") === "asc" ? "desc" : "asc";
Array.prototype.forEach.call(headerRow.children, function (cell) {
cell.removeAttribute("data-sort-dir");
cell.removeAttribute("aria-sort");
});
th.setAttribute("data-sort-dir", nextDir);
th.setAttribute("aria-sort", nextDir === "asc" ? "ascending" : "descending");
var numeric = columnIsNumeric(rows, index);
rows.sort(function (a, b) {
return compareRows(a, b, index, numeric, nextDir);
});
rows.forEach(function (row) {
tbody.appendChild(row);
});
}
document.addEventListener("click", function (event) {
var th = event.target.closest("table.sortable thead th");
if (!th || th.hasAttribute("data-sort-off") || th.textContent.trim() === "") return;
sortTable(th.closest("table"), th);
});
})();
+17
View File
@@ -0,0 +1,17 @@
{%- macro connection_spec(ap) -%}
{%- if ap.iso_thread -%}
M{{ ap.iso_thread.diameter_mm }}×{{ ap.iso_thread.pitch_mm }}
{%- elif ap.unc_thread -%}
{{ ap.unc_thread.diameter_fraction }}-{{ ap.unc_thread.tpi }} {{ ap.unc_thread.series.value }}
{%- elif ap.clamp_connection -%}
{%- if ap.clamp_connection.min_mm is not none -%}
{{ ap.clamp_connection.min_mm }}–{{ ap.clamp_connection.max_mm }} mm
{%- else -%}
{{ ap.clamp_connection.min_in }}–{{ ap.clamp_connection.max_in }} in
{%- endif -%}
{%- elif ap.static_connection -%}
{{ ap.static_connection.od_mm }} mm OD
{%- else -%}
—
{%- endif -%}
{%- endmacro -%}
+3
View File
@@ -145,6 +145,9 @@
{% if su or "audit_log:read" in p %}
<a href="/admin/audit-log">{{ icon("list-checks") }} Audit Log</a>
{% endif %}
{% if su %}
<a href="/admin/danger-zone">{{ icon("alert-circle") }} Danger Zone</a>
{% endif %}
</div>
</div>
{% endif %}
+25 -9
View File
@@ -1,19 +1,30 @@
{% from "_connection_macros.html" import connection_spec %}
{%- macro mag(v) -%}{{ v | int if v == v | int else v | round(1) }}{%- endmacro -%}
{%- macro magnification(info) -%}
{%- macro magnification_range(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) }}×
{{ mag(info.magnification_min) }}-{{ mag(info.magnification_max) }}
{%- elif info and info.magnification_min is not none -%}
{{ mag(info.magnification_min) }}×
{{ mag(info.magnification_min) }}
{%- elif info and info.magnification_max is not none -%}
{{ mag(info.magnification_max) }}×
{{ mag(info.magnification_max) }}
{%- endif -%}
{%- endmacro -%}
{%- macro magnification(info) -%}
{%- set range = magnification_range(info) -%}
{%- if range -%}
{{ range }}x{% if info.objective_lens_diameter_mm is not none %}{{ mag(info.objective_lens_diameter_mm) }}{% endif %}
{%- else -%}
<span class="muted">—</span>
{%- endif -%}
{%- endmacro -%}
<div id="scopes-table-card" class="card" style="overflow-x:auto">
<table class="table">
{% include "admin/_pagination.html" %}
<table class="table sortable">
<thead>
<tr>
{% if can_write_scopes %}
<th data-sort-off><input type="checkbox" id="scopes-select-all" title="Select all"></th>
{% endif %}
<th>Scope</th>
<th>Manufacturer</th>
<th>Model</th>
@@ -21,7 +32,7 @@
<th>Magnification</th>
<th>Objective thread</th>
{% for ct in column_types %}
<th>{{ ct.name }}</th>
<th data-sort-off>{{ ct.name }}</th>
{% endfor %}
</tr>
</thead>
@@ -29,7 +40,10 @@
{% for row in rows %}
{% set scope = row.scope %}
{% set info = row.info %}
<tr>
<tr id="scope-row-{{ scope.id }}">
{% if can_write_scopes %}
<td><input type="checkbox" name="scope_ids" value="{{ scope.id }}"></td>
{% endif %}
<td>
<strong><a href="/admin/augmented-parts/{{ scope.id }}">{{ scope.description }}</a></strong>
{% if scope.sku %}<br><span class="muted">{{ scope.sku }}</span>{% endif %}
@@ -39,7 +53,9 @@
<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 %}
{% if row.objective_attachment_point %}
<code>{{ connection_spec(row.objective_attachment_point) }}</code>
{% elif info and info.objective_thread_pitch %}
{{ info.objective_thread_pitch }}
{% else %}<span class="muted">—</span>{% endif %}
</td>
@@ -61,7 +77,7 @@
</tr>
{% else %}
<tr>
<td colspan="{{ 6 + column_types | length }}" class="muted">
<td colspan="{{ 6 + column_types | length + (1 if can_write_scopes else 0) }}" class="muted">
No scopes match your search.
</td>
</tr>
@@ -1,5 +1,5 @@
<div id="attachment-point-name-list" class="card">
<table class="table">
<table class="table sortable">
<thead>
<tr>
<th>ID</th>
@@ -20,7 +20,6 @@
{% endif %}
<tr id="attachment-point-{{ ap.id }}">
<td>{{ ap.id }}</td>
<td>{{ ap.augmented_part.description | truncate(40) }}</td>
<td>{{ ap.attachment_point_name.name }}</td>
<td>{{ conn_type }}</td>
<td><code>{{ conn_label }}</code></td>
@@ -1,9 +1,8 @@
<div id="attachment-point-list" class="card" style="overflow-x:auto">
<table class="table">
<table class="table sortable">
<thead>
<tr>
<th>ID</th>
<th>Augmented Part</th>
<th>Point Name</th>
<th>Connection Type</th>
<th>Connection</th>
@@ -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>
@@ -45,6 +45,16 @@
</select>
</label>
<div id="ap-{{ augmented_part_id }}-fit-other-points"></div>
<label>Fit type
<select name="label">
<option value="exact" selected>Exact</option>
<option value="possible">Possible</option>
<option value="none">None</option>
</select>
</label>
<label>Note (optional)
<input type="text" name="note" placeholder="Add a note…">
</label>
<button type="submit" class="btn btn-primary">Add fit</button>
</form>
{% endif %}
@@ -0,0 +1,22 @@
<div id="ap-{{ augmented_part_id }}-notes" class="ap-inline-section">
<h3>Notes</h3>
<div class="fit-notes">
{% for note in notes %}
<div class="fit-note">
<span class="fit-note-meta">{{ note.username }} · {{ note.created_at.strftime("%Y-%m-%d %H:%M") }}</span>
<p class="fit-note-body">{{ note.body }}</p>
</div>
{% else %}
<p class="muted" style="font-size:0.85rem;margin:0 0 0.5rem">No notes recorded yet.</p>
{% endfor %}
{% if can_write_notes %}
<form hx-post="/admin/augmented-parts/{{ augmented_part_id }}/notes"
hx-target="#ap-{{ augmented_part_id }}-notes" 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>
@@ -10,6 +10,8 @@
hx-target="#augmented-parts-table-card"
hx-trigger="input delay:300ms, change"
hx-swap="outerHTML">
<input type="hidden" name="sort" value="{{ sort }}">
<input type="hidden" name="dir" value="{{ dir }}">
<input type="text" name="description" placeholder="Filter by description…" value="{{ filter_description or '' }}">
<select name="colour_id">
<option value="">All colours</option>
@@ -1,16 +1,17 @@
{% from "admin/_sort_macros.html" import sort_th %}
<div id="augmented-parts-table-card" class="card">
{% include "admin/_pagination.html" %}
<table class="table">
<thead>
<tr>
<th>ID</th>
<th>SKU</th>
<th>Description</th>
<th>Colour</th>
<th>Manufacturer</th>
<th>Model</th>
<th>Type</th>
<th>Sub Type</th>
{{ sort_th("id", "ID", sort_headers) }}
{{ sort_th("sku", "SKU", sort_headers) }}
{{ sort_th("description", "Description", sort_headers) }}
{{ sort_th("colour", "Colour", sort_headers) }}
{{ sort_th("manufacturer", "Manufacturer", sort_headers) }}
{{ sort_th("scope_model", "Model", sort_headers) }}
{{ sort_th("part_type", "Type", sort_headers) }}
{{ sort_th("sub_part_type", "Sub Type", sort_headers) }}
<th></th>
</tr>
</thead>
@@ -1,5 +1,5 @@
<div id="clamp-connection-list" class="card">
<table class="table">
<table class="table sortable">
<thead>
<tr>
<th>ID</th>
@@ -0,0 +1,14 @@
{% if total %}
<p style="color:var(--color-danger,#c00)">
Purge complete: {{ total }} row(s) deleted.
</p>
<ul class="muted">
{% for label, n in counts.items() %}
{% if n %}
<li>{{ label }}: {{ n }}</li>
{% endif %}
{% endfor %}
</ul>
{% else %}
<p class="muted">Nothing to purge — the database was already clean.</p>
{% endif %}
@@ -1,6 +1,13 @@
{% if points %}
<select name="other_attachment_point_id" required>
<option value="">— select —</option>
{% for p in points %}
<option value="{{ p.id }}">{{ p.attachment_point_name.name }}</option>
{% endfor %}
</select>
{% else %}
<input type="hidden" name="other_attachment_point_id" value="">
<p class="muted" style="font-size:0.85rem;margin:0">
This part has no attachment points yet — one matching yours will be created automatically.
</p>
{% endif %}
@@ -11,6 +11,8 @@
hx-target="#imported-parts-table-card"
hx-trigger="input delay:300ms, change"
hx-swap="outerHTML">
<input type="hidden" name="sort" value="{{ sort }}">
<input type="hidden" name="dir" value="{{ dir }}">
<input type="text" name="description" placeholder="Filter by description…" value="{{ filter_description or '' }}">
</form>
</div>
@@ -1,16 +1,17 @@
{% from "admin/_sort_macros.html" import sort_th %}
<div id="imported-parts-table-card" class="card" style="overflow-x:auto">
{% include "admin/_pagination.html" %}
<table class="table">
<thead>
<tr>
<th>InventoryID</th>
<th>Description</th>
<th>TBX Colour</th>
<th>Max Thread Depth (mm)</th>
<th>Max Thread Depth (in)</th>
<th>OD (mm)</th>
<th>OD (in)</th>
<th>Thread Attribute</th>
{{ sort_th("InventoryID", "InventoryID", sort_headers) }}
{{ sort_th("Description", "Description", sort_headers) }}
{{ sort_th("TBXColour", "TBX Colour", sort_headers) }}
{{ sort_th("ProductMaxThreadDepthmm", "Max Thread Depth (mm)", sort_headers) }}
{{ sort_th("ProductMaxThreadDepthin", "Max Thread Depth (in)", sort_headers) }}
{{ sort_th("ProductODmm", "OD (mm)", sort_headers) }}
{{ sort_th("ProductODin", "OD (in)", sort_headers) }}
{{ sort_th("TBXThreadAttribute", "Thread Attribute", sort_headers) }}
</tr>
</thead>
<tbody>
+7
View File
@@ -0,0 +1,7 @@
<div id="{{ container_id }}"
hx-get="{{ poll_url }}"
hx-trigger="load delay:1s"
hx-swap="outerHTML"
class="{{ extra_class or '' }}">
<p class="muted">⏳ {{ job.message }}</p>
</div>
@@ -0,0 +1,5 @@
<span id="{{ container_id }}"
hx-get="{{ poll_url }}"
hx-trigger="load delay:1s"
hx-swap="outerHTML"
class="muted">⏳ {{ job.message }}</span>
@@ -1,8 +1,8 @@
<div id="manufacturer-list" class="card">
<table class="table">
<table class="table sortable">
<thead>
<tr>
<th>ID</th><th>Name</th><th>Homepage</th><th>Logo</th><th></th>
<th>ID</th><th>Name</th><th>Homepage</th><th data-sort-off>Logo</th><th></th>
</tr>
</thead>
<tbody>
+1 -1
View File
@@ -1,5 +1,5 @@
<div id="part-colour-list" class="card">
<table class="table">
<table class="table sortable">
<thead>
<tr>
<th>ID</th>
+1 -1
View File
@@ -1,5 +1,5 @@
<div id="part-type-list" class="card">
<table class="table">
<table class="table sortable">
<thead>
<tr>
<th>ID</th><th>Name</th><th></th>
+2 -2
View File
@@ -1,7 +1,7 @@
<div id="role-list" class="card">
<table class="table">
<table class="table sortable">
<thead>
<tr><th>ID</th><th>Name</th><th>Description</th><th>Permissions</th><th>Default</th><th></th></tr>
<tr><th>ID</th><th>Name</th><th>Description</th><th data-sort-off>Permissions</th><th data-sort-off>Default</th><th></th></tr>
</thead>
<tbody>
{% for r in roles %}
@@ -1,3 +1,4 @@
<div id="{{ container_id or 'scope-import-result-body' }}">
{% if error %}
<div class="card">
<p style="color:var(--color-danger,#c00)">{{ error }}</p>
@@ -39,3 +40,4 @@
{% endif %}
</div>
{% endif %}
</div>
+1 -1
View File
@@ -1,5 +1,5 @@
<div id="scope-model-list" class="card">
<table class="table">
<table class="table sortable">
<thead>
<tr>
<th>ID</th><th>Name</th><th>Manufacturer</th><th>Product Page</th><th></th>
+13
View File
@@ -0,0 +1,13 @@
{% macro sort_th(key, label, sort_headers) %}
{% if sort_headers and key in sort_headers %}
{% set h = sort_headers[key] %}
<th>
<a href="#" class="sort-link{{ ' active' if h.active else '' }}"
hx-get="{{ h.url }}"
hx-target="{{ h.target_id }}"
hx-swap="outerHTML">{{ label }}{% if h.active %}<span aria-hidden="true">{{ ' ▲' if h.dir == 'asc' else ' ▼' }}</span>{% endif %}</a>
</th>
{% else %}
<th>{{ label }}</th>
{% endif %}
{% endmacro %}
@@ -1,5 +1,5 @@
<div id="static-connection-list" class="card">
<table class="table">
<table class="table sortable">
<thead>
<tr>
<th>ID</th>
@@ -1,5 +1,5 @@
<div id="sub-part-type-list" class="card">
<table class="table">
<table class="table sortable">
<thead>
<tr>
<th>ID</th><th>Name</th><th>Parent Type</th><th></th>
+2 -2
View File
@@ -1,8 +1,8 @@
<div id="user-list" class="card">
<table class="table">
<table class="table sortable">
<thead>
<tr>
<th>ID</th><th>Email</th><th>Name</th><th>Active</th><th>Superuser</th><th>Roles</th><th></th>
<th>ID</th><th>Email</th><th>Name</th><th>Active</th><th>Superuser</th><th data-sort-off>Roles</th><th></th>
</tr>
</thead>
<tbody>
+1 -1
View File
@@ -7,7 +7,7 @@
<div class="card">
{% if entries %}
<table class="table">
<table class="table sortable">
<thead>
<tr>
<th>Time</th>
+16 -26
View File
@@ -3,20 +3,6 @@
{% block body %}
{% include "_nav.html" %}
{%- macro conn_desc(point) -%}
{%- if point.iso_thread -%}
ISO M{{ point.iso_thread.diameter_mm }}×{{ point.iso_thread.pitch_mm }}
{%- elif point.unc_thread -%}
UNC {{ point.unc_thread.diameter_fraction }}-{{ point.unc_thread.tpi }} {{ point.unc_thread.series.value }}
{%- elif point.clamp_connection -%}
Clamp {{ point.clamp_connection.min_mm if point.clamp_connection.min_mm is not none else point.clamp_connection.min_in }}–{{ point.clamp_connection.max_mm if point.clamp_connection.max_mm is not none else point.clamp_connection.max_in }}
{%- elif point.static_connection -%}
Static {{ point.static_connection.od_mm }} mm OD
{%- else -%}
—
{%- endif -%}
{%- endmacro -%}
<main class="container">
<p><a href="/admin/augmented-parts" class="btn btn-ghost">← Augmented Parts</a></p>
@@ -58,28 +44,32 @@
{% else %}—{% endif %}
</dd>
<dt>Model suffix</dt><dd>{{ scope_info.model_suffix or "—" }}</dd>
<dt>Objective lens diameter</dt>
<dd>
{% if scope_info.objective_lens_diameter_mm is not none %}
{{ scope_info.objective_lens_diameter_mm }} mm
{% else %}—{% endif %}
</dd>
<dt>Objective thread</dt><dd>{{ scope_info.objective_thread_pitch or "—" }}</dd>
</dl>
</div>
{% endif %}
<div class="card">
<h2>Attachment points</h2>
{% for point in attachment_points %}
<div class="ap-item">
<span class="ap-item-name">{{ point.attachment_point_name.name }}</span>
<code class="ap-item-conn">{{ conn_desc(point) }}</code>
</div>
{% else %}
<p class="muted">No attachment points recorded.</p>
{% endfor %}
</div>
</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 %}
<div class="card">
{% include "admin/_augmented_part_notes.html" %}
</div>
</main>
{% endblock %}
+32
View File
@@ -0,0 +1,32 @@
{% extends "base.html" %}
{% block title %}Danger Zone · {{ app_name }}{% endblock %}
{% block body %}
{% include "_nav.html" %}
<main class="container">
<h1>Danger Zone</h1>
<div class="card" style="border: 1px solid var(--color-danger, #c00)">
<h2>Purge database</h2>
<p>
Permanently deletes every part, attachment point, fit, and imported
part in the database — manufacturers, scope models, ISO/UNC threads,
clamp/static connections, attachment point names, augmented parts,
scope info, attachment points, fits, fit notes, and imported parts.
</p>
<p>
<strong>Not</strong> deleted: users, roles &amp; permissions, refresh
tokens, the audit log, the OData connection configuration, and part
taxonomy (part colours, part types, sub part types).
</p>
<p class="muted">This cannot be undone. Take a database backup first if you might need this data again.</p>
<button class="btn btn-danger"
hx-post="/admin/danger-zone/purge"
hx-target="#purge-result"
hx-swap="innerHTML"
hx-confirm="This will PERMANENTLY delete all parts, taxonomy, attachment points, fits, and imported parts. Users and the OData configuration are kept. This cannot be undone. Continue?">
Purge all part/fit data
</button>
<div id="purge-result" style="margin-top:1rem"></div>
</div>
</main>
{% endblock %}
+1 -3
View File
@@ -29,11 +29,9 @@
<button class="btn btn-primary"
hx-post="/admin/imported-parts/refresh"
hx-target="#imported-parts-section"
hx-swap="outerHTML"
hx-indicator="#refresh-indicator">
hx-swap="outerHTML">
Refresh from OData
</button>
<span id="refresh-indicator" class="htmx-indicator muted"> Loading…</span>
</div>
{% include "admin/_imported_parts_table.html" %}
+2
View File
@@ -4,10 +4,12 @@
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{% block title %}{{ app_name }}{% endblock %}</title>
<link rel="icon" href="/static/favicon.ico" sizes="any">
<link rel="stylesheet" href="/static/css/style.css">
<script src="https://unpkg.com/htmx.org@1.9.12"
integrity="sha384-ujb1lZYygJmzgSwoxRggbCHcjc0rB2XoQrxeTUQyRjrOnlCoYta87iKBWq3EsdM2"
crossorigin="anonymous"></script>
<script src="/static/js/sortable-tables.js" defer></script>
</head>
<body>
{% block body %}{% 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>
+45 -1
View File
@@ -15,7 +15,7 @@
hx-target="#scopes-table-card"
hx-trigger="input delay:300ms, change"
hx-swap="outerHTML">
<input type="text" name="description" placeholder="Search scopes…"
<input type="text" name="description" placeholder="Search by description or product ID…"
value="{{ filters.description or '' }}">
<select name="manufacturer_id">
<option value="">All manufacturers</option>
@@ -44,6 +44,50 @@
</form>
</div>
{% if can_write_scopes %}
<div class="bulk-actions">
<button id="delete-selected-scopes-btn" type="button" class="btn btn-danger" disabled
hx-post="/scopes/bulk-delete"
hx-include="[name='scope_ids']:checked, [name='description'], [name='manufacturer_id'], [name='scope_model_id'], [name='sub_part_type_id'], [name='fits_part_type_id']"
hx-target="#scopes-table-card"
hx-swap="outerHTML"
hx-confirm="Delete selected scopes?">Delete selected</button>
</div>
{% endif %}
{% include "_scopes_table_card.html" %}
</main>
{% if can_write_scopes %}
<script>
(function () {
function updateDeleteButton() {
var count = document.querySelectorAll('input[name="scope_ids"]:checked').length;
var btn = document.getElementById('delete-selected-scopes-btn');
if (!btn) return;
btn.textContent = count > 0 ? 'Delete selected (' + count + ')' : 'Delete selected';
btn.disabled = count === 0;
btn.setAttribute('hx-confirm', 'Delete ' + count + ' selected scope' + (count === 1 ? '' : 's') + '?');
}
document.addEventListener('change', function (e) {
if (!e.target) return;
if (e.target.id === 'scopes-select-all') {
document.querySelectorAll('input[name="scope_ids"]').forEach(function (cb) {
cb.checked = e.target.checked;
});
}
if (e.target.id === 'scopes-select-all' || e.target.name === 'scope_ids') {
updateDeleteButton();
}
});
// Re-run after every HTMX swap (filter results, post-delete) so the button
// state reflects the freshly rendered checkboxes, which start unchecked.
document.addEventListener('htmx:afterSwap', updateDeleteButton);
}());
</script>
{% endif %}
{% endblock %}
+773 -215
View File
File diff suppressed because it is too large. Load diff
+39 -8
View File
@@ -1,4 +1,6 @@
from fastapi import APIRouter, Depends, Form, HTTPException, Request, status
from datetime import timedelta
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 +16,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,
@@ -21,7 +38,7 @@ def _set_auth_cookie(response: Response, access_token: str) -> None:
httponly=True,
secure=settings.cookie_secure,
samesite="lax",
max_age=settings.access_token_expire_minutes * 60,
max_age=settings.session_timeout_hours * 3600,
path="/",
)
@@ -35,9 +52,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 +65,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 +77,16 @@ 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)
access, _ = await issue_token_pair(
db, user, timedelta(hours=settings.session_timeout_hours)
)
response = RedirectResponse(
url=_safe_next(next_url), status_code=status.HTTP_303_SEE_OTHER
)
_set_auth_cookie(response, access)
return response
@@ -71,6 +96,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 +110,16 @@ 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)
access, _ = await issue_token_pair(
db, user, timedelta(hours=settings.session_timeout_hours)
)
response = RedirectResponse(
url=_safe_next(next_url), status_code=status.HTTP_303_SEE_OTHER
)
_set_auth_cookie(response, access)
return response
+40
View File
@@ -0,0 +1,40 @@
"""Shared pagination helpers for HTMX table cards (admin screens, scopes page)."""
import math
import urllib.parse
def query_url(base_path: str, params: dict) -> str:
clean = {k: v for k, v in params.items() if v not in (None, "", 0)}
return f"{base_path}?{urllib.parse.urlencode(clean)}"
def paginate(
base_path: str,
filter_params: dict,
page: int,
page_size: int,
total: int,
target_id: str,
) -> dict:
total_pages = max(1, math.ceil(total / page_size))
page = max(1, min(page, total_pages))
skip = (page - 1) * page_size
def _url(p: int) -> str:
return query_url(
base_path, {**filter_params, "page": p, "page_size": page_size}
)
return {
"page": page,
"page_size": page_size,
"total": total,
"total_pages": total_pages,
"skip": skip,
"showing_from": skip + 1 if total > 0 else 0,
"showing_to": min(skip + page_size, total),
"prev_url": _url(page - 1) if page > 1 else None,
"next_url": _url(page + 1) if page < total_pages else None,
"target_id": target_id,
}
+102 -7
View File
@@ -5,11 +5,13 @@ 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 import APIRouter, Depends, Form, Request
from fastapi.responses import HTMLResponse
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.crud import audit as audit_crud
from app.crud import augmented_part as augmented_part_crud
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
@@ -17,17 +19,30 @@ 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.pagination import paginate
from app.web.templates import templates
router = APIRouter(include_in_schema=False)
_PAGE_SIZE = 50
_TABLE_TARGET = "#scopes-table-card"
_FILTER_PATH = "/scopes/filter"
def _to_int(value: str) -> int | None:
return int(value) if value.strip() else None
def _can_write_scopes(user: User) -> bool:
return user.is_superuser or "augmented_parts:write" in user.permissions
async def _build(db: AsyncSession, filters: dict) -> dict:
"""Run the guide query for the given raw (string) filter values."""
"""Run the guide query for the given raw (string) filter values.
Always returns the full matching row set — pagination is applied by the
caller by slicing ``rows`` (see ``_paginated_guide``).
"""
return await scope_guide_service.build(
db,
description=filters["description"],
@@ -38,6 +53,22 @@ async def _build(db: AsyncSession, filters: dict) -> dict:
)
async def _paginated_guide(
db: AsyncSession, filters: dict, page: int, page_size: int
) -> tuple[dict, dict]:
"""Run the guide query, then slice its rows to the requested page.
Returns ``(guide, pagination)`` where ``guide["rows"]`` is already sliced.
"""
guide = await _build(db, filters)
all_rows = guide["rows"]
pagination = paginate(
_FILTER_PATH, filters, page, page_size, len(all_rows), _TABLE_TARGET
)
guide["rows"] = all_rows[pagination["skip"] : pagination["skip"] + page_size]
return guide, pagination
async def _filter_options(db: AsyncSession) -> dict:
"""Dropdown option lists for the filter bar."""
scope_type = await part_type_crud.get_by_name(
@@ -73,11 +104,18 @@ async def scopes_page(
"sub_part_type_id": sub_part_type_id,
"fits_part_type_id": fits_part_type_id,
}
guide = await _build(db, filters)
guide, pagination = await _paginated_guide(db, filters, 1, _PAGE_SIZE)
return templates.TemplateResponse(
request,
"scopes.html",
{"user": user, "filters": filters, **guide, **(await _filter_options(db))},
{
"user": user,
"filters": filters,
"can_write_scopes": _can_write_scopes(user),
"pagination": pagination,
**guide,
**(await _filter_options(db)),
},
)
@@ -89,8 +127,10 @@ async def scopes_filter(
scope_model_id: str = "",
sub_part_type_id: str = "",
fits_part_type_id: str = "",
page: int = 1,
page_size: int = _PAGE_SIZE,
db: AsyncSession = Depends(get_db),
_: User = Depends(web_require("augmented_parts:read", "fits:read")),
user: User = Depends(web_require("augmented_parts:read", "fits:read")),
):
filters = {
"description": description,
@@ -99,5 +139,60 @@ async def scopes_filter(
"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)
guide, pagination = await _paginated_guide(db, filters, page, page_size)
return templates.TemplateResponse(
request,
"_scopes_table_card.html",
{
**guide,
"pagination": pagination,
"can_write_scopes": _can_write_scopes(user),
},
)
@router.post("/scopes/bulk-delete", response_class=HTMLResponse)
async def bulk_delete_scopes(
request: Request,
scope_ids: list[int] = Form([]),
description: str = Form(""),
manufacturer_id: str = Form(""),
scope_model_id: str = Form(""),
sub_part_type_id: str = Form(""),
fits_part_type_id: str = Form(""),
db: AsyncSession = Depends(get_db),
actor: User = Depends(web_require("augmented_parts:write")),
):
for augmented_part_id in scope_ids:
ap = await augmented_part_crud.get(db, augmented_part_id)
if ap is None:
continue
label = ap.sku or ap.description
await augmented_part_crud.delete(db, ap)
await audit_crud.log(
db,
actor=actor,
action="delete",
resource_type="AugmentedPart",
resource_id=augmented_part_id,
description=audit_crud.describe_delete(
"AugmentedPart", augmented_part_id, label
),
)
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, pagination = await _paginated_guide(db, filters, 1, _PAGE_SIZE)
return templates.TemplateResponse(
request,
"_scopes_table_card.html",
{
**guide,
"pagination": pagination,
"can_write_scopes": _can_write_scopes(actor),
},
)
+244
View File
@@ -0,0 +1,244 @@
import asyncio
import re
import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from app.core.database import Base
from app.crud import imported_part_config as imported_part_config_crud
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.part_type import PartType
from tests.test_scope_import import _scope_row, _workbook
_JOB_URL_RE = re.compile(r'hx-get="([^"]+/jobs/[^"]+)"')
@pytest_asyncio.fixture
async def db_sessionmaker(tmp_path):
"""Override conftest's in-memory+StaticPool fixture for this module.
These tests run a real background asyncio task (the job) concurrently
with polling requests, both hitting the DB at once. StaticPool pins every
session to one shared raw connection, which can't safely serve two
callers at the same time; a real file-backed DB with its own connection
pool (how SQLite/Postgres actually behave outside tests) can.
"""
db_path = tmp_path / "jobs_test.db"
engine = create_async_engine(f"sqlite+aiosqlite:///{db_path}")
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
maker = async_sessionmaker(engine, expire_on_commit=False)
yield maker
await engine.dispose()
async def _web_login(client, email, password="secret123"):
resp = await client.post("/login", data={"email": email, "password": password})
assert resp.status_code == 303
async def _poll_until_done(client, url, max_tries=50):
"""Follow the same htmx polling loop a browser would, until no further
``hx-get`` job-status link is present in the response."""
body = ""
for _ in range(max_tries):
resp = await client.get(url)
assert resp.status_code == 200
body = resp.text
match = _JOB_URL_RE.search(body)
if not match:
return body
url = match.group(1)
await asyncio.sleep(0.02)
raise AssertionError(f"job did not finish after {max_tries} polls: {body!r}")
@pytest.mark.asyncio
async def test_scope_import_runs_in_background(
client, seeded, make_user, db_sessionmaker, monkeypatch
):
monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker)
async with db_sessionmaker() as db:
perms = seeded["perms"]
editor = await role_crud.create(
db,
name="scope-import-editor",
permission_ids=[perms["augmented_parts:write"].id],
)
editor_id = editor.id
await make_user("importer@example.com", role_ids=[editor_id])
await _web_login(client, "importer@example.com")
data = _workbook([_scope_row(product_number="BG-SCOPE-1")])
resp = await client.post(
"/admin/scope-import",
files={"file": ("scopes.xlsx", data, "application/octet-stream")},
)
assert resp.status_code == 200
assert "scope-import-status" in resp.text
match = _JOB_URL_RE.search(resp.text)
assert match, resp.text
final_body = await _poll_until_done(client, match.group(1))
assert "1</strong> created" in final_body
@pytest.mark.asyncio
async def test_augmented_parts_sync_runs_in_background(
client, seeded, make_user, db_sessionmaker, monkeypatch
):
monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker)
async with db_sessionmaker() as db:
perms = seeded["perms"]
editor = await role_crud.create(
db, name="sync-editor", permission_ids=[perms["augmented_parts:write"].id]
)
editor_id = editor.id
await make_user("syncer@example.com", role_ids=[editor_id])
await _web_login(client, "syncer@example.com")
resp = await client.post("/admin/augmented-parts/sync")
assert resp.status_code == 200
assert 'id="augmented-part-list"' in resp.text
match = _JOB_URL_RE.search(resp.text)
assert match, resp.text
final_body = await _poll_until_done(client, match.group(1))
assert "Sync complete" in final_body
@pytest.mark.asyncio
async def test_fit_recompute_for_part_runs_in_background(
client, seeded, make_user, db_sessionmaker, monkeypatch
):
monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker)
async with db_sessionmaker() as db:
apn = AttachmentPointName(name="Objective")
part_t = PartType(name="Scope")
db.add_all([apn, part_t])
await db.flush()
part = AugmentedPart(
description="Test scope",
colour_id=1,
manufacturer_id=1,
part_type_id=part_t.id,
)
db.add(part)
await db.flush()
db.add(
AttachmentPoint(augmented_part_id=part.id, attachment_point_name_id=apn.id)
)
await db.commit()
part_id = part.id
perms = seeded["perms"]
editor = await role_crud.create(
db,
name="fit-recompute-editor",
permission_ids=[perms["fits:write"].id, perms["augmented_parts:read"].id],
)
editor_id = editor.id
await make_user("recomputer@example.com", role_ids=[editor_id])
await _web_login(client, "recomputer@example.com")
resp = await client.post(f"/admin/augmented-parts/{part_id}/fits/recompute")
assert resp.status_code == 200
assert f'id="ap-{part_id}-fits"' in resp.text
match = _JOB_URL_RE.search(resp.text)
assert match, resp.text
final_body = await _poll_until_done(client, match.group(1))
assert "created" in final_body and "removed" in final_body
@pytest.mark.asyncio
async def test_fit_recompute_all_runs_in_background(
client, seeded, make_user, db_sessionmaker, monkeypatch
):
monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker)
async with db_sessionmaker() as db:
perms = seeded["perms"]
editor = await role_crud.create(
db, name="fit-recompute-all-editor", permission_ids=[perms["fits:write"].id]
)
editor_id = editor.id
await make_user("recomputerall@example.com", role_ids=[editor_id])
await _web_login(client, "recomputerall@example.com")
resp = await client.post("/admin/augmented-parts/fits/recompute")
assert resp.status_code == 200
assert "fits-recompute-status" in resp.text
match = _JOB_URL_RE.search(resp.text)
assert match, resp.text
final_body = await _poll_until_done(client, match.group(1))
assert "Recompute complete" in final_body
@pytest.mark.asyncio
async def test_imported_parts_refresh_runs_in_background(
client, seeded, make_user, db_sessionmaker, monkeypatch
):
monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker)
async def _fake_fetch(url, username, password):
return [{"InventoryID": "INV-1", "Description": "Widget"}]
monkeypatch.setattr("app.web.admin.odata_service.fetch", _fake_fetch)
async with db_sessionmaker() as db:
await imported_part_config_crud.upsert(
db, odata_url="https://example.test/odata", username=None, password=None
)
perms = seeded["perms"]
editor = await role_crud.create(
db,
name="imported-parts-editor",
permission_ids=[perms["imported_parts:write"].id],
)
editor_id = editor.id
await make_user("refresher@example.com", role_ids=[editor_id])
await _web_login(client, "refresher@example.com")
resp = await client.post("/admin/imported-parts/refresh")
assert resp.status_code == 200
assert 'id="imported-parts-section"' in resp.text
match = _JOB_URL_RE.search(resp.text)
assert match, resp.text
final_body = await _poll_until_done(client, match.group(1))
assert "Sync complete: 1 created, 0 updated." in final_body
assert "INV-1" in final_body
@pytest.mark.asyncio
async def test_imported_parts_refresh_without_config_skips_job(
client, seeded, make_user, db_sessionmaker, monkeypatch
):
monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker)
async with db_sessionmaker() as db:
perms = seeded["perms"]
editor = await role_crud.create(
db,
name="imported-parts-editor-noconfig",
permission_ids=[perms["imported_parts:write"].id],
)
editor_id = editor.id
await make_user("refresher2@example.com", role_ids=[editor_id])
await _web_login(client, "refresher2@example.com")
resp = await client.post("/admin/imported-parts/refresh")
assert resp.status_code == 200
assert "OData URL is not configured." in resp.text
# No background job started for this synchronous validation failure.
assert _JOB_URL_RE.search(resp.text) is None
+148
View File
@@ -0,0 +1,148 @@
import pytest
from app.crud import augmented_part as augmented_part_crud
from app.models.augmented_part import AugmentedPart
from app.models.imported_part import ImportedPart
from app.models.manufacturer import Manufacturer
from app.models.part_colour import PartColour
from app.models.part_type import PartType
async def _web_login(client, email, password="secret123"):
resp = await client.post("/login", data={"email": email, "password": password})
assert resp.status_code == 303
async def _seed_augmented_parts(db):
part_t = PartType(name="Scope")
db.add(part_t)
await db.flush()
parts = [
AugmentedPart(
description=desc,
sku=sku,
colour_id=1,
manufacturer_id=1,
part_type_id=part_t.id,
)
for desc, sku in [
("Alpha scope", "SKU-C"),
("Bravo scope", "SKU-A"),
("Charlie scope", "SKU-B"),
]
]
db.add_all(parts)
await db.commit()
async def _seed_imported_parts(db):
db.add_all(
[
ImportedPart(InventoryID="INV-2", Description="Charlie widget"),
ImportedPart(InventoryID="INV-1", Description="Alpha widget"),
ImportedPart(InventoryID="INV-3", Description="Bravo widget"),
]
)
await db.commit()
def _order_of(body: str, needles: list[str]) -> list[str]:
positions = {n: body.find(n) for n in needles}
return sorted(positions, key=lambda n: positions[n])
@pytest.mark.asyncio
async def test_augmented_parts_page_sorts_by_description_desc(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
await _seed_augmented_parts(db)
await make_user("reader@example.com", role_ids=[seeded["roles"]["user"].id])
await _web_login(client, "reader@example.com")
resp = await client.get("/admin/augmented-parts?sort=description&dir=desc")
assert resp.status_code == 200
body = resp.text
assert _order_of(body, ["Charlie scope", "Bravo scope", "Alpha scope"]) == [
"Charlie scope",
"Bravo scope",
"Alpha scope",
]
@pytest.mark.asyncio
async def test_augmented_parts_filter_sorts_by_sku_asc(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
await _seed_augmented_parts(db)
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/filter?sort=sku&dir=asc")
assert resp.status_code == 200
body = resp.text
# SKU-A (Bravo), SKU-B (Charlie), SKU-C (Alpha)
assert _order_of(body, ["Bravo scope", "Charlie scope", "Alpha scope"]) == [
"Bravo scope",
"Charlie scope",
"Alpha scope",
]
@pytest.mark.asyncio
async def test_imported_parts_page_sorts_by_description_desc(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
await _seed_imported_parts(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/imported-parts?sort=Description&dir=desc")
assert resp.status_code == 200
body = resp.text
assert _order_of(body, ["Charlie widget", "Bravo widget", "Alpha widget"]) == [
"Charlie widget",
"Bravo widget",
"Alpha widget",
]
@pytest.mark.asyncio
async def test_list_filtered_sorts_by_joined_colour_name(db_sessionmaker):
async with db_sessionmaker() as db:
red = PartColour(name="Red")
blue = PartColour(name="Blue")
manufacturer = Manufacturer(name="Acme", url="https://acme.example")
part_t = PartType(name="Scope")
db.add_all([red, blue, manufacturer, part_t])
await db.flush()
db.add_all(
[
AugmentedPart(
description="Red one",
colour_id=red.id,
manufacturer_id=manufacturer.id,
part_type_id=part_t.id,
),
AugmentedPart(
description="Blue one",
colour_id=blue.id,
manufacturer_id=manufacturer.id,
part_type_id=part_t.id,
),
]
)
await db.commit()
parts = await augmented_part_crud.list_filtered(
db, sort="colour", direction="asc"
)
assert [p.description for p in parts] == ["Blue one", "Red one"]
parts_desc = await augmented_part_crud.list_filtered(
db, sort="colour", direction="desc"
)
assert [p.description for p in parts_desc] == ["Red one", "Blue one"]
+232 -9
View File
@@ -3,8 +3,9 @@ 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.augmented_part import AugmentedPart, AugmentedPartNote
from app.models.fit import Fit, FitLabel, FitNote
from app.models.iso_thread import IsoThread
from app.models.part_type import PartType
from app.models.scope_info import ScopeInfo
@@ -40,12 +41,8 @@ async def _seed_part_with_fit(db):
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
)
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()
@@ -91,6 +88,37 @@ async def test_detail_shows_attributes_and_fits_readonly(
assert "Add fit" not in body
assert "Mark Exact" not in body
assert "Recompute possible fits" not in body
assert "Add note" 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
@@ -121,6 +149,199 @@ async def test_detail_write_controls_for_fit_editor(
assert "Recompute possible fits" in body
@pytest.mark.asyncio
async def test_create_fit_auto_creates_matching_attachment_point(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as 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()
thread = IsoThread(diameter_mm=42.0, pitch_mm=0.75)
db.add(thread)
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()
own_ap = AttachmentPoint(
augmented_part_id=scope.id,
attachment_point_name_id=apn.id,
iso_thread_id=thread.id,
)
db.add(own_ap)
await db.commit()
scope_id, cover_id, own_ap_id = scope.id, cover.id, own_ap.id
perms = seeded["perms"]
editor = await role_crud.create(
db,
name="fit-editor2",
permission_ids=[
perms["augmented_parts:read"].id,
perms["fits:read"].id,
perms["fits:write"].id,
],
)
editor_id = editor.id
await make_user("editor2@example.com", role_ids=[editor_id])
await _web_login(client, "editor2@example.com")
resp = await client.post(
f"/admin/augmented-parts/{scope_id}/fits",
data={
"own_attachment_point_id": own_ap_id,
"other_augmented_part_id": cover_id,
"other_attachment_point_id": "",
"label": "possible",
},
)
assert resp.status_code == 200
body = resp.text
assert "already exists" not in body
assert "not found" not in body
async with db_sessionmaker() as db:
cover_points = [
ap
for ap in (await db.execute(AttachmentPoint.__table__.select())).all()
if ap.augmented_part_id == cover_id
]
assert len(cover_points) == 1
assert cover_points[0].iso_thread_id == thread.id
assert cover_points[0].attachment_point_name_id == apn.id
fits = (await db.execute(Fit.__table__.select())).all()
assert len(fits) == 1
assert fits[0].label == FitLabel.POSSIBLE
@pytest.mark.asyncio
async def test_create_fit_with_note_creates_fit_note(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as 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()
own_ap = AttachmentPoint(
augmented_part_id=scope.id, attachment_point_name_id=apn.id
)
other_ap = AttachmentPoint(
augmented_part_id=cover.id, attachment_point_name_id=apn.id
)
db.add_all([own_ap, other_ap])
await db.commit()
scope_id, own_ap_id, other_ap_id = scope.id, own_ap.id, other_ap.id
perms = seeded["perms"]
editor = await role_crud.create(
db,
name="fit-editor3",
permission_ids=[
perms["augmented_parts:read"].id,
perms["fits:read"].id,
perms["fits:write"].id,
],
)
editor_id = editor.id
await make_user("editor3@example.com", role_ids=[editor_id])
await _web_login(client, "editor3@example.com")
resp = await client.post(
f"/admin/augmented-parts/{scope_id}/fits",
data={
"own_attachment_point_id": own_ap_id,
"other_attachment_point_id": other_ap_id,
"label": "exact",
"note": "Confirmed fit on the bench.",
},
)
assert resp.status_code == 200
body = resp.text
assert "Confirmed fit on the bench." in body
async with db_sessionmaker() as db:
fits = (await db.execute(Fit.__table__.select())).all()
assert len(fits) == 1
notes = (await db.execute(FitNote.__table__.select())).all()
assert len(notes) == 1
assert notes[0].fit_id == fits[0].id
assert notes[0].body == "Confirmed fit on the bench."
@pytest.mark.asyncio
async def test_create_augmented_part_note(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-note-editor",
permission_ids=[
perms["augmented_parts:read"].id,
perms["augmented_parts:write"].id,
],
)
editor_id = editor.id
await make_user("noteeditor@example.com", role_ids=[editor_id])
await _web_login(client, "noteeditor@example.com")
resp = await client.get(f"/admin/augmented-parts/{ids['scope_id']}")
assert resp.status_code == 200
assert "Add note" in resp.text
resp = await client.post(
f"/admin/augmented-parts/{ids['scope_id']}/notes",
data={"body": "Verified thread depth on the bench."},
)
assert resp.status_code == 200
assert "Verified thread depth on the bench." in resp.text
async with db_sessionmaker() as db:
notes = (await db.execute(AugmentedPartNote.__table__.select())).all()
assert len(notes) == 1
assert notes[0].augmented_part_id == ids["scope_id"]
assert notes[0].body == "Verified thread depth on the bench."
@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])
@@ -130,7 +351,9 @@ async def test_detail_404_for_missing_part(client, seeded, make_user):
@pytest.mark.asyncio
async def test_detail_requires_read_permission(client, seeded, make_user, db_sessionmaker):
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
+158
View File
@@ -0,0 +1,158 @@
import pytest
from app.crud import audit as audit_crud
from app.models.attachment_point import AttachmentPoint
from app.models.attachment_point_name import AttachmentPointName
from app.models.augmented_part import AugmentedPart, AugmentedPartNote
from app.models.clamp_connection import ClampConnection
from app.models.fit import Fit, FitLabel, FitNote
from app.models.imported_part import ImportedPart
from app.models.imported_part_config import ImportedPartConfig
from app.models.iso_thread import IsoThread
from app.models.manufacturer import Manufacturer
from app.models.part_colour import PartColour
from app.models.part_type import PartType
from app.models.scope_info import ScopeInfo
from app.models.scope_model import ScopeModel
from app.models.static_connection import StaticConnection
from app.models.sub_part_type import SubPartType
from app.models.unc_thread import UncSeries, UncThread
async def _seed_domain_data(db):
manufacturer = Manufacturer(name="Acme", url="https://acme.test")
colour = PartColour(name="Black")
scope_t = PartType(name="Scope")
cover_t = PartType(name="Flip Cover")
apn = AttachmentPointName(name="Objective")
db.add_all([manufacturer, colour, scope_t, cover_t, apn])
await db.flush()
sub_t = SubPartType(name="Riflescope", part_type_id=scope_t.id)
scope_model = ScopeModel(name="Alpha", manufacturer_id=manufacturer.id)
iso = IsoThread(diameter_mm=52.0, pitch_mm=0.75)
unc = UncThread(diameter_fraction="1/4", tpi=20, series=UncSeries.UNC)
clamp = ClampConnection(min_mm=25.0, max_mm=30.0)
static = StaticConnection(od_mm=50.0, od_in=1.97)
db.add_all([sub_t, scope_model, iso, unc, clamp, static])
await db.flush()
scope = AugmentedPart(
description="Alpha 3-18x50",
sku="ALP-318",
colour_id=colour.id,
manufacturer_id=manufacturer.id,
part_type_id=scope_t.id,
scope_model_id=scope_model.id,
sub_part_type_id=sub_t.id,
)
cover = AugmentedPart(
description="Flip cover 50mm",
colour_id=colour.id,
manufacturer_id=manufacturer.id,
part_type_id=cover_t.id,
)
db.add_all([scope, cover])
await db.flush()
db.add(AugmentedPartNote(augmented_part_id=scope.id, username="tester", body="note"))
db.add(ScopeInfo(augmented_part_id=scope.id, magnification_min=3, magnification_max=18))
ap_s = AttachmentPoint(
augmented_part_id=scope.id,
attachment_point_name_id=apn.id,
iso_thread_id=iso.id,
)
ap_c = AttachmentPoint(
augmented_part_id=cover.id,
attachment_point_name_id=apn.id,
iso_thread_id=iso.id,
)
db.add_all([ap_s, ap_c])
await db.flush()
fit = 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,
)
db.add(fit)
await db.flush()
db.add(FitNote(fit_id=fit.id, username="tester", body="fits fine"))
imported = ImportedPart(InventoryID="INV-1", Description="Alpha 3-18x50")
db.add(imported)
config = ImportedPartConfig(id=1, odata_url="https://example.test/odata", username="svc")
db.add(config)
await db.commit()
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_non_superuser_forbidden(client, seeded, make_user):
# "admin" role holds every permission but is not a superuser.
await make_user("admin@example.com", role_ids=[seeded["roles"]["admin"].id])
await _web_login(client, "admin@example.com")
assert (await client.get("/admin/danger-zone")).status_code == 403
assert (await client.post("/admin/danger-zone/purge")).status_code == 403
@pytest.mark.asyncio
async def test_purge_wipes_domain_data_but_keeps_users_and_odata_config(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
await _seed_domain_data(db)
await make_user("root@example.com", is_superuser=True)
await _web_login(client, "root@example.com")
assert (await client.get("/admin/danger-zone")).status_code == 200
resp = await client.post("/admin/danger-zone/purge")
assert resp.status_code == 200
assert "Purge complete" in resp.text
async with db_sessionmaker() as db:
for Model in (
AugmentedPart,
AugmentedPartNote,
ScopeInfo,
AttachmentPoint,
Fit,
FitNote,
ImportedPart,
ScopeModel,
AttachmentPointName,
IsoThread,
UncThread,
ClampConnection,
StaticConnection,
Manufacturer,
):
rows = (await db.execute(Model.__table__.select())).all()
assert rows == [], f"{Model.__name__} should be empty after purge"
# Kept: users, RBAC, the OData connection config, and part taxonomy.
from app.crud import user as user_crud
assert await user_crud.get_by_email(db, "root@example.com") is not None
config = await db.get(ImportedPartConfig, 1)
assert config is not None
assert config.odata_url == "https://example.test/odata"
for Model in (PartColour, PartType, SubPartType):
rows = (await db.execute(Model.__table__.select())).all()
assert rows != [], f"{Model.__name__} should survive the purge"
entries = await audit_crud.list_recent(db)
assert any("Purged all domain data" in e.description for e in entries)
+34 -8
View File
@@ -34,20 +34,40 @@ def test_same_label_never_matches_even_with_identical_threads():
def test_unc_thread_exact_match():
a = _ap(1, 1, unc_thread=UncThread(diameter_fraction="1/4", tpi=20, series=UncSeries.UNC))
b = _ap(2, 2, unc_thread=UncThread(diameter_fraction="1/4", tpi=20, series=UncSeries.UNC))
a = _ap(
1,
1,
unc_thread=UncThread(diameter_fraction="1/4", tpi=20, series=UncSeries.UNC),
)
b = _ap(
2,
2,
unc_thread=UncThread(diameter_fraction="1/4", tpi=20, series=UncSeries.UNC),
)
assert attachment_points_match(a, b)
def test_unc_thread_series_mismatch():
a = _ap(1, 1, unc_thread=UncThread(diameter_fraction="1/4", tpi=20, series=UncSeries.UNC))
b = _ap(2, 2, unc_thread=UncThread(diameter_fraction="1/4", tpi=20, series=UncSeries.UNF))
a = _ap(
1,
1,
unc_thread=UncThread(diameter_fraction="1/4", tpi=20, series=UncSeries.UNC),
)
b = _ap(
2,
2,
unc_thread=UncThread(diameter_fraction="1/4", tpi=20, series=UncSeries.UNF),
)
assert not attachment_points_match(a, b)
def test_iso_and_unc_never_match_each_other():
a = _ap(1, 1, iso_thread=IsoThread(diameter_mm=8, pitch_mm=1.25))
b = _ap(2, 2, unc_thread=UncThread(diameter_fraction="1/4", tpi=20, series=UncSeries.UNC))
b = _ap(
2,
2,
unc_thread=UncThread(diameter_fraction="1/4", tpi=20, series=UncSeries.UNC),
)
assert not attachment_points_match(a, b)
@@ -94,9 +114,15 @@ def test_two_bare_attachment_points_never_match():
def test_parts_could_fit_requires_different_types_and_identity():
a = AugmentedPart(id=1, part_type_id=1, description="a", colour_id=1, manufacturer_id=1)
b = AugmentedPart(id=2, part_type_id=1, description="b", colour_id=1, manufacturer_id=1)
c = AugmentedPart(id=3, part_type_id=2, description="c", colour_id=1, manufacturer_id=1)
a = AugmentedPart(
id=1, part_type_id=1, description="a", colour_id=1, manufacturer_id=1
)
b = AugmentedPart(
id=2, part_type_id=1, description="b", colour_id=1, manufacturer_id=1
)
c = AugmentedPart(
id=3, part_type_id=2, description="c", colour_id=1, manufacturer_id=1
)
assert not parts_could_fit(a, b)
assert parts_could_fit(a, c)
assert not parts_could_fit(a, a)
+1 -3
View File
@@ -80,9 +80,7 @@ async def test_ldap_login_is_idempotent(
assert await user_crud.count(db) == 1
async def test_ldap_login_bad_credentials(
client, seeded, monkeypatch, ldap_enabled
):
async def test_ldap_login_bad_credentials(client, seeded, monkeypatch, ldap_enabled):
async def fake_authenticate(username, password):
return None
+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
+276 -6
View File
@@ -1,7 +1,10 @@
import pytest
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.iso_thread import IsoThread
from app.models.part_type import PartType
from app.models.scope_info import ScopeInfo
from app.models.sub_part_type import SubPartType
@@ -31,7 +34,9 @@ async def _seed_guide_data(db):
db.add(p)
return p
s1 = part("Alpha 3-18x50", scope_t.id, manufacturer_id=10, sub_part_type_id=rifle.id)
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)
@@ -43,6 +48,7 @@ async def _seed_guide_data(db):
augmented_part_id=s1.id,
magnification_min=3,
magnification_max=18,
objective_lens_diameter_mm=50,
objective_thread_pitch="M52x0.75",
)
)
@@ -141,6 +147,26 @@ async def test_build_filters_by_description(db_sessionmaker):
assert _descriptions(guide) == ["Alpha 3-18x50"]
@pytest.mark.asyncio
async def test_build_filters_by_sku(db_sessionmaker):
async with db_sessionmaker() as db:
ids = await _seed_guide_data(db)
ids["s1"].sku = "ARM-12345"
await db.commit()
guide = await scope_guide.build(db, description="ARM-12345")
assert _descriptions(guide) == ["Alpha 3-18x50"]
@pytest.mark.asyncio
async def test_build_filters_by_partial_sku(db_sessionmaker):
async with db_sessionmaker() as db:
ids = await _seed_guide_data(db)
ids["s1"].sku = "ARM-12345"
await db.commit()
guide = await scope_guide.build(db, description="12345")
assert _descriptions(guide) == ["Alpha 3-18x50"]
@pytest.mark.asyncio
async def test_build_filters_by_manufacturer(db_sessionmaker):
async with db_sessionmaker() as db:
@@ -169,9 +195,7 @@ async def test_build_filters_by_required_fit_type(db_sessionmaker):
async def _web_login(client, email, password="secret123"):
resp = await client.post(
"/login", data={"email": email, "password": password}
)
resp = await client.post("/login", data={"email": email, "password": password})
assert resp.status_code == 303
@@ -199,7 +223,93 @@ async def test_scopes_page_renders_rows_and_fits(
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
assert "3-18" in body # magnification range, integral values un-padded
@pytest.mark.asyncio
async def test_build_includes_objective_lens_diameter(db_sessionmaker):
async with db_sessionmaker() as db:
await _seed_guide_data(db)
guide = await scope_guide.build(db)
rows = {r["scope"].description: r for r in guide["rows"]}
assert rows["Alpha 3-18x50"]["info"].objective_lens_diameter_mm == 50
assert rows["Zeta 1-6x24"]["info"] is None
@pytest.mark.asyncio
async def test_build_prefers_objective_attachment_point_over_scope_info(
db_sessionmaker,
):
async with db_sessionmaker() as db:
ids = await _seed_guide_data(db)
apn = AttachmentPointName(name="Objective")
db.add(apn)
await db.flush()
iso = IsoThread(diameter_mm=44, pitch_mm=0.5)
db.add(iso)
await db.flush()
db.add(
AttachmentPoint(
augmented_part_id=ids["s1"].id,
attachment_point_name_id=apn.id,
iso_thread_id=iso.id,
)
)
await db.commit()
guide = await scope_guide.build(db)
rows = {r["scope"].description: r for r in guide["rows"]}
s1_ap = rows["Alpha 3-18x50"]["objective_attachment_point"]
assert s1_ap is not None
assert s1_ap.iso_thread.diameter_mm == 44
# No "Objective"-named attachment point for s2 -> falls back to None.
assert rows["Zeta 1-6x24"]["objective_attachment_point"] is None
@pytest.mark.asyncio
async def test_scopes_page_renders_objective_attachment_point_thread(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
ids = await _seed_guide_data(db)
apn = AttachmentPointName(name="Objective")
db.add(apn)
await db.flush()
iso = IsoThread(diameter_mm=44, pitch_mm=0.5)
db.add(iso)
await db.flush()
db.add(
AttachmentPoint(
augmented_part_id=ids["s1"].id,
attachment_point_name_id=apn.id,
iso_thread_id=iso.id,
)
)
await db.commit()
await make_user("reader5@example.com", role_ids=[seeded["roles"]["user"].id])
await _web_login(client, "reader5@example.com")
resp = await client.get("/scopes")
assert resp.status_code == 200
body = resp.text
# The attachment point's thread spec wins over ScopeInfo.objective_thread_pitch.
assert "M44.0×0.5" in body
assert "M52x0.75" not in body
@pytest.mark.asyncio
async def test_scopes_page_renders_objective_lens_diameter_in_magnification(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
await _seed_guide_data(db)
await make_user("reader4@example.com", role_ids=[seeded["roles"]["user"].id])
await _web_login(client, "reader4@example.com")
resp = await client.get("/scopes")
assert resp.status_code == 200
assert "3-18x50" in resp.text
@pytest.mark.asyncio
@@ -238,4 +348,164 @@ async def test_scopes_page_forbidden_without_permission(client, seeded, make_use
async def test_scopes_page_redirects_when_anonymous(client):
resp = await client.get("/scopes")
assert resp.status_code == 302
assert resp.headers["location"] == "/login"
# Deep link is preserved so the user lands back here after logging in.
assert resp.headers["location"] == "/login?next=%2Fscopes"
@pytest.mark.asyncio
async def test_scopes_page_hides_checkboxes_for_read_only_user(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
await _seed_guide_data(db)
await make_user("reader6@example.com", role_ids=[seeded["roles"]["user"].id])
await _web_login(client, "reader6@example.com")
resp = await client.get("/scopes")
assert resp.status_code == 200
assert 'name="scope_ids"' not in resp.text
assert "delete-selected-scopes-btn" not in resp.text
@pytest.mark.asyncio
async def test_scopes_page_shows_checkboxes_for_write_user(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
ids = await _seed_guide_data(db)
await make_user("writer@example.com", is_superuser=True)
await _web_login(client, "writer@example.com")
resp = await client.get("/scopes")
assert resp.status_code == 200
assert f'name="scope_ids" value="{ids["s1"].id}"' in resp.text
assert "delete-selected-scopes-btn" in resp.text
assert 'hx-post="/scopes/bulk-delete"' in resp.text
@pytest.mark.asyncio
async def test_bulk_delete_scopes_removes_selected_augmented_parts(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
ids = await _seed_guide_data(db)
await make_user("writer2@example.com", is_superuser=True)
await _web_login(client, "writer2@example.com")
resp = await client.post(
"/scopes/bulk-delete",
data={"scope_ids": [str(ids["s1"].id), str(ids["s2"].id)]},
)
assert resp.status_code == 200
async with db_sessionmaker() as db:
from app.crud import augmented_part as augmented_part_crud
assert await augmented_part_crud.get(db, ids["s1"].id) is None
assert await augmented_part_crud.get(db, ids["s2"].id) is None
@pytest.mark.asyncio
async def test_bulk_delete_scopes_leaves_unselected_rows(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
ids = await _seed_guide_data(db)
await make_user("writer3@example.com", is_superuser=True)
await _web_login(client, "writer3@example.com")
resp = await client.post(
"/scopes/bulk-delete", data={"scope_ids": [str(ids["s1"].id)]}
)
assert resp.status_code == 200
async with db_sessionmaker() as db:
from app.crud import augmented_part as augmented_part_crud
assert await augmented_part_crud.get(db, ids["s1"].id) is None
assert await augmented_part_crud.get(db, ids["s2"].id) is not None
@pytest.mark.asyncio
async def test_scopes_page_shows_single_page_summary(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
await _seed_guide_data(db) # 2 scopes, well under the default page size
await make_user("pager1@example.com", role_ids=[seeded["roles"]["user"].id])
await _web_login(client, "pager1@example.com")
resp = await client.get("/scopes")
assert resp.status_code == 200
assert "Showing 2 records." in resp.text
assert "Next" not in resp.text
@pytest.mark.asyncio
async def test_scopes_filter_paginates_results(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
scope_t = PartType(name="Scope")
db.add(scope_t)
await db.flush()
for i in range(5):
db.add(
AugmentedPart(
description=f"Scope {i:02d}",
colour_id=1,
manufacturer_id=1,
part_type_id=scope_t.id,
)
)
await db.commit()
await make_user("pager2@example.com", role_ids=[seeded["roles"]["user"].id])
await _web_login(client, "pager2@example.com")
resp = await client.get("/scopes/filter", params={"page_size": 2})
assert resp.status_code == 200
body = resp.text
assert "Scope 00" in body
assert "Scope 01" in body
assert "Scope 02" not in body
assert "Showing 1–2 of 5." in body
resp2 = await client.get("/scopes/filter", params={"page_size": 2, "page": 3})
assert resp2.status_code == 200
body2 = resp2.text
assert "Scope 04" in body2
assert "Scope 00" not in body2
assert "Showing 5–5 of 5." in body2
@pytest.mark.asyncio
async def test_scopes_filter_clamps_out_of_range_page(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
await _seed_guide_data(db) # 2 scopes
await make_user("pager3@example.com", role_ids=[seeded["roles"]["user"].id])
await _web_login(client, "pager3@example.com")
resp = await client.get("/scopes/filter", params={"page_size": 1, "page": 99})
assert resp.status_code == 200
# Clamped to the last real page (2), not an empty page 99.
assert "Zeta 1-6x24" in resp.text
assert "Alpha 3-18x50" not in resp.text
@pytest.mark.asyncio
async def test_bulk_delete_scopes_forbidden_without_write_permission(
client, seeded, make_user, db_sessionmaker
):
async with db_sessionmaker() as db:
ids = await _seed_guide_data(db)
await make_user("reader7@example.com", role_ids=[seeded["roles"]["user"].id])
await _web_login(client, "reader7@example.com")
resp = await client.post(
"/scopes/bulk-delete", data={"scope_ids": [str(ids["s1"].id)]}
)
assert resp.status_code == 403
async with db_sessionmaker() as db:
from app.crud import augmented_part as augmented_part_crud
assert await augmented_part_crud.get(db, ids["s1"].id) is not None
+10 -10
View File
@@ -19,7 +19,7 @@ HEADERS = [
"model",
"magnification_min",
"magnification_max",
"objective_lens_outside_diameter_mm",
"objective_lens_diameter_mm",
"model_suffix",
"sub_part_type",
"product_number",
@@ -52,7 +52,7 @@ def _scope_row(**overrides):
"model": "Viper",
"magnification_min": 5,
"magnification_max": 25,
"objective_lens_outside_diameter_mm": 56,
"objective_lens_diameter_mm": 56,
"model_suffix": "PST",
"sub_part_type": "Riflescope",
"product_number": "SCOPE-A",
@@ -141,6 +141,7 @@ async def test_import_creates_scope_with_info_and_attachment_points(db_sessionma
assert info.magnification_min == 5
assert info.magnification_max == 25
assert info.model_suffix == "PST"
assert info.objective_lens_diameter_mm == 56
assert info.objective_thread_pitch == "0.75"
aps = await attachment_point_crud.list_for_augmented_part(db, part.id)
@@ -151,14 +152,13 @@ async def test_import_creates_scope_with_info_and_attachment_points(db_sessionma
assert len(by_label["Ocular"]) == 1
assert by_label["Ocular"][0].static_connection.od_mm == 40
# The lens diameter is descriptive only: the sole Objective point is the
# thread.
objectives = by_label["Objective"]
assert len(objectives) == 2 # static OD + thread
static_obj = [a for a in objectives if a.static_connection is not None]
thread_obj = [a for a in objectives if a.iso_thread is not None]
assert static_obj and static_obj[0].static_connection.od_mm == 56
assert thread_obj
assert thread_obj[0].iso_thread.diameter_mm == 40
assert thread_obj[0].iso_thread.pitch_mm == 0.75
assert len(objectives) == 1
assert objectives[0].static_connection is None
assert objectives[0].iso_thread.diameter_mm == 40
assert objectives[0].iso_thread.pitch_mm == 0.75
async def test_import_resolves_ocular_and_objective_fits(db_sessionmaker):
@@ -200,7 +200,7 @@ async def test_import_upserts_on_product_number(db_sessionmaker):
# Attachment points were rebuilt, not duplicated.
aps = await attachment_point_crud.list_for_augmented_part(db, matching[0].id)
assert len(aps) == 3
assert len(aps) == 2
async def test_row_without_product_number_is_skipped(db_sessionmaker):