Coverage for server / services / growthbook / context.py: 100%

1""" 

2Targeting context resolution for GrowthBook attribute evaluation. 

3""" 

4 

5import logging 

6from dataclasses import dataclass, field 

7from typing import Any, Optional 

8 

9from beanie import PydanticObjectId 

10from fastapi import HTTPException, Request, status 

11 

12from server.services.growthbook.admin_db import get_feature_db 

13from server.services.growthbook.cache import targeting_context_cache 

14from server.services.growthbook.config import get_growthbook_settings 

15from server.services.growthbook.user_scope import ( 

16 fetch_district_id_for_school, 

17 find_best_user_document, 

18) 

19from server.validators.feature_enum import PlanEnum, normalize_plan_type 

20 

21logger = logging.getLogger(__name__) 

22 

23_RUNTIME_ROLES = frozenset({"teacher", "student"}) 

24 

25_SCOPE_HINTS = { 

26 "user_not_found": ( 

27 "No user profile matched this login. Sign out and sign in again, " 

28 "or ask staff to provision your account." 

29 ), 

30 "missing_school_id": ( 

31 "Your account is not linked to a school. Staff must create or update " 

32 "your teacher/student profile with a valid school_id." 

33 ), 

34 "invalid_school_id": ( 

35 "Your profile has a school_id that is not a valid MongoDB ObjectId. " 

36 "Staff must set school_id to the school document _id from the staff portal." 

37 ), 

38 "school_not_found": ( 

39 "The school on your profile was not found in admin_staff_mongodb. " 

40 "Verify the school exists in the staff portal and STAFF_ADMIN_* env is correct." 

41 ), 

42 "school_missing_district": ( 

43 "The school record exists but has no district_id. " 

44 "Fix the school in the staff portal." 

45 ), 

46 "staff_admin_db_unconfigured": ( 

47 "STAFF_ADMIN_* environment variables are not configured; " 

48 "cannot load schools or districts." 

49 ), 

50} 

51 

52 

53@dataclass 

54class TargetingContext: 

55 """Attributes passed to GrowthBook for plan-based evaluation.""" 

56 

57 user_id: str 

58 role: str 

59 district_id: Optional[str] = None 

60 school_id: Optional[str] = None 

61 plan: str = PlanEnum.FREE.value 

62 attributes: dict[str, Any] = field(default_factory=dict) 

63 

64 def cache_key(self) -> str: 

65 settings = get_growthbook_settings() 

66 district = self.district_id or "-" 

67 school = self.school_id or "-" 

68 return f"gb:ctx:{settings.growthbook_service_name}:{self.user_id}:{district}:{school}" 

69 

70 def to_growthbook_attributes(self) -> dict[str, Any]: 

71 return { 

72 "id": self.user_id, 

73 "role": self.role, 

74 "district_id": self.district_id, 

75 "school_id": self.school_id, 

76 "plan": normalize_plan_type(self.plan), 

77 **self.attributes, 

78 } 

79 

80 

81async def _fetch_district_plan(district_id: str) -> str: 

82 try: 

83 object_id = PydanticObjectId(district_id) 

84 except Exception as exc: 

85 raise HTTPException( 

86 status_code=status.HTTP_400_BAD_REQUEST, 

87 detail=f"Invalid district id: {district_id}", 

88 ) from exc 

89 

90 district = await get_feature_db()["district_collection"].find_one({"_id": object_id}) 

91 if not district: 

92 raise HTTPException( 

93 status_code=status.HTTP_404_NOT_FOUND, 

94 detail=f"District with ID {district_id} not found", 

95 ) 

96 return normalize_plan_type(district.get("feature_plan") or PlanEnum.FREE.value) 

97 

98 

99async def _resolve_scope_from_user_details( 

100 user_details: dict[str, Any], 

101) -> tuple[Optional[str], Optional[str], Optional[str]]: 

102 """ 

103 Walk user → school → district. 

104 

105 Returns (district_id, school_id, failure_reason). 

106 """ 

107 from server.connection.database import staff_admin_db 

108 

109 if staff_admin_db is None: 

110 return None, None, "staff_admin_db_unconfigured" 

111 

112 user_doc = await find_best_user_document( 

113 mongodb_id=user_details.get("mongodb_id") or user_details.get("uuid"), 

114 auth0_user_id=user_details.get("auth0_user_id"), 

115 email=user_details.get("email"), 

116 ) 

