"""Phase A of the enterprise hardening: submission integrity checked on the server (app/crud/form_integrity.py), record-level access policies (app/crud/access_policy.py), field-level permissions (app/crud/ field_access.py), audit of design and data changes, and optimistic concurrency on saves.""" import uuid from sqlalchemy import select from app.core.auth import create_access_token from tests.conftest import AsyncSessionLocal from app.models.audit import AuditLogEntry from app.models.identity import Role, User, UserRole from app.models.workflow import FormSubmission, Workflow, WorkflowInstance FULL = {"masters": ["view", "create", "edit", "delete"], "forms": ["view", "create", "edit", "delete"], "analytics": ["view"]} async def _person(tenant_id: uuid.UUID, name: str, matrix: dict | None = None) -> dict: unique = uuid.uuid4().hex[:8] async with AsyncSessionLocal() as db: async with db.begin(): role = Role(tenant_id=tenant_id, name=f"{name} role {unique}", is_system_role=False, action_matrix=matrix or FULL) user = User(tenant_id=tenant_id, name=f"{name} {unique}", email=f"{name.lower()}-{unique}@example.com", status="active") db.add_all([role, user]) await db.flush() db.add(UserRole(user_id=user.id, role_id=role.id)) ids = {"role_id": str(role.id), "user_id": str(user.id)} token = create_access_token(tenant_id=tenant_id, user_id=uuid.UUID(ids["user_id"]), vendor_id=None) return {**ids, "headers": {"Authorization": f"Bearer {token}"}} async def _form(client, headers, fields, publish=True) -> dict: resp = await client.post("/forms", headers=headers, json={"name": f"Form {uuid.uuid4().hex[:6]}", "field_definitions": fields}) assert resp.status_code == 201, resp.text form = resp.json() if publish: assert (await client.post(f"/forms/{form['form_id']}/publish", headers=headers)).status_code == 200 return form async def _submit(client, headers, form, values): return await client.post("/form-submissions", headers=headers, json={"form_id": form["form_id"], "form_version": form["version"], "field_values": values}) # ---- Submission integrity ------------------------------------------------------------------------- async def test_values_are_typed_normalized_and_limited_to_the_forms_fields(client, auth_headers): form = await _form( client, auth_headers, [ {"name": "qty", "type": "number", "label": "Qty"}, {"name": "agree", "type": "checkbox", "label": "Agree"}, {"name": "when", "type": "date", "label": "When"}, {"name": "tags", "type": "multiselect", "label": "Tags", "options": ["a", "b"]}, {"name": "mail", "type": "email", "label": "Mail"}, ], ) ok = await _submit(client, auth_headers, form, {"qty": "12", "agree": "true", "when": "2026-10-01", "tags": ["a"], "sneaky": "x"}) assert ok.status_code == 201, ok.text assert ok.json()["field_values"] == {"qty": 12, "agree": True, "when": "2026-10-01", "tags": ["a"]} for bad, field in [({"qty": "lots"}, "Qty"), ({"when": "next week"}, "When"), ({"tags": "a"}, "Tags"), ({"tags": ["z"]}, "Tags"), ({"mail": "nope"}, "Mail")]: resp = await _submit(client, auth_headers, form, bad) assert resp.status_code == 422, (bad, resp.text) assert resp.json()["detail"][0]["msg"].startswith(f"{field}: ") async def test_choices_must_be_live_records_passing_the_fields_filters(client, auth_headers, test_tenant): master = (await client.post("/masters", headers=auth_headers, json={"name": f"Offices {uuid.uuid4().hex[:6]}", "field_schema": [{"name": "city", "type": "text"}, {"name": "owner", "type": "text"}]})).json() mine = (await client.post(f"/masters/{master['id']}/records", headers=auth_headers, json={"field_values": {"city": "Kochi", "owner": str(test_tenant["admin_id"])}})).json() other = (await client.post(f"/masters/{master['id']}/records", headers=auth_headers, json={"field_values": {"city": "Pune", "owner": "someone"}})).json() gone = (await client.post(f"/masters/{master['id']}/records", headers=auth_headers, json={"field_values": {"city": "Old", "owner": str(test_tenant["admin_id"])}})).json() await client.delete(f"/masters/{master['id']}/records/{gone['id']}", headers=auth_headers) form = await _form( client, auth_headers, [ { "name": "office", "type": "dropdown", "label": "Office", "master_id": master["id"], "master_display_field": "city", "option_filters": [{"field": "owner", "operator": "equals", "value": "{{current_user.id}}"}], }, {"name": "region", "type": "dropdown", "label": "Region", "master_id": master["id"], "master_display_field": "city", "option_filters": [{"field": "city", "operator": "equals", "value": "{{regionParam}}"}]}, ], ) assert (await _submit(client, auth_headers, form, {"office": mine["id"]})).status_code == 201 # Filtered out for this user, deleted, or not a record at all. for value in (other["id"], gone["id"], str(uuid.uuid4()), "Kochi"): assert (await _submit(client, auth_headers, form, {"office": value})).status_code == 422, value # A filter on a page parameter can't be verified on the server — not enforced. assert (await _submit(client, auth_headers, form, {"region": other["id"]})).status_code == 201 async def test_update_keeps_undeclared_keys_and_rechecks_only_changed_values(client, auth_headers): form = await _form(client, auth_headers, [{"name": "status", "type": "dropdown", "label": "Status", "options": ["open", "closed"]}, {"name": "note", "type": "text", "label": "Note"}]) sub = (await _submit(client, auth_headers, form, {"status": "open"})).json() async with AsyncSessionLocal() as db: async with db.begin(): row = (await db.execute(select(FormSubmission).where(FormSubmission.id == uuid.UUID(sub["id"])))).scalar_one() # A task wrote an extra key; an old value is no longer an option. row.field_values = {"status": "archived", "payment_status": "paid"} resp = await client.patch(f"/form-submissions/{sub['id']}", headers=auth_headers, json={"field_values": {"status": "archived", "note": "hi", "payment_status": "forged", "new_key": 1}}) assert resp.status_code == 200, resp.text assert resp.json()["field_values"] == {"status": "archived", "note": "hi", "payment_status": "paid"} resp = await client.patch(f"/form-submissions/{sub['id']}", headers=auth_headers, json={"field_values": {"status": "bogus"}}) assert resp.status_code == 422 async def test_user_and_file_fields_need_real_people_and_uploads(client, auth_headers, test_tenant): form = await _form(client, auth_headers, [{"name": "who", "type": "user", "label": "Who", "user_type": "internal"}, {"name": "doc", "type": "file_upload", "label": "Doc"}]) assert (await _submit(client, auth_headers, form, {"who": str(test_tenant["admin_id"])})).status_code == 201 assert (await _submit(client, auth_headers, form, {"who": str(uuid.uuid4())})).status_code == 422 assert (await _submit(client, auth_headers, form, {"doc": str(uuid.uuid4())})).status_code == 422 # ---- Record access policies ---------------------------------------------------------------------- async def test_own_records_policy_limits_lists_queries_and_edits(client, auth_headers, test_tenant): clerk = await _person(test_tenant["tenant_id"], "Clerk") master = (await client.post("/masters", headers=auth_headers, json={"name": f"Cases {uuid.uuid4().hex[:6]}", "field_schema": [{"name": "title", "type": "text"}]})).json() base = f"/masters/{master['id']}/records" admin_rec = (await client.post(base, headers=auth_headers, json={"field_values": {"title": "admin's"}})).json() clerk_rec = (await client.post(base, headers=clerk["headers"], json={"field_values": {"title": "clerk's"}})).json() resp = await client.put( f"/masters/{master['id']}/access-policy", headers=auth_headers, json={"enabled": True, "rules": [{"role_ids": [clerk["role_id"]], "scope": "own"}]}, ) assert resp.status_code == 200, resp.text def titles(resp): return sorted(r.get("title") or r["field_values"]["title"] for r in resp.json() if True) if isinstance(resp.json(), list) else None assert titles(await client.get(base, headers=clerk["headers"])) == ["clerk's"] assert (await client.get(f"{base}/count", headers=clerk["headers"])).json()["total"] == 1 assert titles(await client.get(base, headers=auth_headers)) == ["admin's", "clerk's"] # administrators aren't restricted query = await client.post("/datasets/query", headers=clerk["headers"], json={"source_type": "master", "source_id": master["id"]}) assert [r["title"] for r in query.json()["rows"]] == ["clerk's"] assert (await client.patch(f"{base}/{admin_rec['id']}", headers=clerk["headers"], json={"field_values": {"title": "x"}})).status_code == 404 assert (await client.delete(f"{base}/{admin_rec['id']}", headers=clerk["headers"])).status_code == 404 assert (await client.patch(f"{base}/{clerk_rec['id']}", headers=clerk["headers"], json={"field_values": {"title": "edited"}})).status_code == 200 # View but not edit/delete. await client.put( f"/masters/{master['id']}/access-policy", headers=auth_headers, json={"enabled": True, "rules": [{"role_ids": [clerk["role_id"]], "scope": "all", "can_edit": False, "can_delete": False}]}, ) assert titles(await client.get(base, headers=clerk["headers"])) == ["admin's", "edited"] assert (await client.patch(f"{base}/{admin_rec['id']}", headers=clerk["headers"], json={"field_values": {"title": "x"}})).status_code == 403 assert (await client.delete(f"{base}/{clerk_rec['id']}", headers=clerk["headers"])).status_code == 403 # No rule for this person = nothing visible. other = await _person(test_tenant["tenant_id"], "Other") assert (await client.get(base, headers=other["headers"])).json() == [] async def test_filtered_policy_on_form_submissions_uses_viewer_tokens(client, auth_headers, test_tenant): agent = await _person(test_tenant["tenant_id"], "Agent") form = await _form(client, auth_headers, [{"name": "assignee", "type": "text", "label": "Assignee"}, {"name": "title", "type": "text", "label": "Title"}]) a = (await _submit(client, auth_headers, form, {"assignee": agent["user_id"], "title": "for agent"})).json() b = (await _submit(client, auth_headers, form, {"assignee": "nobody", "title": "not for agent"})).json() resp = await client.put( f"/forms/{form['form_id']}/access-policy", headers=auth_headers, json={"enabled": True, "rules": [{"scope": "filtered", "filters": [{"field": "assignee", "operator": "equals", "value": "{{current_user.id}}"}]}]}, ) assert resp.status_code == 200, resp.text listed = await client.get(f"/forms/{form['form_id']}/submissions", headers=agent["headers"]) assert [s["id"] for s in listed.json()] == [a["id"]] assert (await client.get(f"/form-submissions/{b['id']}", headers=agent["headers"])).status_code == 404 assert (await client.get(f"/form-submissions/{a['id']}", headers=agent["headers"])).status_code == 200 policy = (await client.get(f"/forms/{form['form_id']}/access-policy", headers=auth_headers)).json() assert policy["enabled"] and policy["rules"][0]["scope"] == "filtered" # Saving over a newer version of the policy is refused. stale = await client.put( f"/forms/{form['form_id']}/access-policy", headers=auth_headers, json={"enabled": False, "rules": [], "expected_updated_at": "2000-01-01T00:00:00+00:00"}, ) assert stale.status_code == 409 # ---- Field-level permissions --------------------------------------------------------------------- async def test_hidden_and_read_only_fields_per_role(client, auth_headers, test_tenant): clerk = await _person(test_tenant["tenant_id"], "Clerk") form = await _form( client, auth_headers, [ {"name": "title", "type": "text", "label": "Title"}, {"name": "salary", "type": "number", "label": "Salary", "access_rules": [{"role_ids": [clerk["role_id"]], "access": "hidden"}]}, {"name": "grade", "type": "text", "label": "Grade", "required": True, "access_rules": [{"role_ids": [clerk["role_id"]], "access": "read_only"}]}, ], ) sub = (await _submit(client, auth_headers, form, {"title": "Job", "salary": 900, "grade": "A"})).json() access = (await client.get(f"/forms/{form['form_id']}/field-access", headers=clerk["headers"])).json() assert access["fields"] == {"salary": "hidden", "grade": "read_only"} assert (await client.get(f"/forms/{form['form_id']}/field-access", headers=auth_headers)).json()["fields"] == {} got = (await client.get(f"/form-submissions/{sub['id']}", headers=clerk["headers"])).json() assert "salary" not in got["field_values"] and got["field_values"]["grade"] == "A" listed = (await client.get(f"/forms/{form['form_id']}/submissions", headers=clerk["headers"])).json() assert "salary" not in listed[0]["field_values"] query = await client.post("/datasets/query", headers=clerk["headers"], json={"source_type": "form", "source_id": form["form_id"]}) assert "salary" not in query.json()["rows"][0] and "salary" not in query.json()["columns"] probe = await client.post( "/datasets/query", headers=clerk["headers"], json={"source_type": "form", "source_id": form["form_id"], "filters": [{"field": "salary", "operator": "gt", "value": 500}]}, ) assert probe.status_code == 403 assert (await client.get(f"/forms/{form['form_id']}/submissions", headers=clerk["headers"], params={"q": "900"})).json() == [] # The clerk's edit can't change either field; the stored values stay. resp = await client.patch(f"/form-submissions/{sub['id']}", headers=clerk["headers"], json={"field_values": {"title": "Job 2", "salary": 1, "grade": "Z"}}) assert resp.status_code == 200, resp.text full = (await client.get(f"/form-submissions/{sub['id']}", headers=auth_headers)).json()["field_values"] assert full == {"title": "Job 2", "salary": 900, "grade": "A"} async def test_step_rules_follow_the_submissions_workflow_step(client, auth_headers, test_tenant): form = await _form( client, auth_headers, [{"name": "amount", "type": "number", "label": "Amount", "access_rules": [{"steps": ["approve"], "access": "read_only"}]}], ) sub = (await _submit(client, auth_headers, form, {"amount": 10})).json() async with AsyncSessionLocal() as db: async with db.begin(): wf = Workflow(tenant_id=test_tenant["tenant_id"], name="Approvals", step_definitions=[], status="published") db.add(wf) await db.flush() inst = WorkflowInstance(tenant_id=test_tenant["tenant_id"], workflow_id=wf.workflow_id, workflow_version=wf.version, current_step="approve") db.add(inst) await db.flush() row = (await db.execute(select(FormSubmission).where(FormSubmission.id == uuid.UUID(sub["id"])))).scalar_one() row.workflow_instance_id = inst.id inst_id = inst.id access = (await client.get(f"/forms/{form['form_id']}/field-access", headers=auth_headers, params={"submission_id": sub["id"]})).json() assert access == {"fields": {"amount": "read_only"}, "step": "approve"} resp = await client.patch(f"/form-submissions/{sub['id']}", headers=auth_headers, json={"field_values": {"amount": 99}}) assert resp.json()["field_values"]["amount"] == 10 async with AsyncSessionLocal() as db: async with db.begin(): (await db.get(WorkflowInstance, inst_id)).current_step = "done" resp = await client.patch(f"/form-submissions/{sub['id']}", headers=auth_headers, json={"field_values": {"amount": 99}}) assert resp.json()["field_values"]["amount"] == 99 # ---- Audit and concurrency ----------------------------------------------------------------------- async def _audit(tenant_id, action): async with AsyncSessionLocal() as db: rows = await db.execute(select(AuditLogEntry).where(AuditLogEntry.tenant_id == tenant_id, AuditLogEntry.action == action).order_by(AuditLogEntry.occurred_at)) return list(rows.scalars()) async def test_design_and_data_changes_are_audited(client, auth_headers, test_tenant): form = await _form(client, auth_headers, [{"name": "a", "type": "text", "label": "A"}], publish=False) resp = await client.patch( f"/forms/{form['form_id']}", headers=auth_headers, json={"field_definitions": [{"name": "a", "type": "text", "label": "A2"}, {"name": "b", "type": "number", "label": "B"}]}, ) assert resp.status_code == 200 entry = (await _audit(test_tenant["tenant_id"], "form.update"))[-1] assert entry.details["fields_added"] == ["b"] and entry.details["fields_changed"] == ["a"] await client.post(f"/forms/{form['form_id']}/publish", headers=auth_headers) published = (await client.get(f"/forms/{form['form_id']}", headers=auth_headers)).json() sub = (await _submit(client, auth_headers, published, {"a": "x", "b": 1})).json() await client.patch(f"/form-submissions/{sub['id']}", headers=auth_headers, json={"field_values": {"a": "y", "b": 1}}) entry = (await _audit(test_tenant["tenant_id"], "submission.update"))[-1] assert entry.details["changes"] == {"a": ["x", "y"]} assert await _audit(test_tenant["tenant_id"], "form.publish") async def test_stale_saves_are_refused(client, auth_headers): form = await _form(client, auth_headers, [{"name": "a", "type": "text", "label": "A"}], publish=False) first = await client.patch(f"/forms/{form['form_id']}", headers=auth_headers, json={"name": "One", "expected_updated_at": form["updated_at"]}) assert first.status_code == 200, first.text stale = await client.patch(f"/forms/{form['form_id']}", headers=auth_headers, json={"name": "Two", "expected_updated_at": form["updated_at"]}) assert stale.status_code == 409 fresh = await client.patch(f"/forms/{form['form_id']}", headers=auth_headers, json={"name": "Two", "expected_updated_at": first.json()["updated_at"]}) assert fresh.status_code == 200 page = (await client.post("/pages", headers=auth_headers, json={"name": "P"})).json() ok = await client.patch(f"/pages/{page['id']}", headers=auth_headers, json={"name": "P1", "expected_updated_at": page["updated_at"]}) assert ok.status_code == 200 assert (await client.patch(f"/pages/{page['id']}", headers=auth_headers, json={"name": "P2", "expected_updated_at": page["updated_at"]})).status_code == 409 master = (await client.post("/masters", headers=auth_headers, json={"name": f"M {uuid.uuid4().hex[:6]}", "field_schema": [{"name": "t", "type": "text"}]})).json() rec = (await client.post(f"/masters/{master['id']}/records", headers=auth_headers, json={"field_values": {"t": "1"}})).json() ok = await client.patch(f"/masters/{master['id']}/records/{rec['id']}", headers=auth_headers, json={"field_values": {"t": "2"}, "expected_updated_at": rec["updated_at"]}) assert ok.status_code == 200 stale = await client.patch(f"/masters/{master['id']}/records/{rec['id']}", headers=auth_headers, json={"field_values": {"t": "3"}, "expected_updated_at": rec["updated_at"]}) assert stale.status_code == 409 async def test_field_permissions_can_change_on_a_live_form(client, auth_headers, test_tenant): clerk = await _person(test_tenant["tenant_id"], "Clerk") form = await _form(client, auth_headers, [{"name": "salary", "type": "number", "label": "Salary"}]) url = f"/forms/{form['form_id']}/fields/salary/access-rules" resp = await client.put(url, headers=auth_headers, json={"access_rules": [{"role_ids": [clerk["role_id"]], "access": "hidden"}]}) assert resp.status_code == 200, resp.text assert resp.json()["status"] == "published" and resp.json()["version"] == 1 access = (await client.get(f"/forms/{form['form_id']}/field-access", headers=clerk["headers"])).json() assert access["fields"] == {"salary": "hidden"} assert (await client.put(f"/forms/{form['form_id']}/fields/nope/access-rules", headers=auth_headers, json={"access_rules": []})).status_code == 404 entry = (await _audit(test_tenant["tenant_id"], "form.field_access"))[-1] assert entry.details["field"] == "salary" and entry.details["before"] == []