MediaRouter / tests /test_api_contract_regression.py
basyx's picture
Upload 437 files
7cc81cb verified
Raw
History Blame Contribute Delete
4.4 kB
from __future__ import annotations
import ast
import re
from pathlib import Path
import unittest
ROUTES = {
"brand_api": Path("app/brand/api.py"),
"projects_api": Path("app/projects/api.py"),
}
FRONTEND_CALLS = {
"brand_api": Path("frontend/features/brand-kits/api/index.ts"),
"collaboration_api": Path("frontend/features/workspace/collaboration/api/collaboration.ts"),
}
BRAND_ROUTES = {
"router.post('', response_model=BrandKitResponse, status_code=status.HTTP_201_CREATED)": "/v1/brand POST",
"router.get('', response_model=list[BrandKitResponse])": "/v1/brand GET",
"router.patch('/{brand_kit_id}', response_model=BrandKitResponse)": "/v1/brand PATCH",
"router.delete('/{brand_kit_id}', status_code=status.HTTP_204_NO_CONTENT)": "/v1/brand DELETE",
}
EXPECTED_BRAND_FRONTEND_CALLS = [
"await apiClient.get('/v1/brand');",
"await apiClient.post('/v1/brand', payload);",
]
EXPECTED_COLLABORATION_FRONTEND_CALLS = [
"await apiClient.get('/v1/projects/workspace/teams');",
"await apiClient.post('/v1/projects/workspace/teams', payload);",
"await apiClient.post('/v1/projects/workspace/invitations', payload);",
"await apiClient.get('/v1/projects/workspace/members');",
"await apiClient.delete(`/v1/projects/workspace/members/${userId}`);",
"await apiClient.patch(`/v1/projects/workspace/members/${userId}/role?new_role=${newRole}`);",
"await apiClient.get(`/v1/projects/workspace/workflows/${workflowId}/requests`);",
"await apiClient.post(`/v1/projects/workspace/workflows/${workflowId}/requests`, { project_id: projectId });",
"await apiClient.post(`/v1/projects/workspace/requests/${requestId}/approve`);",
"await apiClient.post(`/v1/projects/workspace/requests/${requestId}/reject`);",
"await apiClient.post(`/v1/projects/workspace/requests/${requestId}/comments?content=${encodeURIComponent(content)}`);",
"await apiClient.get(`/v1/projects/${encodeURIComponent(projectId)}/collaborators`);",
"await apiClient.post(`/v1/projects/${encodeURIComponent(projectId)}/collaborators?user_id=${encodeURIComponent(userId)}&role=${encodeURIComponent(role)}`);",
"await apiClient.delete(`/v1/projects/${encodeURIComponent(projectId)}/collaborators/${encodeURIComponent(userId)}`);",
]
def _route_decorators(path: Path) -> list[str]:
tree = ast.parse(path.read_text())
calls = []
for node in tree.body:
if not isinstance(node, ast.AsyncFunctionDef):
continue
for decorator in node.decorator_list:
if isinstance(decorator, ast.Call) and isinstance(decorator.func, ast.Attribute) and decorator.func.attr in {"get", "post", "patch", "delete"}:
calls.append(ast.unparse(decorator))
return calls
def _frontend_calls(path: Path) -> list[str]:
return [re.sub(r"^\s*const\s+\{[^}]*\}\s+=\s+", "", line.strip()) for line in path.read_text().splitlines() if "apiClient." in line]
def test_brand_kit_routes_match_expected_contract() -> None:
assert _route_decorators(ROUTES["brand_api"]) == list(BRAND_ROUTES.keys())
def test_approval_request_create_route_is_exposed() -> None:
decorators = _route_decorators(ROUTES["projects_api"])
assert any(
decorator == "router.post('/workspace/workflows/{workflow_id}/requests', response_model=ApprovalRequest, status_code=status.HTTP_201_CREATED)"
for decorator in decorators
)
def test_brand_kit_frontend_uses_expected_backend_routes() -> None:
assert _frontend_calls(FRONTEND_CALLS["brand_api"]) == EXPECTED_BRAND_FRONTEND_CALLS
def test_collaboration_frontend_uses_expected_backend_routes() -> None:
assert _frontend_calls(FRONTEND_CALLS["collaboration_api"]) == EXPECTED_COLLABORATION_FRONTEND_CALLS
class ApiContractRegressionTests(unittest.TestCase):
def test_brand_kit_routes_match_expected_contract(self) -> None:
test_brand_kit_routes_match_expected_contract()
def test_brand_kit_frontend_uses_expected_backend_routes(self) -> None:
test_brand_kit_frontend_uses_expected_backend_routes()
def test_collaboration_frontend_uses_expected_backend_routes(self) -> None:
test_collaboration_frontend_uses_expected_backend_routes()
def test_approval_request_create_route_is_exposed(self) -> None:
test_approval_request_create_route_is_exposed()
if __name__ == "__main__":
unittest.main()