Merge pull request #3830 from esl/graphql-subscriptions

GraphQL subscriptions for stanzas

big_tests/default.spec

@@ -26,6 +26,7 @@
 {suites, "tests", extdisco_SUITE}.
 {suites, "tests", gdpr_SUITE}.
 {suites, "tests", graphql_SUITE}.
+{suites, "tests", graphql_sse_SUITE}.
 {suites, "tests", graphql_account_SUITE}.
 {suites, "tests", graphql_domain_SUITE}.
 {suites, "tests", graphql_inbox_SUITE}.

big_tests/dynamic_domains.spec

@@ -42,6 +42,7 @@
  "at the moment mod_pubsub doesn't support dynamic domains"}.
 
 {suites, "tests", graphql_SUITE}.
+{suites, "tests", graphql_sse_SUITE}.
 {suites, "tests", graphql_account_SUITE}.
 {suites, "tests", graphql_domain_SUITE}.
 {suites, "tests", graphql_inbox_SUITE}.

big_tests/tests/graphql_helper.erl

@@ -17,20 +17,40 @@ execute(EpName, Body, Creds) ->
 -spec execute(node(), atom(), binary(), {binary(), binary()} | undefined) ->
     {Status :: tuple(), Data :: map()}.
 execute(Node, EpName, Body, Creds) ->
-    Request =
-      #{port => get_listener_port(Node, EpName),
-        role => {graphql, EpName},
-        method => <<"POST">>,
-        return_maps => true,
-        creds => Creds,
-        path => "/graphql",
-        body => Body},
+    Request = build_request(Node, EpName, Body, Creds),
     rest_helper:make_request(Request).
 
