File size: 4,400 Bytes
7cc81cb
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
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()