Coverage for apps/accounts/admin_api.py: 87%

1"""Accounts self (/me) + RBAC admin router (mounted at /api/accounts). 

2 

3Reuses the RBAC foundation unchanged: `StaffAuth`/`requires` guards, `assign_user_branch`, and 

4the principal already resolved onto the request. Pure-Lumen. 

5""" 

6 

7from __future__ import annotations 

8 

9from datetime import date 

10 

11from django.conf import settings 

12from django.db import transaction as db_transaction 

13from django.utils.crypto import get_random_string 

14from ninja import Query, Router 

15 

16from apps.accounts.admin_schemas import ( 

17 AgentConfigOut, 

18 AgentConfigPatch, 

19 ApiClientCreate, 

20 ApiClientOut, 

21 BillingOut, 

22 CompanyOut, 

23 CompanyPatch, 

24 ComplianceCheckInput, 

25 ComplianceCheckOut, 

26 FieldConfigsOut, 

27 LeadFormConfigOut, 

28 LeadFormConfigPatch, 

29 LicenseCreate, 

30 LicenseOut, 

31 MeOut, 

32 NewApiCredentialsOut, 

33 NonDelegatedConfigOut, 

34 NonDelegatedConfigPatch, 

35 OrgInfoOut, 

36 OrgUnitCreate, 

37 OrgUnitOut, 

38 PermissionOut, 

39 PosConfigOut, 

40 PosConfigPatch, 

41 ProvisionApplyInput, 

42 ProvisionApplyOut, 

43 ProvisionPlanOut, 

44 ProvisionPreviewInput, 

45 RoleAssignmentCreate, 

46 RoleAssignmentOut, 

47 RoleAssignmentReplace, 

48 RoleMatrixOut, 

49 RoleOut, 

50 StateComplianceOut, 

51 UserActiveUpdate, 

52 UserDetailOut, 

53 UserInviteInput, 

54 UserInviteOut, 

55) 

56from apps.accounts.auth import StaffAuth, assign_user_branch, requires 

57from apps.accounts.models import ( 

58 ApiClient, 

59 Company, 

60 License, 

61 OrgUnit, 

62 Permission, 

63 Role, 

64 User, 

65 UserRole, 

66) 

67from apps.integrations.encryption import encrypt_secret 

68from domain.field_configs import get_field_configs 

69from domain.provisioning import EXAMPLE_CONFIG, summarize_plan, validate_client_config 

70from domain.rbac import PERMISSIONS, ROLE_PERMISSIONS, ROLES, permission_meta 

71from domain.result import AppError, DomainError 

72from domain.state_compliance import STATE_COMPLIANCE, validate_state_compliance 

73 

74router = Router(tags=["accounts"]) 

75# Mounted at /api/admin (not /api/accounts) — the Roles & Permissions tab's own prefix 

76# (add-admin-roles-matrix-api). A separate router instance since it shares no other paths 

77# with apps.integrations.admin_api's router, which already owns /api/admin. 

78admin_console_router = Router(tags=["admin-console"]) 

79compliance_router = Router(tags=["compliance"]) 

80billing_router = Router(tags=["billing"]) 

81provision_router = Router(tags=["provisioning"]) 

82 

83 

84def _user(user_id: int) -> User: 

85 user = User.objects.filter(pk=user_id).first() 

86 if user is None: 86 ↛ 87line 86 didn't jump to line 87 because the condition on line 86 was never true

87 raise DomainError([AppError("NOT_FOUND", "User not found")]) 

88 return user 

89 

90 

91def _resolve_role(role_name: str) -> Role: 

92 role = Role.objects.filter(name=role_name).first() 

93 if role is None: 

94 raise DomainError([AppError("NOT_FOUND", f"Role not found: {role_name}")]) 

95 return role 

96 

97 

98def _resolve_org_unit(org_unit_id: int | None) -> OrgUnit | None: 

99 if org_unit_id is None: 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true

