Coverage for api/roles/services.py: 94.27%

1from sqlalchemy import and_, delete, func, select 

2from sqlalchemy.ext.asyncio import AsyncSession 

3 

4from core.config import settings 

5from core.rbac import ( 

6 check_user_has_super_role, 

7 get_user_role_id, 

8 get_user_role_level, 

9 is_super_admin_role_name, 

10 user_has_role, 

11) 

12from models.role_attributes import RoleAttributes 

13from models.role_attributes_mapper import RoleAttributesMapper 

14from models.role_mapper import RoleMapper 

15from models.roles import Roles 

16from utils.custom_exception import ( 

17 AuthorizationException, 

18 ConflictException, 

19 NotFoundException, 

20 ServerException, 

21) 

22 

23from .schema import ( 

24 AttributeMappingResult, 

25 PermissionCheckResponse, 

26 RoleAttributeDetail, 

27 RoleAttributeMappingBatchResponse, 

28 RoleAttributesGroup, 

29 RoleAttributesGroupedResponse, 

30 RoleCreate, 

31 RoleResponse, 

32 RolesListResponse, 

33 RoleUpdate, 

34) 

35 

36 

37def _ensure_role_is_mutable(role: Roles) -> None: 

38 """Block mutations of the system super-admin role.""" 

39 if is_super_admin_role_name(role.name): 

40 raise AuthorizationException("Cannot modify the system super-admin role") 

41 

42 

43def _to_role_response(role: Roles) -> RoleResponse: 

44 return RoleResponse( 

45 id=role.id, 

46 name=role.name, 

47 description=role.description, 

48 level=role.level, 

49 ) 

50 

51 

52async def _assert_can_manage_role_level( 

53 db: AsyncSession, 

54 actor_user_id: str, 

55 *, 

56 target_level: int, 

57 existing_role_level: int | None = None, 

58) -> int: 

59 """ 

60 Allow managing roles at the same level or lower. 

61 Block managing / assigning levels strictly higher than the actor. 

62 """ 

63 actor_level = await get_user_role_level(actor_user_id, db) 

64 if existing_role_level is not None and existing_role_level > actor_level: 

65 raise AuthorizationException("Cannot manage a role with higher level than your own") 

66 if target_level > actor_level: 

67 raise AuthorizationException("Cannot assign a role level higher than your own") 

68 return actor_level 

69 

70 

71async def _assert_not_own_role( 

72 db: AsyncSession, 

73 actor_user_id: str, 

74 role_id: str, 

75) -> None: 

76 """Block editing/deleting the role currently assigned to the actor.""" 

77 if await user_has_role(actor_user_id, role_id, db): 

78 raise AuthorizationException("Cannot modify or delete your own role") 

79 

80 

81async def get_all_roles(db: AsyncSession, actor_user_id: str) -> RolesListResponse: 

82 """Get assignable roles (excludes the system super-admin role).""" 

83 try: 

84 actor_level = await get_user_role_level(actor_user_id, db) 

85 actor_role_id = await get_user_role_id(actor_user_id, db) 

86 roles_query = ( 

87 select(Roles) 

88 .where(Roles.name != settings.DEFAULT_SUPER_ADMIN_ROLE) 

89 .order_by(Roles.level.desc(), Roles.name.asc()) 

90 ) 

91 roles_result = await db.execute(roles_query) 

92 roles = roles_result.scalars().all() 

93 

94 role_responses = [_to_role_response(role) for role in roles] 

95 return RolesListResponse( 

96 roles=role_responses, 

97 actor_level=actor_level, 

98 actor_role_id=actor_role_id, 

99 ) 

100 

101 except Exception as e: 

102 raise ServerException(f"Failed to retrieve roles: {str(e)}") 

103 

104 

105async def create_role( 

106 db: AsyncSession, 

107 role_data: RoleCreate, 

108 actor_user_id: str, 

109) -> RoleResponse: 

