Coverage for apps / core / api.py: 88%

1"""System API for administrative operations like database reset.""" 

2 

3import logging 

4import os 

5import shutil 

6 

7from django_ratelimit.decorators import ratelimit 

8 

9logger = logging.getLogger(__name__) 

10security_logger = logging.getLogger("security") 

11 

12from django.conf import settings 

13from django.core.cache import cache 

14from django.core.management import call_command 

15from django.contrib.sessions.models import Session 

16from ninja import Router, Schema, Status 

17 

18from apps.profiles.models import Profile 

19from apps.recipes.models import ( 

20 Recipe, 

21 RecipeFavorite, 

22 RecipeCollection, 

23 RecipeCollectionItem, 

24 RecipeViewHistory, 

25 SearchSource, 

26 ServingAdjustment, 

27 CachedSearchImage, 

28) 

29from apps.ai.models import AIDiscoverySuggestion, AIPrompt 

30from apps.core.auth import HomeOnlyAuth, SessionAuth 

31 

32router = Router(tags=["system"]) 

33 

34 

35class HealthSchema(Schema): 

36 status: str 

37 

38 

39class ReadySchema(Schema): 

40 status: str 

41 database: str 

42 

43 

44@router.get("/mode/", response={200: dict}) 

45def get_mode(request): 

46 """Return the current operating mode. Also ensures CSRF cookie is set.""" 

47 from django.middleware.csrf import get_token 

48 

49 get_token(request) # Forces Django to set the CSRF cookie 

50 # `version` was removed in v1.42.0 to eliminate deployment fingerprinting. 

51 # Operators check the version via `python manage.py cookie_admin status --json`. 

52 result = {"mode": settings.AUTH_MODE} 

53 if settings.AUTH_MODE == "passkey": 

54 result["registration_enabled"] = True 

55 return result 

56 

57 

58@router.get("/health/", response=HealthSchema) 

59def health_check(request): 

60 """Liveness probe — confirms the process is running. No dependency checks.""" 

61 from django.middleware.csrf import get_token 

62 

63 get_token(request) # Ensures CSRF cookie is set for SPA 

64 return {"status": "healthy"} 

65 

66 

67@router.get("/ready/", response={200: ReadySchema, 503: ReadySchema}) 

68def readiness_check(request): 

69 """Readiness probe — checks database connectivity.""" 

70 from django.db import connection 

71 

72 try: 

73 with connection.cursor() as cursor: 

74 cursor.execute("SELECT 1") 

75 return Status(200, {"status": "ready", "database": "ok"}) 

76 except Exception: 

77 return Status(503, {"status": "not_ready", "database": "error"}) 

78 

79 

80class DataCountsSchema(Schema): 

81 profiles: int 

82 recipes: int 

83 recipe_images: int 

84 favorites: int 

85 collections: int 

86 collection_items: int 

87 view_history: int 

88 ai_suggestions: int 

89 serving_adjustments: int 

90 cached_search_images: int 

91 

92 

93class ResetPreviewSchema(Schema): 

94 data_counts: DataCountsSchema 

95 preserved: list[str] 

96 warnings: list[str] 

97 

98 

99class ResetConfirmSchema(Schema): 

100 confirmation_text: str # Must be "RESET" 

101 

102 

103class ErrorSchema(Schema): 

104 error: str 

105 message: str 

106 

107 

108class ResetSuccessSchema(Schema): 

109 success: bool 

110 message: str 

111 actions_performed: list[str] 

112 

113 

114@router.get("/reset-preview/", response={200: ResetPreviewSchema}, auth=HomeOnlyAuth()) 

115def get_reset_preview(request): 

116 """Get summary of data that will be deleted on reset. Home mode only — 404 in passkey mode.""" 

117 return { 

118 "data_counts": { 

119 "profiles": Profile.objects.count(), 

120 "recipes": Recipe.objects.count(), 

121 "recipe_images": Recipe.objects.exclude(image="").exclude(image__isnull=True).count(), 

122 "favorites": RecipeFavorite.objects.count(), 

123 "collections": RecipeCollection.objects.count(), 

124 "collection_items": RecipeCollectionItem.objects.count(), 

125 "view_history": RecipeViewHistory.objects.count(), 

126 "ai_suggestions": AIDiscoverySuggestion.objects.count(), 

127 "serving_adjustments": ServingAdjustment.objects.count(), 

128 "cached_search_images": CachedSearchImage.objects.count(), 

129 }, 

130 "preserved": [ 

131 "Search source configurations", 

132 "AI prompt templates", 

133 "Application settings", 

134 ], 

135 "warnings": [ 

136 "All user data will be permanently deleted", 

137 "All recipe images will be removed from storage", 

138 "This action cannot be undone", 

139 ], 

140 } 

