You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tooling.apache.org by ar...@apache.org on 2026/03/19 14:37:34 UTC
(tooling-trusted-releases) branch arm updated: Move most of the admin routes over to "typed" decorators for param validation. Add test to make sure future decorator changes don't break routing. Simplify param checking.
This is an automated email from the ASF dual-hosted git repository.
arm pushed a commit to branch arm
in repository https://gitbox.apache.org/repos/asf/tooling-trusted-releases.git
The following commit(s) were added to refs/heads/arm by this push:
new b451f601 Move most of the admin routes over to "typed" decorators for param validation. Add test to make sure future decorator changes don't break routing. Simplify param checking.
b451f601 is described below
commit b451f601081456f8f6fd60ef5da289a6830a44c3
Author: Alastair McFarlane <ar...@apache.org>
AuthorDate: Thu Mar 19 14:37:11 2026 +0000
Move most of the admin routes over to "typed" decorators for param validation. Add test to make sure future decorator changes don't break routing. Simplify param checking.
---
atr/admin/__init__.py | 409 +++++++++++++++++++++++++++++-------------
atr/blueprints/admin.py | 128 ++++++++++++-
atr/blueprints/api.py | 67 ++-----
atr/blueprints/common.py | 133 ++++++++++++--
atr/get/test.py | 7 +-
tests/unit/test_blueprints.py | 62 +++++++
6 files changed, 602 insertions(+), 204 deletions(-)
diff --git a/atr/admin/__init__.py b/atr/admin/__init__.py
index 3ceb3bf1..1ce05888 100644
--- a/atr/admin/__init__.py
+++ b/atr/admin/__init__.py
@@ -53,6 +53,7 @@ import atr.mapping as mapping
import atr.models.safe as safe
import atr.models.session
import atr.models.sql as sql
+import atr.models.unsafe as unsafe
import atr.paths as paths
import atr.principal as principal
import atr.storage as storage
@@ -126,17 +127,25 @@ class SessionDataCommon(NamedTuple):
projects: list[str]
-@admin.get("/all-releases")
-async def all_releases(session: web.Committer) -> str:
- """Display a list of all releases across all phases."""
+@admin.typed
+async def all_releases(_session: web.Committer, _all_releases: Literal["all/releases"]) -> str:
+ """
+ URL: GET /all/releases
+
+ Display a list of all releases across all phases.
+ """
async with db.session() as data:
releases = await data.release(_project=True, _committee=True).order_by(sql.Release.key).all()
return await template.render("all-releases.html", releases=releases, release_as_url=mapping.release_as_url)
-@admin.get("/browse-as")
-async def browse_as_get(session: web.Committer) -> str | web.WerkzeugResponse:
- """Allows an admin to browse as another user."""
+@admin.typed
+async def browse_as_get(_session: web.Committer, _browse_as: Literal["browse-as"]) -> str | web.WerkzeugResponse:
+ """
+ URL: GET /browse-as
+
+ Allows an admin to browse as another user.
+ """
rendered_form = form.render(
model_cls=BrowseAsUserForm,
submit_label="Browse as this user",
@@ -144,10 +153,15 @@ async def browse_as_get(session: web.Committer) -> str | web.WerkzeugResponse:
return await template.render("browse-as.html", form=rendered_form)
-@admin.post("/browse-as")
-@admin.form(BrowseAsUserForm)
-async def browse_as_post(session: web.Committer, browse_form: BrowseAsUserForm) -> str | web.WerkzeugResponse:
- """Allows an admin to browse as another user."""
+@admin.typed
+async def browse_as_post(
+ session: web.Committer, _browse_as: Literal["browse-as"], browse_form: BrowseAsUserForm
+) -> str | web.WerkzeugResponse:
+ """
+ URL: POST /browse-as
+
+ Allows an admin to browse as another user.
+ """
# We specifically need to use this on the production server
import atr.get.root as root
@@ -194,9 +208,13 @@ async def browse_as_post(session: web.Committer, browse_form: BrowseAsUserForm)
return await session.redirect(root.index)
-@admin.get("/configuration")
-async def configuration(session: web.Committer) -> web.QuartResponse:
- """Display the current application configuration values."""
+@admin.typed
+async def configuration(_session: web.Committer, _configuration: Literal["configuration"]) -> web.QuartResponse:
+ """
+ URL: GET /configuration
+
+ Display the current application configuration values.
+ """
sensitive_config_patterns = ("_PASSWORD", "_KEY", "_TOKEN", "_SECRET")
@@ -219,9 +237,13 @@ async def configuration(session: web.Committer) -> web.QuartResponse:
return web.TextResponse("\n".join(values))
-@admin.get("/consistency")
-async def consistency(session: web.Committer) -> web.TextResponse:
- """Check for consistency between the database and the filesystem."""
+@admin.typed
+async def consistency(_session: web.Committer, _consistency: Literal["consistency"]) -> web.TextResponse:
+ """
+ URL: GET /consistency
+
+ Check for consistency between the database and the filesystem.
+ """
# Get all releases from the database
async with db.session() as data:
releases = await data.release().all()
@@ -264,19 +286,34 @@ async def consistency(session: web.Committer) -> web.TextResponse:
)
-@admin.get("/data")
-async def data(session: web.Committer) -> str:
- return await _data(session, "Committee")
+@admin.typed
+async def data(session: web.Committer, _data: Literal["data"]) -> str:
+ """
+ URL: GET /data
+ """
+ return await _data_browse(session, "Committee")
-@admin.get("/data/<model>")
-async def data_model(session: web.Committer, model: str = "Committee") -> str:
- return await _data(session, model)
+@admin.typed
+async def data_model(
+ session: web.Committer, _data: Literal["data"], model: unsafe.UnsafeStr = unsafe.UnsafeStr("Committee")
+) -> str:
+ """
+ URL: GET /data/<model>
+ """
+ return await _data_browse(session, str(model))
-@admin.get("/delete-committee-keys")
-async def delete_committee_keys_get(session: web.Committer) -> str | web.WerkzeugResponse:
- """Display the form to delete committee keys."""
+
+@admin.typed
+async def delete_committee_keys_get(
+ _session: web.Committer, _delete_committee_keys: Literal["delete-committee-keys"]
+) -> str | web.WerkzeugResponse:
+ """
+ URL: GET /delete-committee-keys
+
+ Display the form to delete committee keys.
+ """
async with db.session() as data:
all_committees = await data.committee(_public_signing_keys=True).order_by(sql.Committee.key).all()
committees_with_keys = [c for c in all_committees if c.public_signing_keys]
@@ -291,12 +328,17 @@ async def delete_committee_keys_get(session: web.Committer) -> str | web.Werkzeu
return await template.render("delete-committee-keys.html", form=rendered_form)
-@admin.post("/delete-committee-keys")
-@admin.form(DeleteCommitteeKeysForm)
+@admin.typed
async def delete_committee_keys_post(
- session: web.Committer, delete_form: DeleteCommitteeKeysForm
+ session: web.Committer,
+ _delete_committee_keys: Literal["delete-committee-keys"],
+ delete_form: DeleteCommitteeKeysForm,
) -> str | web.WerkzeugResponse:
- """Delete all keys for selected committee."""
+ """
+ URL: POST /delete-committee-keys
+
+ Delete all keys for selected committee.
+ """
committee_key = delete_form.committee_key
async with db.session() as data:
@@ -336,9 +378,15 @@ async def delete_committee_keys_post(
return await session.redirect(delete_committee_keys_get)
-@admin.get("/delete-release")
-async def delete_release_get(session: web.Committer) -> str | web.WerkzeugResponse:
- """Display the form to delete releases."""
+@admin.typed
+async def delete_release_get(
+ _session: web.Committer, _delete_release_get: Literal["delete-release"]
+) -> str | web.WerkzeugResponse:
+ """
+ URL: GET /delete-release
+
+ Display the form to delete releases.
+ """
async with db.session() as data:
releases = await data.release(_project=True).order_by(sql.Release.key).all()
@@ -378,20 +426,31 @@ async def delete_release_get(session: web.Committer) -> str | web.WerkzeugRespon
return await template.render("delete-release.html", form=rendered_form)
-@admin.post("/delete-release")
-@admin.form(DeleteReleaseForm)
-async def delete_release_post(session: web.Committer, delete_form: DeleteReleaseForm) -> str | web.WerkzeugResponse:
- """Delete selected releases and their associated data and files."""
+@admin.typed
+async def delete_release_post(
+ session: web.Committer, _delete_release_get: Literal["delete-release"], delete_form: DeleteReleaseForm
+) -> str | web.WerkzeugResponse:
+ """
+ URL: POST /delete-release
+
+ Delete selected releases and their associated data and files.
+ """
await _delete_releases(session, delete_form.releases_to_delete)
return await session.redirect(delete_release_get)
-@admin.get("/delete-test-openpgp-keys")
-async def delete_test_openpgp_keys_get(session: web.Committer) -> web.Response:
- """Display the form to delete test user OpenPGP keys."""
+@admin.typed
+async def delete_test_openpgp_keys_get(
+ _session: web.Committer, _delete_test_openpgp_keys: Literal["delete-test-openpgp-keys"]
+) -> web.Response:
+ """
+ URL: GET /delete-test-openpgp-keys
+
+ Display the form to delete test user OpenPGP keys.
+ """
if not config.get().ALLOW_TESTS:
- raise base.ASFQuartException("Test operations are disabled in this environment", errorcode=403)
+ return quart.abort(404)
rendered_form = form.render(
model_cls=form.Empty,
@@ -401,12 +460,17 @@ async def delete_test_openpgp_keys_get(session: web.Committer) -> web.Response:
return web.ElementResponse(rendered_form)
-@admin.post("/delete-test-openpgp-keys")
-@admin.empty()
-async def delete_test_openpgp_keys_post(session: web.Committer) -> web.Response:
- """Delete all test user OpenPGP keys and their links."""
+@admin.typed
+async def delete_test_openpgp_keys_post(
+ session: web.Committer, _delete_test_openpgp_keys: Literal["delete-test-openpgp-keys"], _form: form.Empty
+) -> web.Response:
+ """
+ URL: POST /delete-test-openpgp-keys
+
+ Delete all test user OpenPGP keys and their links.
+ """
if not config.get().ALLOW_TESTS:
- raise base.ASFQuartException("Test operations are disabled in this environment", errorcode=403)
+ return quart.abort(404)
test_uid = "test"
try:
@@ -427,18 +491,26 @@ async def delete_test_openpgp_keys_post(session: web.Committer) -> web.Response:
return await session.redirect(get.keys.keys)
-@admin.get("/env")
-async def env(session: web.Committer) -> web.QuartResponse:
- """Display the environment variables."""
+@admin.typed
+async def env(_session: web.Committer, _env: Literal["env"]) -> web.QuartResponse:
+ """
+ URL: GET /env
+
+ Display the environment variables.
+ """
env_vars = []
for key, value in os.environ.items():
env_vars.append(f"{key}={value}")
return web.TextResponse("\n".join(env_vars))
-@admin.get("/keys/check")
-async def keys_check_get(session: web.Committer) -> web.QuartResponse:
- """Check public signing key details."""
+@admin.typed
+async def keys_check_get(_session: web.Committer, _keys_check: Literal["keys/check"]) -> web.QuartResponse:
+ """
+ URL: GET /keys/check
+
+ Check public signing key details.
+ """
rendered_form = form.render(
model_cls=form.Empty,
submit_label="Check public signing key details",
@@ -447,6 +519,7 @@ async def keys_check_get(session: web.Committer) -> web.QuartResponse:
return web.ElementResponse(rendered_form)
+# TODO: AM Why has this no empty form?
@admin.post("/keys/check")
async def keys_check_post(session: web.Committer) -> web.QuartResponse:
"""Check public signing key details."""
@@ -458,9 +531,15 @@ async def keys_check_post(session: web.Committer) -> web.QuartResponse:
return web.TextResponse(f"Exception during key check: {e!s}")
-@admin.get("/keys/regenerate-all")
-async def keys_regenerate_all_get(session: web.Committer) -> web.QuartResponse:
- """Display the form to regenerate KEYS files."""
+@admin.typed
+async def keys_regenerate_all_get(
+ _session: web.Committer, _keys_regenerate_all: Literal["keys/regenerate-all"]
+) -> web.QuartResponse:
+ """
+ URL: GET /keys/regenerate-all
+
+ Display the form to regenerate KEYS files.
+ """
rendered_form = form.render(
model_cls=form.Empty,
submit_label="Regenerate all KEYS files",
@@ -469,6 +548,7 @@ async def keys_regenerate_all_get(session: web.Committer) -> web.QuartResponse:
return web.ElementResponse(rendered_form)
+# TODO: AM Why has this no empty form?
@admin.post("/keys/regenerate-all")
async def keys_regenerate_all_post(session: web.Committer) -> web.QuartResponse:
"""Regenerate the KEYS file for all committees."""
@@ -493,9 +573,15 @@ async def keys_regenerate_all_post(session: web.Committer) -> web.QuartResponse:
return web.TextResponse("\n".join(response_lines))
-@admin.get("/keys/update")
-async def keys_update_get(session: web.Committer) -> str | web.WerkzeugResponse | tuple[Mapping[str, Any], int]:
- """Update keys from remote data."""
+@admin.typed
+async def keys_update_get(
+ _session: web.Committer, _keys_update: Literal["keys/update"]
+) -> str | web.WerkzeugResponse | tuple[Mapping[str, Any], int]:
+ """
+ URL: GET /keys/update
+
+ Update keys from remote data.
+ """
rendered_form = form.render(
model_cls=form.Empty,
submit_label="Update keys",
@@ -512,6 +598,7 @@ async def keys_update_get(session: web.Committer) -> str | web.WerkzeugResponse
return await template.render("update-keys.html", empty_form=rendered_form, previous_output=previous_output)
+# TODO: AM Why has this no empty form?
@admin.post("/keys/update")
async def keys_update_post(session: web.Committer) -> str | web.WerkzeugResponse | tuple[Mapping[str, Any], int]:
"""Update keys from remote data."""
@@ -531,8 +618,11 @@ async def keys_update_post(session: web.Committer) -> str | web.WerkzeugResponse
}, 200
-@admin.get("/ldap")
-async def ldap_get(session: web.Committer) -> str:
+@admin.typed
+async def ldap_get(session: web.Committer, _ldap: Literal["ldap"]) -> str:
+ """
+ URL: GET /ldap
+ """
rendered_form = form.render(
model_cls=LdapLookupForm,
submit_label="Lookup",
@@ -547,9 +637,11 @@ async def ldap_get(session: web.Committer) -> str:
)
-@admin.post("/ldap")
-@admin.form(LdapLookupForm)
-async def ldap_post(session: web.Committer, lookup_form: LdapLookupForm) -> str:
+@admin.typed
+async def ldap_post(session: web.Committer, _ldap: Literal["ldap"], lookup_form: LdapLookupForm) -> str:
+ """
+ URL POST /ldap
+ """
# TODO: This is one case where we should perhaps allow str | None on the form
uid_query = lookup_form.uid if lookup_form.uid else None
email_query = lookup_form.email if lookup_form.email else None
@@ -586,8 +678,11 @@ async def ldap_post(session: web.Committer, lookup_form: LdapLookupForm) -> str:
)
-@admin.get("/logs")
-async def logs(session: web.Committer) -> web.QuartResponse:
+@admin.typed
+async def logs(_session: web.Committer, _logs: Literal["logs"]) -> web.QuartResponse:
+ """
+ URL: GET /logs
+ """
_require_debug_and_allow_tests()
recent_logs = log.get_recent_logs()
if recent_logs is None:
@@ -595,29 +690,35 @@ async def logs(session: web.Committer) -> web.QuartResponse:
return web.TextResponse("\n".join(recent_logs))
-@admin.get("/ongoing-tasks/<project_key>/<version_key>/<revision>")
+@admin.typed
async def ongoing_tasks_get(
- session: web.Committer, project_key: str, version_key: str, revision: str
+ session: web.Committer,
+ _ongoing_tasks: Literal["ongoing-tasks"],
+ project_key: safe.ProjectKey,
+ version_key: safe.VersionKey,
+ revision: safe.RevisionNumber,
) -> web.QuartResponse:
- project = safe.ProjectKey(project_key)
- version = safe.VersionKey(version_key)
- revision_number = safe.RevisionNumber(revision)
- return await _ongoing_tasks(session, project, version, revision_number)
+ """
+ URL: GET /ongoing-tasks
+ """
+ return await _fetch_ongoing_tasks(session, project_key, version_key, revision)
+# TODO: AM Why has this no empty form?
@admin.post("/ongoing-tasks/<project_key>/<version_key>/<revision>")
async def ongoing_tasks_post(
- session: web.Committer, project_key: str, version_key: str, revision: str
+ session: web.Committer, project_key: safe.ProjectKey, version_key: safe.VersionKey, revision: safe.RevisionNumber
) -> web.QuartResponse:
- project = safe.ProjectKey(project_key)
- version = safe.VersionKey(version_key)
- revision_number = safe.RevisionNumber(revision)
- return await _ongoing_tasks(session, project, version, revision_number)
+ return await _fetch_ongoing_tasks(session, project_key, version_key, revision)
+
+@admin.typed
+async def performance(_session: web.Committer, _performance: Literal["performance"]) -> str:
+ """
+ URL: GET /performance
-@admin.get("/performance")
-async def performance(session: web.Committer) -> str:
- """Display performance statistics for all routes."""
+ Display performance statistics for all routes.
+ """
app = asfquart.APP
if app is ...:
@@ -696,9 +797,15 @@ async def performance(session: web.Committer) -> str:
return await template.render("performance.html", stats=sorted_summary)
-@admin.get("/projects/update")
-async def projects_update_get(session: web.Committer) -> str | web.WerkzeugResponse | tuple[Mapping[str, Any], int]:
- """Update projects from remote data."""
+@admin.typed
+async def projects_update_get(
+ _session: web.Committer, _projects_update: Literal["projects/update"]
+) -> str | web.WerkzeugResponse | tuple[Mapping[str, Any], int]:
+ """
+ URL: GET /projects/update
+
+ Update projects from remote data.
+ """
rendered_form = form.render(
model_cls=form.Empty,
submit_label="Update projects",
@@ -708,6 +815,7 @@ async def projects_update_get(session: web.Committer) -> str | web.WerkzeugRespo
return await template.render("update-projects.html", empty_form=rendered_form)
+# TODO: AM Why has this no empty form?
@admin.post("/projects/update")
async def projects_update_post(session: web.Committer) -> str | web.WerkzeugResponse | tuple[Mapping[str, Any], int]:
"""Update projects from remote data."""
@@ -725,14 +833,21 @@ async def projects_update_post(session: web.Committer) -> str | web.WerkzeugResp
}, 200
-@admin.get("/raise-error")
-async def raise_error(session: web.Committer) -> web.QuartResponse:
+@admin.typed
+async def raise_error(_session: web.Committer, _raise_error: Literal["raise-error"]) -> web.QuartResponse:
+ """
+ URL: GET /raise-error
+ """
raise RuntimeError("Admin test route deliberately raised an unhandled error")
-@admin.get("/revoke-user-tokens")
-async def revoke_user_tokens_get(session: web.Committer) -> str:
- """Revoke all Personal Access Tokens for a specified user."""
+@admin.typed
+async def revoke_user_tokens_get(_session: web.Committer, _revoke_user_tokens: Literal["revoke-user-tokens"]) -> str:
+ """
+ URL: GET /revoke-user-tokens
+
+ Revoke all Personal Access Tokens for a specified user.
+ """
token_counts: list[tuple[str, int]] = []
async with db.session() as data:
stmt = (
@@ -757,12 +872,15 @@ async def revoke_user_tokens_get(session: web.Committer) -> str:
)
-@admin.post("/revoke-user-tokens")
-@admin.form(RevokeUserTokensForm)
+@admin.typed
async def revoke_user_tokens_post(
- session: web.Committer, revoke_form: RevokeUserTokensForm
+ session: web.Committer, _revoke_user_tokens: Literal["revoke-user-tokens"], revoke_form: RevokeUserTokensForm
) -> str | web.WerkzeugResponse:
- """Revoke all Personal Access Tokens for a specified user."""
+ """
+ URL: POST /revoke-user-tokens
+
+ Revoke all Personal Access Tokens for a specified user.
+ """
target_uid = revoke_form.asf_uid
async with storage.write(session) as write:
@@ -777,8 +895,11 @@ async def revoke_user_tokens_post(
return await session.redirect(revoke_user_tokens_get)
-@admin.get("/rotate-jwt-key")
-async def rotate_jwt_key_get(session: web.Committer) -> str:
+@admin.typed
+async def rotate_jwt_key_get(_session: web.Committer, _rotate_jwt_key: Literal["rotate-jwt-key"]) -> str:
+ """
+ URL: GET /rotate-jwt-key
+ """
rendered_form = form.render(
model_cls=RotateJwtKeyForm,
submit_label="Rotate JWT key",
@@ -786,9 +907,13 @@ async def rotate_jwt_key_get(session: web.Committer) -> str:
return await _rotate_jwt_key_page(rendered_form)
-@admin.post("/rotate-jwt-key")
-@admin.form(RotateJwtKeyForm)
-async def rotate_jwt_key_post(session: web.Committer, _rotate_form: RotateJwtKeyForm) -> str | web.WerkzeugResponse:
+@admin.typed
+async def rotate_jwt_key_post(
+ session: web.Committer, _rotate_jwt_key: Literal["rotate-jwt-key"], _rotate_form: RotateJwtKeyForm
+) -> str | web.WerkzeugResponse:
+ """
+ URL: POST /rotate-jwt-key
+ """
async with storage.write(session) as write:
wafa = write.as_foundation_admin()
await wafa.tokens.rotate_jwt_signing_key()
@@ -796,16 +921,23 @@ async def rotate_jwt_key_post(session: web.Committer, _rotate_form: RotateJwtKey
return await session.redirect(rotate_jwt_key_get)
-@admin.get("/task-times/<project_key>/<version_key>/<revision_number>")
+@admin.typed
async def task_times(
- session: web.Committer, project_key: str, version_key: str, revision_number: str
+ _session: web.Committer,
+ _task_times: Literal["task-times"],
+ project_key: safe.ProjectKey,
+ version_key: safe.VersionKey,
+ revision_number: safe.RevisionNumber,
) -> web.QuartResponse:
+ """
+ URL: GET /task-times/<project_key>/<version_key>/<revision_number>
+ """
values = []
async with db.session() as data:
tasks = await data.task(
- project_key=project_key,
- version_key=version_key,
- revision_number=revision_number,
+ project_key=str(project_key),
+ version_key=str(version_key),
+ revision_number=str(revision_number),
).all()
for task in tasks:
if (task.started is None) or (task.completed is None):
@@ -816,9 +948,13 @@ async def task_times(
return web.TextResponse("\n".join(values))
-@admin.get("/tasks/recent/<int:minutes>")
-async def tasks_recent(session: web.Committer, minutes: int) -> str:
- """Display tasks from the last N minutes."""
+@admin.typed
+async def tasks_recent(_session: web.Committer, _tasks_recent: Literal["tasks/recent"], minutes: int) -> str:
+ """
+ URL: GET /tasks/recent/<int:minutes>
+
+ Display tasks from the last N minutes.
+ """
if minutes < 1:
minutes = 1
if minutes > 1440:
@@ -929,9 +1065,13 @@ async def tasks_recent(session: web.Committer, minutes: int) -> str:
return await template.blank(f"Recent Tasks ({minutes}m)", content=page.collect())
-@admin.get("/toggle-view")
-async def toggle_view_get(session: web.Committer) -> str:
- """Display the page with a button to toggle between admin and user views."""
+@admin.typed
+async def toggle_view_get(_session: web.Committer, _toggle_view: Literal["toggle-view"]) -> str:
+ """
+ URL: GET /toggle-view
+
+ Display the page with a button to toggle between admin and user views.
+ """
rendered_form = form.render(
model_cls=form.Empty,
submit_label="Toggle view",
@@ -941,10 +1081,15 @@ async def toggle_view_get(session: web.Committer) -> str:
return await template.render("toggle-admin-view.html", empty_form=rendered_form)
-@admin.post("/toggle-view")
-@admin.empty()
-async def toggle_view_post(session: web.Committer) -> web.WerkzeugResponse:
- """Toggle between admin and user views."""
+@admin.typed
+async def toggle_view_post(
+ _session: web.Committer, _toggle_view: Literal["toggle-view"], _form: form.Empty
+) -> web.WerkzeugResponse:
+ """
+ URL: POST /toggle-view
+
+ Toggle between admin and user views.
+ """
app = asfquart.APP
if (not hasattr(app, "app_id")) or (not isinstance(app.app_id, str)):
raise TypeError("Internal error: APP has no valid app_id")
@@ -962,9 +1107,13 @@ async def toggle_view_post(session: web.Committer) -> web.WerkzeugResponse:
return quart.redirect("https://" + quart.request.host + "/")
-@admin.get("/validate")
-async def validate_(session: web.Committer) -> str:
- """Run validators and display any divergences."""
+@admin.typed
+async def validate_(_session: web.Committer, _validate: Literal["validate"]) -> str:
+ """
+ URL: GET /validate
+
+ Run validators and display any divergences.
+ """
async with db.session() as data:
divergences = [d async for d in validate.everything(data)]
@@ -975,9 +1124,15 @@ async def validate_(session: web.Committer) -> str:
)
-@admin.get("/validate-jwt")
-async def validate_jwt_get(session: web.Committer) -> str:
- _require_debug_and_allow_tests()
+@admin.typed
+async def validate_jwt_get(_session: web.Committer, _validate_jwt: Literal["validate-jwt"]) -> str:
+ """
+ URL: GET /validate-jwt
+ """
+ try:
+ _require_debug_and_allow_tests()
+ except base.ASFQuartException:
+ return quart.abort(404)
rendered_form = form.render(
model_cls=ValidateJwtForm,
submit_label="Validate JWT",
@@ -986,10 +1141,18 @@ async def validate_jwt_get(session: web.Committer) -> str:
return await _validate_jwt_page(rendered_form, result=None)
-@admin.post("/validate-jwt")
-@admin.form(ValidateJwtForm)
-async def validate_jwt_post(session: web.Committer, validate_form: ValidateJwtForm) -> str:
- _require_debug_and_allow_tests()
+@admin.typed
+async def validate_jwt_post(
+ _session: web.Committer, _validate_jwt: Literal["validate-jwt"], validate_form: ValidateJwtForm
+) -> str:
+ """
+ URL: POST /validate-jwt
+ """
+ try:
+ _require_debug_and_allow_tests()
+ except base.ASFQuartException:
+ return quart.abort(404)
+
token = validate_form.token
result: dict[str, Any] = {"token_length": len(token), "valid": False}
@@ -1044,7 +1207,7 @@ async def _check_keys(fix: bool = False) -> str:
return message
-async def _data(session: web.Committer, model: str = "Committee") -> str:
+async def _data_browse(_session: web.Committer, model: str = "Committee") -> str:
"""Browse all records in the database."""
async with db.session() as data:
# Map of model names to their classes
@@ -1175,8 +1338,8 @@ async def _get_filesystem_dirs_unfinished(filesystem_dirs: list[str]) -> None:
filesystem_dirs.append(version_dir_path)
-async def _ongoing_tasks(
- session: web.Committer,
+async def _fetch_ongoing_tasks(
+ _session: web.Committer,
project_key: safe.ProjectKey,
version_key: safe.VersionKey,
revision: safe.RevisionNumber,
diff --git a/atr/blueprints/admin.py b/atr/blueprints/admin.py
index a6f93ae9..5ef56003 100644
--- a/atr/blueprints/admin.py
+++ b/atr/blueprints/admin.py
@@ -14,23 +14,27 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-
import json
+import time
from collections.abc import Awaitable, Callable
from types import ModuleType
-from typing import Any
+from typing import Any, Concatenate, overload
import asfquart.base as base
import asfquart.session
import pydantic
import quart
+import quart_schema
+import atr.blueprints.common as common
import atr.form
+import atr.log as log
import atr.user as user
import atr.web as web
_BLUEPRINT_NAME = "admin_blueprint"
_BLUEPRINT = quart.Blueprint(_BLUEPRINT_NAME, __name__, url_prefix="/admin", template_folder="../admin/templates")
+_routes: list[str] = []
def empty() -> Callable[[Callable[..., Awaitable[Any]]], Callable[..., Awaitable[Any]]]:
@@ -131,7 +135,125 @@ def register(app: base.QuartApp) -> tuple[ModuleType, list[str]]:
import atr.admin as admin
app.register_blueprint(_BLUEPRINT)
- return admin, []
+ return admin, _routes
+
+
+@overload
+def typed[**P, R](func: Callable[Concatenate[web.Committer, P], Awaitable[R]]) -> web.RouteFunction[R]: ...
+
+
+@overload
+def typed[**P, R](func: Callable[Concatenate[web.Public, P], Awaitable[R]]) -> web.RouteFunction[R]: ... # pyright: ignore[reportOverlappingOverload]
+
+
+def typed(func: Callable[..., Any]) -> Callable[..., Any]:
+ """Decorator that derives the URL path from the function's type annotations.
+
+ - Literal["..."] parameters become literal path segments
+ - safe.ProjectName / safe.VersionName parameters are validated via cache/DB
+ - pydantic.BaseModel subclass parameters are parsed from the JSON request body
+ - dataclass parameters are parsed from the query string
+ - str | None parameters create optional URL segments (two routes registered)
+ - int, float, str use Quart's built-in type converters
+ - HTTP method is POST if a body param is present, GET otherwise
+ """
+ path, validated_params, literal_params, body_param, form_param, query_param, optional_params = (
+ common.build_api_path(func)
+ )
+ method = "POST" if (body_param is not None or form_param is not None) else "GET"
+ body_safe_params = common.safe_params_for_type(body_param[1]) if body_param is not None else []
+ form_safe_params = common.safe_params_for_type(form_param[1]) if form_param is not None else []
+ query_safe_params = common.safe_params_for_type(query_param[1]) if query_param is not None else []
+
+ async def wrapper(*_args: Any, **kwargs: Any) -> Any:
+ enhanced_session = await common.authenticate()
+ await common.validate_params(kwargs, validated_params)
+ kwargs.update(literal_params)
+
+ if body_param is not None:
+ await common.parse_body(body_param, body_safe_params, kwargs)
+
+ if form_param is not None:
+ form_param_name, form_cls = form_param
+ context: dict[str, Any] = {"kwargs": kwargs, "session": enhanced_session}
+ try:
+ kwargs[form_param_name] = await enhanced_session.form_validate(form_cls, context)
+ except pydantic.ValidationError as e:
+ errors = e.errors()
+ if len(errors) == 0:
+ raise RuntimeError("Validation failed, but no errors were reported")
+ form_data_raw = await atr.form.quart_request()
+ flash_data = atr.form.flash_error_data(form_cls, errors, form_data_raw)
+ summary = atr.form.flash_error_summary(errors, flash_data)
+ await quart.flash(summary, category="error")
+ await quart.flash(json.dumps(flash_data), category="form-error-data")
+ return quart.redirect(quart.request.path)
+ if form_safe_params:
+ await common.validate_safe_fields(kwargs[form_param_name], form_safe_params, kwargs)
+
+ if query_param is not None:
+ await common.parse_query(query_param, query_safe_params, kwargs)
+
+ start_time_ns = time.perf_counter_ns()
+ response = await func(enhanced_session, **kwargs)
+ end_time_ns = time.perf_counter_ns()
+ total_ms = (end_time_ns - start_time_ns) // 1_000_000
+ log.performance(f"API {method} {path} {func.__name__} = 0 0 {total_ms}")
+
+ return response
+
+ endpoint = func.__module__.replace(".", "_") + "_" + func.__name__
+ wrapper.__name__ = func.__name__
+ wrapper.__doc__ = func.__doc__
+ wrapper.__annotations__["endpoint"] = _BLUEPRINT_NAME + "." + endpoint
+
+ # Replace the original quart request decorators
+ if query_param is not None:
+ wrapper = quart_schema.validate_querystring(query_param[1])(wrapper)
+ if body_param is not None:
+ wrapper = quart_schema.validate_request(body_param[1])(wrapper)
+
+ # Examine `func` for quart attributes and re-attach to the wrapped function
+ # This makes sure the OpenAPI documentation is preserved
+ # Note: we don't update querystring or request as they're processed above using our detected types
+ _copy_quart_attributes(func, wrapper)
+
+ # If there are optional params, we need two routes, one with the optional params omitted
+ # and one with them all present.
+ # AM 26/03/03: This actually only handles the case where there's some required and a single optional, but
+ # that's the only case that existed in the original code. Theoretically we could count the optional params and
+ # generate the correct number of routes, but that's lot of effort for little gain right now
+ _add_url_rules(wrapper, path, endpoint, method, optional_params)
+
+ common.register_route(func, "api", _routes)
+ return wrapper
+
+
+def _add_url_rules(
+ wrapper: Callable[..., Any],
+ path: str,
+ endpoint: str,
+ method: str,
+ optional_params: list[str],
+) -> None:
+ """Register URL rules for the wrapper, handling optional path params with a default short route."""
+ if optional_params:
+ required_segments = [
+ seg for seg in path.strip("/").split("/") if not any(seg == f"<{name}>" for name in optional_params)
+ ]
+ short_path = "/" + "/".join(required_segments)
+ defaults = {name: None for name in optional_params}
+ _BLUEPRINT.add_url_rule(short_path, endpoint=endpoint, view_func=wrapper, methods=[method], defaults=defaults)
+ _BLUEPRINT.add_url_rule(path, endpoint=endpoint + "_full", view_func=wrapper, methods=[method])
+ else:
+ _BLUEPRINT.add_url_rule(path, endpoint=endpoint, view_func=wrapper, methods=[method])
+
+
+def _copy_quart_attributes(src: Callable[..., Any], dst: Callable[..., Any]) -> None:
+ """Copy quart schema attributes from src to dst to preserve OpenAPI documentation."""
+ for attr in common.QUART_ATTRIBUTES:
+ if hasattr(src, attr):
+ setattr(dst, attr, getattr(src, attr))
@_BLUEPRINT.before_request
diff --git a/atr/blueprints/api.py b/atr/blueprints/api.py
index f04428cb..e335072b 100644
--- a/atr/blueprints/api.py
+++ b/atr/blueprints/api.py
@@ -14,12 +14,11 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-import dataclasses
import datetime
import inspect
import sys
import time
-from collections.abc import Callable
+from collections.abc import Awaitable, Callable
from types import ModuleType
from typing import Any
@@ -33,19 +32,11 @@ import werkzeug.exceptions as exceptions
import atr.blueprints.common as common
import atr.log as log
+import atr.web as web
_BLUEPRINT_NAME = "api_blueprint"
_BLUEPRINT = quart.Blueprint(_BLUEPRINT_NAME, __name__, url_prefix="/api")
_routes: list[str] = []
-_QUART_ATTRIBUTES = [
- quart_schema.validation.QUART_SCHEMA_HEADERS_ATTRIBUTE,
- quart_schema.validation.QUART_SCHEMA_RESPONSE_ATTRIBUTE,
- quart_schema.openapi.QUART_SCHEMA_SECURITY_ATTRIBUTE,
- quart_schema.openapi.QUART_SCHEMA_TAG_ATTRIBUTE,
- quart_schema.openapi.QUART_SCHEMA_HIDDEN_ATTRIBUTE,
- quart_schema.openapi.QUART_SCHEMA_DEPRECATED_ATTRIBUTE,
- quart_schema.openapi.QUART_SCHEMA_OPERATION_ID_ATTRIBUTE,
-]
def register(app: base.QuartApp) -> tuple[ModuleType, list[str]]:
@@ -55,9 +46,10 @@ def register(app: base.QuartApp) -> tuple[ModuleType, list[str]]:
return api, _routes
-def typed(func: Callable[..., Any]) -> Callable[..., Any]:
+def typed(func: Callable[..., Awaitable[Any]]) -> web.RouteFunction[Any]:
"""Decorator that derives the URL path from the function's type annotations.
+ - Arguments after session are joined with / to make the web path
- Literal["..."] parameters become literal path segments
- safe.ProjectName / safe.VersionName parameters are validated via cache/DB
- pydantic.BaseModel subclass parameters are parsed from the JSON request body
@@ -67,7 +59,9 @@ def typed(func: Callable[..., Any]) -> Callable[..., Any]:
- HTTP method is POST if a body param is present, GET otherwise
"""
original = inspect.unwrap(func)
- path, validated_params, literal_params, body_param, query_param, optional_params = common.build_api_path(original)
+ path, validated_params, literal_params, body_param, _, query_param, optional_params = common.build_api_path(
+ original
+ )
method = "POST" if body_param is not None else "GET"
body_safe_params = common.safe_params_for_type(body_param[1]) if body_param is not None else []
query_safe_params = common.safe_params_for_type(query_param[1]) if query_param is not None else []
@@ -77,22 +71,10 @@ def typed(func: Callable[..., Any]) -> Callable[..., Any]:
kwargs.update(literal_params)
if body_param is not None:
- body_name, body_cls = body_param
- json_data = await quart.request.get_json()
- try:
- body_instance = body_cls.model_validate(json_data)
- except pydantic.ValidationError as e:
- raise quart_schema.RequestSchemaValidationError(e) from e
- if body_safe_params:
- await common.validate_safe_fields(body_instance, body_safe_params, kwargs)
- kwargs[body_name] = body_instance
+ await common.parse_body(body_param, body_safe_params, kwargs)
if query_param is not None:
- query_name, query_cls = query_param
- query_instance = _parse_query_args(query_cls, quart.request.args)
- if query_safe_params:
- await common.validate_safe_fields(query_instance, query_safe_params, kwargs)
- kwargs[query_name] = query_instance
+ await common.parse_query(query_param, query_safe_params, kwargs)
start_time_ns = time.perf_counter_ns()
response = await func(**kwargs)
@@ -157,7 +139,7 @@ def _add_url_rules(
def _copy_quart_attributes(src: Callable[..., Any], dst: Callable[..., Any]) -> None:
"""Copy quart schema attributes from src to dst to preserve OpenAPI documentation."""
- for attr in _QUART_ATTRIBUTES:
+ for attr in common.QUART_ATTRIBUTES:
if hasattr(src, attr):
setattr(dst, attr, getattr(src, attr))
@@ -197,35 +179,6 @@ async def _handle_request_validation(err: quart_schema.RequestSchemaValidationEr
return _json_error("Input validation failed", 400, {"validation_details": verr.errors()})
-def _coerce_query_field(raw: str, field_type: Any, field_name: str) -> Any:
- """Coerce a raw query string value to the expected field type."""
- if field_type is str or field_type == "str":
- return raw
- if field_type is int or field_type == "int":
- try:
- return int(raw)
- except ValueError:
- raise exceptions.BadRequest(f"Query parameter {field_name!r} must be an integer")
- if field_type is bool or field_type == "bool":
- return raw.lower() in ("true", "1", "yes")
- return raw
-
-
-def _parse_query_args(query_cls: type, args: Any) -> Any:
- """Parse query string parameters into a dataclass instance."""
- field_values: dict[str, Any] = {}
- for field in dataclasses.fields(query_cls):
- raw = args.get(field.name)
- if raw is None:
- if field.default is not dataclasses.MISSING:
- field_values[field.name] = field.default
- elif field.default_factory is not dataclasses.MISSING:
- field_values[field.name] = field.default_factory()
- continue
- field_values[field.name] = _coerce_query_field(raw, field.type, field.name)
- return query_cls(**field_values)
-
-
def _json_error(
message: str, status_code: int | None, extra: dict[str, Any] | None = None
) -> tuple[quart.Response, int]:
diff --git a/atr/blueprints/common.py b/atr/blueprints/common.py
index ee69be5a..8457a9ab 100644
--- a/atr/blueprints/common.py
+++ b/atr/blueprints/common.py
@@ -25,6 +25,9 @@ from typing import Annotated, Any, Literal, TypeAliasType, get_args, get_origin,
import asfquart.base as base
import asfquart.session
import pydantic
+import quart
+import quart_schema
+import werkzeug.exceptions as exceptions
import atr.form as form
import atr.ldap as ldap
@@ -89,7 +92,7 @@ def build_path(
segments: list[str] = []
validated_params: list[tuple[str, type]] = []
literal_params: dict[str, str] = {}
- form_param: tuple[str, type] | None = None
+ unique = _UniqueParams()
for ix, param_name in enumerate(params):
hint = hints.get(param_name)
@@ -102,16 +105,13 @@ def build_path(
public = hint is web.Public
continue
- if _is_form_type(hint):
- if form_param is not None:
- raise TypeError(f"Parameter {param_name!r} in {func.__name__}: only one Form is allowed")
- form_param = (param_name, hint)
+ if unique.check(hint, param_name, func.__name__):
continue
_classify_url_param(param_name, hint, func.__name__, segments, validated_params, literal_params)
path = "/" + "/".join(segments)
- return path, validated_params, literal_params, form_param, public
+ return path, validated_params, literal_params, unique.form, public
def build_api_path(
@@ -122,6 +122,7 @@ def build_api_path(
dict[str, str],
tuple[str, type[pydantic.BaseModel]] | None,
tuple[str, type] | None,
+ tuple[str, type] | None,
list[str],
]:
"""Inspect a function's type hints to build a URL path for an API route.
@@ -144,25 +145,20 @@ def build_api_path(
segments: list[str] = []
validated_params: list[tuple[str, type]] = []
literal_params: dict[str, str] = {}
- body_param: tuple[str, type[pydantic.BaseModel]] | None = None
- query_param: tuple[str, type] | None = None
+ unique = _UniqueParams()
optional_params: list[str] = []
- for param_name in params:
+ for ix, param_name in enumerate(params):
hint = hints.get(param_name)
if hint is None:
raise TypeError(f"Parameter {param_name!r} in {func.__name__} has no type annotation")
- if _is_body_type(hint):
- if body_param is not None:
- raise TypeError(f"Parameter {param_name!r} in {func.__name__}: only one body type is allowed")
- body_param = (param_name, hint)
+ if (hint is web.Public) or (hint is web.Committer):
+ if ix != 0:
+ raise TypeError(f"Parameter {param_name!r} in {func.__name__} must be first")
continue
- if _is_query_type(hint):
- if query_param is not None:
- raise TypeError(f"Parameter {param_name!r} in {func.__name__}: only one query type is allowed")
- query_param = (param_name, hint)
+ if unique.check(hint, param_name, func.__name__):
continue
inner, is_optional = _unwrap_optional(hint)
@@ -175,7 +171,7 @@ def build_api_path(
_classify_url_param(param_name, hint, func.__name__, segments, validated_params, literal_params)
path = "/" + "/".join(segments)
- return path, validated_params, literal_params, body_param, query_param, optional_params
+ return path, validated_params, literal_params, unique.body, unique.form, unique.query, optional_params
def register_route(func: Callable[..., Any], prefix: str, routes: list[str]) -> None:
@@ -236,6 +232,96 @@ async def validate_safe_fields(
setattr(instance, name, temp[name])
+async def parse_body(
+ body_param: tuple[str, type[pydantic.BaseModel]],
+ safe_params: list[tuple[str, type]],
+ kwargs: dict[str, Any],
+) -> None:
+ """Parse and validate a JSON body parameter, adding it to kwargs."""
+ body_name, body_cls = body_param
+ json_data = await quart.request.get_json()
+ try:
+ body_instance = body_cls.model_validate(json_data)
+ except pydantic.ValidationError as e:
+ raise quart_schema.RequestSchemaValidationError(e) from e
+ if safe_params:
+ await validate_safe_fields(body_instance, safe_params, kwargs)
+ kwargs[body_name] = body_instance
+
+
+async def parse_query(
+ query_param: tuple[str, type],
+ safe_params: list[tuple[str, type]],
+ kwargs: dict[str, Any],
+) -> None:
+ """Parse and validate query string parameters, adding them to kwargs."""
+ query_name, query_cls = query_param
+ query_instance = _parse_query_args(query_cls, quart.request.args)
+ if safe_params:
+ await validate_safe_fields(query_instance, safe_params, kwargs)
+ kwargs[query_name] = query_instance
+
+
+def _coerce_query_field(raw: str, field_type: Any, field_name: str) -> Any:
+ """Coerce a raw query string value to the expected field type."""
+ if field_type is str or field_type == "str":
+ return raw
+ if field_type is int or field_type == "int":
+ try:
+ return int(raw)
+ except ValueError:
+ raise exceptions.BadRequest(f"Query parameter {field_name!r} must be an integer")
+ if field_type is bool or field_type == "bool":
+ return raw.lower() in ("true", "1", "yes")
+ return raw
+
+
+def _parse_query_args(query_cls: type, args: Any) -> Any:
+ """Parse query string parameters into a dataclass instance."""
+ field_values: dict[str, Any] = {}
+ for field in dataclasses.fields(query_cls):
+ raw = args.get(field.name)
+ if raw is None:
+ if field.default is not dataclasses.MISSING:
+ field_values[field.name] = field.default
+ elif field.default_factory is not dataclasses.MISSING:
+ field_values[field.name] = field.default_factory()
+ continue
+ field_values[field.name] = _coerce_query_field(raw, field.type, field.name)
+ return query_cls(**field_values)
+
+
+@dataclasses.dataclass
+class _UniqueParams:
+ """Tracks the at-most-one body, form, and query parameters during path building."""
+
+ body: tuple[str, type[pydantic.BaseModel]] | None = None
+ form: tuple[str, type] | None = None
+ query: tuple[str, type] | None = None
+
+ def check(self, hint: Any, param_name: str, func_name: str) -> bool:
+ """If hint is a body/form/query type, store it (ensuring uniqueness). Return True if matched."""
+ if _is_body_type(hint):
+ if self.body is not None:
+ raise TypeError(f"Parameter {param_name!r} in {func_name}: only one body type is allowed")
+ self.body = (param_name, hint)
+ return True
+
+ if _is_form_type(hint):
+ if self.form is not None:
+ raise TypeError(f"Parameter {param_name!r} in {func_name}: only one Form is allowed")
+ self.form = (param_name, hint)
+ return True
+
+ if _is_query_type(hint):
+ if self.query is not None:
+ raise TypeError(f"Parameter {param_name!r} in {func_name}: only one query type is allowed")
+ self.query = (param_name, hint)
+ return True
+
+ return False
+
+
def _classify_url_param(
param_name: str,
hint: Any,
@@ -309,3 +395,14 @@ def _unwrap_optional(hint: Any) -> tuple[Any, bool]:
if len(non_none) == 1 and type(None) in args:
return non_none[0], True
return hint, False
+
+
+QUART_ATTRIBUTES = [
+ quart_schema.validation.QUART_SCHEMA_HEADERS_ATTRIBUTE,
+ quart_schema.validation.QUART_SCHEMA_RESPONSE_ATTRIBUTE,
+ quart_schema.openapi.QUART_SCHEMA_SECURITY_ATTRIBUTE,
+ quart_schema.openapi.QUART_SCHEMA_TAG_ATTRIBUTE,
+ quart_schema.openapi.QUART_SCHEMA_HIDDEN_ATTRIBUTE,
+ quart_schema.openapi.QUART_SCHEMA_DEPRECATED_ATTRIBUTE,
+ quart_schema.openapi.QUART_SCHEMA_OPERATION_ID_ATTRIBUTE,
+]
diff --git a/atr/get/test.py b/atr/get/test.py
index d5d98521..f1ccba41 100644
--- a/atr/get/test.py
+++ b/atr/get/test.py
@@ -21,6 +21,7 @@ from typing import Literal
import aiofiles
import asfquart.base as base
+import quart
import werkzeug.wrappers.response as response
import atr.blueprints.get as get
@@ -68,7 +69,7 @@ async def test_login(_session: web.Public, _test_login: Literal["test/login"]) -
URL: /test/login
"""
if not config.get().ALLOW_TESTS:
- return await web.redirect(root.notfound)
+ return quart.abort(404)
session_data = atr.models.session.CookieData(
uid="test",
@@ -96,7 +97,7 @@ async def test_merge(
URL: /test/merge/<project_key>/<version_key>
"""
if not config.get().ALLOW_TESTS:
- raise base.ASFQuartException("Test routes not enabled", errorcode=404)
+ return quart.abort(404)
async with storage.write(session) as write_n:
wacp_n = await write_n.as_project_committee_participant(project_key)
@@ -213,7 +214,7 @@ async def test_vote(
URL: /test/vote/<category>/<project_key>/<version_key>
"""
if not config.get().ALLOW_TESTS:
- raise base.ASFQuartException("Test routes not enabled", errorcode=404)
+ return quart.abort(404)
category_map = {
"unauthenticated": vote.UserCategory.UNAUTHENTICATED,
diff --git a/tests/unit/test_blueprints.py b/tests/unit/test_blueprints.py
new file mode 100644
index 00000000..af268bfe
--- /dev/null
+++ b/tests/unit/test_blueprints.py
@@ -0,0 +1,62 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import asfquart
+import pytest
+
+import atr.blueprints as blueprints
+import atr.util as util
+
+_TESTED_BLUEPRINTS = frozenset({"get_blueprint", "post_blueprint", "admin_blueprint"})
+
+
+@pytest.mark.asyncio
+async def test_all_routes_support_url_construction(monkeypatch):
+ # Prevent writing routes.json to a real directory
+ monkeypatch.setattr("atr.blueprints._export_routes", lambda _: None)
+ # We don't need a functional .APP for this test but we do need it reset afterwards
+ monkeypatch.setattr("asfquart.APP", None)
+
+ app = asfquart.construct("test")
+ blueprints.register(app)
+
+ failures: list[str] = []
+
+ async with app.test_request_context("/"):
+ for rule in app.url_map.iter_rules():
+ blueprint_name = rule.endpoint.rsplit(".", 1)[0] if "." in rule.endpoint else None
+ if blueprint_name not in _TESTED_BLUEPRINTS:
+ continue
+
+ view_func = app.view_functions[rule.endpoint]
+
+ # Build dummy kwargs so url_for can construct the URL
+ kwargs = {}
+ for arg in rule.arguments:
+ converter = rule._converters.get(arg)
+ if converter is not None and type(converter).__name__ == "IntegerConverter":
+ kwargs[arg] = 1
+ else:
+ kwargs[arg] = "test"
+
+ try:
+ util.as_url(view_func, **kwargs)
+ except Exception as e:
+ failures.append(f"{rule.endpoint} ({rule.rule}): {e}")
+
+ if failures:
+ raise AssertionError("Routes incompatible with as_url:\n" + "\n".join(failures))
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org