Coverage for domain/provisioning.py: 77%
103 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-04 14:06 +0000
1"""Client provisioning / installer domain service.
3Validates incoming ClientConfig, generates ProvisionPlan summary, and provides EXAMPLE_CONFIG.
4Matches admin.types.ts and arive-spec/src/domain/provisioning.ts.
5"""
7from __future__ import annotations
9from typing import Any
11from domain.rbac import ROLES
12from domain.result import AppError, Result, err, ok
13from domain.state_compliance import get_state_rule
15# Representative seed counts (products, rules, templates)
16SEED_PRODUCTS_COUNT = 6
17SEED_RULES_COUNT = 8
18SEED_TEMPLATES_COUNT = 5
20EXAMPLE_CONFIG: dict[str, Any] = {
21 "company": {"name": "Summit Peak Lending", "nmls_id": "1234567", "type": "Mortgage Broker"},
22 "branches": [{"id": "HQ", "label": "Headquarters"}, {"id": "BR1", "label": "Denver Branch"}],
23 "people": [
24 {"name": "Dana Owner", "email": "owner@summitpeak.com", "password": "changeme", "roles": [{"role": "Owner", "org_unit_id": "HQ"}]},
25 {"name": "Sam Admin", "email": "admin@summitpeak.com", "password": "changeme", "roles": [{"role": "Company Admin", "org_unit_id": "HQ"}]},
26 {"name": "Riley Officer", "email": "riley@summitpeak.com", "password": "changeme", "roles": [{"role": "Loan Officer", "org_unit_id": "BR1"}], "nmls_id": "2000111", "licenses": [{"state_code": "CO", "sponsored": True}, {"state_code": "TX", "sponsored": True}]},
27 {"name": "Pat Processor", "email": "pat@summitpeak.com", "password": "changeme", "roles": [{"role": "Processor", "org_unit_id": "BR1"}]},
28 ],
29 "api_client": {"client_id": "summitpeak-demo", "client_secret": "change-this-secret", "org_id": "BR1"},
30 "options": {"seed_products": True, "seed_rules": True, "seed_templates": True},
31}
34def _get_val(d: dict[str, Any], snake_k: str, camel_k: str, default: Any = None) -> Any:
35 if snake_k in d:
36 return d[snake_k]
37 if camel_k in d: 37 ↛ 38line 37 didn't jump to line 38 because the condition on line 37 was never true
38 return d[camel_k]
39 return default
42def validate_client_config(cfg: dict[str, Any]) -> Result:
43 errors: list[str] = []
44 company = cfg.get("company") or {}
45 company_name = str(company.get("name") or "").strip()
46 company_nmls = str(_get_val(company, "nmls_id", "nmlsId") or "").strip()
48 if not company_name: 48 ↛ 49line 48 didn't jump to line 49 because the condition on line 48 was never true
49 errors.append("company.name is required")
50 if not company_nmls: 50 ↛ 51line 50 didn't jump to line 51 because the condition on line 50 was never true
51 errors.append("company.nmlsId is required")
53 branches = cfg.get("branches") or []
54 if not branches: 54 ↛ 55line 54 didn't jump to line 55 because the condition on line 54 was never true
55 errors.append("at least one branch (org unit) is required")
56 branch_ids = {str(b.get("id")) for b in branches if b.get("id")}
58 people = cfg.get("people") or []
59 if not people: 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true
60 errors.append("at least one person is required")
62 has_admin = False
63 for p in people:
64 for r in p.get("roles") or []:
65 role_name = r.get("role")
66 if role_name in ("Owner", "Company Admin"):
67 has_admin = True
68 break
69 if not has_admin: 69 ↛ 70line 69 didn't jump to line 70 because the condition on line 69 was never true
70 errors.append("at least one Owner or Company Admin is required")
72 emails: set[str] = set()
73 for i, p in enumerate(people):
74 who = p.get("name") or f"person[{i}]"
75 name = str(p.get("name") or "").strip()
76 email = str(p.get("email") or "").strip()
77 password = str(p.get("password") or "")
79 if not name: 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true
80 errors.append(f"person[{i}].name is required")
81 if not email: 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true
82 errors.append(f"{who}: email is required")
83 else:
84 k = email.lower()
85 if k in emails: 85 ↛ 86line 85 didn't jump to line 86 because the condition on line 85 was never true
86 errors.append(f"{who}: duplicate email {email}")
87 emails.add(k)
89 if not password or len(password) < 4: 89 ↛ 90line 89 didn't jump to line 90 because the condition on line 89 was never true
90 errors.append(f"{who}: password must be at least 4 characters")
92 roles = p.get("roles") or []
93 if not roles: 93 ↛ 94line 93 didn't jump to line 94 because the condition on line 93 was never true
94 errors.append(f"{who}: at least one role is required")
95 for r in roles:
96 role_name = r.get("role")
97 org_unit_id = str(_get_val(r, "org_unit_id", "orgUnitId") or "")
98 if role_name not in ROLES: 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true
99 errors.append(f'{who}: unknown role "{role_name}"')
100 if org_unit_id not in branch_ids: 100 ↛ 101line 100 didn't jump to line 101 because the condition on line 100 was never true
101 errors.append(f'{who}: role references unknown branch "{org_unit_id}"')
103 licenses = p.get("licenses") or []
104 nmls_id = _get_val(p, "nmls_id", "nmlsId")
105 for lic in licenses:
106 state_code = str(_get_val(lic, "state_code", "stateCode") or "")
107 if not get_state_rule(state_code): 107 ↛ 108line 107 didn't jump to line 108 because the condition on line 107 was never true
108 errors.append(f'{who}: unknown license state "{state_code}"')
109 if licenses and not nmls_id: 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true
110 errors.append(f"{who}: nmlsId is required for a licensed loan officer")
112 if errors: 112 ↛ 113line 112 didn't jump to line 113 because the condition on line 112 was never true
113 return err(*[AppError("PROVISION_INVALID", msg) for msg in errors])
114 return ok(True)
117def summarize_plan(cfg: dict[str, Any]) -> dict[str, Any]:
118 company = cfg.get("company") or {}
119 company_name = str(company.get("name") or "")
120 company_nmls = str(_get_val(company, "nmls_id", "nmlsId") or "")
121 branches = cfg.get("branches") or []
122 people = cfg.get("people") or []
124 los = [p for p in people if _get_val(p, "nmls_id", "nmlsId")]
125 licensed_states: set[str] = set()
126 warnings: list[str] = []
128 total_licenses = 0
129 for lo in los:
130 for lic in lo.get("licenses") or []:
131 sc = str(_get_val(lic, "state_code", "stateCode") or "").upper()
132 if sc: 132 ↛ 135line 132 didn't jump to line 135 because the condition on line 132 was always true
133 licensed_states.add(sc)
134 total_licenses += 1
135 if lic.get("sponsored") is False: 135 ↛ 136line 135 didn't jump to line 136 because the condition on line 135 was never true
136 warnings.append(
137 f"{lo.get('name')} is licensed in {sc} but not sponsored there (loans in that state will fail ORG-00002)."
138 )
140 if not los: 140 ↛ 141line 140 didn't jump to line 141 because the condition on line 140 was never true
141 warnings.append("No licensed loan officers — no loans can be originated until one is added.")
143 options = cfg.get("options") or {}
144 seed_products = _get_val(options, "seed_products", "seedProducts", True) is not False
145 seed_rules = _get_val(options, "seed_rules", "seedRules", True) is not False
146 seed_templates = _get_val(options, "seed_templates", "seedTemplates", True) is not False
147 api_client = cfg.get("api_client") or cfg.get("apiClient")
149 return {
150 "company": company_name,
151 "nmls_id": company_nmls,
152 "branches": len(branches),
153 "staff": len(people),
154 "loan_officers": len(los),
155 "licenses": total_licenses,
156 "licensed_states": sorted(list(licensed_states)),
157 "products": SEED_PRODUCTS_COUNT if seed_products else 0,
158 "rules": SEED_RULES_COUNT if seed_rules else 0,
159 "templates": SEED_TEMPLATES_COUNT if seed_templates else 0,
160 "api_client": bool(api_client),
161 "warnings": warnings,
162 }