Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2318a6daeb | ||
|
|
6d7c5c5604 | ||
|
|
7a15d9f7c0 | ||
|
|
a9c4c16d4f | ||
|
|
14aa2ac5d6 | ||
|
|
f4771e33d4 |
No files matched your search
@@ -37,7 +37,9 @@
|
||||
"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(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)"
|
||||
]
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
VERSION := $(shell git describe --tags --always)
|
||||
VERSION := $(shell git tag | tail -1)
|
||||
IMAGE_NAME := fitguide
|
||||
|
||||
sync:
|
||||
|
||||
@@ -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 |
|
||||
|
||||
+5
-1
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
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
|
||||
@@ -89,10 +89,19 @@ async def list_filtered(
|
||||
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:
|
||||
|
||||
+12
-3
@@ -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,
|
||||
|
||||
@@ -33,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::
|
||||
|
||||
@@ -69,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:
|
||||
|
||||
@@ -567,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;
|
||||
|
||||
@@ -18,9 +18,13 @@
|
||||
{%- endif -%}
|
||||
{%- endmacro -%}
|
||||
<div id="scopes-table-card" class="card" style="overflow-x:auto">
|
||||
{% 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>
|
||||
@@ -36,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 %}
|
||||
@@ -70,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>
|
||||
|
||||
@@ -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 %}
|
||||
+7
-44
@@ -1,8 +1,5 @@
|
||||
"""HTMX-driven admin dashboard for users, roles, and thread standards."""
|
||||
|
||||
import math
|
||||
import urllib.parse
|
||||
|
||||
from fastapi import (
|
||||
APIRouter,
|
||||
Depends,
|
||||
@@ -17,9 +14,14 @@ from fastapi.responses import HTMLResponse, RedirectResponse, Response
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.database import AsyncSessionLocal, get_db
|
||||
from app.crud import attachment_point as attachment_point_crud
|
||||
from app.crud import attachment_point_name as apn_crud
|
||||
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.crud import clamp_connection as clamp_connection_crud
|
||||
from app.crud import fit as fit_crud
|
||||
from app.crud import fit_note as fit_note_crud
|
||||
from app.crud import imported_part as imported_part_crud
|
||||
from app.crud import imported_part_config as imported_part_config_crud
|
||||
from app.crud import iso_thread as iso_thread_crud
|
||||
@@ -29,11 +31,6 @@ from app.crud import part_type as part_type_crud
|
||||
from app.crud import role as role_crud
|
||||
from app.crud import scope_info as scope_info_crud
|
||||
from app.crud import scope_model as scope_model_crud
|
||||
from app.crud import attachment_point as attachment_point_crud
|
||||
from app.crud import attachment_point_name as apn_crud
|
||||
from app.crud import clamp_connection as clamp_connection_crud
|
||||
from app.crud import fit as fit_crud
|
||||
from app.crud import fit_note as fit_note_crud
|
||||
from app.crud import static_connection as static_connection_crud
|
||||
from app.crud import sub_part_type as sub_part_type_crud
|
||||
from app.crud import unc_thread as unc_thread_crud
|
||||
@@ -48,6 +45,8 @@ from app.services import fit_matching as fit_matching_service
|
||||
from app.services import jobs
|
||||
from app.services import odata as odata_service
|
||||
from app.services import scope_import as scope_import_service
|
||||
from app.web.pagination import paginate as _paginate
|
||||
from app.web.pagination import query_url as _query_url
|
||||
from app.web.templates import templates
|
||||
|
||||
router = APIRouter(prefix="/admin", include_in_schema=False)
|
||||
@@ -55,42 +54,6 @@ router = APIRouter(prefix="/admin", include_in_schema=False)
|
||||
_PAGE_SIZE = 50
|
||||
|
||||
|
||||
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,
|
||||
}
|
||||
|
||||
|
||||
def _sort_headers(
|
||||
base_path: str,
|
||||
filter_params: dict,
|
||||
|
||||
+9
-3
@@ -1,3 +1,5 @@
|
||||
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
|
||||
@@ -36,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="/",
|
||||
)
|
||||
|
||||
@@ -79,7 +81,9 @@ async def login_submit(
|
||||
},
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
)
|
||||
access, _ = await issue_token_pair(db, user)
|
||||
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
|
||||
)
|
||||
@@ -110,7 +114,9 @@ async def login_ldap_submit(
|
||||
},
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
)
|
||||
access, _ = await issue_token_pair(db, user)
|
||||
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
|
||||
)
|
||||
|
||||
@@ -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
@@ -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),
|
||||
},
|
||||
)
|
||||
@@ -147,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:
|
||||
@@ -330,3 +350,162 @@ async def test_scopes_page_redirects_when_anonymous(client):
|
||||
assert resp.status_code == 302
|
||||
# Deep link is preserved so the user lands back here after logging in.
|
||||
assert resp.headers["location"] == "/login?next=%2Fscopes"
|
||||
|
||||
|
||||
@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
|
||||
Reference in new issue
Block a user