bvandeusenandClaude Opus 5.5
3369288e1d
Run 9095: with the Archives template at version 2, the library test's "bad" copy no longer had a bad version, saved, and took the name the real save then wanted. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
178 lines
8.2 KiB
Python
178 lines
8.2 KiB
Python
"""Integration: the widget template library (#5692) against Postgres. An admin
|
||
copies a built-in template, saves it under its own name, edits, previews,
|
||
exports, uploads and deletes; the widget picker lists it; each change is
|
||
audited; a viewer can look but not change. Requires STEWARD_DATABASE_URL."""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import io
|
||
import json
|
||
import os
|
||
import re
|
||
import time
|
||
import uuid
|
||
|
||
import pytest
|
||
|
||
pytestmark = pytest.mark.integration
|
||
|
||
CSRF = {"X-CSRF-Token": "test-token"}
|
||
|
||
|
||
@pytest.fixture
|
||
def app():
|
||
if not os.environ.get("STEWARD_DATABASE_URL"):
|
||
pytest.skip("needs Postgres")
|
||
from steward.app import create_app
|
||
return create_app(testing=False)
|
||
|
||
|
||
async def _client(app, role: str):
|
||
from steward.models.users import User, UserRole
|
||
uid = str(uuid.uuid4())
|
||
name = f"tl-{uid[:8]}"
|
||
async with app.db_sessionmaker() as s:
|
||
async with s.begin():
|
||
s.add(User(id=uid, username=name, email=f"{name}@test", password_hash="x",
|
||
role=UserRole(role), is_active=True))
|
||
client = app.test_client()
|
||
async with client.session_transaction() as sess:
|
||
sess.update(user_id=uid, user_role=role, username=name, _csrf="test-token",
|
||
login_at=time.time(), tz="UTC")
|
||
return client, uid
|
||
|
||
|
||
async def _audits(app, uid: str) -> list[tuple[str, dict]]:
|
||
from sqlalchemy import select
|
||
from steward.models.audit import AuditEvent
|
||
async with app.db_sessionmaker() as s:
|
||
rows = (await s.execute(select(AuditEvent).where(AuditEvent.user_id == uid)
|
||
.order_by(AuditEvent.timestamp))).scalars().all()
|
||
return [(r.action, json.loads(r.detail_json or "{}")) for r in rows]
|
||
|
||
|
||
def _prom(sub):
|
||
"""A submodule of the prometheus plugin as the app loaded it: importing
|
||
plugins.prometheus beside it would define its tables twice."""
|
||
import importlib
|
||
from steward.core.plugin_manager import get_loaded_plugin_modules
|
||
loaded = get_loaded_plugin_modules().get("prometheus")
|
||
return importlib.import_module(f"{loaded.__name__ if loaded else 'plugins.prometheus'}.{sub}")
|
||
|
||
|
||
def _version(text: str, value: str) -> str:
|
||
"""`text` with its top-level version: line set to `value`, whatever the
|
||
bundled template's own version is."""
|
||
changed, n = re.subn(r"(?m)^version:.*$", f"version: {value}", text, count=1)
|
||
assert n == 1
|
||
return changed
|
||
|
||
|
||
def _text(page: str) -> str:
|
||
return " ".join(page.split())
|
||
|
||
|
||
def test_an_admin_copies_edits_previews_exports_and_deletes_a_template(app):
|
||
from steward.core.widgets import WIDGET_REGISTRY
|
||
from steward.dashboard.routes import with_options
|
||
|
||
tag = uuid.uuid4().hex[:8]
|
||
mine = f"Archives {tag}"
|
||
|
||
async def go():
|
||
admin, uid = await _client(app, "admin")
|
||
out = {}
|
||
|
||
listing = _text(await (await admin.get("/plugins/prometheus/templates/")).get_data(as_text=True))
|
||
out["listed_bundled"] = "Fabled Archives" in listing and "built in" in listing
|
||
out["tab_from_targets"] = 'href="/plugins/prometheus/templates/"' in _text(
|
||
await (await admin.get("/plugins/prometheus/")).get_data(as_text=True))
|
||
|
||
# Duplicate: an editor holding a copy under a name nobody has, nothing saved yet.
|
||
copy = await (await admin.get("/plugins/prometheus/templates/new?from=bundled:fabled_archives")
|
||
).get_data(as_text=True)
|
||
out["copy_named"] = "Fabled Archives (copy" in copy and "A copy of “Fabled Archives”" in copy
|
||
|
||
library = _prom("library")
|
||
text = library.renamed(library.bundled()[0].text, mine)
|
||
|
||
# A template with a problem isn't saved, and says where.
|
||
bad = await admin.post("/plugins/prometheus/templates/new", headers=CSRF,
|
||
form={"text": _version(text, "one")})
|
||
out["bad"] = (bad.status_code, "Not saved." in _text(await bad.get_data(as_text=True)))
|
||
# The bundled template's name is taken.
|
||
clash = await admin.post("/plugins/prometheus/templates/new", headers=CSRF,
|
||
form={"text": library.bundled()[0].text})
|
||
out["clash"] = (clash.status_code, "is already named" in await clash.get_data(as_text=True))
|
||
|
||
saved = await admin.post("/plugins/prometheus/templates/new", headers=CSRF, form={"text": text})
|
||
out["saved"] = saved.status_code
|
||
where = saved.headers["Location"]
|
||
key = where.rsplit("/", 1)[1]
|
||
out["key"] = key
|
||
|
||
async with app.db_sessionmaker() as db:
|
||
picker = await with_options(db, WIDGET_REGISTRY["prometheus"])
|
||
out["picker"] = {"value": key, "label": mine} in picker["params"][1]["options"]
|
||
|
||
page = _text(await (await admin.get(where)).get_data(as_text=True))
|
||
out["editable"] = f'action="/plugins/prometheus/templates/{key}/edit"' in page and "Your own" in page
|
||
|
||
renamed = _version(library.renamed(text, f"{mine} v2"), "9")
|
||
edit = await admin.post(f"/plugins/prometheus/templates/{key}/edit", headers=CSRF, form={"text": renamed})
|
||
out["edited"] = edit.status_code
|
||
|
||
exported = await admin.get(f"/plugins/prometheus/templates/{key}/export")
|
||
out["export"] = (exported.status_code, exported.headers.get("Content-Disposition"),
|
||
await exported.get_data(as_text=True) == renamed)
|
||
|
||
preview = await admin.post("/plugins/prometheus/templates/preview", headers=CSRF,
|
||
form={"text": "name: x\n"})
|
||
out["preview_bad"] = "to fix before it draws" in await preview.get_data(as_text=True)
|
||
preview = await admin.post("/plugins/prometheus/templates/preview", headers=CSRF,
|
||
form={"text": renamed, "target": "no-such-target"})
|
||
out["preview_ok"] = "passes the checker" in await preview.get_data(as_text=True)
|
||
|
||
from werkzeug.datastructures import FileStorage
|
||
upload = await admin.post("/plugins/prometheus/templates/upload", headers=CSRF, files={
|
||
"file": FileStorage(io.BytesIO(("" + renamed).encode()), filename="mine.yaml")})
|
||
up = await upload.get_data(as_text=True)
|
||
out["upload"] = (upload.status_code, "Read from mine.yaml" in up, f"{mine} v2" in up)
|
||
|
||
gone = await admin.post(f"/plugins/prometheus/templates/{key}/delete", headers=CSRF)
|
||
out["deleted"] = gone.status_code
|
||
out["after"] = (await admin.get(where)).status_code
|
||
out["audits"] = [(a, d.get("name"), d.get("old_name")) for a, d in await _audits(app, uid)]
|
||
out["mine"] = mine
|
||
return out
|
||
|
||
out = asyncio.run(go())
|
||
mine = out["mine"]
|
||
assert out["listed_bundled"] and out["tab_from_targets"] and out["copy_named"]
|
||
assert out["bad"] == (422, True) and out["clash"] == (422, True)
|
||
assert out["saved"] == 302 and out["key"].startswith("custom:")
|
||
assert out["picker"] and out["editable"] and out["edited"] == 302
|
||
assert out["export"] == (200, f'attachment; filename="archives-{mine.split()[1]}-v2.yaml"', True)
|
||
assert out["preview_bad"] and out["preview_ok"]
|
||
assert out["upload"] == (200, True, True)
|
||
assert out["deleted"] == 302 and out["after"] == 404
|
||
assert out["audits"] == [("prometheus.template_created", mine, None),
|
||
("prometheus.template_updated", f"{mine} v2", mine),
|
||
("prometheus.template_deleted", f"{mine} v2", None)]
|
||
|
||
|
||
def test_a_viewer_looks_and_exports_but_changes_nothing(app):
|
||
async def go():
|
||
viewer, _ = await _client(app, "viewer")
|
||
listing = await viewer.get("/plugins/prometheus/templates/")
|
||
page = await viewer.get("/plugins/prometheus/templates/bundled:fabled_archives")
|
||
exported = await viewer.get("/plugins/prometheus/templates/bundled:fabled_archives/export")
|
||
new = await viewer.get("/plugins/prometheus/templates/new")
|
||
post = await viewer.post("/plugins/prometheus/templates/new", headers=CSRF, form={"text": "x"})
|
||
return (listing.status_code, page.status_code, " readonly" in await page.get_data(as_text=True),
|
||
exported.status_code, new.status_code, post.status_code)
|
||
|
||
listing, page, readonly, exported, new, post = asyncio.run(go())
|
||
assert (listing, page, readonly, exported) == (200, 200, True, 200)
|
||
assert new == 403 and post == 403
|