100 return None 

101 org = OrgUnit.objects.filter(pk=org_unit_id).first() 

102 if org is None: 102 ↛ 103line 102 didn't jump to line 103 because the condition on line 102 was never true

103 raise DomainError([AppError("NOT_FOUND", "Org unit not found")]) 

104 return org 

105 

106 

107def _role_bindings(user: User) -> list[dict]: 

108 """Uses ``user.user_roles.all()`` (not a fresh queryset) so a prefetch on ``user`` 

109 (e.g. ``list_users``) avoids N+1 queries across many users.""" 

110 bindings = user.user_roles.all() 

111 return [{"id": b.id, "role": b.role.name, "org_unit_id": b.org_unit_id} for b in bindings] 

112 

113 

114def _user_detail(user: User) -> dict: 

115 return { 

116 "id": user.id, 

117 "email": user.email, 

118 "first_name": user.first_name, 

119 "last_name": user.last_name, 

120 "title": user.title, 

121 "phone": user.phone, 

122 "nmls_id": user.nmls_id, 

123 "org_unit_id": user.org_unit_id, 

124 "is_active": user.is_active, 

125 "roles": _role_bindings(user), 

126 "licenses": list(user.licenses.all()), 

127 } 

128 

129 

130# --- Self ---------------------------------------------------------------------- 

131@router.get("/me", response=MeOut, auth=StaffAuth()) 

132def me(request): 

133 p = request.principal 

134 user = User.objects.filter(pk=p.user_id).first() 

135 bindings = UserRole.objects.filter(user_id=p.user_id).select_related("role") 

136 return { 

137 "user_id": p.user_id, 

138 "email": p.email, 

139 "first_name": user.first_name if user else "", 

140 "last_name": user.last_name if user else "", 

141 "is_hq": p.is_hq, 

142 "permissions": sorted(p.permissions), 

143 "org_unit_ids": sorted(p.org_unit_ids), 

144 "roles": [{"role": b.role.name, "org_unit_id": b.org_unit_id} for b in bindings], 

145 } 

146 

147 

148# --- Org units ----------------------------------------------------------------- 

149@router.get("/org-units", response=list[OrgUnitOut], auth=requires("MANAGE_USER_ACCOUNT")) 

150def list_org_units(request): 

151 return list(OrgUnit.objects.all()) 

152 

153 

154@router.post("/org-units", response={201: OrgUnitOut}, auth=requires("MANAGE_USER_ACCOUNT")) 

155def create_org_unit(request, payload: OrgUnitCreate): 

156 if payload.parent_id is not None and not OrgUnit.objects.filter(pk=payload.parent_id).exists(): 156 ↛ 157line 156 didn't jump to line 157 because the condition on line 156 was never true

157 raise DomainError([AppError("NOT_FOUND", "Parent org unit not found")]) 

158 org = OrgUnit.objects.create( 

159 name=payload.name, 

160 parent_id=payload.parent_id, 

161 nmls_id=payload.nmls_id, 

162 is_headquarters=payload.is_headquarters, 

163 ) 

164 return 201, org 

165 

166 

167# --- Roles + permissions ------------------------------------------------------- 

168@router.get("/roles", response=list[RoleOut], auth=requires("MANAGE_USER_ACCOUNT")) 

169def list_roles(request): 

170 out = [] 

171 for r in Role.objects.prefetch_related("role_permissions__permission"): 

172 codes = sorted(rp.permission.code for rp in r.role_permissions.all()) 

173 out.append({"name": r.name, "permissions": codes}) 

174 return out 

175 

176 

177@router.get("/permissions", response=list[PermissionOut], auth=requires("MANAGE_USER_ACCOUNT")) 

178def list_permissions(request): 

179 return list(Permission.objects.all()) 

180 

181 

182@admin_console_router.get( 

183 "/roles", response=RoleMatrixOut, auth=requires("MANAGE_USER_ACCOUNT") 

184) 

