WIP3 happy path with tests

big_tests/tests/graphql_helper.erl

@@ -27,10 +27,24 @@ execute(Node, EpName, Body, Creds) ->
         body => Body},
     rest_helper:make_request(Request).
 
+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 +54,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 +93,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(),

big_tests/tests/graphql_stanza_SUITE.erl

@@ -1,14 +1,15 @@
 -module(graphql_stanza_SUITE).
 
--include_lib("common_test/include/ct.hrl").
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("exml/include/exml.hrl").
 
 -compile([export_all, nowarn_export_all]).
 
 -import(distributed_helper, [mim/0, require_rpc_nodes/1]).
 -import(graphql_helper, [execute_user_command/5, execute_command/4,
-                         get_ok_value/2, get_err_code/1, get_err_msg/1, get_coercion_err_msg/1,
+                         execute_user_command_sse/5, execute_command_sse/4,
+                         get_ok_value/2, get_value/2,
+                         get_err_code/1, get_err_msg/1, get_coercion_err_msg/1,
                          get_unauthorized/1, get_not_loaded/1]).
 
 suite() ->
@@ -22,19 +23,25 @@ all() ->
 
 groups() ->
     [{admin_stanza_http, [], [{group, admin_mam},
-                              {group, admin_no_mam}]},
+                              {group, admin_no_mam_http}]},
      {admin_stanza_cli, [], [{group, admin_mam},
-                             {group, admin_no_mam}]},
+                             {group, admin_no_mam_cli}]},
      {domain_admin_stanza, [], [{group, admin_mam}, % same as for admin
                                 {group, domain_admin_no_mam}]},
      {user_stanza, [], [{group, user_mam},
                         {group, user_no_mam}]},
      {admin_mam, [parallel], admin_mam_cases()},
-     {admin_no_mam, [parallel], admin_stanza_cases() ++ admin_no_mam_cases()},
+     {admin_no_mam_http, [parallel], admin_stanza_cases(http) ++ admin_no_mam_cases()},
+     {admin_no_mam_cli, [parallel], admin_stanza_cases(cli) ++ admin_no_mam_cases()},
      {domain_admin_no_mam, [parallel], domain_admin_stanza_cases() ++ admin_no_mam_cases()},
      {user_mam, [parallel], user_mam_cases()},
      {user_no_mam, [parallel], user_stanza_cases() ++ user_no_mam_cases()}].
 
+admin_stanza_cases(cli) ->
+    admin_stanza_cases();
+admin_stanza_cases(http) ->
+    admin_stanza_cases() ++ admin_sse_cases().
+
 admin_stanza_cases() ->
     [admin_send_message,
      admin_send_message_to_unparsable_jid,
@@ -44,6 +51,9 @@ admin_stanza_cases() ->
      admin_send_stanza_from_unknown_user,
      admin_send_stanza_from_unknown_domain].
 
