proxy: add tests for batch http sql (#4793)

This PR adds an integration test case for batch HTTP SQL endpoint.
https://github.com/neondatabase/neon/pull/4654/ should be merged first
before we land this PR.

---------

Signed-off-by: Alex Chi Z <chi@neon.tech>

proxy/src/http/sql_over_http.rs

@@ -223,7 +223,7 @@ pub async fn handle(
                 }
             }
             transaction.commit().await?;
-            Ok(Value::Array(results))
+            Ok(json!({ "results": results }))
         }
     };
 

test_runner/regress/test_proxy.py

@@ -1,6 +1,6 @@
 import json
 import subprocess
-from typing import Any, List, Optional
+from typing import Any, List, Optional, Tuple
 
 import psycopg2
 import pytest
@@ -260,3 +260,54 @@ def test_sql_over_http_output_options(static_proxy: NeonProxy):
 
     rows = q("select 1 as n, 'a' as s, '{1,2,3}'::int4[] as arr", True, True)["rows"]
     assert rows == [["1", "a", "{1,2,3}"]]
+
+
+def test_sql_over_http_batch(static_proxy: NeonProxy):
+    static_proxy.safe_psql("create role http with login password 'http' superuser")
+
+    def qq(queries: List[Tuple[str, Optional[List[Any]]]]) -> Any:
+        connstr = f"postgresql://http:http@{static_proxy.domain}:{static_proxy.proxy_port}/postgres"
+        response = requests.post(
+            f"https://{static_proxy.domain}:{static_proxy.external_http_port}/sql",
+            data=json.dumps(list(map(lambda x: {"query": x[0], "params": x[1] or []}, queries))),
+            headers={"Content-Type": "application/sql", "Neon-Connection-String": connstr},
+            verify=str(static_proxy.test_output_dir / "proxy.crt"),
+        )
+        assert response.status_code == 200
+        return response.json()["results"]
+
+    result = qq(
+        [
+            ("select 42 as answer", None),
+            ("select $1 as answer", [42]),
+            ("select $1 * 1 as answer", [42]),
+            ("select $1::int[] as answer", [[1, 2, 3]]),
+            ("select $1::json->'a' as answer", [{"a": {"b": 42}}]),
+            ("select * from pg_class limit 1", None),
+            ("create table t(id serial primary key, val int)", None),
+            ("insert into t(val) values (10), (20), (30) returning id", None),
+            ("select * from t", None),
+            ("drop table t", None),
+        ]
+    )
+
+    assert result[0]["rows"] == [{"answer": 42}]
+    assert result[1]["rows"] == [{"answer": "42"}]
+    assert result[2]["rows"] == [{"answer": 42}]
+    assert result[3]["rows"] == [{"answer": [1, 2, 3]}]
+    assert result[4]["rows"] == [{"answer": {"b": 42}}]
+    assert len(result[5]["rows"]) == 1
+    res = result[6]
+    assert res["command"] == "CREATE"
+    assert res["rowCount"] is None
+    res = result[7]
+    assert res["command"] == "INSERT"
+    assert res["rowCount"] == 3
+    assert res["rows"] == [{"id": 1}, {"id": 2}, {"id": 3}]
+    res = result[8]
+    assert res["command"] == "SELECT"
+    assert res["rowCount"] == 3
+    res = result[9]
+    assert res["command"] == "DROP"
+    assert res["rowCount"] is None
+    assert len(result) == 10