185def role_matrix(request): 

186 """The role x permission matrix (Admin Console → Roles & Permissions), pure data 

187 from domain.rbac — no DB query needed.""" 

188 permissions = [ 

189 {"permission": code, "roles": ROLE_PERMISSIONS.get(code, []), **permission_meta(code)} 

190 for code in PERMISSIONS 

191 ] 

192 return {"roles": list(ROLES), "permissions": permissions} 

193 

194 

195def _get_company() -> Company: 

196 """The one Company row (add-admin-organization-api) — lazily created on first read, 

197 same singleton pattern as core_los.LoanNumberSequence.""" 

198 company = Company.objects.first() 

199 if company is None: 

200 company = Company.objects.create() 

201 return company 

202 

203 

204def _company_out(company: Company) -> dict: 

205 return { 

206 "name": company.name, 

207 "nmls_id": company.nmls_id, 

208 "type": company.type, 

209 "dba": company.dba or None, 

210 "phone": company.phone or None, 

211 "mers_id": company.mers_id or None, 

212 "address_line": company.address_line or None, 

213 "city": company.city or None, 

214 "state_code": company.state_code or None, 

215 "postal_code": company.postal_code or None, 

216 "business_hours": company.business_hours or None, 

217 } 

218 

219 

220@admin_console_router.get("/org", response=OrgInfoOut, auth=requires("MANAGE_USER_ACCOUNT")) 

221def org_info(request): 

222 org_units = [ 

223 {"id": str(u.id), "label": u.name} 

224 for u in OrgUnit.objects.all().order_by("-is_headquarters", "name") 

225 ] 

226 

227 loan_officers = [] 

228 officers = ( 

229 User.objects.filter(user_roles__role__name="Loan Officer") 

230 .exclude(nmls_id="") 

231 .distinct() 

232 .prefetch_related("licenses") 

233 ) 

234 for user in officers: 

235 states = [ 

236 lic.state_code if lic.active and lic.sponsored 

237 else f"{lic.state_code} ({'unsponsored' if lic.active else 'inactive'})" 

238 for lic in user.licenses.all() 

239 ] 

240 loan_officers.append( 

241 { 

242 "id": user.id, 

243 "name": f"{user.first_name} {user.last_name}", 

244 "nmls_id": user.nmls_id, 

245 "states": states, 

246 } 

247 ) 

248 

249 return { 

250 "company": _company_out(_get_company()), 

251 "org_units": org_units, 

252 "user_count": User.objects.filter(is_staff=True).count(), 

253 "loan_officers": loan_officers, 

254 } 

255 

256 

257@admin_console_router.patch( 

258 "/company", response=CompanyOut, auth=requires("MANAGE_USER_ACCOUNT") 

259) 

260def update_company(request, payload: CompanyPatch): 

261 data = payload.model_dump(exclude_unset=True) 

262 if "name" in data and not data["name"]: 

263 raise DomainError([AppError("VALIDATION", "Company name is required.", "name")]) 

264 company = _get_company() 

265 for field, value in data.items(): 

266 setattr(company, field, value if value is not None else "") 

267 company.save() 

268 return _company_out(company) 

269 

270 

271DEFAULT_SECTIONS = [ 

272 "mortgage", "personal", "coborrower", "employment", "income", 

273 "assets", "liabilities", "reo", "declarations", "demographics", 

274 "additional", "credit", "review", 

275] 

276 

277 

278def _pos_config_out(company: Company) -> dict: 

279 c = company.pos_config or {} 

280 sections = DEFAULT_SECTIONS 

281 languages = c.get("languages") or ["English", "Spanish"] 

282 default_language = c.get("default_language") or c.get("defaultLanguage") or "English" 

283 programs = c.get("programs") or ["Standard", "DSCR", "VA", "FHA"] 