117 if not user_doc: 

118 return None, None, "user_not_found" 

119 

120 school_id_raw = user_doc.get("school_id") 

121 if not school_id_raw: 

122 return None, None, "missing_school_id" 

123 

124 try: 

125 school_oid = ( 

126 school_id_raw 

127 if isinstance(school_id_raw, PydanticObjectId) 

128 else PydanticObjectId(str(school_id_raw)) 

129 ) 

130 except Exception: 

131 return None, str(school_id_raw), "invalid_school_id" 

132 

133 school_id = str(school_oid) 

134 district_id = await fetch_district_id_for_school(school_oid) 

135 if district_id: 

136 return district_id, school_id, None 

137 

138 admin_school = await staff_admin_db["school_collection"].find_one({"_id": school_oid}) 

139 if not admin_school: 

140 return None, school_id, "school_not_found" 

141 

142 if not admin_school.get("district_id"): 

143 return None, school_id, "school_missing_district" 

144 

145 return None, school_id, "school_not_found" 

146 

147 

148async def resolve_targeting_context( 

149 request: Request, 

150 *, 

151 district_id: Optional[str] = None, 

152 school_id: Optional[str] = None, 

153 role: Optional[str] = None, 

154 use_cache: bool = True, 

155 require_district: bool = False, 

156) -> TargetingContext: 

157 """ 

158 Build targeting context from Auth0 user details and the user's school chain. 

159 

160 Plan is always loaded from Mongo `district_collection.feature_plan` when 

161 district_id is known — never from client-supplied values. 

162 """ 

163 user_details = getattr(request.state, "user_details", None) or {} 

164 user_id = str( 

165 user_details.get("mongodb_id") 

166 or user_details.get("uuid") 

167 or user_details.get("auth0_user_id") 

168 or "" 

169 ) 

170 if not user_id: 

171 raise HTTPException( 

172 status_code=status.HTTP_401_UNAUTHORIZED, 

173 detail="Not authenticated", 

174 ) 

175 

176 resolved_role = (role or user_details.get("role") or "teacher").lower() 

177 profile_district, profile_school, scope_reason = await _resolve_scope_from_user_details( 

178 user_details 

179 ) 

180 resolved_district = district_id or profile_district 

181 resolved_school = school_id or profile_school 

182 

183 if require_district and resolved_role in _RUNTIME_ROLES and not resolved_district: 

184 logger.warning( 

185 "feature_scope_denied reason=%s user_id=%s email=%s role=%s school_id=%s", 

186 scope_reason or "unknown", 

187 user_id, 

188 user_details.get("email"), 

189 resolved_role, 

190 profile_school, 

191 ) 

192 reason = scope_reason or "missing_school_id" 

193 raise HTTPException( 

194 status_code=status.HTTP_403_FORBIDDEN, 

195 detail={ 

196 "message": "District scope is required for feature evaluation.", 

197 "reason": reason, 

198 "hint": _SCOPE_HINTS.get(reason, _SCOPE_HINTS["missing_school_id"]), 

199 }, 

200 ) 

201 

202 plan = PlanEnum.FREE.value 

203 if resolved_district: 

204 plan = await _fetch_district_plan(resolved_district) 

205 

206 ctx = TargetingContext( 

207 user_id=user_id, 

208 role=resolved_role, 

209 district_id=str(resolved_district) if resolved_district else None, 

210 school_id=str(resolved_school) if resolved_school else None, 

211 plan=plan, 

212 ) 

213 

214 if use_cache: 

215 cached = await targeting_context_cache.get(ctx.cache_key()) 

216 if cached: 

217 return TargetingContext(**cached) 

218 

219 payload = { 

220 "user_id": ctx.user_id, 

221 "role": ctx.role, 

222 "district_id": ctx.district_id, 

223 "school_id": ctx.school_id, 

224 "plan": ctx.plan, 

225 "attributes": ctx.attributes, 

226 } 

227 await targeting_context_cache.set(ctx.cache_key(), payload) 

228 return ctx 

229 

230 

231async def invalidate_targeting_context_for_district(district_id: str) -> None: 

232 """Call after district plan change.""" 

233 await targeting_context_cache.delete_by_district(district_id)