141 

142 

143@router.post("/reset/", response={200: ResetSuccessSchema, 400: ErrorSchema, 429: dict}, auth=HomeOnlyAuth()) 

144@ratelimit(key="ip", rate="1/h", method="POST", block=False) 

145def reset_database(request, data: ResetConfirmSchema): 

146 """ 

147 Completely reset the database to factory state. 

148 

149 Requires confirmation_text="RESET" to proceed. 

150 Rate limited to 1 request per hour per IP. 

151 Home mode only — 404 in passkey mode (use CLI: python manage.py cookie_admin reset). 

152 """ 

153 if getattr(request, "limited", False): 

154 security_logger.warning("Rate limit hit: /system/reset/ from %s", request.META.get("REMOTE_ADDR")) 

155 return Status(429, {"error": "rate_limited", "message": "Too many requests. Please try again later."}) 

156 if data.confirmation_text != "RESET": 

157 return Status( 

158 400, 

159 { 

160 "error": "invalid_confirmation", 

161 "message": "Type RESET to confirm", 

162 }, 

163 ) 

164 

165 client_ip = request.META.get("REMOTE_ADDR") 

166 user_info = getattr(request, "auth", None) 

167 security_logger.warning( 

168 "DATABASE RESET initiated by %s from %s", 

169 user_info, 

170 client_ip, 

171 ) 

172 

173 try: 

174 # 1. Clear database tables (order matters for FK constraints) 

175 # Start with leaf tables that depend on others 

176 AIDiscoverySuggestion.objects.all().delete() 

177 ServingAdjustment.objects.all().delete() 

178 RecipeViewHistory.objects.all().delete() 

179 RecipeCollectionItem.objects.all().delete() 

180 RecipeCollection.objects.all().delete() 

181 RecipeFavorite.objects.all().delete() 

182 CachedSearchImage.objects.all().delete() 

183 

184 # Delete all recipes (this will cascade to related items) 

185 Recipe.objects.all().delete() 

186 

187 # Delete all profiles 

188 Profile.objects.all().delete() 

189 

190 # Reset SearchSource failure counters (keep selectors) 

191 SearchSource.objects.all().update( 

192 consecutive_failures=0, 

193 needs_attention=False, 

194 last_validated_at=None, 

195 ) 

196 

197 # 2. Clear recipe images 

198 images_dir = os.path.join(settings.MEDIA_ROOT, "recipe_images") 

199 if os.path.exists(images_dir): 

200 shutil.rmtree(images_dir) 

201 os.makedirs(images_dir) # Recreate empty directory 

202 

203 # Clear cached search images 

204 search_images_dir = os.path.join(settings.MEDIA_ROOT, "search_images") 

205 if os.path.exists(search_images_dir): 

206 shutil.rmtree(search_images_dir) 

207 os.makedirs(search_images_dir) # Recreate empty directory 

208 

209 # 3. Clear Django cache 

210 cache.clear() 

211 

212 # 4. Clear all sessions 

213 Session.objects.all().delete() 

214 

215 # 5. Re-run migrations (ensures clean state) 

216 call_command("migrate", verbosity=0) 

217 

218 # 6. Re-seed default data 

219 try: 

220 call_command("seed_search_sources", verbosity=0) 

221 except Exception: 

222 logger.debug("seed_search_sources command not available, skipping") 

223 

224 try: 

225 call_command("seed_ai_prompts", verbosity=0) 

226 except Exception: 

227 logger.debug("seed_ai_prompts command not available, skipping") 

228 

229 security_logger.warning( 

230 "DATABASE RESET completed successfully by %s from %s", 

231 user_info, 

232 client_ip, 

233 ) 

234 

235 return { 

236 "success": True, 

237 "message": "Database reset complete", 

238 "actions_performed": [ 

239 "Deleted all user profiles", 

240 "Deleted all recipes and images", 

241 "Cleared all favorites and collections", 

242 "Cleared all view history", 

243 "Cleared all AI cache data", 

244 "Cleared all cached search images", 

245 "Reset search source counters", 

246 "Cleared application cache", 

247 "Cleared all sessions", 

248 "Re-ran database migrations", 

249 "Restored default seed data", 

250 ], 

251 } 

252 

253 except Exception as e: 

254 logger.error("Database reset failed: %s", str(e), exc_info=True) 

255 return Status(400, {"error": "reset_failed", "message": "Database reset failed. Check server logs."})