284 sections_by_program = c.get("sections_by_program") or { 

285 "Standard": sections, 

286 "DSCR": [s for s in sections if s not in ("employment", "income")], 

287 "VA": sections, 

288 "FHA": sections, 

289 } 

290 features = { 

291 "credit_auth_required": c.get("credit_auth_required", True), 

292 "allow_submit_without_credit_consent": c.get("allow_submit_without_credit_consent", False), 

293 "generate_needs_on_submit": c.get("generate_needs_on_submit", True), 

294 "co_borrower_invite": True, 

295 "doc_upload": True, 

296 "e_consent": True, 

297 } 

298 branding = { 

299 "dba": company.dba or None, 

300 "use_dba_in_disclosures": False, 

301 } 

302 return { 

303 "host": "posapi.my1003app.com", 

304 "sections": sections, 

305 "languages": languages, 

306 "default_language": default_language, 

307 "programs": programs, 

308 "sections_by_program": sections_by_program, 

309 "features": features, 

310 "branding": branding, 

311 } 

312 

313 

314def _lead_form_config_out(company: Company) -> dict: 

315 c = company.lead_form_config or {} 

316 return { 

317 "sources": c.get("sources") or ["Website", "Referral", "Zillow", "Repeat", "Realtor", "SocialMedia", "Other"], 

318 "statuses": c.get("statuses") or ["New", "Contacted", "Qualified", "Nurture", "Converted", "Lost"], 

319 "tags": c.get("tags") or ["Hot", "Warm", "Cold", "Pre-approved", "VIP"], 

320 "required_fields": c.get("required_fields") or ["firstName", "lastName"], 

321 } 

322 

323 

324def _agent_config_out(company: Company) -> dict: 

325 c = company.agent_config or {} 

326 return { 

327 "agent_portal_enabled": c.get("agent_portal_enabled", True), 

328 "co_brand_letters": c.get("co_brand_letters", False), 

329 "show_agent_on_letter": c.get("show_agent_on_letter", True), 

330 "realtor_domain": c.get("realtor_domain", "myagentloans.com"), 

331 } 

332 

333 

334def _non_delegated_config_out(company: Company) -> dict: 

335 c = company.non_delegated_config or {} 

336 return { 

337 "enabled": c.get("enabled", False), 

338 "sponsor_name": c.get("sponsor_name") or None, 

339 "sponsor_ein": c.get("sponsor_ein") or None, 

340 "underwriting_contact": c.get("underwriting_contact") or None, 

341 } 

342 

343 

344# --- 1003 Field Config --------------------------------------------------------- 

345@admin_console_router.get("/field-configs", response=FieldConfigsOut, auth=StaffAuth()) 

346def field_configs(request): 

347 return get_field_configs() 

348 

349 

350# --- POS App Config ------------------------------------------------------------ 

351@admin_console_router.get("/pos-config", response=PosConfigOut, auth=StaffAuth()) 

352def get_pos_config(request): 

353 return _pos_config_out(_get_company()) 

354 

355 

356@admin_console_router.patch("/pos-config", response=PosConfigOut, auth=requires("EDIT_ORG_POS_APP_CONFIG")) 

357def update_pos_config(request, payload: PosConfigPatch): 

358 company = _get_company() 

359 current = dict(company.pos_config or {}) 

360 for k, v in payload.model_dump(exclude_unset=True).items(): 

361 current[k] = v 

362 company.pos_config = current 

363 company.save(update_fields=["pos_config"]) 

364 return _pos_config_out(company) 

365 

366 

367# --- Lead Form Config ---------------------------------------------------------- 

368@admin_console_router.get("/lead-form-config", response=LeadFormConfigOut, auth=StaffAuth()) 

369def get_lead_form_config(request): 

370 return _lead_form_config_out(_get_company()) 

371 

372 

373@admin_console_router.patch("/lead-form-config", response=LeadFormConfigOut, auth=requires("EDIT_LEAD_DETAILS")) 

