Coverage for server / services / growthbook / organization.py: 78%

1""" 

2Load school and district records for the authenticated teacher/student. 

3""" 

4 

5from typing import Any, Optional 

6 

7from beanie import PydanticObjectId 

8from fastapi import HTTPException, Request, status 

9 

10from server.connection.database import staff_admin_db 

11from server.services.growthbook.dependencies import get_targeting_context 

12from server.services.growthbook.context import resolve_targeting_context 

13from server.validators.feature_enum import ( 

14 PLAN_DESCRIPTIONS, 

15 PLAN_DISPLAY_NAMES, 

16 PlanEnum, 

17 normalize_plan_type, 

18) 

19 

20_SCHOOL_FIELDS = ( 

21 "school_name", 

22 "address", 

23 "state", 

24 "city", 

25 "zip_code", 

26 "street", 

27 "phone", 

28 "email", 

29 "school_logo", 

30 "status", 

31) 

32 

33_DISTRICT_FIELDS = ( 

34 "district_name", 

35 "district_address", 

36 "district_logo", 

37 "contact_first_name", 

38 "contact_middle_name", 

39 "contact_last_name", 

40 "contact_email", 

41 "contact_phone", 

42) 

43 

44 

45def _require_admin_db(): 

46 if staff_admin_db is None: 

47 raise HTTPException( 

48 status_code=status.HTTP_503_SERVICE_UNAVAILABLE, 

49 detail={ 

50 "message": "Admin database is not configured.", 

51 "reason": "staff_admin_db_unconfigured", 

52 }, 

53 ) 

54 return staff_admin_db 

55 

56 

57def _serialize_document( 

58 doc: Optional[dict[str, Any]], 

59 *, 

60 fields: tuple[str, ...], 

61) -> Optional[dict[str, Any]]: 

62 if not doc: 

63 return None 

64 payload: dict[str, Any] = {"id": str(doc["_id"])} 

65 for field in fields: 

66 if field in doc and doc[field] is not None: 

67 payload[field] = doc[field] 

68 return payload 

69 

70 

71async def fetch_school_document(school_id: str) -> Optional[dict[str, Any]]: 

72 try: 

73 school_oid = PydanticObjectId(school_id) 

74 except Exception as exc: 

75 raise HTTPException( 

76 status_code=status.HTTP_400_BAD_REQUEST, 

77 detail=f"Invalid school id: {school_id}", 

78 ) from exc 

79 

80 admin_db = _require_admin_db() 

81 return await admin_db["school_collection"].find_one({"_id": school_oid}) 

82 

83 

84async def fetch_district_document(district_id: str) -> Optional[dict[str, Any]]: 

85 try: 

86 district_oid = PydanticObjectId(district_id) 

87 except Exception as exc: 

88 raise HTTPException( 

89 status_code=status.HTTP_400_BAD_REQUEST, 

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

91 ) from exc 

92 

93 admin_db = _require_admin_db() 

94 return await admin_db["district_collection"].find_one({"_id": district_oid}) 

95 

96 

97def _plan_payload(plan: str) -> dict[str, str]: 

98 normalized = normalize_plan_type(plan) 

99 return { 

100 "current_plan": normalized, 

101 "plan_name": PLAN_DISPLAY_NAMES.get( 

102 normalized, 

103 normalized.replace("_", " ").title(), 

104 ), 

105 "plan_description": PLAN_DESCRIPTIONS.get(normalized, ""), 

106 } 

107 

108 

109 

110# --------------------------------------------------------------------------- 

111# Accounts with no district scope 

112# --------------------------------------------------------------------------- 

113# 

114# These three are READS of the caller's own organisation. They used to answer 

115# 403 for an account with no school_id, because they resolved scope through 

116# get_targeting_context(require_district=True). 

117# 

118# 403 means the server understood and REFUSES (RFC 9110 15.5.4) — a decision of 

119# "no". Nothing is refused here; there is simply no school linked yet, which is 

120# a data-completeness condition rather than an authorization one. And the SPA 

121# fails OPEN when these calls error, so an unprovisioned account hit that path 

122# permanently and was handed a fully unlocked UI in which every action then 

123# 403'd — the worst of both answers. This mirrors what #331 did for /effective. 

124# 

125# THE ENVELOPE DOES NOT CHANGE, only the values. A caller iterating a known 

126# shape keeps working; it reads null where it used to read an object. AIP-156 

127# treats a singleton like this as always resolving — "my school" is not a keyed 

