"""Detailed portal role permissions (Role.portal_permissions): per granted datasource, which of view/create/edit/delete it allows; per granted form, which of view/submit/edit/delete — enforced by the API. An item with no detailed setting allows everything (roles saved before these settings).""" import uuid from app.core.auth import create_access_token from app.models.identity import Role, User, UserRole, Vendor from tests.conftest import AsyncSessionLocal async def _portal_user(tenant_id: uuid.UUID, *roles: dict) -> dict[str, str]: """A portal user holding one role per {action_matrix, portal_permissions}.""" tag = uuid.uuid4().hex[:6] async with AsyncSessionLocal() as db: async with db.begin(): org = Vendor(tenant_id=tenant_id, organization_name=f"Org {tag}") db.add(org) await db.flush() user = User(tenant_id=tenant_id, vendor_id=org.id, name="Public", email=f"public-{tag}@example.com", status="active") db.add(user) await db.flush() for i, r in enumerate(roles): role = Role(tenant_id=tenant_id, name=f"Role {i} {tag}", scope="portal", action_matrix=r["grants"], portal_permissions=r.get("permissions")) db.add(role) await db.flush() db.add(UserRole(user_id=user.id, role_id=role.id)) return {"Authorization": f"Bearer {create_access_token(tenant_id=tenant_id, user_id=user.id, vendor_id=org.id)}"} async def _office(client, headers) -> tuple[str, str]: m = await client.post("/masters", headers=headers, json={"name": f"Office {uuid.uuid4().hex[:6]}", "field_schema": [{"name": "office_name", "type": "text"}]}) assert m.status_code == 201, m.text master_id = m.json()["id"] r = await client.post(f"/masters/{master_id}/records", headers=headers, json={"field_values": {"office_name": "Adoor"}}) assert r.status_code == 201, r.text return master_id, r.json()["id"] async def test_a_view_only_datasource_can_be_read_but_not_changed(client, auth_headers, test_tenant): master_id, record_id = await _office(client, auth_headers) me = await _portal_user(test_tenant["tenant_id"], {"grants": {"masters": [master_id]}, "permissions": {"masters": {master_id: ["view"]}}}) listed = await client.get(f"/masters/{master_id}/records", headers=me) assert listed.status_code == 200 and listed.json() added = await client.post(f"/masters/{master_id}/records", headers=me, json={"field_values": {"office_name": "Pala"}}) assert added.status_code == 403 assert "add records" in added.json()["detail"] assert (await client.patch(f"/masters/{master_id}/records/{record_id}", headers=me, json={"field_values": {"office_name": "X"}})).status_code == 403 assert (await client.delete(f"/masters/{master_id}/records/{record_id}", headers=me)).status_code == 403 async def test_without_detailed_settings_every_action_is_allowed(client, auth_headers, test_tenant): master_id, record_id = await _office(client, auth_headers) me = await _portal_user(test_tenant["tenant_id"], {"grants": {"masters": [master_id]}}) assert (await client.post(f"/masters/{master_id}/records", headers=me, json={"field_values": {"office_name": "Pala"}})).status_code == 201 assert (await client.delete(f"/masters/{master_id}/records/{record_id}", headers=me)).status_code == 204 async def test_roles_add_up(client, auth_headers, test_tenant): master_id, record_id = await _office(client, auth_headers) me = await _portal_user( test_tenant["tenant_id"], {"grants": {"masters": [master_id]}, "permissions": {"masters": {master_id: ["view"]}}}, {"grants": {"masters": [master_id]}, "permissions": {"masters": {master_id: ["edit"]}}}, ) assert (await client.patch(f"/masters/{master_id}/records/{record_id}", headers=me, json={"field_values": {"office_name": "Adoor Depot"}})).status_code == 200 assert (await client.delete(f"/masters/{master_id}/records/{record_id}", headers=me)).status_code == 403 async def _form(client, headers) -> dict: resp = await client.post("/forms", headers=headers, json={"name": f"Application {uuid.uuid4().hex[:6]}", "field_definitions": [{"name": "subject", "type": "text", "label": "Subject"}], "settings": {"allow_drafts": True}}) assert resp.status_code == 201, resp.text return (await client.post(f"/forms/{resp.json()['form_id']}/publish", headers=headers)).json() async def test_a_form_that_can_be_submitted_but_not_edited_or_deleted(client, auth_headers, test_tenant): form = await _form(client, auth_headers) fid = form["form_id"] me = await _portal_user(test_tenant["tenant_id"], {"grants": {"forms": [fid]}, "permissions": {"forms": {fid: ["submit"]}}}) body = {"form_id": fid, "form_version": form["version"], "field_values": {"subject": "Bus timings"}} sent = await client.post("/form-submissions", headers=me, json=body) assert sent.status_code == 201, sent.text sid = sent.json()["id"] assert (await client.patch(f"/form-submissions/{sid}", headers=me, json={"field_values": {"subject": "Fares"}})).status_code == 403 assert (await client.delete(f"/form-submissions/{sid}", headers=me)).status_code == 403 # Their own draft is still theirs to finish or throw away. draft = await client.post("/form-submissions", headers=me, json={**body, "draft": True}) assert draft.status_code == 201, draft.text finished = await client.patch(f"/form-submissions/{draft.json()['id']}", headers=me, json={"field_values": {"subject": "Fares"}, "submit": True}) assert finished.status_code == 200, finished.text async def test_a_view_only_form_cant_be_submitted(client, auth_headers, test_tenant): form = await _form(client, auth_headers) fid = form["form_id"] me = await _portal_user(test_tenant["tenant_id"], {"grants": {"forms": [fid]}, "permissions": {"forms": {fid: []}}}) assert (await client.get(f"/forms/{fid}", headers=me)).status_code == 200 sent = await client.post("/form-submissions", headers=me, json={"form_id": fid, "form_version": form["version"], "field_values": {"subject": "x"}}) assert sent.status_code == 403 assert "submit" in sent.json()["detail"] async def test_settings_are_checked_and_kept_tidy(client, auth_headers): master_id = str(uuid.uuid4()) page_id = str(uuid.uuid4()) made = await client.post( "/roles", headers=auth_headers, json={ "name": f"Detailed {uuid.uuid4().hex[:6]}", "scope": "portal", "action_matrix": {"masters": [master_id], "pages": [page_id]}, "portal_permissions": {"masters": {master_id.upper(): ["delete", "view"]}, "hidden_buttons": {page_id: ["btn1", "list1:add", "list1:row:ra1"]}}, }, ) assert made.status_code == 201, made.text perms = made.json()["portal_permissions"] assert perms == {"masters": {master_id: ["view", "delete"]}, "hidden_buttons": {page_id: ["btn1", "list1:add", "list1:row:ra1"]}} bad = await client.patch(f"/roles/{made.json()['id']}", headers=auth_headers, json={"portal_permissions": {"masters": {master_id: ["approve"]}}}) assert bad.status_code == 422 internal = await client.post("/roles", headers=auth_headers, json={"name": f"Internal {uuid.uuid4().hex[:6]}", "portal_permissions": {"masters": {master_id: ["view"]}}}) assert internal.status_code == 422 cleared = await client.patch(f"/roles/{made.json()['id']}", headers=auth_headers, json={"portal_permissions": {}}) assert cleared.status_code == 200 assert not cleared.json()["portal_permissions"] # ---- import and export --------------------------------------------------- CSV = b"Office\nPala\nKottayam\n" async def _import(client, headers, master_id, commit=True): return await client.post( f"/masters/{master_id}/bulk-import", headers=headers, files={"file": ("offices.csv", CSV, "text/csv")}, data={"column_mapping": '{"Office": "office_name"}', "commit": "true" if commit else "false"}, ) async def _export(client, headers, master_id): return await client.post("/datasets/export.xlsx", headers=headers, json={"source_type": "master", "source_id": master_id}) async def test_import_and_export_are_off_unless_given(client, auth_headers, test_tenant): """A role saved before these settings keeps what it had: no import, and export only with the Report Builder.""" master_id, _ = await _office(client, auth_headers) plain = await _portal_user(test_tenant["tenant_id"], {"grants": {"masters": [master_id]}}) assert (await _import(client, plain, master_id)).status_code == 403 assert (await _export(client, plain, master_id)).status_code == 403 builder = await _portal_user(test_tenant["tenant_id"], {"grants": {"masters": [master_id], "reports": ["enabled"]}}) assert (await _export(client, builder, master_id)).status_code == 200 assert (await _import(client, builder, master_id)).status_code == 403 async def test_a_role_given_import_and_export_can_use_them(client, auth_headers, test_tenant): master_id, _ = await _office(client, auth_headers) me = await _portal_user(test_tenant["tenant_id"], {"grants": {"masters": [master_id]}, "permissions": {"masters": {master_id: ["import", "export"]}}}) imported = await _import(client, me, master_id) assert imported.status_code == 200, imported.text assert imported.json()["committed"] and imported.json()["valid_rows"] == 2 exported = await _export(client, me, master_id) assert exported.status_code == 200 assert exported.headers["content-type"].startswith("application/vnd.openxmlformats") # Import doesn't bring Add with it. assert (await client.post(f"/masters/{master_id}/records", headers=me, json={"field_values": {"office_name": "X"}})).status_code == 403 async def test_a_granted_reports_export_can_be_withheld(client, auth_headers, test_tenant): form = await _form(client, auth_headers) fid = form["form_id"] tpl = {"name": f"Claims {uuid.uuid4().hex[:6]}", "sections": [{"id": "t", "kind": "table", "source_type": "form", "source_id": fid, "columns": ["subject"]}]} report_id = (await client.post("/report-templates", headers=auth_headers, json=tpl)).json()["id"] grants = {"forms": [fid], "report_templates": [report_id]} allowed = await _portal_user(test_tenant["tenant_id"], {"grants": grants}) assert (await client.post(f"/report-templates/{report_id}/export.xlsx", headers=allowed, json={})).status_code == 200 withheld = await _portal_user(test_tenant["tenant_id"], {"grants": grants, "permissions": {"report_templates": {report_id: []}}}) assert (await client.post(f"/report-templates/{report_id}/run", headers=withheld, json={})).status_code == 200 blocked = await client.post(f"/report-templates/{report_id}/export.xlsx", headers=withheld, json={}) assert blocked.status_code == 403 assert "export" in blocked.json()["detail"]