374def update_lead_form_config(request, payload: LeadFormConfigPatch): 

375 company = _get_company() 

376 current = dict(company.lead_form_config or {}) 

377 for k, v in payload.model_dump(exclude_unset=True).items(): 

378 current[k] = v 

379 company.lead_form_config = current 

380 company.save(update_fields=["lead_form_config"]) 

381 return _lead_form_config_out(company) 

382 

383 

384# --- Agent Config -------------------------------------------------------------- 

385@admin_console_router.get("/agent-config", response=AgentConfigOut, auth=StaffAuth()) 

386def get_agent_config(request): 

387 return _agent_config_out(_get_company()) 

388 

389 

390@admin_console_router.patch("/agent-config", response=AgentConfigOut, auth=requires("EDIT_ORG_POS_APP_CONFIG")) 

391def update_agent_config(request, payload: AgentConfigPatch): 

392 company = _get_company() 

393 current = dict(company.agent_config or {}) 

394 for k, v in payload.model_dump(exclude_unset=True).items(): 

395 current[k] = v 

396 company.agent_config = current 

397 company.save(update_fields=["agent_config"]) 

398 return _agent_config_out(company) 

399 

400 

401# --- Non-Delegated Config ------------------------------------------------------ 

402@admin_console_router.get("/non-delegated-config", response=NonDelegatedConfigOut, auth=StaffAuth()) 

403def get_non_delegated_config(request): 

404 return _non_delegated_config_out(_get_company()) 

405 

406 

407@admin_console_router.patch("/non-delegated-config", response=NonDelegatedConfigOut, auth=requires("EDIT_AUS_CREDENTIALS")) 

408def update_non_delegated_config(request, payload: NonDelegatedConfigPatch): 

409 company = _get_company() 

410 current = dict(company.non_delegated_config or {}) 

411 for k, v in payload.model_dump(exclude_unset=True).items(): 

412 current[k] = v 

413 company.non_delegated_config = current 

414 company.save(update_fields=["non_delegated_config"]) 

415 return _non_delegated_config_out(company) 

416 

417 

418# --- API Clients --------------------------------------------------------------- 

419@admin_console_router.get("/api-clients", response=list[ApiClientOut], auth=requires("MANAGE_USER_ACCOUNT")) 

420def list_api_clients(request): 

421 clients = ApiClient.objects.all() 

422 return [ 

423 { 

424 "client_id": c.client_id, 

425 "org_id": str(c.org_unit_id or "HQ"), 

426 "label": c.label or None, 

427 "has_key": bool(c.api_key_enc), 

428 } 

429 for c in clients 

430 ] 

431 

432 

433@admin_console_router.post("/api-clients", response={201: NewApiCredentialsOut}, auth=requires("MANAGE_USER_ACCOUNT")) 

434def create_api_client(request, payload: ApiClientCreate): 

435 client_id = f"client_{get_random_string(16)}" 

436 client_secret = get_random_string(32) 

437 api_key = f"lmn_{get_random_string(32)}" 

438 ApiClient.objects.create( 

439 client_id=client_id, 

440 label=payload.label or "Zapier", 

441 secret_enc=encrypt_secret(client_secret), 

442 api_key_enc=encrypt_secret(api_key), 

443 ) 

444 return 201, { 

445 "client_id": client_id, 

446 "client_secret": client_secret, 

447 "api_key": api_key, 

448 } 

449 

450 

451# --- State Compliance ---------------------------------------------------------- 

452@compliance_router.get("/compliance/states", response=list[StateComplianceOut], auth=StaffAuth()) 

453def list_compliance_states(request): 

454 return STATE_COMPLIANCE 

455 

456 

457@compliance_router.post("/compliance/validate", response=ComplianceCheckOut, auth=StaffAuth()) 

458def validate_compliance(request, payload: ComplianceCheckInput): 

459 lo_dict = None 

460 if payload.loan_officer_id: 