+admin_sse_cases() ->
+    [admin_get_messages].
+
 admin_mam_cases() ->
     [admin_get_last_messages,
      admin_get_last_messages_for_unknown_user,
@@ -69,6 +79,7 @@ domain_admin_stanza_cases() ->
 
 user_stanza_cases() ->
     [user_send_message,
+     user_get_messages,
      user_send_message_without_from,
      user_send_message_with_spoofed_from,
      user_send_message_headline,
@@ -105,7 +116,8 @@ init_per_group(GN, Config) when GN =:= admin_mam;
                                 GN =:= domain_admin_mam;
                                 GN =:= user_mam ->
     init_mam(Config);
-init_per_group(GN, Config) when GN =:= admin_no_mam;
+init_per_group(GN, Config) when GN =:= admin_no_mam_http;
+                                GN =:= admin_no_mam_cli;
                                 GN =:= domain_admin_no_mam;
                                 GN =:= user_no_mam ->
     Mods = [{mod_mam, stopped}],
@@ -154,22 +166,21 @@ admin_send_message_story(Config, Alice, Bob) ->
     escalus:assert(is_message, escalus:wait_for_stanza(Bob)).
 
 admin_get_messages(Config) ->
-    escalus:fresh_story_with_config(Config, [{alice, 1}, {bob, 1}],
-                                    fun admin_send_message_story/3).
+    FreshConfig = escalus_fresh:create_users(Config, [{alice, 1}, {bob, 1}]),
+    escalus:story(FreshConfig, [{alice, 1}],
+                  fun(Alice) -> admin_get_messages_story(Config, Alice) end).
 
-admin_get_messages_story(Config, Alice, Bob) ->
+admin_get_messages_story(Config, Alice) ->
     From = escalus_client:full_jid(Alice),
-    To = escalus_client:short_jid(Bob),
-
-
+    To = escalus_users:get_jid(Config, bob),
+    {200, Stream} = get_messages(To, Config),
     escalus:send(Alice, escalus_stanza:chat(From, To, <<"Hi!">>)),
-
-
-
-    Res = send_message(From, To, <<"Hi!">>, Config),
-    #{<<"id">> := StanzaId} = get_ok_value([data, stanza, sendMessage], Res),
-    assert_not_empty(StanzaId),
-    escalus:assert(is_message, escalus:wait_for_stanza(Bob)).
+    Event = sse_helper:wait_for_event(Stream),
+    #{<<"stanza">> := StanzaBin, <<"sender">> := From} =
+        graphql_helper:get_value([data, stanza, getMessages], Event),
+    {ok, Stanza} = exml:parse(StanzaBin),
+    escalus:assert(is_message, Stanza),
+    sse_helper:stop_sse(Stream).
 
 user_send_message(Config) ->
     escalus:fresh_story_with_config(Config, [{alice, 1}, {bob, 1}],
@@ -183,6 +194,22 @@ user_send_message_story(Config, Alice, Bob) ->
     assert_not_empty(StanzaId),
     escalus:assert(is_message, escalus:wait_for_stanza(Bob)).
 
+user_get_messages(Config) ->
+    escalus:fresh_story_with_config(Config, [{alice, 1}, {bob, 1}],
+                                    fun user_get_messages_story/3).
+
+user_get_messages_story(Config, Alice, Bob) ->
+    From = escalus_client:full_jid(Alice),
+    To = escalus_client:short_jid(Bob),
+    {200, Stream} = user_get_messages(Bob, Config),
+    escalus:send(Alice, escalus_stanza:chat(From, To, <<"Hi!">>)),
+    Event = sse_helper:wait_for_event(Stream),
+    #{<<"stanza">> := StanzaBin, <<"sender">> := From} =
+        graphql_helper:get_value([data, stanza, getMessages], Event),
+    {ok, Stanza} = exml:parse(StanzaBin),
+    escalus:assert(is_message, Stanza),
+    sse_helper:stop_sse(Stream).
+
 user_send_message_without_from(Config) ->
     escalus:fresh_story_with_config(Config, [{alice, 1}, {bob, 1}],
                                     fun user_send_message_without_from_story/3).
@@ -545,6 +572,13 @@ user_send_stanza(User, Stanza, Config) ->
     Vars = #{stanza => Stanza},
     execute_user_command(<<"stanza">>, <<"sendStanza">>, User, Vars, Config).
 
+get_messages(Caller, Config) ->
+    Vars = #{caller => Caller},
+    execute_command_sse(<<"stanza">>, <<"getMessages">>, Vars, Config).
+
+user_get_messages(User, Config) ->
+    execute_user_command_sse(<<"stanza">>, <<"getMessages">>, User, #{}, Config).
+
 get_last_messages(Caller, With, Before, Config) ->
     Vars = #{caller => Caller, with => With, before => Before},
     execute_command(<<"stanza">>, <<"getLastMessages">>, Vars, Config).

big_tests/tests/rest_client_SUITE.erl

@@ -228,15 +228,12 @@ msg_is_sent_and_delivered_over_sse(ConfigIn) ->
     Bob = escalus_users:get_userspec(Config, bob),
     Alice = escalus_users:get_userspec(Config, alice),
 
-    Conn = connect_to_sse({alice, Alice}),
+    {200, Conn} = connect_to_sse({alice, Alice}),
     M = send_message(bob, Bob, Alice),
 
-    Event = wait_for_event(Conn),
-    Data = jiffy:decode(maps:get(data, Event), [return_maps]),
-
-    assert_json_message(M, Data),
-
-    stop_sse(Conn).
+    Event = sse_helper:wait_for_event(Conn),
+    assert_json_message(M, Event),
+    sse_helper:stop_sse(Conn).
 
 message_sending_errors(Config) ->
     escalus:fresh_story(Config, [{alice, 1}, {bob, 1}], fun(Alice, Bob) ->
@@ -258,31 +255,27 @@ room_msg_is_sent_and_delivered_over_sse(ConfigIn) ->
     RoomID = given_new_room_with_users({alice, Alice}, [{bob, Bob}]),
     RoomInfo = get_room_info({alice, Alice}, RoomID),
     true = is_participant(Bob, <<"member">>, RoomInfo),
-    Conn = connect_to_sse({bob, Bob}),
+    {200, Conn} = connect_to_sse({bob, Bob}),
     Message = given_message_sent_to_room(RoomID, {alice, Alice}),
-    Event = wait_for_event(Conn),
-    Data = jiffy:decode(maps:get(data, Event), [return_maps]),
-    assert_json_room_sse_message(Message#{room => RoomID, type => <<"message">>},
-                                 Data),
-    stop_sse(Conn).
+    Event = sse_helper:wait_for_event(Conn),
+    assert_json_room_sse_message(Message#{room => RoomID, type => <<"message">>}, Event),
+    sse_helper:stop_sse(Conn).
 
 aff_change_msg_is_delivered_over_sse(ConfigIn) ->
     Config = escalus_fresh:create_users(ConfigIn, [{alice, 1}, {bob, 1}]),
     Bob = escalus_users:get_userspec(Config, bob),
     Alice = escalus_users:get_userspec(Config, alice),
     RoomID = given_new_room({alice, Alice}),
-    Conn = connect_to_sse({bob, Bob}),
+    {200, Conn} = connect_to_sse({bob, Bob}),
     given_user_invited({alice, Alice}, RoomID, Bob),
-    Event = wait_for_event(Conn),
-    Data = jiffy:decode(maps:get(data, Event), [return_maps]),
+    Event = sse_helper:wait_for_event(Conn),
     BobJID = user_jid(Bob),
     RoomJID = room_jid(RoomID, Config),
     assert_json_room_sse_message(#{room => RoomID,
                                    from => RoomJID,
                                    type => <<"affiliation">>,
-                                   user => BobJID},
-                                 Data),
-    stop_sse(Conn).
+                                   user => BobJID}, Event),
+    sse_helper:stop_sse(Conn).
 
 all_messages_are_archived(Config) ->
     escalus:fresh_story(Config, [{alice, 1}, {bob, 1}, {kate, 1}], fun(Alice, Bob, Kate) ->
@@ -1132,32 +1125,7 @@ is_participant(User, Role, RoomInfo) ->
 
 connect_to_sse(User) ->
     Port = ct:get_config({hosts, mim, http_api_client_endpoint_port}),
-    {Username, Password} = credentials(User),
-    Base64 = base64:encode(binary_to_list(Username) ++ [$: | binary_to_list(Password)]),
-    Headers = [{<<"authorization">>, <<"basic ", Base64/binary>>},
-               {<<"host">>, <<"localhost">>},
-               {<<"accept">>, <<"text/event-stream">>}],
-    {ok, ConnPid} = gun:open("localhost", Port, #{
-        transport => tls,
-        protocols => [http],
-        http_opts => #{content_handlers => [gun_sse_h, gun_data_h]}
-    }),
-    {ok, _} = gun:await_up(ConnPid),
-    StreamRef = gun:get(ConnPid, "/api/sse", Headers),
-    #{pid => ConnPid, stream_ref => StreamRef}.
-
-wait_for_event(#{pid := Pid, stream_ref := StreamRef} = Opts) ->
-    case gun:await(Pid, StreamRef) of
-        {response, nofin, _Status, _} ->
-            wait_for_event(Opts);
-        {sse, #{data := [Response]}} ->
-          Opts#{data => Response};
-        Error ->
-            Error
-    end.
-
-stop_sse(#{pid := Pid}) ->
-    gun:close(Pid).
+    sse_helper:connect_to_sse(Port, "/api/sse", credentials(User), #{transport => tls}).
 
 assert_json_message(Sent, Received) ->
     #{<<"body">> := Body,

priv/graphql/schemas/user/stanza.gql

@@ -18,3 +18,7 @@ type StanzaUserMutation @protected{
   "Send an arbitrary stanza"
   sendStanza(stanza: Stanza): SendStanzaPayload
 }
+
+type StanzaUserSubscription {
+  getMessages: StanzaMap
+}

priv/graphql/schemas/user/user_schema.gql

@@ -1,6 +1,7 @@
 schema{
   query: UserQuery,
-  mutation: UserMutation
+  mutation: UserMutation,
+  subscription: UserSubscription
 }
 
 """
@@ -58,3 +59,7 @@ type UserMutation @protected{
   "OAUTH token management"
   token: TokenUserMutation
 }
+
+type UserSubscription {
+  stanza: StanzaUserSubscription
+}

rebar.lock

@@ -63,7 +63,7 @@
  {<<"jsx">>,{pkg,<<"jsx">>,<<"2.9.0">>},1},
  {<<"jwerl">>,
   {git,"https://gitlab.com/vkatsuba/jwerl.git",
-       {ref,"d03607fd14a6a7556f01014af84903a3df60ff5d"}},
+       {ref,"cd1cdfc393bc7865f6a412a2c5e015729397e607"}},
   0},
  {<<"lager">>,{pkg,<<"lager">>,<<"3.9.2">>},0},
  {<<"lasse">>,{pkg,<<"lasse">>,<<"1.2.0">>},0},

src/graphql/admin/mongoose_graphql_stanza_admin_subscription.erl

@@ -12,14 +12,9 @@
 execute(Ctx, _Obj, <<"getMessages">>, Args) ->
     get_messages(Ctx, Args).
 
-get_messages(#{event := Event}, _) ->
-    case mongoose_graphql_stanza_helper:handle_event(Event) of
-        {EventName, Data} ->
-            {ok, Data, [{event, EventName}]};
-        no_event ->
-            {ok, null, [{event, none}]}
-    end;
 get_messages(#{event := terminate, stream := Session}, _) ->
     {ok, null, [{stream, mongoose_stanza_event_api:close_session(Session)}]};
+get_messages(#{event := Event}, _) ->
+    mongoose_graphql_stanza_helper:handle_event(Event);
 get_messages(_Ctx, #{<<"caller">> := Jid}) ->
     {ok, null, [{stream, mongoose_stanza_event_api:open_session(Jid)}]}.