Compare commits

...
3 Commits
Author SHA1 Message Date
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
12 changed files with 374 additions and 114 deletions

No files matched your search

+4 -1
View File
@@ -34,7 +34,10 @@
"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 \"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)"
]
}
}
@@ -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 ###
+2
View File
@@ -14,6 +14,7 @@ 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
@@ -40,6 +41,7 @@ __all__ = [
"ImportedPart",
"ImportedPartConfig",
"IsoThread",
"Job",
"Manufacturer",
"PartColour",
"PartType",
+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
)
-3
View File
@@ -16,12 +16,9 @@ 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.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 UncThread
# Children before the RESTRICT-protected reference tables they point to.
+74 -52
View File
@@ -1,11 +1,12 @@
"""Lightweight in-process tracking for long-running admin operations.
"""DB-backed tracking for long-running admin operations.
Jobs live in a module-level dict, so this only works correctly for a single
OS process (the app's current deployment: one uvicorn worker, no multi-process
or multi-replica setup — see ``entrypoint.sh`` / ``docker-compose.yml``). If
the app is ever scaled to multiple workers/replicas, this needs to move to
shared storage (a DB table, Redis) so a poll request landing on a different
process can still see the job.
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
@@ -13,62 +14,83 @@ from __future__ import annotations
import asyncio
import uuid
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
from datetime import UTC, datetime
from datetime import UTC, datetime, timedelta
_MAX_FINISHED_AGE_SECONDS = 3600
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)
@dataclass
class Job:
id: str
kind: str
status: str = "running" # running | done | error
message: str = "Working…"
result: object | None = None
created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
finished_at: datetime | None = None
async def get(db: AsyncSession, job_id: str) -> Job | None:
return await db.get(Job, job_id)
_jobs: dict[str, Job] = {}
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.
def get(job_id: str) -> Job | None:
return _jobs.get(job_id)
def _evict_stale() -> None:
now = datetime.now(UTC)
stale = [
job_id
for job_id, job in _jobs.items()
if job.finished_at
and (now - job.finished_at).total_seconds() > _MAX_FINISHED_AGE_SECONDS
]
for job_id in stale:
del _jobs[job_id]
def start(kind: str, run: Callable[[Job], Awaitable[None]]) -> Job:
"""Create a job, launch ``run(job)`` in the background, and return it immediately.
``run`` is responsible for updating ``job.status``/``job.message``/``job.result``
to reflect the outcome, including catching 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.
``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.
"""
_evict_stale()
job = Job(id=uuid.uuid4().hex, kind=kind)
_jobs[job.id] = job
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)
await run(job.id)
except Exception as exc:
job.status = "error"
job.message = f"{kind} failed: {exc}"
finally:
job.finished_at = datetime.now(UTC)
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()
+6 -6
View File
@@ -8,15 +8,15 @@
<div class="card" style="border: 1px solid var(--color-danger, #c00)">
<h2>Purge database</h2>
<p>
Permanently deletes every part, taxonomy entry, attachment point, fit,
and imported part in the database — manufacturers, part colours, part
types, sub part types, scope models, ISO/UNC threads, clamp/static
connections, attachment point names, augmented parts, scope info,
attachment points, fits, fit notes, and imported parts.
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, and the OData connection configuration.
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"
+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" %}
+118 -44
View File
@@ -1366,6 +1366,19 @@ async def save_imported_part_config(
return HTMLResponse("Configuration saved.", status_code=200)
_IMPORTED_PARTS_REFRESH_TARGET_ID = "imported-parts-section"
async def _imported_parts_table_ctx(
db: AsyncSession, *, sync_result: str | None = None, sync_error: str | None = None
) -> dict:
return {
**await _default_imported_parts_table_ctx(db),
"sync_result": sync_result,
"sync_error": sync_error,
}
@router.post("/imported-parts/refresh", response_class=HTMLResponse)
async def refresh_imported_parts(
request: Request,
@@ -1373,45 +1386,94 @@ async def refresh_imported_parts(
actor: User = Depends(web_require("imported_parts:write")),
):
config = await imported_part_config_crud.get(db)
async def _imported_table_ctx(sync_result=None, sync_error=None):
return {
**await _default_imported_parts_table_ctx(db),
"sync_result": sync_result,
"sync_error": sync_error,
}
if not config or not config.odata_url:
return templates.TemplateResponse(
request,
"admin/_imported_parts_table.html",
await _imported_table_ctx(sync_error="OData URL is not configured."),
await _imported_parts_table_ctx(
db, sync_error="OData URL is not configured."
),
)
actor_id = actor.id
async def _run(job_id: str) -> None:
async with AsyncSessionLocal() as db:
config = await imported_part_config_crud.get(db)
if not config or not config.odata_url:
await jobs.mark_error(job_id, "OData URL is not configured.")
return
try:
records = await odata_service.fetch(
config.odata_url, config.username, config.password
)
created, updated = await imported_part_crud.upsert_bulk(db, records)
except Exception as exc:
await jobs.mark_error(job_id, f"Sync failed: {exc}")
return
actor_row = await user_crud.get(db, actor_id)
await audit_crud.log(
db,
actor=actor,
actor=actor_row,
action="update",
resource_type="ImportedPart",
resource_id=None,
description=f"OData sync: {created} created, {updated} updated ({len(records)} total records)",
)
return templates.TemplateResponse(
request,
"admin/_imported_parts_table.html",
await _imported_table_ctx(
sync_result=f"Sync complete: {created} created, {updated} updated."
description=(
f"OData sync: {created} created, {updated} updated "
f"({len(records)} total records)"
),
)
except Exception as exc:
await jobs.mark_done(
job_id,
message=f"Sync complete: {created} created, {updated} updated.",
)
job = await jobs.start("imported_parts_refresh", _run)
return templates.TemplateResponse(
request,
"admin/_job_progress.html",
{
"container_id": _IMPORTED_PARTS_REFRESH_TARGET_ID,
"poll_url": f"/admin/imported-parts/refresh/jobs/{job.id}",
"job": job,
},
)
@router.get("/imported-parts/refresh/jobs/{job_id}", response_class=HTMLResponse)
async def imported_parts_refresh_job_status(
request: Request,
job_id: str,
db: AsyncSession = Depends(get_db),
_: User = Depends(web_require("imported_parts:write")),
):
job = await jobs.get(db, job_id)
if job is None:
return templates.TemplateResponse(
request,
"admin/_imported_parts_table.html",
await _imported_table_ctx(sync_error=f"Sync failed: {exc}"),
await _imported_parts_table_ctx(
db, sync_error="Refresh job not found — it may have expired."
),
)
if job.status == "running":
return templates.TemplateResponse(
request,
"admin/_job_progress.html",
{
"container_id": _IMPORTED_PARTS_REFRESH_TARGET_ID,
"poll_url": f"/admin/imported-parts/refresh/jobs/{job_id}",
"job": job,
},
)
return templates.TemplateResponse(
request,
"admin/_imported_parts_table.html",
await _imported_parts_table_ctx(
db,
sync_result=job.message if job.status == "done" else None,
sync_error=job.message if job.status == "error" else None,
),
)
@@ -1511,13 +1573,12 @@ async def upload_scope_import(
data = await file.read()
actor_id = actor.id
async def _run(job: jobs.Job) -> None:
async def _run(job_id: str) -> None:
async with AsyncSessionLocal() as db:
try:
result = await scope_import_service.import_from_bytes(db, data)
except Exception as exc: # noqa: BLE001
job.status = "error"
job.message = f"Could not read spreadsheet: {exc}"
await jobs.mark_error(job_id, f"Could not read spreadsheet: {exc}")
return
actor_row = await user_crud.get(db, actor_id)
await audit_crud.log(
@@ -1532,10 +1593,16 @@ async def upload_scope_import(
f"{result.errors} errors"
),
)
job.status = "done"
job.result = result
await jobs.mark_done(
job_id,
message=(
f"Scope import complete: {result.created} created, "
f"{result.updated} updated."
),
result=result.model_dump(mode="json"),
)
job = jobs.start("scope_import", _run)
job = await jobs.start("scope_import", _run)
return templates.TemplateResponse(
request, "admin/_job_progress.html", _scope_import_progress_ctx(job)
)
@@ -1545,9 +1612,10 @@ async def upload_scope_import(
async def scope_import_job_status(
request: Request,
job_id: str,
db: AsyncSession = Depends(get_db),
_: User = Depends(web_require("augmented_parts:write")),
):
job = jobs.get(job_id)
job = await jobs.get(db, job_id)
if job is None:
return templates.TemplateResponse(
request,
@@ -1718,13 +1786,12 @@ async def sync_augmented_parts(
):
actor_id = actor.id
async def _run(job: jobs.Job) -> None:
async def _run(job_id: str) -> None:
async with AsyncSessionLocal() as db:
try:
result = await augmented_part_sync_service.sync(db)
except Exception as exc:
job.status = "error"
job.message = f"Sync failed: {exc}"
await jobs.mark_error(job_id, f"Sync failed: {exc}")
return
actor_row = await user_crud.get(db, actor_id)
await audit_crud.log(
@@ -1738,18 +1805,21 @@ async def sync_augmented_parts(
f"{result.skipped} skipped (no type match)"
),
)
job.status = "done"
job.message = (
f"Sync complete: {result.created} created, {result.deleted} deleted"
await jobs.mark_done(
job_id,
message=(
f"Sync complete: {result.created} created, "
f"{result.deleted} deleted"
+ (
f", {result.skipped} skipped (type not recognised)"
if result.skipped
else ""
)
+ "."
),
)
job = jobs.start("augmented_part_sync", _run)
job = await jobs.start("augmented_part_sync", _run)
return templates.TemplateResponse(
request,
"admin/_job_progress.html",
@@ -1768,7 +1838,7 @@ async def augmented_parts_sync_job_status(
db: AsyncSession = Depends(get_db),
_: User = Depends(web_require("augmented_parts:write")),
):
job = jobs.get(job_id)
job = await jobs.get(db, job_id)
if job is None:
form_ctx = await _augmented_part_form_ctx(db)
return templates.TemplateResponse(
@@ -2483,7 +2553,7 @@ async def recompute_augmented_part_fits(
):
actor_id = actor.id
async def _run(job: jobs.Job) -> None:
async def _run(job_id: str) -> None:
async with AsyncSessionLocal() as db:
result = await fit_matching_service.recompute_for_part(
db, augmented_part_id
@@ -2500,10 +2570,11 @@ async def recompute_augmented_part_fits(
f"{result.created} created, {result.removed} removed"
),
)
job.status = "done"
job.message = f"{result.created} created, {result.removed} removed."
await jobs.mark_done(
job_id, f"{result.created} created, {result.removed} removed."
)
job = jobs.start("fit_recompute_part", _run)
job = await jobs.start("fit_recompute_part", _run)
return templates.TemplateResponse(
request,
"admin/_job_progress.html",
@@ -2527,7 +2598,7 @@ async def augmented_part_fits_recompute_job_status(
db: AsyncSession = Depends(get_db),
actor: User = Depends(web_require("fits:write")),
):
job = jobs.get(job_id)
job = await jobs.get(db, job_id)
if job is None:
ctx = await _augmented_part_fits_ctx(
db,
@@ -2774,7 +2845,7 @@ async def recompute_all_fits(
):
actor_id = actor.id
async def _run(job: jobs.Job) -> None:
async def _run(job_id: str) -> None:
async with AsyncSessionLocal() as db:
result = await fit_matching_service.recompute_all(db)
actor_row = await user_crud.get(db, actor_id)
@@ -2789,13 +2860,15 @@ async def recompute_all_fits(
f"{result.removed} removed"
),
)
job.status = "done"
job.message = (
await jobs.mark_done(
job_id,
message=(
f"Recompute complete: {result.created} created, "
f"{result.removed} removed."
),
)
job = jobs.start("fit_recompute_all", _run)
job = await jobs.start("fit_recompute_all", _run)
return templates.TemplateResponse(
request,
"admin/_job_progress_inline.html",
@@ -2813,9 +2886,10 @@ async def recompute_all_fits(
async def recompute_all_fits_job_status(
request: Request,
job_id: str,
db: AsyncSession = Depends(get_db),
_: User = Depends(web_require("fits:write")),
):
job = jobs.get(job_id)
job = await jobs.get(db, job_id)
if job is None:
return HTMLResponse("Recompute job not found — it may have expired.")
if job.status == "running":
+88
View File
@@ -2,7 +2,11 @@ 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
@@ -13,6 +17,25 @@ 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
@@ -39,6 +62,7 @@ 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(
@@ -69,6 +93,7 @@ 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(
@@ -93,6 +118,7 @@ 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")
@@ -137,6 +163,7 @@ 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(
@@ -154,3 +181,64 @@ async def test_fit_recompute_all_runs_in_background(
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
+5 -4
View File
@@ -132,20 +132,17 @@ async def test_purge_wipes_domain_data_but_keeps_users_and_odata_config(
FitNote,
ImportedPart,
ScopeModel,
SubPartType,
PartType,
AttachmentPointName,
IsoThread,
UncThread,
ClampConnection,
StaticConnection,
PartColour,
Manufacturer,
):
rows = (await db.execute(Model.__table__.select())).all()
assert rows == [], f"{Model.__name__} should be empty after purge"
# Kept: users, RBAC, and the OData connection config.
# 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
@@ -153,5 +150,9 @@ async def test_purge_wipes_domain_data_but_keeps_users_and_odata_config(
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)
+1 -1
View File
@@ -275,7 +275,7 @@ async def test_scopes_page_renders_objective_attachment_point_thread(
assert resp.status_code == 200
body = resp.text
# The attachment point's thread spec wins over ScopeInfo.objective_thread_pitch.
assert "M44×0.5" in body
assert "M44.0×0.5" in body
assert "M52x0.75" not in body