461 user = User.objects.filter(pk=payload.loan_officer_id).prefetch_related("licenses").first() 

462 if user: 462 ↛ 471line 462 didn't jump to line 471 because the condition on line 462 was always true

463 lo_dict = { 

464 "id": user.id, 

465 "name": f"{user.first_name} {user.last_name}".strip() or user.email, 

466 "licenses": [ 

467 {"state_code": lic.state_code, "active": lic.active, "sponsored": lic.sponsored} 

468 for lic in user.licenses.all() 

469 ], 

470 } 

471 res = validate_state_compliance( 

472 state_code=payload.state_code, 

473 lo=lo_dict, 

474 processor=payload.processor.dict() if payload.processor else None, 

475 ) 

476 if not res.ok: 476 ↛ 477line 476 didn't jump to line 477 because the condition on line 476 was never true

477 raise DomainError(res.errors) 

478 return res.value 

479 

480 

481# --- Plans & Billing ----------------------------------------------------------- 

482@billing_router.get("/billing", response=BillingOut, auth=StaffAuth()) 

483def get_billing(request): 

484 seats = User.objects.filter(is_active=True, is_staff=True).count() or 1 

485 price_per_seat = 110 

486 mrr = seats * price_per_seat 

487 today = date.today() 

488 if today.month == 12: 488 ↛ 489line 488 didn't jump to line 489 because the condition on line 488 was never true

489 next_invoice = date(today.year + 1, 1, 1) 

490 else: 

491 next_invoice = date(today.year, today.month + 1, 1) 

492 return { 

493 "provider": "unconfigured", 

494 "plan": "Pro", 

495 "status": "active", 

496 "seats": seats, 

497 "price_per_seat": price_per_seat, 

498 "mrr": mrr, 

499 "next_invoice_date": next_invoice.isoformat(), 

500 "features": [ 

501 "Loan Origination & Pipeline", 

502 "1003 Application & Credit Ingestion", 

503 "Pricing Engine & Lenders", 

504 "Document Management & E-Signatures", 

505 "State Compliance Engine", 

506 "Partner Automation API", 

507 ], 

508 } 

509 

510 

511# --- Client Provisioning ------------------------------------------------------- 

512@provision_router.get("/provision/example", response=dict, auth=StaffAuth()) 

513def get_provision_example(request): 

514 return EXAMPLE_CONFIG 

515 

516 

517@provision_router.post("/provision/preview", response=ProvisionPlanOut, auth=requires("MANAGE_USER_ACCOUNT")) 

518def preview_provision(request, payload: ProvisionPreviewInput): 

519 val = validate_client_config(payload.config) 

520 if not val.ok: 520 ↛ 521line 520 didn't jump to line 521 because the condition on line 520 was never true

521 raise DomainError(val.errors) 

522 return summarize_plan(payload.config) 

523 

524 

525@provision_router.post("/provision/apply", response={201: ProvisionApplyOut}, auth=requires("MANAGE_USER_ACCOUNT")) 

526def apply_provision(request, payload: ProvisionApplyInput): 

527 cfg = payload.config 

528 val = validate_client_config(cfg) 

529 if not val.ok: 529 ↛ 530line 529 didn't jump to line 530 because the condition on line 529 was never true

530 raise DomainError(val.errors) 

531 

532 if payload.clear_loans: 532 ↛ 533line 532 didn't jump to line 533 because the condition on line 532 was never true

533 from apps.core_los.models import Loan 

534 Loan.objects.all().delete() 

535 

536 company_data = cfg.get("company", {}) 

537 company = _get_company() 

538 company.name = company_data.get("name", company.name) 

539 company.nmls_id = company_data.get("nmls_id") or company_data.get("nmlsId") or company.nmls_id 

540 if company_data.get("type"): 540 ↛ 542line 540 didn't jump to line 542 because the condition on line 540 was always true

541 company.type = company_data["type"] 

542 company.save() 

