Coverage for src / lexigram / admin / controllers / form_validation.py: 0%

1"""Form validation and autocomplete controller for real-time features.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6from typing import Any 

7 

8from starlette.responses import HTMLResponse, JSONResponse 

9 

10from lexigram.admin.controllers.base import AdminController 

11from lexigram.contracts.web import get, post 

12from lexigram.di.decorators import inject 

13 

14 

15@inject 

16class FormValidationController(AdminController): 

17 """Controller for real-time form validation and autocomplete.""" 

18 

19 def __init__(self, renderer: Any) -> None: 

20 super().__init__(renderer) 

21 

22 @post("/api/forms/validate/{field_name}") 

23 async def validate_field(self, field_name: str, request) -> Any: 

24 """Validate a single field in real-time.""" 

25 try: 

26 # Get form data from request 

27 if request.headers.get("content-type") == "application/json": 

28 data = await request.json() 

29 else: 

30 form_data = await request.form() 

31 data = dict(form_data) 

32 

33 field_value = data.get(field_name) 

34 

35 # For now, return success - in a real implementation, 

36 # you'd look up the field from a form registry and validate it 

37 # This is a placeholder for the real-time validation endpoint 

38 

39 # Simulate async validation delay 

40 if field_name == "email": 

41 await asyncio.sleep(0.1) # Simulate API call 

42 if field_value and "@" not in field_value: 

43 return HTMLResponse( 

44 content='<span class="text-destructive text-sm">Invalid email address</span>', 

45 status_code=200, 

46 ) 

47 

48 return HTMLResponse(content="", status_code=200) # Clear any previous error 

49 

50 except Exception as e: # noqa: BLE001 — controller boundary; all errors become HTTP 400 responses 

51 from lexigram.logging import get_logger 

52 

53 logger = get_logger(__name__) 

54 logger.exception("Field validation error for %s", field_name) 

55 return HTMLResponse( 

56 content=f'<span class="text-destructive text-sm">Validation error: {e!s}</span>', 

57 status_code=400, 

58 ) 

59 

60 @get("/api/forms/autocomplete/{field_name}") 

61 async def autocomplete_field(self, field_name: str, request) -> Any: 

62 """Provide autocomplete suggestions for a field.""" 

63 try: 

64 query = request.query_params.get("q", "") 

65 

66 # Simulate autocomplete data based on field name 

67 suggestions = [] 

68 

69 if field_name == "country": 

70 countries = [ 

71 "United States", 

72 "United Kingdom", 

73 "Canada", 

74 "Australia", 

75 "Germany", 

76 "France", 

77 ] 

78 suggestions = list( 

79 filter(lambda c: query.lower() in c.lower(), countries), 

80 ) 

81 elif field_name == "city": 

82 cities = ["New York", "London", "Toronto", "Sydney", "Berlin", "Paris"] 

83 suggestions = list(filter(lambda c: query.lower() in c.lower(), cities)) 

84 elif field_name == "tags": 

85 tags = [ 

86 "urgent", 

87 "important", 

88 "review", 

89 "draft", 

90 "published", 

91 "archived", 

92 ] 

93 suggestions = list(filter(lambda t: query.lower() in t.lower(), tags)) 

94 

95 # Return HTML for HTMX to swap in 

96 if suggestions: 

97 items_html = "".join( 

98 [ 

99 f'<div class="px-3 py-2 hover:bg-muted cursor-pointer" ' 

100 f"onclick=\"selectAutocomplete('{field_name}', '{suggestion}')\">{suggestion}</div>" 

101 for suggestion in suggestions[:5] # Limit to 5 suggestions 

102 ], 

103 ) 

104 return HTMLResponse( 

105 content=f'<div class="absolute z-10 bg-card border border-border rounded-md shadow-lg max-h-40 overflow-y-auto">{items_html}</div>', 

106 status_code=200, 

107 ) 

108 return HTMLResponse(content="", status_code=200) 

109 

110 except Exception as e: # noqa: BLE001 — controller boundary; all errors become HTTP 400 responses 

111 from lexigram.logging import get_logger 

112 

113 logger = get_logger(__name__) 

114 logger.exception("Autocomplete error for %s", field_name) 

115 return HTMLResponse( 

116 content=f'<div class="text-destructive text-sm">Autocomplete error: {e!s}</div>', 

117 status_code=400, 

118 ) 

119 

120 @post("/api/forms/async-validate/{field_name}") 

121 async def async_validate_field(self, field_name: str, request) -> Any: 

122 """Handle async validation for fields like username availability.""" 

123 try: 

124 if request.headers.get("content-type") == "application/json": 

125 data = await request.json() 

126 else: 

127 form_data = await request.form() 

128 data = dict(form_data) 

129 

130 field_value = data.get(field_name) 

131 

132 # Simulate async validation (e.g., checking username availability) 

133 await asyncio.sleep(0.5) # Simulate API/database call 

134 

135 if field_name == "username": 

136 # Simulate checking if username is taken 

137 taken_usernames = ["admin", "root", "user", "test"] 

138 if field_value in taken_usernames: 

139 return JSONResponse( 

140 {"valid": False, "message": "Username is already taken"}, 

141 ) 

142 

143 return JSONResponse({"valid": True, "message": "Available"}) 

144 

145 except Exception as e: # noqa: BLE001 — controller boundary; all errors become HTTP 400 responses 

146 from lexigram.logging import get_logger 

147 

148 logger = get_logger(__name__) 

149 logger.exception("Async validation error for %s", field_name) 

150 return JSONResponse( 

151 {"valid": False, "message": f"Validation failed: {e!s}"}, 

152 status_code=400, 

153 )