128# lookup that can fail to find anything, it is a relationship that may be unset. 

129# 

130# A 404 IS STILL A 404 WHEN AN ID IS SET AND THE DOCUMENT IS MISSING. That is a 

131# genuine not-found and a data-integrity problem worth surfacing; it is a 

132# different condition from "no school is linked", and the two must not collapse 

133# into one another. 

134 

135 

136_NO_PLAN = {"current_plan": None, "plan_name": None, "plan_description": None} 

137 

138 

139async def _scope(request: Request): 

140 """Resolve the caller's scope WITHOUT demanding a district.""" 

141 return await resolve_targeting_context(request, require_district=False) 

142 

143 

144async def build_organization_payload(request: Request) -> dict[str, Any]: 

145 ctx = await _scope(request) 

146 if not ctx.district_id and not ctx.school_id: 

147 return { 

148 "user_id": ctx.user_id, 

149 "role": ctx.role, 

150 "school_id": None, 

151 "district_id": None, 

152 **_NO_PLAN, 

153 "school": None, 

154 "district": None, 

155 } 

156 

157 school_doc = await fetch_school_document(ctx.school_id) if ctx.school_id else None 

158 district_doc = ( 

159 await fetch_district_document(ctx.district_id) if ctx.district_id else None 

160 ) 

161 

162 if ctx.school_id and not school_doc: 

163 raise HTTPException( 

164 status_code=status.HTTP_404_NOT_FOUND, 

165 detail=f"School with ID {ctx.school_id} not found", 

166 ) 

167 if ctx.district_id and not district_doc: 

168 raise HTTPException( 

169 status_code=status.HTTP_404_NOT_FOUND, 

170 detail=f"District with ID {ctx.district_id} not found", 

171 ) 

172 

173 plan_info = _plan_payload(ctx.plan) 

174 district_info = _serialize_document(district_doc, fields=_DISTRICT_FIELDS) 

175 if district_info is not None: 

176 district_info.update(plan_info) 

177 district_info["feature_plan"] = ( 

178 district_doc.get("feature_plan") if district_doc else PlanEnum.FREE.value 

179 ) 

180 

181 return { 

182 "user_id": ctx.user_id, 

183 "role": ctx.role, 

184 "school_id": ctx.school_id, 

185 "district_id": ctx.district_id, 

186 **plan_info, 

187 "school": _serialize_document(school_doc, fields=_SCHOOL_FIELDS), 

188 "district": district_info, 

189 } 

190 

191 

192async def build_school_payload(request: Request) -> dict[str, Any]: 

193 ctx = await _scope(request) 

194 if not ctx.school_id: 

195 return { 

196 "user_id": ctx.user_id, 

197 "school_id": None, 

198 "district_id": ctx.district_id, 

199 "school": None, 

200 } 

201 

202 school_doc = await fetch_school_document(ctx.school_id) 

203 if not school_doc: 

204 raise HTTPException( 

205 status_code=status.HTTP_404_NOT_FOUND, 

206 detail=f"School with ID {ctx.school_id} not found", 

207 ) 

208 

209 return { 

210 "user_id": ctx.user_id, 

211 "school_id": ctx.school_id, 

212 "district_id": ctx.district_id, 

213 "school": _serialize_document(school_doc, fields=_SCHOOL_FIELDS), 

214 } 

215 

216 

217async def build_district_plan_payload(request: Request) -> dict[str, Any]: 

218 ctx = await _scope(request) 

219 if not ctx.district_id: 

220 return { 

221 "user_id": ctx.user_id, 

222 "school_id": ctx.school_id, 

223 "district_id": None, 

224 **_NO_PLAN, 

225 "district": None, 

226 } 

227 

228 district_doc = await fetch_district_document(ctx.district_id) 

229 if not district_doc: 

230 raise HTTPException( 

231 status_code=status.HTTP_404_NOT_FOUND, 

232 detail=f"District with ID {ctx.district_id} not found", 

233 ) 

234 

235 plan_info = _plan_payload(ctx.plan) 

236 district_info = _serialize_document(district_doc, fields=_DISTRICT_FIELDS) or {} 

237 district_info.update(plan_info) 

238 district_info["feature_plan"] = district_doc.get("feature_plan") 

239 

240 return { 

241 "user_id": ctx.user_id, 

242 "school_id": ctx.school_id, 

243 "district_id": ctx.district_id, 

244 **plan_info, 

245 "district": district_info, 

246 }