543 

544 # Provision branches 

545 branch_map = {} 

546 for b in cfg.get("branches", []): 

547 unit, _ = OrgUnit.objects.get_or_create( 

548 name=b.get("label") or b.get("id"), 

549 defaults={"is_headquarters": b.get("id") == "HQ"}, 

550 ) 

551 branch_map[b.get("id")] = unit 

552 

553 # Provision people 

554 for p in cfg.get("people", []): 

555 email = p.get("email", "").strip().lower() 

556 if not email: 556 ↛ 557line 556 didn't jump to line 557 because the condition on line 556 was never true

557 continue 

558 first_name, _, last_name = (p.get("name") or "User").partition(" ") 

559 user, created = User.objects.get_or_create( 

560 email=email, 

561 defaults={ 

562 "first_name": first_name, 

563 "last_name": last_name, 

564 "is_staff": True, 

565 "nmls_id": p.get("nmls_id") or p.get("nmlsId") or "", 

566 }, 

567 ) 

568 if created and p.get("password"): 568 ↛ 573line 568 didn't jump to line 573 because the condition on line 568 was always true

569 user.set_password(p["password"]) 

570 user.save() 

571 

572 # Roles 

573 for r in p.get("roles", []): 

574 role_obj = Role.objects.filter(name=r.get("role")).first() 

575 org_id = r.get("org_unit_id") or r.get("orgUnitId") 

576 org_unit = branch_map.get(org_id) 

577 if role_obj: 577 ↛ 573line 577 didn't jump to line 573 because the condition on line 577 was always true

578 UserRole.objects.get_or_create(user=user, role=role_obj, org_unit=org_unit) 

579 

580 # Licenses 

581 for lic in p.get("licenses", []): 

582 sc = (lic.get("state_code") or lic.get("stateCode") or "").upper() 

583 if sc: 583 ↛ 581line 583 didn't jump to line 581 because the condition on line 583 was always true

584 License.objects.update_or_create( 

585 user=user, 

586 state_code=sc, 

587 defaults={ 

588 "active": lic.get("active", True), 

589 "sponsored": lic.get("sponsored", True), 

590 }, 

591 ) 

592 

593 if cfg.get("api_client") or cfg.get("apiClient"): 593 ↛ 603line 593 didn't jump to line 603 because the condition on line 593 was always true

594 client_data = cfg.get("api_client") or cfg.get("apiClient") or {} 

595 cid = client_data.get("client_id") or client_data.get("clientId") 

596 secret = client_data.get("client_secret") or client_data.get("clientSecret") 

597 if cid: 597 ↛ 603line 597 didn't jump to line 603 because the condition on line 597 was always true

598 ApiClient.objects.update_or_create( 

599 client_id=cid, 

600 defaults={"secret_enc": encrypt_secret(secret or "secret")}, 

601 ) 

602 

603 return 201, {"plan": summarize_plan(cfg)} 

604 

605 

606# --- Users --------------------------------------------------------------------- 

607@router.get("/users", response=list[UserDetailOut], auth=requires("MANAGE_USER_ACCOUNT")) 

608def list_users(request, org_unit_id: int | None = Query(None)): 

609 qs = User.objects.filter(is_staff=True).prefetch_related("user_roles__role", "licenses") 

610 if org_unit_id is not None: 

611 qs = qs.filter(org_unit_id=org_unit_id) 

612 return [_user_detail(user) for user in qs] 

613 

614 

615@router.post( 

616 "/users", 

617 response={201: UserInviteOut}, 

618 auth=requires("MANAGE_USER_ACCOUNT"), 

619 exclude_none=True, 

620) 

621def invite_user(request, payload: UserInviteInput): 

622 name = payload.name.strip() 

623 email = payload.email.strip() 

624 if not name or not email: 624 ↛ 625line 624 didn't jump to line 625 because the condition on line 624 was never true

625 raise DomainError([AppError("VALIDATION", "Name and email are required.")]) 

