"""Builds the FleetSecure GRC app in the GRC tenant, through the API. Order matters: datasources (and their sample records) first, then the portal roles the notification tasks address, the tasks themselves, the forms that run them, the portal organisation and its two users, the forms' sample submissions (made by a portal user, so the portal can see them), the datasources that link to those submissions, and finally pages, menus and role grants (pages.py). Re-running it is safe: everything is looked up by name first and only created when missing; sample data is only added to an empty datasource or form. """ import copy import secrets import string from .client import Api, ApiError from .model_core import CORE, R from .model_tenant import FORMS, FORMS_AFTER, TENANT, TENANT_AFTER_FORMS ROLE_CORE = "GRC Tenant Admin" ROLE_TENANT = "Implementation Team" VENDOR_NAME = "FleetSecureGRC" class Build: def __init__(self, admin: Api, log=print): self.api = admin self.log = log self.portal: Api | None = None # set once the portal users exist self.masters: dict[str, dict] = {} # key -> master (id, name, field_schema) self.forms: dict[str, dict] = {} # key -> form (form_id, version, name) self.rows: dict[str, dict[str, str]] = {} # ds/form key -> row key -> record id self.roles: dict[str, dict] = {} self.tasks: dict[str, dict] = {} self.defs = {d["key"]: d for d in [*CORE, *TENANT, *TENANT_AFTER_FORMS]} self.form_defs = {f["key"]: f for f in [*FORMS, *FORMS_AFTER]} # ---- datasources ---------------------------------------------------------------- def _field_payload(self, f: dict, master_key: str, with_self: bool) -> dict | None: f = {k: v for k, v in f.items()} target = f.pop("_target", None) if target is None: return f if target == "self": if not with_self: return None f["reference_master_id"] = self.masters[master_key]["id"] elif target.startswith("form:"): f["reference_form_id"] = self.forms[target[5:]]["form_id"] else: f["reference_master_id"] = self.masters[target]["id"] return f def ensure_datasource(self, d: dict) -> None: existing = {m["name"]: m for m in self.api.get("/masters", limit=500)} key = d["key"] has_self = any(f.get("_target") == "self" for f in d["fields"]) if d["name"] in existing: m = existing[d["name"]] self.masters[key] = m schema = [p for f in d["fields"] if (p := self._field_payload(f, key, True)) is not None] current = m["field_schema"] if [x["name"] for x in current] != [x["name"] for x in schema] or any( any(a.get(k) != v for k, v in b.items()) for a, b in zip(current, schema) ): m = self.api.patch(f"/masters/{m['id']}", {"field_schema": schema}) self.log(f" updated datasource {d['name']}") if m.get("storage_tier") == "flexible": self.api.post(f"/masters/{m['id']}/storage", {"tier": "table"}) m["storage_tier"] = "table" self.masters[key] = m return schema = [p for f in d["fields"] if (p := self._field_payload(f, key, False)) is not None] m = self.api.post("/masters", {"name": d["name"], "field_schema": schema, "storage_tier": "table"}) self.masters[key] = m if has_self: full = [p for f in d["fields"] if (p := self._field_payload(f, key, True)) is not None] m = self.api.patch(f"/masters/{m['id']}", {"field_schema": full}) self.masters[key] = m self.log(f" created datasource {d['name']} (table)") def _resolve(self, value): if isinstance(value, R): ids = self.rows.get(value.ds, {}) if value.key not in ids: raise ApiError(f"Unresolved link {value.ds}:{value.key}") return ids[value.key] if isinstance(value, list): return [self._resolve(v) for v in value] return value def _values(self, row: dict) -> dict: return {k: self._resolve(v) for k, v in row.items() if not k.startswith("_")} @staticmethod def _find(existing: list[dict], values: dict) -> str | None: """The already-created record holding exactly these values (re-runs).""" def same(a, b): return (sorted(map(str, a)) == sorted(map(str, b))) if isinstance(b, list) else str(a) == str(b) for r in existing: if all(same(r["field_values"].get(k), v) for k, v in values.items()): return r["id"] return None def seed_datasource(self, d: dict) -> None: key = d["key"] m = self.masters[key] existing = self.api.get(f"/masters/{m['id']}/records", limit=500) self.rows.setdefault(key, {}) added = 0 for row in d["rows"]: values = self._values(row) rid = self._find(existing, values) if existing else None if rid is None and not existing: rid = self.api.post(f"/masters/{m['id']}/records", {"field_values": values})["id"] added += 1 if rid and row.get("_k"): self.rows[key][row["_k"]] = rid if added: self.log(f" {d['name']}: {added} sample records") # ---- roles, vendor, users ------------------------------------------------------- def ensure_roles(self) -> None: existing = {r["name"]: r for r in self.api.get("/roles")} for name in (ROLE_CORE, ROLE_TENANT): if name not in existing: existing[name] = self.api.post("/roles", {"name": name, "scope": "portal", "action_matrix": {}}) self.log(f" created portal role {name}") self.roles[name] = existing[name] def ensure_vendor_and_users(self, users: list[dict], password: str) -> dict[str, dict]: vendors = {v["organization_name"]: v for v in self.api.get("/vendors")} vendor = vendors.get(VENDOR_NAME) or self.api.post("/vendors", {"organization_name": VENDOR_NAME}) existing = {u["email"].lower(): u for u in self.api.get("/users", limit=500)} out = {} for spec in users: u = existing.get(spec["email"].lower()) if u is None: u = self.api.post("/users", {"name": spec["name"], "email": spec["email"], "password": password, "vendor_id": vendor["id"], "phone": spec.get("phone")}) u = u.get("user", u) self.log(f" created portal user {spec['name']}") self.api.put(f"/users/{u['id']}/roles/{self.roles[spec['role']]['id']}") out[spec["role"]] = u return out # ---- tasks ------------------------------------------------------------------------ def ensure_tasks(self) -> None: roles = [self.roles[ROLE_CORE]["id"], self.roles[ROLE_TENANT]["id"]] tenant_role = [self.roles[ROLE_TENANT]["id"]] core_role = [self.roles[ROLE_CORE]["id"]] def inputs(*names): return [{"name": n, "type": "string", "source_field": n, "required": False} for n in names] def lookup_node(key, master_key, source_field, output, label): return { "key": key, "type": "data_get", "label": label, "datasource_type": "master", "master_id": self.masters[master_key]["id"], "match_field": "id", "input_params": [{"name": "match_value", "type": "string", "source_field": source_field, "required": False}], "output_params": [{"name": output, "type": "string", "source": output}], "next_nodes": [{"to": "notify"}], } def notify(subject, body, names, role_ids): return {"key": "notify", "type": "create_notification", "label": "Notify", "subject": subject, "body": body, "input_params": inputs(*names), "notify_role_ids": role_ids, "next_nodes": []} specs = { "kpi_assigned": ( "Notify — KPI assigned", "kpi", lambda: [ {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "kpi"}]}, lookup_node("kpi", "kpi", "kpi", "kpi_title", "Find the KPI"), notify("KPI assigned: {{kpi_title}}", "A {{frequency}} KPI ({{level}}) starts on {{start_date}}. Open KPI Assignment to review it.", ["kpi_title", "frequency", "level", "start_date"], tenant_role), ], ), "kpi_result": ( "Notify — KPI result recorded", "kpi", lambda: [ {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "kpi"}]}, lookup_node("kpi", "kpi", "kpi", "kpi_title", "Find the KPI"), notify("KPI result: {{kpi_title}} — {{outcome}}", "Result {{result_value}} recorded for {{period_label}} ({{entry_status}}).", ["kpi_title", "outcome", "result_value", "period_label", "entry_status"], tenant_role), ], ), "audit_planned": ( "Notify — audit planned", None, lambda: [ {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "notify"}]}, notify("Audit planned: {{audit_title}}", "{{audit_type}} from {{start_date}} to {{end_date}} — status {{status}}.", ["audit_title", "audit_type", "start_date", "end_date", "status"], roles), ], ), "assessment": ( "Notify — audit assessment submitted", "audit_task", lambda: [ {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "task"}]}, lookup_node("task", "audit_task", "audit_task", "task_title", "Find the audit task"), notify("Assessment submitted: {{task_title}}", "Overall result {{result}} (design {{design_effectiveness}}, operation {{operational_effectiveness}}).", ["task_title", "result", "design_effectiveness", "operational_effectiveness"], tenant_role), ], ), "soa_saved": ( "Notify — SOA saved", None, lambda: [ {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "notify"}]}, notify("SOA {{status}}: {{soa_title}}", "{{revision}} — {{scope}} scope, {{applicability}}.", ["status", "soa_title", "revision", "scope", "applicability"], roles), ], ), "tenant_saved": ( "Notify — tenant onboarding saved", None, lambda: [ {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "notify"}]}, notify("Tenant {{status}}: {{legal_company_name}}", "{{display_name}} — {{deployment}}; admin {{admin_full_name}} ({{admin_work_email}}).", ["status", "legal_company_name", "display_name", "deployment", "admin_full_name", "admin_work_email"], core_role), ], ), } existing = {t["name"]: t for t in self.api.get("/tasks")} for key, (name, needs, make_nodes) in specs.items(): if key in self.tasks or (needs and needs not in self.masters): continue nodes = make_nodes() for i, n in enumerate(nodes): n.setdefault("position_x", 120 + 260 * i) n.setdefault("position_y", 160) t = existing.get(name) if t is None: t = self.api.post("/tasks", {"name": name, "nodes": nodes}) t = self.api.post(f"/tasks/{t['id']}/publish") self.log(f" created task {name}") self.tasks[key] = t # ---- forms ------------------------------------------------------------------------ FORM_TASKS = {"client_tenant": "tenant_saved", "soa": "soa_saved", "kpi_assignment": "kpi_assigned", "kpi_result": "kpi_result", "audit_plan": "audit_planned", "audit_assessment": "assessment"} def ensure_form(self, f: dict) -> None: existing = {x["name"]: x for x in self.api.get("/forms")} if f["name"] in existing: self.forms[f["key"]] = existing[f["name"]] return fields = [] for fd in copy.deepcopy(f["fields"]): target = fd.pop("_master", None) if target: fd["master_id"] = self.masters[target]["id"] fields.append(fd) task = self.tasks[self.FORM_TASKS[f["key"]]] events = [{"trigger": "on_submit", "action_type": "run_task", "task_id": task["id"]}] form = self.api.post("/forms", {"name": f["name"], "field_definitions": fields, "events": events, "storage_tier": "table"}) form = self.api.post(f"/forms/{form['form_id']}/publish") self.forms[f["key"]] = form self.log(f" created form {f['name']} (table, published)") def seed_form(self, f: dict, as_api: Api) -> None: form = self.forms[f["key"]] existing = as_api.get(f"/forms/{form['form_id']}/submissions", limit=500) self.rows.setdefault(f["key"], {}) added = 0 for row in f["rows"]: values = self._values(row) sid = self._find(existing, values) if existing else None if sid is None and not existing: sid = as_api.post("/form-submissions", {"form_id": form["form_id"], "form_version": form["version"], "field_values": values})["id"] added += 1 if sid and row.get("_k"): self.rows[f["key"]][row["_k"]] = sid if added: self.log(f" {f['name']}: {added} sample submissions") # ---- the whole data layer ------------------------------------------------------- def build_data(self, users: list[dict], password: str) -> dict[str, dict]: self.log("Datasources") for d in [*CORE, *TENANT]: self.ensure_datasource(d) self.seed_datasource(d) self.log("Portal roles") self.ensure_roles() self.log("Tasks") self.ensure_tasks() self.log("Forms") for f in FORMS: self.ensure_form(f) self.log("Portal organisation and users") accounts = self.ensure_vendor_and_users(users, password) return accounts def build_after_forms(self) -> None: self.log("Datasources linked to submissions") for d in TENANT_AFTER_FORMS: self.ensure_datasource(d) self.seed_datasource(d) self.ensure_tasks() for f in FORMS_AFTER: self.ensure_form(f) def make_password() -> str: """16 characters: letters, digits and a symbol — meets any sign-in policy.""" alphabet = string.ascii_letters + string.digits core = "".join(secrets.choice(alphabet) for _ in range(13)) return f"Grc{core}!"