diff --git a/scripts/check_repo.py b/scripts/check_repo.py index f1ece8f45..ebf2c86df 100644 --- a/scripts/check_repo.py +++ b/scripts/check_repo.py @@ -177,9 +177,25 @@ def check_trust_roots(root: Path) -> list[str]: return failures +def router_bindings(tree: ast.AST) -> set[str]: + # ponytail: literal `name = APIRouter(...)` only; add alias tracking when routers get re-bound. + names: set[str] = set() + for node in ast.walk(tree): + if not isinstance(node, ast.Assign) or not isinstance(node.value, ast.Call): + continue + callee = node.value.func + called = callee.attr if isinstance(callee, ast.Attribute) else getattr(callee, "id", None) + if called != "APIRouter": + continue + names.update(target.id for target in node.targets if isinstance(target, ast.Name)) + return names + + def declared_routes(source: str) -> set[tuple[str, str]]: routes = set() - for function in ast.walk(ast.parse(source)): + tree = ast.parse(source) + routers = router_bindings(tree) + for function in ast.walk(tree): if not isinstance(function, ast.FunctionDef | ast.AsyncFunctionDef): continue for node in function.decorator_list: @@ -187,12 +203,20 @@ def declared_routes(source: str) -> set[tuple[str, str]]: continue if node.func.attr not in {"get", "post", "put", "patch", "delete", "head", "options"}: continue - if ( - node.args - and isinstance(node.args[0], ast.Constant) - and isinstance(node.args[0].value, str) + # Named paths need a receiver bound to APIRouter(); positional receivers stay unchecked. + if not node.args and not ( + isinstance(node.func.value, ast.Name) and node.func.value.id in routers ): - routes.add((node.func.attr, node.args[0].value)) + continue + path = ( + node.args[0] + if node.args + else next( + (keyword.value for keyword in node.keywords if keyword.arg == "path"), None + ) + ) + if isinstance(path, ast.Constant) and isinstance(path.value, str): + routes.add((node.func.attr, path.value)) return routes diff --git a/tests/test_repo_contract.py b/tests/test_repo_contract.py index d861e0e96..ca759cd1b 100644 --- a/tests/test_repo_contract.py +++ b/tests/test_repo_contract.py @@ -72,13 +72,47 @@ def test_proportional_trust_template_requires_new_owner_version(tmp_path, versio assert (CHECK["check_trust_roots"](tmp_path) == []) is accepted -def test_public_api_must_be_an_actual_route_not_a_comment_or_mapping_access(): +@pytest.mark.parametrize("method", ["get", "post"]) +@pytest.mark.parametrize("path_argument", ['"/v1/status"', 'path="/v1/status"']) +def test_public_api_must_be_an_actual_route_not_a_comment_or_mapping_access(method, path_argument): source = ( + "from fastapi import APIRouter\n" + "router = APIRouter()\n" '# @router.post("/v1/submissions")\n' 'mapping.get("/v1/proof/topics")\n' - '@router.get("/v1/status")\nasync def status(): return {}\n' + f"@router.{method}({path_argument})\nasync def status(): return {{}}\n" ) - assert CHECK["declared_routes"](source) == {("get", "/v1/status")} + assert CHECK["declared_routes"](source) == {(method, "/v1/status")} + + +@pytest.mark.parametrize("method", ["get", "post"]) +@pytest.mark.parametrize( + "receiver,assignment,named_credited", + [ + ("router", "router = APIRouter()", True), + ("router_status", "router_status = APIRouter(prefix='/v1')", True), + ("api", "api = fastapi.APIRouter()", True), + ("router", "router = object()", False), + ("router_fake", "router_fake = object()", False), + ("other", "other = object()", False), + ("mapping", "mapping = {}", False), + ("routerlike", "routerlike = object()", False), + ("holder.router", "", False), + ], +) +def test_named_routes_require_apirouter_bindings_but_positional_receivers_stay_compatible( + method, receiver, assignment, named_credited +): + source = ( + "import fastapi\nfrom fastapi import APIRouter\n" + f"{assignment}\n" + f'@{receiver}.{method}("/v1/positional")\nasync def positional(): return {{}}\n' + f'@{receiver}.{method}(path="/v1/named")\nasync def named(): return {{}}\n' + ) + expected = {(method, "/v1/positional")} + if named_credited: + expected.add((method, "/v1/named")) + assert CHECK["declared_routes"](source) == expected def test_miner_contract_detects_removed_route_and_undocumented_nonce(tmp_path):