242 lines
8.5 KiB
Python
242 lines
8.5 KiB
Python
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.augmented_part import AugmentedPart
|
|
from app.models.part_type import PartType
|
|
|
|
_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.
|
|
|
|
The dashboard's sync/recompute actions run a real background asyncio task
|
|
concurrently with polling requests, both hitting the DB at once. A real
|
|
file-backed DB (its own connection pool) supports that, unlike StaticPool's
|
|
single shared connection (see tests/test_admin_background_jobs.py).
|
|
"""
|
|
db_path = tmp_path / "dashboard_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_dashboard_shows_scope_and_skipped_counts(
|
|
client, seeded, make_user, db_sessionmaker
|
|
):
|
|
async with db_sessionmaker() as db:
|
|
scope_t = PartType(name="Scope")
|
|
db.add(scope_t)
|
|
await db.flush()
|
|
db.add_all(
|
|
[
|
|
AugmentedPart(
|
|
description="Alpha 3-18x50",
|
|
colour_id=1,
|
|
manufacturer_id=1,
|
|
part_type_id=scope_t.id,
|
|
),
|
|
AugmentedPart(
|
|
description="Zeta 1-6x24",
|
|
colour_id=1,
|
|
manufacturer_id=1,
|
|
part_type_id=scope_t.id,
|
|
),
|
|
]
|
|
)
|
|
await db.commit()
|
|
await make_user("reader@example.com", role_ids=[seeded["roles"]["user"].id])
|
|
await _web_login(client, "reader@example.com")
|
|
|
|
resp = await client.get("/")
|
|
assert resp.status_code == 200
|
|
assert "Data sync" in resp.text
|
|
assert "<dd>2</dd>" in resp.text # scope count
|
|
assert "<dd>0</dd>" in resp.text # nothing imported/skipped yet
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dashboard_hides_action_buttons_without_permissions(
|
|
client, seeded, make_user
|
|
):
|
|
await make_user("plain@example.com", role_ids=[seeded["roles"]["user"].id])
|
|
await _web_login(client, "plain@example.com")
|
|
|
|
resp = await client.get("/")
|
|
assert resp.status_code == 200
|
|
assert "Refresh from OData" not in resp.text
|
|
assert "Sync augmented parts" not in resp.text
|
|
assert "Recompute possible fits" not in resp.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dashboard_shows_action_buttons_for_superuser(client, make_user):
|
|
await make_user("boss@example.com", is_superuser=True)
|
|
await _web_login(client, "boss@example.com")
|
|
|
|
resp = await client.get("/")
|
|
assert resp.status_code == 200
|
|
assert "Refresh from OData" in resp.text
|
|
assert "Sync augmented parts" in resp.text
|
|
assert "Recompute possible fits" in resp.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dashboard_shows_only_permitted_button(
|
|
client, seeded, make_user, db_sessionmaker
|
|
):
|
|
async with db_sessionmaker() as db:
|
|
perms = seeded["perms"]
|
|
editor = await role_crud.create(
|
|
db, name="fits-only-editor", permission_ids=[perms["fits:write"].id]
|
|
)
|
|
editor_id = editor.id
|
|
await make_user("fitsonly@example.com", role_ids=[editor_id])
|
|
await _web_login(client, "fitsonly@example.com")
|
|
|
|
resp = await client.get("/")
|
|
assert resp.status_code == 200
|
|
assert "Recompute possible fits" in resp.text
|
|
assert "Refresh from OData" not in resp.text
|
|
assert "Sync augmented parts" not in resp.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dashboard_refresh_odata_without_config_skips_job(
|
|
client, seeded, make_user, db_sessionmaker
|
|
):
|
|
async with db_sessionmaker() as db:
|
|
perms = seeded["perms"]
|
|
editor = await role_crud.create(
|
|
db,
|
|
name="dash-refresh-noconfig-editor",
|
|
permission_ids=[perms["imported_parts:write"].id],
|
|
)
|
|
editor_id = editor.id
|
|
await make_user("dashrefresher2@example.com", role_ids=[editor_id])
|
|
await _web_login(client, "dashrefresher2@example.com")
|
|
|
|
resp = await client.post("/dashboard/refresh-odata")
|
|
assert resp.status_code == 200
|
|
assert "OData URL is not configured." in resp.text
|
|
assert _JOB_URL_RE.search(resp.text) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dashboard_sync_augmented_parts_runs_in_background(
|
|
client, seeded, make_user, db_sessionmaker, monkeypatch
|
|
):
|
|
monkeypatch.setattr("app.web.pages.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="dash-sync-editor",
|
|
permission_ids=[perms["augmented_parts:write"].id],
|
|
)
|
|
editor_id = editor.id
|
|
await make_user("dashsyncer@example.com", role_ids=[editor_id])
|
|
await _web_login(client, "dashsyncer@example.com")
|
|
|
|
resp = await client.post("/dashboard/sync-augmented-parts")
|
|
assert resp.status_code == 200
|
|
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
|
|
assert "dashboard-sync-card" in final_body
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dashboard_recompute_fits_runs_in_background(
|
|
client, seeded, make_user, db_sessionmaker, monkeypatch
|
|
):
|
|
monkeypatch.setattr("app.web.pages.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="dash-recompute-editor", permission_ids=[perms["fits:write"].id]
|
|
)
|
|
editor_id = editor.id
|
|
await make_user("dashrecomputer@example.com", role_ids=[editor_id])
|
|
await _web_login(client, "dashrecomputer@example.com")
|
|
|
|
resp = await client.post("/dashboard/recompute-fits")
|
|
assert resp.status_code == 200
|
|
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_dashboard_refresh_odata_runs_in_background(
|
|
client, seeded, make_user, db_sessionmaker, monkeypatch
|
|
):
|
|
monkeypatch.setattr("app.web.pages.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.pages.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="dash-refresh-editor",
|
|
permission_ids=[perms["imported_parts:write"].id],
|
|
)
|
|
editor_id = editor.id
|
|
await make_user("dashrefresher@example.com", role_ids=[editor_id])
|
|
await _web_login(client, "dashrefresher@example.com")
|
|
|
|
resp = await client.post("/dashboard/refresh-odata")
|
|
assert resp.status_code == 200
|
|
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
|