Add missing call to withConstructorFn in live queries (fix hasura#3239)

server/src-lib/Hasura/GraphQL/Execute/LiveQuery/Plan.hs

@@ -120,11 +120,13 @@ resolveMultiplexedValue = \case
   GR.UVSQL sqlExp -> pure sqlExp
   GR.UVSession -> pure $ fromResVars (PGTypeScalar PGJSON) ["session"]
   where
-    fromResVars ty jPath =
-      flip S.SETyAnn (S.mkTypeAnn ty) $ S.SEOpApp (S.SQLOp "#>>")
+    fromResVars pgType jPath = addTypeAnnotation pgType $ S.SEOpApp (S.SQLOp "#>>")
       [ S.SEQIden $ S.QIden (S.QualIden (Iden "_subs") Nothing) (Iden "result_vars")
       , S.SEArray $ map S.SELit jPath
       ]
+    addTypeAnnotation pgType = flip S.SETyAnn (S.mkTypeAnn pgType) . case pgType of
+      PGTypeScalar scalarType -> withConstructorFn scalarType
+      PGTypeArray _           -> id
 
 newtype CohortId = CohortId { unCohortId :: UUID }
   deriving (Show, Eq, Hashable, J.ToJSON, Q.FromCol)

server/tests-py/queries/graphql_query/boolexp/postgis/query_cast_geography_to_geometry.yaml

@@ -7,7 +7,7 @@ response:
     - name: New York
 query:
   query: |
-    {
+    query {
       geog_table(where: {
         geog_col: {
           _cast: {

server/tests-py/queries/graphql_query/boolexp/postgis/query_cast_geometry_to_geography.yaml

@@ -8,7 +8,7 @@ response:
     - name: Paris
 query:
   query: |
-    {
+    query {
       geog_as_geom_table(where: {
         geom_col: {
           _cast: {

server/tests-py/queries/graphql_query/boolexp/postgis/query_illegal_cast_is_not_allowed.yaml

@@ -9,7 +9,7 @@ response:
     message: 'field "integer" not found in type: ''geography_cast_exp'''
 query:
   query: |
-    {
+    query {
       geog_table(where: {geog_col: {_cast: {integer: {_eq: 0}}}}) {
         name
       }

server/tests-py/test_graphql_queries.py

@@ -18,7 +18,7 @@ def test_select_query_author_with_skip_directive(self, hge_ctx, transport):
 
     def test_select_query_author_with_include_directive(self, hge_ctx, transport):
         check_query_f(hge_ctx, self.dir() + '/select_query_author_include_directive.yaml', transport)
-    
+
     # Can't run server upgrade tests, as this test has a schema change
     @pytest.mark.skip_server_upgrade_test
     def test_select_various_postgres_types(self, hge_ctx, transport):
@@ -388,7 +388,7 @@ def test_jsonb_has_keys_all_touchscreen_ram(self, hge_ctx, transport):
     def dir(cls):
         return 'queries/graphql_query/boolexp/jsonb'
 
-@pytest.mark.parametrize("transport", ['http', 'websocket'])
+@pytest.mark.parametrize("transport", ['http', 'websocket', 'subscription'])
 @usefixtures('per_class_tests_db_state')
 class TestGraphQLQueryBoolExpPostGIS:
 

server/tests-py/validate.py

@@ -10,6 +10,7 @@
 import base64
 import jsondiff
 import jwt
+import queue
 import random
 import warnings
 
@@ -194,27 +195,31 @@ def check_query(hge_ctx, conf, transport='http', add_auth=True, claims_namespace
             test_forbidden_when_admin_secret_reqd(hge_ctx, conf)
             headers['X-Hasura-Admin-Secret'] = hge_ctx.hge_key
 
-    assert transport in ['websocket', 'http'], "Unknown transport type " + transport
-    if transport == 'websocket':
-        assert 'response' in conf
-        assert conf['url'].endswith('/graphql')
-        print('running on websocket')
-        return validate_gql_ws_q(
-            hge_ctx,
-            conf['url'],
-            conf['query'],
-            headers,
-            conf['response'],
-            True
-        )
-    elif transport == 'http':
+    assert transport in ['http', 'websocket', 'subscription'], "Unknown transport type " + transport
+    if transport == 'http':
         print('running on http')
         return validate_http_anyq(hge_ctx, conf['url'], conf['query'], headers,
                                   conf['status'], conf.get('response'))
+    elif transport == 'websocket':
+        print('running on websocket')
+        return validate_gql_ws_q(hge_ctx, conf, headers, retry=True)
+    elif transport == 'subscription':
+        print('running via subscription')
+        return validate_gql_ws_q(hge_ctx, conf, headers, retry=True, via_subscription=True)
+
 
+def validate_gql_ws_q(hge_ctx, conf, headers, retry=False, via_subscription=False):
+    assert 'response' in conf
+    assert conf['url'].endswith('/graphql')
+    endpoint = conf['url']
+    query = conf['query']
+    exp_http_response = conf['response']
 
+    if via_subscription:
+        query_text = query['query']
+        assert query_text.startswith('query '), query_text
+        query['query'] = 'subscription' + query_text[len('query'):]
 
-def validate_gql_ws_q(hge_ctx, endpoint, query, headers, exp_http_response, retry=False):
     if endpoint == '/v1alpha1/graphql':
         ws_client = GQLWsClient(hge_ctx, '/v1alpha1/graphql')
     else:
@@ -223,7 +228,7 @@ def validate_gql_ws_q(hge_ctx, endpoint, query, headers, exp_http_response, retr
     if not headers or len(headers) == 0:
         ws_client.init({})
 
-    query_resp = ws_client.send_query(query, headers=headers, timeout=15)
+    query_resp = ws_client.send_query(query, query_id='hge_test', headers=headers, timeout=15)
     resp = next(query_resp)
     print('websocket resp: ', resp)
 
@@ -240,10 +245,15 @@ def validate_gql_ws_q(hge_ctx, endpoint, query, headers, exp_http_response, retr
         assert resp['type'] in ['data', 'error'], resp
     else:
         assert resp['type'] == 'data', resp
-
     assert 'payload' in resp, resp
-    resp_done = next(query_resp)
-    assert resp_done['type'] == 'complete'
+
+    if via_subscription:
+        ws_client.send({ 'id': 'hge_test', 'type': 'stop' })
+        with pytest.raises(queue.Empty):
+            ws_client.get_ws_event(0)
+    else:
+        resp_done = next(query_resp)
+        assert resp_done['type'] == 'complete'
 
     return assert_graphql_resp_expected(resp['payload'], exp_http_response, query, skip_if_err_msg=hge_ctx.avoid_err_msg_checks)