"""A form's submissions stored as a table: it follows the live version (the latest published one, or the draft of a form never published), holds submitted submissions only (never drafts), migrates when a new version is published (or refuses the publish when submissions don't fit) — and queries over it read the typed columns, giving exactly what the JSON gives, a cleared number's or checkbox's "" included.""" import pytest import pytest_asyncio from sqlalchemy import text from tests.conftest import AsyncSessionLocal from app.crud import dataset_query, entity_storage, field_indexes, record_sql from app.models.workflow import Form, FormSubmission from tests.test_record_sql import _both, python_only # noqa: F401 — python_only is a fixture FIELDS = [ {"name": "applicant", "type": "text", "label": "Applicant"}, {"name": "amount", "type": "number", "label": "Amount"}, {"name": "agree", "type": "checkbox", "label": "Agree"}, {"name": "kind", "type": "dropdown", "label": "Kind", "options": ["new", "renewal"]}, {"name": "on", "type": "date", "label": "On"}, {"name": "tags", "type": "multiselect", "label": "Tags", "options": ["a", "b"]}, ] async def _form(client, headers, fields=FIELDS, allow_drafts=True): form = (await client.post("/forms", headers=headers, json={"name": "Grant", "field_definitions": fields, "settings": {"allow_drafts": allow_drafts}})).json() assert (await client.post(f"/forms/{form['form_id']}/publish", headers=headers)).status_code == 200 return form async def _submit(client, headers, form, values, draft=False): resp = await client.post("/form-submissions", headers=headers, json={"form_id": form["form_id"], "form_version": form["version"], "field_values": values, "draft": draft}) assert resp.status_code == 201, resp.text return resp.json() async def _rows(table: str) -> list[dict]: async with AsyncSessionLocal() as db: result = await db.execute(text(f'SELECT to_jsonb(t) FROM entity_data."{table.split(".", 1)[1]}" t ORDER BY submitted_at')) return [r[0] for r in result] async def _store(client, headers, form_id): await field_indexes.wait_for_pending() return await client.post(f"/forms/{form_id}/storage", headers=headers, json={"tier": "table"}) async def test_submitted_submissions_only_and_kept_in_step(client, auth_headers): form = await _form(client, auth_headers) first = await _submit(client, auth_headers, form, {"applicant": "Asha", "amount": 1200.5, "agree": True, "kind": "new", "on": "2026-10-01", "tags": ["a"]}) draft = await _submit(client, auth_headers, form, {"applicant": "Draft only"}, draft=True) resp = await _store(client, auth_headers, form["form_id"]) assert resp.status_code == 200, resp.text info = resp.json() assert info["tier"] == "table" and info["row_count"] == 1 assert {c["field"]: c["sql_type"] for c in info["columns"]} == {"applicant": "text", "amount": "numeric", "agree": "boolean", "kind": "text", "on": "date", "tags": "jsonb"} [row] = await _rows(info["table"]) assert row["id"] == first["id"] and row["amount"] == 1200.5 and row["agree"] is True and row["on"] == "2026-10-01" assert row["form_version"] == form["version"] # A cleared number is "" — empty in the table, not an error. await _submit(client, auth_headers, form, {"applicant": "Ravi", "amount": "", "agree": False}) # The draft joins once it's submitted; a deleted submission leaves. resp = await client.patch(f"/form-submissions/{draft['id']}", headers=auth_headers, json={"field_values": {"applicant": "Draft only", "amount": 10}, "submit": True}) assert resp.status_code == 200, resp.text assert (await client.delete(f"/form-submissions/{first['id']}", headers=auth_headers)).status_code == 204 rows = await _rows(info["table"]) assert sorted(r["applicant"] for r in rows) == ["Draft only", "Ravi"] assert [r["amount"] for r in rows if r["applicant"] == "Ravi"] == [None] async def _columns(table: str) -> dict[str, str]: async with AsyncSessionLocal() as db: rows = await db.execute(text( "SELECT column_name, data_type FROM information_schema.columns WHERE table_schema = 'entity_data' AND table_name = :t" ), {"t": table.split(".", 1)[1]}) return dict(rows.all()) async def test_a_form_can_be_created_as_a_table(client, auth_headers): """The table exists from the start; until the first publish it follows the draft.""" resp = await client.post("/forms", headers=auth_headers, json={"name": "Born typed", "field_definitions": FIELDS[:2], "storage_tier": "table"}) assert resp.status_code == 201, resp.text form = resp.json() assert form["storage_tier"] == "table" and form["status"] == "draft" info = (await client.get(f"/forms/{form['form_id']}/storage", headers=auth_headers)).json() assert info["tier"] == "table" assert (await _columns(info["table"]))["amount"] == "numeric" # Draft edits reach the table… resp = await client.patch(f"/forms/{form['form_id']}", headers=auth_headers, json={"field_definitions": FIELDS}) assert resp.status_code == 200, resp.text assert (await _columns(info["table"]))["on"] == "date" # …and so do submissions once it's published. assert (await client.post(f"/forms/{form['form_id']}/publish", headers=auth_headers)).status_code == 200 await _submit(client, auth_headers, form, {"applicant": "Asha", "amount": 7, "on": "2026-10-01"}) [row] = await _rows(info["table"]) assert row["applicant"] == "Asha" and row["amount"] == 7 listed = {f["form_id"]: f for f in (await client.get("/forms", headers=auth_headers)).json()} assert listed[form["form_id"]]["storage_tier"] == "table" plain = (await client.post("/forms", headers=auth_headers, json={"name": "Born flexible"})).json() assert plain["storage_tier"] == "flexible" async def test_a_never_published_form_can_be_stored_as_a_table(client, auth_headers): form = (await client.post("/forms", headers=auth_headers, json={"name": "Unpublished", "field_definitions": FIELDS})).json() resp = await _store(client, auth_headers, form["form_id"]) assert resp.status_code == 200, resp.text assert resp.json()["tier"] == "table" async def test_publishing_a_new_version_migrates_or_is_refused(client, auth_headers): form = await _form(client, auth_headers) await _submit(client, auth_headers, form, {"applicant": "Asha", "amount": 5}) info = (await _store(client, auth_headers, form["form_id"])).json() # A new version adds a field: the table gains its column on publish, not before. v2 = (await client.post(f"/forms/{form['form_id']}/new-version", headers=auth_headers)).json() fields = [*FIELDS, {"name": "score", "type": "rating", "label": "Score"}] assert (await client.patch(f"/forms/{form['form_id']}", headers=auth_headers, json={"field_definitions": fields})).status_code == 200 assert "score" not in (await _rows(info["table"]))[0] assert (await client.post(f"/forms/{form['form_id']}/publish", headers=auth_headers)).status_code == 200 assert (await _rows(info["table"]))[0]["score"] is None # A version that turns Amount into text doesn't fit the submissions (5 is a number) — refused, still a draft. await client.post(f"/forms/{form['form_id']}/new-version", headers=auth_headers) changed = [{**f, "type": "text"} if f["name"] == "amount" else f for f in fields] await client.patch(f"/forms/{form['form_id']}", headers=auth_headers, json={"field_definitions": changed}) resp = await client.post(f"/forms/{form['form_id']}/publish", headers=auth_headers) assert resp.status_code == 422, resp.text assert resp.json()["detail"]["issues"][0]["field"] == "amount" current = (await client.get(f"/forms/{form['form_id']}", headers=auth_headers)).json() assert current["status"] == "draft" and current["version"] == v2["version"] + 1 history = (await client.get(f"/forms/{form['form_id']}/storage", headers=auth_headers)).json()["history"] assert [h["action"] for h in history] == ["migrate", "promote"] model = (await client.get("/app-model", headers=auth_headers)).json() assert [e["storage"]["tier"] for e in model["entities"] if e["id"] == form["form_id"]] == ["table"] assert (await client.post(f"/forms/{form['form_id']}/storage", headers=auth_headers, json={"tier": "flexible"})).status_code == 200 async with AsyncSessionLocal() as db: assert (await db.execute(text("SELECT to_regclass(:t)"), {"t": info["table"]})).scalar() is None # ---- typed reads over a form ------------------------------------------------------------------ APPLICANTS = ["apple", "Apple", "", None, "10", "zeta", "Éclair", ""] AMOUNTS = [5, 5.0, 12.5, -3, 0, "", None, "", 100] AGREE = [True, False, "", None, ""] KINDS = ["new", "renewal", "", None] @pytest_asyncio.fixture async def form_dataset(test_tenant): tenant_id = test_tenant["tenant_id"] async with AsyncSessionLocal() as db: async with db.begin(): form = Form(tenant_id=tenant_id, name="Typed form", field_definitions=FIELDS, status="published", is_current=True) db.add(form) await db.flush() for i in range(30): values = { "applicant": APPLICANTS[i % len(APPLICANTS)], "amount": AMOUNTS[i % len(AMOUNTS)], "agree": AGREE[i % len(AGREE)], "kind": KINDS[i % len(KINDS)], "tags": ["a"] if i % 2 else [], } values = {k: v for k, v in values.items() if v != ""} db.add(FormSubmission(tenant_id=tenant_id, form_id=form.form_id, form_version=form.version, field_values=values, status="draft" if i % 7 == 6 else "submitted")) await db.flush() source, _ = await entity_storage.form_source(db, tenant_id, form.form_id) await entity_storage.promote(db, tenant_id, None, source) return {"tenant_id": tenant_id, "form_id": form.form_id} def _cases() -> list[dict]: ops_values = { "equals": ["apple", "", 5, 12.5, 0, True, False, "new", None], "not_equals": ["apple", 5, True, ""], "gt": ["b", 5, -3, "", None], "lte": [5, "renewal"], "in": [["apple", 5, True], [0, 100], ["", "new"], []], "contains": ["a", "5", "true", "", "new"], "between": [[0, 12.5], ["2026-09-01", "2026-09-30"]], "is_empty": [None], "is_not_empty": [None], } return [{"field": f, "operator": op, "value": v} for f in ["applicant", "amount", "agree", "kind", "tags"] for op, vals in ops_values.items() for v in vals] async def test_form_typed_filters_match_python(form_dataset, python_only): ds = form_dataset async with AsyncSessionLocal() as db: src = await record_sql.load_source(db, ds["tenant_id"], "form", ds["form_id"], typed=True) assert {k: v[1] for k, v in src.typed.items()} == {"applicant": "text", "amount": "numeric_blank", "agree": "boolean_blank", "kind": "text"} for case in _cases(): async def run(db, case=case): rows, total, _ = await dataset_query.resolve_rows(db, ds["tenant_id"], "form", ds["form_id"], [case], None, "asc", ["id"]) return sorted(r["id"] for r in rows), total fast, slow = await _both(python_only, run) assert fast == slow, case @pytest.mark.parametrize("sort_field", ["applicant", "amount", "agree", "kind"]) async def test_form_sorting_matches_python(form_dataset, python_only, sort_field): ds = form_dataset async def run(db): return await dataset_query.resolve_rows(db, ds["tenant_id"], "form", ds["form_id"], [], sort_field, "desc", [sort_field, "applicant"], limit=12) fast, slow = await _both(python_only, run) assert fast == slow @pytest.mark.parametrize( "group_by,aggregate_field,aggregate_fn", [("kind", "amount", "sum"), ("agree", None, "count"), ("amount", None, "count"), ("amount", "amount", "max"), ("applicant", "amount", "avg")], ) async def test_form_groups_match_python(form_dataset, python_only, group_by, aggregate_field, aggregate_fn): ds = form_dataset async def run(db): rows, total, cols = await dataset_query.resolve_rows( db, ds["tenant_id"], "form", ds["form_id"], [], None, "desc", [], group_by=group_by, aggregate_field=aggregate_field, aggregate_fn=aggregate_fn ) return sorted(rows, key=lambda r: str(r[group_by])), total, cols fast, slow = await _both(python_only, run) assert fast[1:] == slow[1:] for a, b in zip(fast[0], slow[0]): for key in a: assert a[key] == pytest.approx(b[key]) if isinstance(b[key], float) else a[key] == b[key], (a, b) @pytest.mark.parametrize("x_field", ["kind", "agree", "amount"]) async def test_form_chart_series_match_python(form_dataset, python_only, x_field): ds = form_dataset series = [{"field": "amount", "aggregate": "sum"}, {"aggregate": "count", "label": "Rows"}] async def run(db): rows, labels = await dataset_query.resolve_grouped_series(db, ds["tenant_id"], "form", ds["form_id"], [], x_field, "none", series) return sorted(rows, key=lambda r: str(r["x"])), labels fast, slow = await _both(python_only, run) assert fast[1] == slow[1] assert [r["x"] for r in fast[0]] == [r["x"] for r in slow[0]] for a, b in zip(fast[0], slow[0]): assert a == pytest.approx(b)