110 """Create a new role""" 

111 try: 

112 if is_super_admin_role_name(role_data.name): 

113 raise AuthorizationException("Cannot create the system super-admin role") 

114 

115 await _assert_can_manage_role_level( 

116 db, 

117 actor_user_id, 

118 target_level=role_data.level, 

119 ) 

120 

121 existing_role = await db.execute(select(Roles).where(Roles.name == role_data.name)) 

122 if existing_role.scalar_one_or_none(): 

123 raise ConflictException("Role name already exists") 

124 

125 role = Roles( 

126 name=role_data.name, 

127 description=role_data.description, 

128 level=role_data.level, 

129 ) 

130 db.add(role) 

131 await db.commit() 

132 await db.refresh(role) 

133 

134 return _to_role_response(role) 

135 

136 except ConflictException, AuthorizationException: 

137 raise 

138 except Exception as e: 

139 raise ServerException(f"Failed to create role: {str(e)}") 

140 

141 

142async def update_role( 

143 db: AsyncSession, 

144 role_id: str, 

145 role_data: RoleUpdate, 

146 actor_user_id: str, 

147) -> RoleResponse: 

148 """Update role information""" 

149 try: 

150 role_result = await db.execute(select(Roles).where(Roles.id == role_id)) 

151 role = role_result.scalar_one_or_none() 

152 if not role: 

153 raise NotFoundException("Role not found") 

154 

155 _ensure_role_is_mutable(role) 

156 

157 if role_data.name and is_super_admin_role_name(role_data.name): 

158 raise AuthorizationException("Cannot rename a role to the system super-admin role") 

159 

160 await _assert_not_own_role(db, actor_user_id, role_id) 

161 

162 update_data = role_data.model_dump(exclude_unset=True) 

163 target_level = update_data.get("level", role.level) 

164 await _assert_can_manage_role_level( 

165 db, 

166 actor_user_id, 

167 target_level=target_level, 

168 existing_role_level=role.level, 

169 ) 

170 

171 if role_data.name and role_data.name != role.name: 

172 existing_role = await db.execute( 

173 select(Roles).where(Roles.name == role_data.name, Roles.id != role_id) 

174 ) 

175 if existing_role.scalar_one_or_none(): 

176 raise ConflictException("Role name already exists") 

177 

178 for field, value in update_data.items(): 

179 setattr(role, field, value) 

180 

181 await db.commit() 

182 await db.refresh(role) 

183 

184 return _to_role_response(role) 

185 

186 except ConflictException, NotFoundException, AuthorizationException: 

187 raise 

188 except Exception as e: 

189 raise ServerException(f"Failed to update role: {str(e)}") 

190 

191 

192async def delete_role(db: AsyncSession, role_id: str, actor_user_id: str) -> bool: 

193 """Delete a role""" 

194 try: 

195 role_result = await db.execute(select(Roles).where(Roles.id == role_id)) 

196 role = role_result.scalar_one_or_none() 

197 if not role: 

198 raise NotFoundException("Role not found") 

199 

200 _ensure_role_is_mutable(role) 

201 await _assert_not_own_role(db, actor_user_id, role_id) 

202 await _assert_can_manage_role_level( 

203 db, 

204 actor_user_id, 

205 target_level=role.level, 

206 existing_role_level=role.level, 

207 ) 

208 

209 user_count = await db.execute( 

210 select(func.count(RoleMapper.user_id)).where(RoleMapper.role_id == role_id) 

211 ) 

212 if user_count.scalar() > 0: 

213 raise ConflictException("Cannot delete role that is assigned to users") 

214 

215 await db.execute( 

216 delete(RoleAttributesMapper).where(RoleAttributesMapper.role_id == role_id) 

217 ) 

218 

219 await db.execute(delete(Roles).where(Roles.id == role_id)) 

220 await db.commit() 

221 

222 return True 

223 

