"""Public (signed-out) read access to a datasource or a form's submissions (app/crud/public_data.py): off until turned on, then only the chosen fields and only the rows the fixed filter allows — and data blocks on public pages only over such sources, read-only.""" import uuid import pytest from app.core import throttle @pytest.fixture(autouse=True) def _fresh_rate_limits(): throttle.reset() yield def _slug() -> str: return f"pd{uuid.uuid4().hex[:8]}" async def _site(client, headers) -> str: slug = _slug() resp = await client.put("/tenants/me/site", headers=headers, json={"enabled": True, "slug": slug, "site_name": "Kerala Lines"}) assert resp.status_code == 200, resp.text return slug async def _buses(client, headers) -> tuple[str, str]: depots = (await client.post("/masters", headers=headers, json={"name": f"Pytest Depots {uuid.uuid4().hex[:6]}", "field_schema": [{"name": "depot_name", "type": "varchar"}]})).json() ekm = (await client.post(f"/masters/{depots['id']}/records", headers=headers, json={"field_values": {"depot_name": "Ernakulam"}})).json() buses = await client.post( "/masters", headers=headers, json={ "name": f"Pytest Buses {uuid.uuid4().hex[:6]}", "field_schema": [ {"name": "reg_no", "type": "varchar", "alias": "Registration"}, {"name": "route", "type": "varchar"}, {"name": "status", "type": "single_select", "options": ["Running", "Retired"]}, {"name": "depot", "type": "reference", "reference_master_id": depots["id"], "reference_display_field": "depot_name"}, {"name": "driver_phone", "type": "varchar"}, {"name": "photo", "type": "file"}, ], }, ) assert buses.status_code == 201, buses.text bid = buses.json()["id"] for reg, route, status in (("KL-07-1001", "Aluva - Fort Kochi", "Running"), ("KL-07-1002", "Aluva - Vyttila", "Running"), ("KL-07-0001", "Old route", "Retired")): r = await client.post(f"/masters/{bid}/records", headers=headers, json={"field_values": {"reg_no": reg, "route": route, "status": status, "depot": ekm["id"], "driver_phone": "98470 00000"}}) assert r.status_code == 201, r.text return bid, ekm["id"] PUBLIC = {"enabled": True, "fields": ["reg_no", "route", "depot", "status"], "filters": [{"field": "status", "operator": "equals", "value": "Running"}], "allow_search": True} async def test_public_reads_only_what_was_opened(client, auth_headers): slug = await _site(client, auth_headers) bid, _ = await _buses(client, auth_headers) q = {"source_type": "master", "source_id": bid} # Off by default. assert (await client.post(f"/public/sites/{slug}/data/query", json=q)).status_code == 404 detail = (await client.get(f"/public-data/master/{bid}", headers=auth_headers)).json() assert detail["enabled"] is False assert "photo" not in [f["name"] for f in detail["offerable"]] # files never are bad = await client.put(f"/public-data/master/{bid}", headers=auth_headers, json={**PUBLIC, "fields": ["reg_no", "photo"]}) assert bad.status_code == 422 assert (await client.put(f"/public-data/master/{bid}", headers=auth_headers, json=PUBLIC)).status_code == 200 schema = (await client.get(f"/public/sites/{slug}/data/master/{bid}/schema")).json() assert [(f["name"], f["type"]) for f in schema["fields"]] == [("reg_no", "varchar"), ("route", "varchar"), ("depot", "varchar"), ("status", "single_select")] assert schema["fields"][0]["label"] == "Registration" result = (await client.post(f"/public/sites/{slug}/data/query", json={**q, "sort_field": "reg_no"})).json() # The retired bus is filtered out; a hidden field is never sent; a link reads as its name. assert result["total"] == 2 assert [r["reg_no"] for r in result["rows"]] == ["KL-07-1001", "KL-07-1002"] assert all("driver_phone" not in r and r["depot"] == "Ernakulam" and r["id"] for r in result["rows"]) # Paging and search (over the public text fields only). page2 = (await client.post(f"/public/sites/{slug}/data/query", json={**q, "sort_field": "reg_no", "skip": 1, "limit": 1})).json() assert [r["reg_no"] for r in page2["rows"]] == ["KL-07-1002"] found = (await client.post(f"/public/sites/{slug}/data/query", json={**q, "q": "vyttila"})).json() assert [r["reg_no"] for r in found["rows"]] == ["KL-07-1002"] assert (await client.post(f"/public/sites/{slug}/data/query", json={**q, "q": "98470"})).json()["rows"] == [] # A hidden field can't be shown, filtered, sorted or grouped on — that would leak it. for probe in ( {"columns": ["driver_phone"]}, {"filters": [{"field": "driver_phone", "operator": "contains", "value": "98"}]}, {"filters": [{"match": "any", "filters": [{"field": "driver_phone", "operator": "is_not_empty"}]}]}, {"sort_field": "driver_phone"}, {"group_by": "driver_phone"}, {"columns": ["depot.depot_name"]}, ): assert (await client.post(f"/public/sites/{slug}/data/query", json={**q, **probe})).status_code == 403, probe # The fixed filter can't be widened by the block's own. widened = await client.post(f"/public/sites/{slug}/data/query", json={**q, "filters": [{"field": "status", "operator": "equals", "value": "Retired"}]}) assert widened.json()["total"] == 0 # Charts: grouped counts over a public field. grouped = (await client.post(f"/public/sites/{slug}/data/query", json={**q, "group_by": "route", "aggregate_fn": "count"})).json() assert len(grouped["rows"]) == 2 # Turning it off takes effect at once. await client.put(f"/public-data/master/{bid}", headers=auth_headers, json={**PUBLIC, "enabled": False}) assert (await client.post(f"/public/sites/{slug}/data/query", json=q)).status_code == 404 async def test_public_pages_may_hold_read_only_data_blocks(client, auth_headers): await _site(client, auth_headers) bid, _ = await _buses(client, auth_headers) block = {"id": "l1", "type": "data_list", "data_source": f"master:{bid}", "columns": ["reg_no", "route"], "layout": {"x": 0, "y": 0, "w": 600, "h": 300}} page = (await client.post("/pages", headers=auth_headers, json={"name": "Buses (public)", "blocks": [block]})).json() refused = await client.patch(f"/pages/{page['id']}/site", headers=auth_headers, json={"is_public": True}) assert refused.status_code == 422 and "isn't public" in refused.json()["detail"]["message"] await client.put(f"/public-data/master/{bid}", headers=auth_headers, json=PUBLIC) ok = await client.patch(f"/pages/{page['id']}/site", headers=auth_headers, json={"is_public": True}) assert ok.status_code == 200, ok.text # Read-only: a list that edits can't be published to the public site. await client.patch(f"/pages/{page['id']}", headers=auth_headers, json={"blocks": [{**block, "allow_edit": True}]}) published = await client.post(f"/pages/{page['id']}/publish", headers=auth_headers) assert published.status_code == 422 and "can only be read" in published.json()["detail"]["message"] async def test_form_submissions_public_and_site_must_be_on(client, auth_headers): created = await client.post( "/forms", headers=auth_headers, json={"name": f"Pytest Notices {uuid.uuid4().hex[:6]}", "field_definitions": [{"name": "title", "type": "text", "label": "Title"}, {"name": "internal_note", "type": "text", "label": "Note"}]}, ) form = (await client.post(f"/forms/{created.json()['form_id']}/publish", headers=auth_headers)).json() for title in ("Fare revision", "Route change"): await client.post("/form-submissions", headers=auth_headers, json={"form_id": form["form_id"], "form_version": form["version"], "field_values": {"title": title, "internal_note": "x"}}) await client.put(f"/public-data/form/{form['form_id']}", headers=auth_headers, json={"enabled": True, "fields": ["title", "created_at"]}) q = {"source_type": "form", "source_id": form["form_id"], "sort_field": "title"} slug = _slug() # The website is off: nothing is served. assert (await client.post(f"/public/sites/{slug}/data/query", json=q)).status_code == 404 await client.put("/tenants/me/site", headers=auth_headers, json={"enabled": True, "slug": slug}) rows = (await client.post(f"/public/sites/{slug}/data/query", json=q)).json()["rows"] assert [r["title"] for r in rows] == ["Fare revision", "Route change"] assert all(set(r) == {"id", "title", "created_at"} for r in rows)