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

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 AdminAuth, 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 result = {"mode": settings.AUTH_MODE, "version": settings.COOKIE_VERSION} 

51 if settings.AUTH_MODE == "passkey": 

52 result["registration_enabled"] = True 

53 return result 

54 

55 

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

57def health_check(request): 

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

59 from django.middleware.csrf import get_token 

60 

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

62 return {"status": "healthy"} 

63 

64 

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

66def readiness_check(request): 

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

68 from django.db import connection 

69 

70 try: 

71 with connection.cursor() as cursor: 

72 cursor.execute("SELECT 1") 

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

74 except Exception: 

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

76 

77 

78class DataCountsSchema(Schema): 

79 profiles: int 

80 recipes: int 

81 recipe_images: int 

82 favorites: int 

83 collections: int 

84 collection_items: int 

85 view_history: int 

86 ai_suggestions: int 

87 serving_adjustments: int 

88 cached_search_images: int 

89 

90 

91class ResetPreviewSchema(Schema): 

92 data_counts: DataCountsSchema 

93 preserved: list[str] 

94 warnings: list[str] 

95 

96 

97class ResetConfirmSchema(Schema): 

98 confirmation_text: str # Must be "RESET" 

99 

100 

101class ErrorSchema(Schema): 

102 error: str 

103 message: str 

104 

105 

106class ResetSuccessSchema(Schema): 

107 success: bool 

108 message: str 

109 actions_performed: list[str] 

110 

111 

112@router.get("/reset-preview/", response={200: ResetPreviewSchema, 403: ErrorSchema}, auth=AdminAuth()) 

113def get_reset_preview(request): 

114 """Get summary of data that will be deleted on reset. Disabled in passkey mode.""" 

115 if settings.AUTH_MODE == "passkey": 

116 return Status(403, {"error": "disabled", "message": "Database reset is disabled in passkey mode. Use the CLI: python manage.py cookie_admin reset"}) 

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, 403: ErrorSchema, 429: dict}, auth=AdminAuth()) 

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 Disabled in passkey mode — use CLI: python manage.py cookie_admin reset 

152 """ 

153 if settings.AUTH_MODE == "passkey": 

154 security_logger.warning("Blocked /system/reset/ attempt in passkey mode from %s", request.META.get("REMOTE_ADDR")) 

155 return Status(403, {"error": "disabled", "message": "Database reset is disabled in passkey mode. Use the CLI: python manage.py cookie_admin reset"}) 

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

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

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

159 if data.confirmation_text != "RESET": 

160 return Status( 

161 400, 

162 { 

163 "error": "invalid_confirmation", 

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

165 }, 

166 ) 

167 

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

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

170 security_logger.warning( 

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

172 user_info, 

173 client_ip, 

174 ) 

175 

176 try: 

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

178 # Start with leaf tables that depend on others 

179 AIDiscoverySuggestion.objects.all().delete() 

180 ServingAdjustment.objects.all().delete() 

181 RecipeViewHistory.objects.all().delete() 

182 RecipeCollectionItem.objects.all().delete() 

183 RecipeCollection.objects.all().delete() 

184 RecipeFavorite.objects.all().delete() 

185 CachedSearchImage.objects.all().delete() 

186 

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

188 Recipe.objects.all().delete() 

189 

190 # Delete all profiles 

191 Profile.objects.all().delete() 

192 

193 # In passkey mode, also delete all user accounts and device codes 

194 if settings.AUTH_MODE == "passkey": 

195 from django.contrib.auth.models import User 

196 from apps.core.models import DeviceCode 

197 

198 DeviceCode.objects.all().delete() 

199 User.objects.all().delete() 

200 

201 # Reset SearchSource failure counters (keep selectors) 

202 SearchSource.objects.all().update( 

203 consecutive_failures=0, 

204 needs_attention=False, 

205 last_validated_at=None, 

206 ) 

207 

208 # 2. Clear recipe images 

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

210 if os.path.exists(images_dir): 

211 shutil.rmtree(images_dir) 

212 os.makedirs(images_dir) # Recreate empty directory 

213 

214 # Clear cached search images 

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

216 if os.path.exists(search_images_dir): 

217 shutil.rmtree(search_images_dir) 

218 os.makedirs(search_images_dir) # Recreate empty directory 

219 

220 # 3. Clear Django cache 

221 cache.clear() 

222 

223 # 4. Clear all sessions 

224 Session.objects.all().delete() 

225 

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

227 call_command("migrate", verbosity=0) 

228 

229 # 6. Re-seed default data 

230 try: 

231 call_command("seed_search_sources", verbosity=0) 

232 except Exception: 

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

234 

235 try: 

236 call_command("seed_ai_prompts", verbosity=0) 

237 except Exception: 

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

239 

240 security_logger.warning( 

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

242 user_info, 

243 client_ip, 

244 ) 

245 

246 return { 

247 "success": True, 

248 "message": "Database reset complete", 

249 "actions_performed": [ 

250 "Deleted all user profiles", 

251 "Deleted all recipes and images", 

252 "Cleared all favorites and collections", 

253 "Cleared all view history", 

254 "Cleared all AI cache data", 

255 "Cleared all cached search images", 

256 "Reset search source counters", 

257 "Cleared application cache", 

258 "Cleared all sessions", 

259 "Re-ran database migrations", 

260 "Restored default seed data", 

261 ], 

262 } 

263 

264 except Exception as e: 

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

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