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

1""" 

2Resolve teacher/student user documents and school → district chain. 

3 

4Schools and districts live in admin_staff_mongodb; users live in teacher_student_db. 

5When duplicate profiles share auth0_user_id or email, prefer the one whose 

6school_id resolves in the admin database. 

7""" 

8 

9from typing import Any, Optional 

10 

11from beanie import PydanticObjectId 

12 

13from server.connection.database import staff_admin_db 

14from server.models.clients import School 

15from server.models.users import User 

16from server.services.growthbook.admin_db import get_feature_db 

17 

18 

19async def fetch_district_id_for_school(school_oid: PydanticObjectId) -> Optional[str]: 

20 """Load school from admin_staff_mongodb first, then teacher_student_db fallback.""" 

21 if staff_admin_db is not None: 

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

23 if school_doc and school_doc.get("district_id"): 

24 return str(school_doc["district_id"]) 

25 

26 school_doc = await get_feature_db()["school_collection"].find_one({"_id": school_oid}) 

27 if school_doc and school_doc.get("district_id"): 

28 return str(school_doc["district_id"]) 

29 

30 school = await School.get(school_oid) 

31 if school and school.district_id: 

32 return str(school.district_id) 

33 

34 return None 

35 

36 

37async def _district_for_user_doc(user_doc: dict[str, Any]) -> Optional[str]: 

38 school_id_raw = user_doc.get("school_id") 

39 if not school_id_raw: 

40 return None 

41 try: 

42 school_oid = ( 

43 school_id_raw 

44 if isinstance(school_id_raw, PydanticObjectId) 

45 else PydanticObjectId(str(school_id_raw)) 

46 ) 

47 except Exception: 

48 return None 

49 return await fetch_district_id_for_school(school_oid) 

50 

51 

52async def find_best_user_document( 

53 *, 

54 mongodb_id: Optional[str] = None, 

55 auth0_user_id: Optional[str] = None, 

56 email: Optional[str] = None, 

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

58 """ 

59 Return the best matching user document for feature scope. 

60 

61 When multiple profiles match (duplicate auth0/email), prefer the document 

62 whose school_id resolves to a district in admin_staff_mongodb. 

63 """ 

64 collection = User.get_pymongo_collection() 

65 candidates: list[dict[str, Any]] = [] 

66 seen_ids: set[str] = set() 

67 

68 def _add(doc: Optional[dict[str, Any]]) -> None: 

69 if not doc: 

70 return 

71 doc_id = str(doc["_id"]) 

72 if doc_id in seen_ids: 

73 return 

74 seen_ids.add(doc_id) 

75 candidates.append(doc) 

76 

77 if mongodb_id: 

78 try: 

79 _add(await collection.find_one({"_id": PydanticObjectId(str(mongodb_id))})) 

80 except Exception: 

81 pass 

82 

83 if auth0_user_id: 

84 async for doc in collection.find({"auth0_user_id": auth0_user_id}): 

85 _add(doc) 

86 

87 if email: 

88 async for doc in collection.find({"email": email}): 

89 _add(doc) 

90 

91 if not candidates: 

92 return None 

93 

94 provisioned: list[tuple[dict[str, Any], str]] = [] 

95 for doc in candidates: 

96 district_id = await _district_for_user_doc(doc) 

97 if district_id: 

98 provisioned.append((doc, district_id)) 

99 

100 if provisioned: 

101 # Prefer explicit mongodb_id match when it has a valid chain 

102 if mongodb_id: 

103 for doc, _ in provisioned: 

104 if str(doc["_id"]) == str(mongodb_id): 

105 return doc 

106 return provisioned[0][0] 

107 

108 return candidates[0]