Coverage for src/lexigram/web/debug/routes.py: 0%

1"""Debug routes for development. 

2 

3Provides /debug/routes and /debug/di-graph endpoints. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import Any 

9 

10from lexigram.web.exceptions import ForbiddenError 

11 

12 

13async def _check_admin(request: Any) -> None: 

14 """Ensure the current user is an admin.""" 

15 user = request.scope.get("extensions", {}).get("user") 

16 if not user: 

17 raise ForbiddenError("Authentication required") 

18 

19 role = user.get("role") 

20 roles = user.get("roles", []) 

21 if role != "admin" and "admin" not in roles: 

22 raise ForbiddenError("Admin access required") 

23 

24 

25async def debug_routes(request: Any) -> Any: 

26 """List all registered routes in a table format.""" 

27 await _check_admin(request) 

28 from starlette.responses import JSONResponse 

29 

30 # Get routes from the app 

31 routes = [] 

32 

33 if hasattr(request.app, "routes"): 

34 for route in request.app.routes: 

35 route_info = { 

36 "path": getattr(route, "path", "/"), 

37 "methods": getattr(route, "methods", []), 

38 } 

39 

40 # Get handler info 

41 if hasattr(route, "endpoint"): 

42 endpoint = route.endpoint 

43 route_info["handler"] = getattr(endpoint, "__name__", str(endpoint)) 

44 route_info["module"] = getattr(endpoint, "__module__", "") 

45 

46 routes.append(route_info) 

47 

48 return JSONResponse( 

49 { 

50 "routes": routes, 

51 "count": len(routes), 

52 }, 

53 ) 

54 

55 

56async def debug_di_graph(request: Any) -> Any: 

57 """Generate a DI dependency graph in DOT/Mermaid format.""" 

58 await _check_admin(request) 

59 from starlette.responses import JSONResponse 

60 

61 format = request.query_params.get("format", "json") 

62 

63 # Try to get container from app state 

64 container = None 

65 if hasattr(request.app, "state"): 

66 container = getattr(request.app.state, "container", None) 

67 

68 if not container: 

69 return JSONResponse( 

70 {"error": "Container not available"}, 

71 status_code=404, 

72 ) 

73 

74 # Build dependency graph 

75 graph: dict[str, list[Any]] = { 

76 "nodes": [], 

77 "edges": [], 

78 } 

79 

80 if hasattr(container, "_registry"): 

81 for descriptor in container._registry.all(): 

82 service_name = getattr( 

83 descriptor.service_type, 

84 "__name__", 

85 str(descriptor.service_type), 

86 ) 

87 node = { 

88 "id": service_name, 

89 "name": service_name, 

90 "lifetime": descriptor.scope.value 

91 if hasattr(descriptor.scope, "value") 

92 else str(descriptor.scope), 

93 } 

94 graph["nodes"].append(node) 

95 

96 if format == "mermaid": 

97 # Generate Mermaid graph 

98 lines = ["graph LR"] 

99 for node in graph["nodes"]: 

100 lifetime = node.get("lifetime", "") 

101 lines.append(f' {node["id"]}["{node["name"]} ({lifetime})"]') 

102 

103 output = "\n".join(lines) 

104 return JSONResponse({"mermaid": output}) 

105 

106 return JSONResponse(graph) 

107 

108 

109async def debug_container(request: Any) -> Any: 

110 """List all registered services in the DI container.""" 

111 await _check_admin(request) 

112 from starlette.responses import JSONResponse 

113 

114 container = None 

115 if hasattr(request.app, "state"): 

116 container = getattr(request.app.state, "container", None) 

117 

118 if not container: 

119 return JSONResponse( 

120 {"error": "Container not available"}, 

121 status_code=404, 

122 ) 

123 

124 services = [] 

125 

126 if hasattr(container, "_registry"): 

127 for descriptor in container._registry.all(): 

128 services.append( 

129 { 

130 "name": getattr( 

131 descriptor.service_type, 

132 "__name__", 

133 str(descriptor.service_type), 

134 ), 

135 "lifetime": descriptor.scope.value 

136 if hasattr(descriptor.scope, "value") 

137 else str(descriptor.scope), 

138 "type": str(descriptor.implementation), 

139 }, 

140 ) 

141 

142 return JSONResponse( 

143 { 

144 "services": services, 

145 "count": len(services), 

146 }, 

147 )