224 except ConflictException, NotFoundException, AuthorizationException: 

225 raise 

226 except Exception as e: 

227 raise ServerException(f"Failed to delete role: {str(e)}") 

228 

229 

230async def get_role_attribute_mapping( 

231 db: AsyncSession, role_id: str 

232) -> RoleAttributesGroupedResponse: 

233 """Get role attributes mapping grouped by group and category (left join).""" 

234 try: 

235 role_result = await db.execute(select(Roles).where(Roles.id == role_id)) 

236 role = role_result.scalar_one_or_none() 

237 if not role: 

238 raise NotFoundException("Role not found") 

239 

240 _ensure_role_is_mutable(role) 

241 

242 # Use LEFT JOIN to get all attributes and their mappings 

243 query = ( 

244 select( 

245 RoleAttributes.name, 

246 RoleAttributes.group, 

247 RoleAttributes.category, 

248 RoleAttributesMapper.value, 

249 ) 

250 .select_from(RoleAttributes) 

251 .outerjoin( 

252 RoleAttributesMapper, 

253 and_( 

254 RoleAttributes.id == RoleAttributesMapper.attributes_id, 

255 RoleAttributesMapper.role_id == role_id, 

256 ), 

257 ) 

258 .order_by(RoleAttributes.id) 

259 ) 

260 

261 result = await db.execute(query) 

262 rows = result.all() 

263 

264 grouped: dict[str, dict[str, list[RoleAttributeDetail]]] = {} 

265 for row in rows: 

266 group = row.group or "default" 

267 category = row.category or "uncategorized" 

268 grouped.setdefault(group, {}).setdefault(category, []).append( 

269 RoleAttributeDetail( 

270 name=row.name, 

271 value=row.value if row.value is not None else False, 

272 ) 

273 ) 

274 

275 groups = [] 

276 for group, categories in grouped.items(): 

277 groups.append( 

278 RoleAttributesGroup( 

279 group=group, 

280 categories=categories, 

281 ) 

282 ) 

283 

284 return RoleAttributesGroupedResponse(groups=groups) 

285 

286 except NotFoundException, AuthorizationException: 

287 raise 

288 except Exception as e: 

289 raise ServerException(f"Failed to get role attributes: {str(e)}") 

290 

291 

292async def update_role_attribute_mapping( 

293 db: AsyncSession, 

294 role_id: str, 

295 attributes_data: dict[str, bool], 

296 actor_user_id: str, 

297) -> RoleAttributeMappingBatchResponse: 

298 """Batch update role and attributes mapping with detailed results""" 

299 try: 

300 role_result = await db.execute(select(Roles).where(Roles.id == role_id)) 

301 role = role_result.scalar_one_or_none() 

302 if not role: 

303 raise NotFoundException("Role not found") 

304 

305 _ensure_role_is_mutable(role) 

306 await _assert_not_own_role(db, actor_user_id, role_id) 

307 await _assert_can_manage_role_level( 

308 db, 

309 actor_user_id, 

310 target_level=role.level, 

311 existing_role_level=role.level, 

312 ) 

313 

314 results = [] 

315 success_count = 0 

316 failed_count = 0 

317 

318 # Get mapping from attribute names to IDs 

319 attribute_names = list(attributes_data.keys()) 

320 name_to_id_map = {} 

321 invalid_names = set() 

322 

323 if attribute_names: 

324 existing_attributes = await db.execute( 

325 select(RoleAttributes.id, RoleAttributes.name).where( 

326 RoleAttributes.name.in_(attribute_names) 

327 ) 

328 ) 

329 for row in existing_attributes: 

330 name_to_id_map[row.name] = row.id 

331 

332 # Handle invalid attribute names 

333 invalid_names = set(attribute_names) - set(name_to_id_map.keys()) 

334 for invalid_name in invalid_names: 