+build_request(Node, EpName, Body, Creds) ->
+    #{port => get_listener_port(Node, EpName),
+      role => {graphql, EpName},
+      method => <<"POST">>,
+      return_maps => true,
+      creds => Creds,
+      path => "/graphql",
+      body => Body}.
+
+execute_sse(EpName, Params, Creds) ->
+    #{node := Node} = mim(),
+    execute_sse(Node, EpName, Params, Creds).
+
+execute_sse(Node, EpName, Params, Creds) ->
+    Port = get_listener_port(Node, EpName),
+    Path = "/api/graphql/sse",
+    QS = uri_string:compose_query([{atom_to_binary(K), encode_sse_value(V)}
+                                   || {K, V} <- maps:to_list(Params)]),
+    sse_helper:connect_to_sse(Port, [Path, "?", QS], Creds, #{}).
+
+encode_sse_value(M) when is_map(M) -> jiffy:encode(M);
+encode_sse_value(V) when is_binary(V) -> V.
+
 execute_user_command(Category, Command, User, Args, Config) ->
-    #{Category := #{commands := #{Command := #{doc := Doc}}}} = get_specs(),
+    Doc = get_doc(Category, Command),
     execute_user(#{query => Doc, variables => Args}, User, Config).
 
+execute_user_command_sse(Category, Command, User, Args, Config) ->
+    Doc = get_doc(Category, Command),
+    execute_user_sse(#{query => Doc, variables => Args}, User, Config).
+
 execute_command(Category, Command, Args, Config) ->
     #{node := Node} = mim(),
     Protocol = ?config(protocol, Config),
@@ -40,16 +60,24 @@ execute_command(Node, Category, Command, Args, Config) ->
     Protocol = ?config(protocol, Config),
     execute_command(Node, Category, Command, Args, Config, Protocol).
 
+execute_command_sse(Category, Command, Args, Config) ->
+    Doc = get_doc(Category, Command),
+    execute_auth_sse(#{query => Doc, variables => Args}, Config).
+
 %% Admin commands can be executed as GraphQL over HTTP or with CLI (mongooseimctl)
 execute_command(Node, Category, Command, Args, Config, http) ->
-    #{Category := #{commands := #{Command := #{doc := Doc}}}} = get_specs(),
+    Doc = get_doc(Category, Command),
     execute_auth(Node, #{query => Doc, variables => Args}, Config);
 execute_command(Node, Category, Command, Args, Config, cli) ->
     CLIArgs = encode_cli_args(Args),
     {Result, Code}
         = mongooseimctl_helper:mongooseimctl(Node, Category, [Command | CLIArgs], Config),
     {{exit_status, Code}, rest_helper:decode(Result, #{return_maps => true})}.
 
+get_doc(Category, Command) ->
+    #{Category := #{commands := #{Command := #{doc := Doc}}}} = get_specs(),
+    Doc.
+
 encode_cli_args(Args) ->
     lists:flatmap(fun({Name, Value}) -> encode_cli_arg(Name, Value) end, maps:to_list(Args)).
 encode_cli_arg(_Name, null) ->
@@ -71,21 +99,35 @@ execute_auth(Body, Config) ->
     execute_auth(Node, Body, Config).
 
 execute_auth(Node, Body, Config) ->
-    case Ep = ?config(schema_endpoint, Config) of
-        admin ->
-            #{username := Username, password := Password} = get_listener_opts(Ep),
-            execute(Node, Ep, Body, {Username, Password});
-        domain_admin ->
-            Creds = ?config(domain_admin, Config),
-            execute(Node, Ep, Body, Creds)
-    end.
+    Ep = ?config(schema_endpoint, Config),
+    execute(Node, Ep, Body, make_admin_creds(Ep, Config)).
+
+execute_auth_sse(Body, Config) ->
+    #{node := Node} = mim(),
+    execute_auth_sse(Node, Body, Config).
+
+execute_auth_sse(Node, Body, Config) ->
+    Ep = ?config(schema_endpoint, Config),
+    execute_sse(Node, Ep, Body, make_admin_creds(Ep, Config)).
+
+make_admin_creds(admin = Ep, _Config) ->
+    #{username := Username, password := Password} = get_listener_opts(Ep),
+    {Username, Password};
+make_admin_creds(domain_admin, Config) ->
+    ?config(domain_admin, Config).
 
 execute_user(Body, User, Config) ->
     Ep = ?config(schema_endpoint, Config),
     Creds = make_creds(User),
     #{node := Node} = mim(),
     execute(Node, Ep, Body, Creds).
 
+execute_user_sse(Body, User, Config) ->
+    Ep = ?config(schema_endpoint, Config),
+    Creds = make_creds(User),
+    #{node := Node} = mim(),
+    execute_sse(Node, Ep, Body, Creds).
+
 -spec get_listener_port(binary()) -> integer().
 get_listener_port(EpName) ->
     #{node := Node} = mim(),
@@ -173,6 +215,9 @@ get_unauthorized({Code, #{<<"errors">> := Errors}}) ->
 get_bad_request({Code, _Msg}) ->
     assert_response_code(bad_request, Code).
 
+get_method_not_allowed({Code, _Msg}) ->
+    assert_response_code(method_not_allowed, Code).
+
 get_coercion_err_msg({Code, #{<<"errors">> := [Error]}}) ->
     assert_response_code(bad_request, Code),
     ?assertEqual(<<"input_coercion">>, get_value([extensions, code], Error)),
@@ -196,8 +241,14 @@ get_ok_value(Path, {Code, Data}) ->
 
 assert_response_code(bad_request, {<<"400">>, <<"Bad Request">>}) -> ok;
 assert_response_code(unauthorized, {<<"401">>, <<"Unauthorized">>}) -> ok;
+assert_response_code(method_not_allowed, {<<"405">>, <<"Method Not Allowed">>}) -> ok;
 assert_response_code(error, {<<"200">>, <<"OK">>}) -> ok;
 assert_response_code(ok, {<<"200">>, <<"OK">>}) -> ok;
+assert_response_code(bad_request, 400) -> ok;
+assert_response_code(unauthorized, 401) -> ok;
+assert_response_code(method_not_allowed, 405) -> ok;
+assert_response_code(error, 200) -> ok;
+assert_response_code(ok, 200) -> ok;
 assert_response_code(bad_request, {exit_status, 1}) -> ok;
 assert_response_code(error, {exit_status, 1}) -> ok;
 assert_response_code(ok, {exit_status, 0}) -> ok;

big_tests/tests/graphql_sse_SUITE.erl

@@ -0,0 +1,138 @@
+%% @doc Tests for the SSE handling of GraphQL subscriptions
+-module(graphql_sse_SUITE).
+
+-compile([export_all, nowarn_export_all]).
+
+-import(distributed_helper, [mim/0, require_rpc_nodes/1, rpc/4]).
+-import(graphql_helper, [get_bad_request/1, get_unauthorized/1, get_method_not_allowed/1,
+                         build_request/4, make_creds/1, execute_auth/2,
+                         execute_sse/3, execute_user_sse/3, execute_auth_sse/2]).
+
+%% common_test callbacks
+
+suite() ->
+    require_rpc_nodes([mim]) ++ escalus:suite().
+
+all() ->
+    [{group, admin},
+     {group, user}].
+
+groups() ->
+    [{admin, [parallel], admin_tests()},
+     {user, [parallel], user_tests()}].
+
+init_per_suite(Config) ->
+    Config1 = escalus:init_per_suite(Config),
+    application:ensure_all_started(gun),
+    Config1.
+
+end_per_suite(Config) ->
+    escalus:end_per_suite(Config).
+
+init_per_group(user, Config) ->
+    graphql_helper:init_user(Config);
+init_per_group(admin, Config) ->
+    graphql_helper:init_admin_handler(Config).
+
+end_per_group(user, _Config) ->
+    escalus_fresh:clean(),
+    graphql_helper:clean();
+end_per_group(admin, _Config) ->
+    graphql_helper:clean().
+
+init_per_testcase(CaseName, Config) ->
+    escalus:init_per_testcase(CaseName, Config).
+
+end_per_testcase(CaseName, Config) ->
+    escalus:end_per_testcase(CaseName, Config).
+
+admin_tests() ->
+    [admin_missing_query,
+     admin_invalid_query_string,
+     admin_missing_creds,
+     admin_invalid_creds,
+     admin_invalid_method,
+     admin_invalid_operation_type].
+
+user_tests() ->
+    [user_missing_query,
+     user_invalid_query_string,
+     user_missing_creds,
+     user_invalid_creds,
+     user_invalid_method,
+     user_invalid_operation_type].
+
+%% Test cases and stories
+
+admin_missing_query(Config) ->
+    get_bad_request(execute_auth_sse(#{}, Config)).
+
+user_missing_query(Config) ->
+    escalus:fresh_story_with_config(Config, [{alice, 1}], fun user_missing_query_story/2).
+
+user_missing_query_story(Config, Alice) ->
+    get_bad_request(execute_user_sse(#{}, Alice, Config)).
+
+admin_invalid_query_string(_Config) ->
+    Port = graphql_helper:get_listener_port(admin),
+    get_bad_request(sse_helper:connect_to_sse(Port, "/api/graphql/sse?=invalid", undefined, #{})).
+
+user_invalid_query_string(Config) ->
+    escalus:fresh_story(Config, [{alice, 1}], fun user_invalid_query_string_story/1).
+
+user_invalid_query_string_story(Alice) ->
+    Port = graphql_helper:get_listener_port(user),
+    Creds = make_creds(Alice),
+    get_bad_request(sse_helper:connect_to_sse(Port, "/api/graphql/sse?=invalid", Creds, #{})).
+
+admin_missing_creds(_Config) ->
+    get_unauthorized(execute_sse(admin, #{query => doc(), variables => args()}, undefined)).
+
+user_missing_creds(_Config) ->
+    get_unauthorized(execute_sse(user, #{query => doc()}, undefined)).
+
+admin_invalid_creds(_Config) ->
+    Creds = {<<"invalid">>, <<"creds">>},
+    get_unauthorized(execute_sse(admin, #{query => doc(), variables => args()}, Creds)).
+
+user_invalid_creds(_Config) ->
+    get_unauthorized(execute_sse(user, #{query => doc()}, {<<"invalid">>, <<"creds">>})).
+
+admin_invalid_method(_Config) ->
+    #{node := Node} = mim(),
+    Request = build_request(Node, admin, #{query => doc(), variables => args()}, undefined),
+    %% POST was used, while SSE accepts only GET
+    get_method_not_allowed(rest_helper:make_request(Request#{path => "/graphql/sse"})).
+
+user_invalid_method(Config) ->
+    escalus:fresh_story(Config, [{alice, 1}], fun user_invalid_method_story/1).
+
+user_invalid_method_story(Alice) ->
+    #{node := Node} = mim(),
+    Request = build_request(Node, user, #{query => doc()}, make_creds(Alice)),
+    %% POST was used, while SSE accepts only GET
+    get_method_not_allowed(rest_helper:make_request(Request#{path => "/graphql/sse"})).
+
+admin_invalid_operation_type(Config) ->
+    Creds = graphql_helper:make_admin_creds(admin, Config),
+    get_bad_request(execute_sse(admin, #{query => query_doc(), variables => args()}, Creds)).
+
+user_invalid_operation_type(Config) ->
+    escalus:fresh_story(Config, [{alice, 1}], fun user_invalid_operation_type_story/1).
+
+user_invalid_operation_type_story(Alice) ->
+    get_bad_request(execute_sse(user, #{query => query_doc()}, make_creds(Alice))).
+
+%% Helpers
+
+%% Subscription - works only with the SSE handler
+doc() ->
+    graphql_helper:get_doc(<<"stanza">>, <<"subscribeForMessages">>).
+
+%% Query - works only with the REST handler
+query_doc() ->
+    graphql_helper:get_doc(<<"stanza">>, <<"getLastMessages">>).
+
+%% Same args used by both operations - only for Admin
+args() ->
+    #{caller => <<"alice@localhost">>}.