Compare commits

..
6 Commits
17 changed files with 444 additions and 71 deletions

No files matched your search

+3 -1
View File
@@ -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)"
]
}
}
+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
+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 |
+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
+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
+11 -2
View File
@@ -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
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,
+7 -3
View File
@@ -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:
+5
View File
@@ -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;
+9 -2
View File
@@ -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>
+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 %}
+7 -44
View File
@@ -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
View File
@@ -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
)
+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),
},
)
+179
View File
@@ -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