335 results.append( 

336 AttributeMappingResult( 

337 attribute_id=invalid_name, # Keep name for error reporting 

338 status="failed", 

339 message="Invalid attribute name", 

340 ) 

341 ) 

342 failed_count += 1 

343 

344 # Process valid attributes 

345 for attribute_name, value in attributes_data.items(): 

346 if attribute_name in invalid_names: 

347 continue 

348 

349 attribute_id = name_to_id_map.get(attribute_name) 

350 if not attribute_id: 

351 continue 

352 

353 try: 

354 existing_mapping = await db.execute( 

355 select(RoleAttributesMapper).where( 

356 and_( 

357 RoleAttributesMapper.role_id == role_id, 

358 RoleAttributesMapper.attributes_id == attribute_id, 

359 ) 

360 ) 

361 ) 

362 mapping = existing_mapping.scalar_one_or_none() 

363 

364 if mapping: 

365 # Update existing mapping 

366 mapping.value = value 

367 else: 

368 # Create new mapping 

369 new_mapping = RoleAttributesMapper( 

370 role_id=role_id, attributes_id=attribute_id, value=value 

371 ) 

372 db.add(new_mapping) 

373 

374 results.append( 

375 AttributeMappingResult( 

376 attribute_id=attribute_name, 

377 status="success", 

378 message="Updated successfully", 

379 ) 

380 ) 

381 success_count += 1 

382 

383 except Exception as e: 

384 results.append( 

385 AttributeMappingResult( 

386 attribute_id=attribute_name, 

387 status="failed", 

388 message=f"Failed to process: {str(e)}", 

389 ) 

390 ) 

391 failed_count += 1 

392 

393 await db.commit() 

394 

395 return RoleAttributeMappingBatchResponse( 

396 results=results, 

397 total_attributes=len(attributes_data), 

398 success_count=success_count, 

399 failed_count=failed_count, 

400 ) 

401 

402 except NotFoundException, AuthorizationException: 

403 raise 

404 except Exception as e: 

405 raise ServerException(f"Failed to update role attributes mapping: {str(e)}") 

406 

407 

408async def check_user_permissions( 

409 db: AsyncSession, user_id: str, required_attributes: list[str] = None 

410) -> PermissionCheckResponse: 

411 """Check if user has required permission attributes.""" 

412 try: 

413 # Get all available attributes 

414 all_attributes_result = await db.execute(select(RoleAttributes.name)) 

415 all_attributes = [row[0] for row in all_attributes_result.fetchall()] 

416 

417 if await check_user_has_super_role(user_id, db): 

418 if required_attributes: 

419 permissions = {attr: True for attr in required_attributes} 

420 else: 

421 permissions = {attr: True for attr in all_attributes} 

422 return PermissionCheckResponse(permissions=permissions) 

423 

424 user_role_query = select(RoleMapper.role_id).where(RoleMapper.user_id == user_id) 

425 user_role_result = await db.execute(user_role_query) 

426 user_role_id = user_role_result.scalar_one_or_none() 

427 

428 user_attributes_set = set() 

429 if user_role_id: 

430 attributes_query = ( 

431 select(RoleAttributes.name) 

432 .join(RoleAttributesMapper, RoleAttributes.id == RoleAttributesMapper.attributes_id) 

433 .where( 

434 and_( 

435 RoleAttributesMapper.role_id == user_role_id, 

436 RoleAttributesMapper.value, 

437 ) 

438 ) 

439 ) 

440 attributes_result = await db.execute(attributes_query) 

441 user_attributes_set = {row[0] for row in attributes_result.fetchall()} 

442 

443 if not required_attributes: 

444 permissions = {attr: attr in user_attributes_set for attr in all_attributes} 

445 else: 

446 permissions = {attr: attr in user_attributes_set for attr in required_attributes} 

447 

448 return PermissionCheckResponse(permissions=permissions) 

449 

450 except Exception as e: 

451 raise ServerException(f"Failed to check user permissions: {str(e)}")