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 <[email protected]>
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]
[email protected]("/all-releases")
-async def all_releases(session: web.Committer) -> str:
- """Display a list of all releases across all phases."""
[email protected]
+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)
[email protected]("/browse-as")
-async def browse_as_get(session: web.Committer) -> str | web.WerkzeugResponse:
- """Allows an admin to browse as another user."""
[email protected]
+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)
[email protected]("/browse-as")
[email protected](BrowseAsUserForm)
-async def browse_as_post(session: web.Committer, browse_form:
BrowseAsUserForm) -> str | web.WerkzeugResponse:
- """Allows an admin to browse as another user."""
[email protected]
+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)
[email protected]("/configuration")
-async def configuration(session: web.Committer) -> web.QuartResponse:
- """Display the current application configuration values."""
[email protected]
+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))
[email protected]("/consistency")
-async def consistency(session: web.Committer) -> web.TextResponse:
- """Check for consistency between the database and the filesystem."""
[email protected]
+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:
)
[email protected]("/data")
-async def data(session: web.Committer) -> str:
- return await _data(session, "Committee")
[email protected]
+async def data(session: web.Committer, _data: Literal["data"]) -> str:
+ """
+ URL: GET /data
+ """
+ return await _data_browse(session, "Committee")
[email protected]("/data/<model>")
-async def data_model(session: web.Committer, model: str = "Committee") -> str:
- return await _data(session, model)
[email protected]
+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))
[email protected]("/delete-committee-keys")
-async def delete_committee_keys_get(session: web.Committer) -> str |
web.WerkzeugResponse:
- """Display the form to delete committee keys."""
+
[email protected]
+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)
[email protected]("/delete-committee-keys")
[email protected](DeleteCommitteeKeysForm)
[email protected]
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)
[email protected]("/delete-release")
-async def delete_release_get(session: web.Committer) -> str |
web.WerkzeugResponse:
- """Display the form to delete releases."""
[email protected]
+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)
[email protected]("/delete-release")
[email protected](DeleteReleaseForm)
-async def delete_release_post(session: web.Committer, delete_form:
DeleteReleaseForm) -> str | web.WerkzeugResponse:
- """Delete selected releases and their associated data and files."""
[email protected]
+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)
[email protected]("/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."""
[email protected]
+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)
[email protected]("/delete-test-openpgp-keys")
[email protected]()
-async def delete_test_openpgp_keys_post(session: web.Committer) ->
web.Response:
- """Delete all test user OpenPGP keys and their links."""
[email protected]
+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)
[email protected]("/env")
-async def env(session: web.Committer) -> web.QuartResponse:
- """Display the environment variables."""
[email protected]
+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))
[email protected]("/keys/check")
-async def keys_check_get(session: web.Committer) -> web.QuartResponse:
- """Check public signing key details."""
[email protected]
+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}")
[email protected]("/keys/regenerate-all")
-async def keys_regenerate_all_get(session: web.Committer) -> web.QuartResponse:
- """Display the form to regenerate KEYS files."""
[email protected]
+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))
[email protected]("/keys/update")
-async def keys_update_get(session: web.Committer) -> str |
web.WerkzeugResponse | tuple[Mapping[str, Any], int]:
- """Update keys from remote data."""
[email protected]
+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
[email protected]("/ldap")
-async def ldap_get(session: web.Committer) -> str:
[email protected]
+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:
)
[email protected]("/ldap")
[email protected](LdapLookupForm)
-async def ldap_post(session: web.Committer, lookup_form: LdapLookupForm) ->
str:
[email protected]
+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:
)
[email protected]("/logs")
-async def logs(session: web.Committer) -> web.QuartResponse:
[email protected]
+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))
[email protected]("/ongoing-tasks/<project_key>/<version_key>/<revision>")
[email protected]
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)
+
[email protected]
+async def performance(_session: web.Committer, _performance:
Literal["performance"]) -> str:
+ """
+ URL: GET /performance
[email protected]("/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)
[email protected]("/projects/update")
-async def projects_update_get(session: web.Committer) -> str |
web.WerkzeugResponse | tuple[Mapping[str, Any], int]:
- """Update projects from remote data."""
[email protected]
+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
[email protected]("/raise-error")
-async def raise_error(session: web.Committer) -> web.QuartResponse:
[email protected]
+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")
[email protected]("/revoke-user-tokens")
-async def revoke_user_tokens_get(session: web.Committer) -> str:
- """Revoke all Personal Access Tokens for a specified user."""
[email protected]
+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:
)
[email protected]("/revoke-user-tokens")
[email protected](RevokeUserTokensForm)
[email protected]
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)
[email protected]("/rotate-jwt-key")
-async def rotate_jwt_key_get(session: web.Committer) -> str:
[email protected]
+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)
[email protected]("/rotate-jwt-key")
[email protected](RotateJwtKeyForm)
-async def rotate_jwt_key_post(session: web.Committer, _rotate_form:
RotateJwtKeyForm) -> str | web.WerkzeugResponse:
[email protected]
+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)
[email protected]("/task-times/<project_key>/<version_key>/<revision_number>")
[email protected]
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))
[email protected]("/tasks/recent/<int:minutes>")
-async def tasks_recent(session: web.Committer, minutes: int) -> str:
- """Display tasks from the last N minutes."""
[email protected]
+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())
[email protected]("/toggle-view")
-async def toggle_view_get(session: web.Committer) -> str:
- """Display the page with a button to toggle between admin and user
views."""
[email protected]
+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)
[email protected]("/toggle-view")
[email protected]()
-async def toggle_view_post(session: web.Committer) -> web.WerkzeugResponse:
- """Toggle between admin and user views."""
[email protected]
+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 + "/")
[email protected]("/validate")
-async def validate_(session: web.Committer) -> str:
- """Run validators and display any divergences."""
[email protected]
+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:
)
[email protected]("/validate-jwt")
-async def validate_jwt_get(session: web.Committer) -> str:
- _require_debug_and_allow_tests()
[email protected]
+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)
[email protected]("/validate-jwt")
[email protected](ValidateJwtForm)
-async def validate_jwt_post(session: web.Committer, validate_form:
ValidateJwtForm) -> str:
- _require_debug_and_allow_tests()
[email protected]
+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)
+
+
[email protected]
+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"})
+
+
[email protected]
+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: [email protected]
For additional commands, e-mail: [email protected]