626 if User.objects.filter(email__iexact=email).exists(): 

627 raise DomainError( 

628 [AppError("VALIDATION", "An account with that email already exists.", "email")] 

629 ) 

630 role = _resolve_role(payload.role_name) 

631 org = _resolve_org_unit(payload.org_unit_id) 

632 

633 first_name, _, last_name = name.partition(" ") 

634 temp_password = "lumen-" + get_random_string( 

635 6, allowed_chars="abcdefghijklmnopqrstuvwxyz0123456789" 

636 ) 

637 from apps.communications import emails 

638 from apps.communications.delivery import EmailSendError, deliver_email 

639 

640 company = _get_company() 

641 company_label = company.dba or company.name or "Lumen Lending" 

642 with db_transaction.atomic(): 

643 user = User.objects.create_user( 

644 email=email, 

645 password=temp_password, 

646 first_name=first_name, 

647 last_name=last_name or "", 

648 org_unit=org, 

649 is_staff=True, 

650 nmls_id=payload.nmls_id, 

651 ) 

652 UserRole.objects.get_or_create(user=user, role=role, org_unit=org) 

653 try: 

654 # The user exists only if the invite email went out (add-sendgrid-email). 

655 subject, content = emails.staff_invite( 

656 company=company_label, 

657 first_name=first_name, 

658 role=role.name, 

659 login_url=f"{settings.STAFF_LOS_URL}/login", 

660 email=email, 

661 temp_password=temp_password, 

662 ) 

663 deliver_email( 

664 to=email, 

665 subject=subject, 

666 content=content, 

667 trigger="user:invite", 

668 automated=True, 

669 track_clicks=False, 

670 raise_on_failure=True, 

671 ) 

672 except EmailSendError as exc: 

673 raise DomainError( 

674 [AppError("EMAIL_SEND_FAILED", "We couldn't send the invite email. Try again.")] 

675 ) from exc 

676 # The password is only in the email; development also returns it for local testing. 

677 result = {"user": _user_detail(user), "email_sent": True} 

678 if settings.DEBUG: 678 ↛ 679line 678 didn't jump to line 679 because the condition on line 678 was never true

679 result["devTempPassword"] = temp_password 

680 return 201, result 

681 

682 

683@router.patch( 

684 "/users/{user_id}/active", response=UserDetailOut, auth=requires("MANAGE_USER_ACCOUNT") 

685) 

686def set_user_active(request, user_id: int, payload: UserActiveUpdate): 

687 user = _user(user_id) 

688 user.is_active = payload.is_active 

689 user.save(update_fields=["is_active"]) 

690 return _user_detail(user) 

691 

692 

693@router.get("/users/{user_id}", response=UserDetailOut, auth=requires("MANAGE_USER_ACCOUNT")) 

694def get_user(request, user_id: int): 

695 return _user_detail(_user(user_id)) 

696 

697 

698# --- Role + branch assignment -------------------------------------------------- 

699@router.post( 

700 "/users/{user_id}/role-assignments", 

701 response={201: RoleAssignmentOut}, 

702 auth=requires("MANAGE_USER_ACCOUNT"), 

703) 

704def assign_role(request, user_id: int, payload: RoleAssignmentCreate): 

705 user = _user(user_id) 

706 role = Role.objects.filter(name=payload.role_name).first() 

707 if role is None: 

708 raise DomainError([AppError("NOT_FOUND", f"Role not found: {payload.role_name}")]) 

709 org = None 

710 if payload.org_unit_id is not None: 710 ↛ 714line 710 didn't jump to line 714 because the condition on line 710 was always true

711 org = OrgUnit.objects.filter(pk=payload.org_unit_id).first() 

712 if org is None: 712 ↛ 713line 712 didn't jump to line 713 because the condition on line 712 was never true

713 raise DomainError([AppError("NOT_FOUND", "Org unit not found")])