%%
%%
%% Copyright WhatsApp Inc. and its affiliates. All rights reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%     http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%
%%-------------------------------------------------------------------
%% @author Maxim Fedorov <maximfca@gmail.com>
%%     Process Groups smoke test.
-module(pg_SUITE).
-author("maximfca@gmail.com").

%% Test server callbacks
-export([
    suite/0,
    all/0,
    groups/0,
    init_per_testcase/2,
    end_per_testcase/2,
    stop_proc/1,
    ensure_peers_info/2
]).

%% Test cases exports
-export([
    pg/0, pg/1,
    errors/0, errors/1,
    leave_exit_race/0, leave_exit_race/1,
    dyn_distribution/0, dyn_distribution/1,
    process_owner_check/0, process_owner_check/1,
    overlay_missing/0, overlay_missing/1,
    single/0, single/1,
    two/1,
    empty_group_by_remote_leave/0, empty_group_by_remote_leave/1,
    thundering_herd/0, thundering_herd/1,
    initial/1,
    netsplit/1,
    trisplit/1,
    foursplit/1,
    exchange/1,
    nolocal/1,
    double/1,
    scope_restart/1,
    missing_scope_join/1,
    disconnected_start/1,
    forced_sync/0, forced_sync/1,
    group_leave/1
]).

-include_lib("common_test/include/ct.hrl").
-include_lib("stdlib/include/assert.hrl").

suite() ->
    [{timetrap, {seconds, 60}}].

init_per_testcase(TestCase, Config) ->
    {ok, _Pid} = pg:start_link(TestCase),
    Config.

end_per_testcase(TestCase, _Config) ->
    gen_server:stop(TestCase),
    ok.

all() ->
    [dyn_distribution, {group, basic}, {group, cluster}, {group, performance}].

groups() ->
    [
        {basic, [parallel], [errors, pg, leave_exit_race, single, overlay_missing]},
        {performance, [sequential], [thundering_herd]},
        {cluster, [parallel], [process_owner_check, two, initial, netsplit, trisplit, foursplit,
            exchange, nolocal, double, scope_restart, missing_scope_join, empty_group_by_remote_leave,
            disconnected_start, forced_sync, group_leave]}
    ].

%%--------------------------------------------------------------------
%% TEST CASES

pg() ->
    [{doc, "This test must be names pg, to stay inline with default scope"}].

pg(_Config) ->
    ?assertNotEqual(undefined, whereis(?FUNCTION_NAME)), %% ensure scope was started
    ?assertEqual(ok, pg:join(?FUNCTION_NAME, self())),
    ?assertEqual([self()], pg:get_local_members(?FUNCTION_NAME)),
    ?assertEqual([?FUNCTION_NAME], pg:which_groups()),
    ?assertEqual([?FUNCTION_NAME], pg:which_local_groups()),
    ?assertEqual(ok, pg:leave(?FUNCTION_NAME, self())),
    ?assertEqual([], pg:get_members(?FUNCTION_NAME)),
    ?assertEqual([], pg:which_groups(?FUNCTION_NAME)),
    ?assertEqual([], pg:which_local_groups(?FUNCTION_NAME)).

errors() ->
    [{doc, "Tests that errors are handled as expected, for example pg server crashes when it needs to"}].

errors(_Config) ->
    %% kill with 'info' and 'cast'
    ?assertException(error, badarg, pg:handle_info(garbage, garbage)),
    ?assertException(error, badarg, pg:handle_cast(garbage, garbage)),
    %% kill with call
    {ok, _Pid} = pg:start(second),
    ?assertException(exit, {{badarg, _}, _}, gen_server:call(second, garbage, 100)).

leave_exit_race() ->
    [{doc, "Tests that pg correctly handles situation when leave and 'DOWN' messages are both in pg queue"}].

leave_exit_race(Config) when is_list(Config) ->
    process_flag(priority, high),
    [
        begin
            Pid = spawn(fun () -> ok end),
            pg:join(leave_exit_race, test, Pid),
            pg:leave(leave_exit_race, test, Pid)
        end
        || _ <- lists:seq(1, 100)].

single() ->
    [{doc, "Tests single node groups"}, {timetrap, {seconds, 5}}].

single(Config) when is_list(Config) ->
    ?assertEqual(ok, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, self())),
    ?assertEqual(ok, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, [self(), self()])),
    ?assertEqual([self(), self(), self()], pg:get_local_members(?FUNCTION_NAME, ?FUNCTION_NAME)),
    ?assertEqual([self(), self(), self()], pg:get_members(?FUNCTION_NAME, ?FUNCTION_NAME)),
    ?assertEqual(not_joined, pg:leave(?FUNCTION_NAME, '$missing$', self())),
    ?assertEqual(ok, pg:leave(?FUNCTION_NAME, ?FUNCTION_NAME, [self(), self()])),
    ?assertEqual(ok, pg:leave(?FUNCTION_NAME, ?FUNCTION_NAME, self())),
    ?assertEqual([], pg:which_groups(?FUNCTION_NAME)),
    ?assertEqual([], pg:get_local_members(?FUNCTION_NAME, ?FUNCTION_NAME)),
    ?assertEqual([], pg:get_members(?FUNCTION_NAME, ?FUNCTION_NAME)),
    %% double
    ?assertEqual(ok, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, self())),
    Pid = erlang:spawn(forever()),
    ?assertEqual(ok, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, Pid)),
    Expected = lists:sort([Pid, self()]),
    ?assertEqual(Expected, lists:sort(pg:get_members(?FUNCTION_NAME, ?FUNCTION_NAME))),
    ?assertEqual(Expected, lists:sort(pg:get_local_members(?FUNCTION_NAME, ?FUNCTION_NAME))),

    stop_proc(Pid),
    sync(?FUNCTION_NAME),
    ?assertEqual([self()], pg:get_local_members(?FUNCTION_NAME, ?FUNCTION_NAME)),
    ?assertEqual(ok, pg:leave(?FUNCTION_NAME, ?FUNCTION_NAME, self())),
    ok.

dyn_distribution() ->
    [{doc, "Tests that local node when distribution is started dynamically is not treated as remote node"}].

dyn_distribution(Config) when is_list(Config) ->
    %% When distribution is started or stopped dynamically,
    %%  there is a nodeup/nodedown message delivered to pg
    %% It is possible but non-trivial to simulate this
    %%  behaviour with starting slave nodes being not
    %%  distributed, and calling net_kernel:start/1, however
    %%  the effect is still the same as simply sending nodeup,
    %%  which is also documented.
    ?FUNCTION_NAME ! {nodeup, node()},
    %%
    ?assertEqual(ok, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, self())),
    ?assertEqual([self()], pg:get_members(?FUNCTION_NAME, ?FUNCTION_NAME)),
    ok.

process_owner_check() ->
    [{doc, "Tests that process owner is local node"}].

process_owner_check(Config) when is_list(Config) ->
    {Peer, Node} = spawn_node(?FUNCTION_NAME),
    %% spawn remote process
    LocalPid = erlang:spawn(forever()),
    RemotePid = erlang:spawn(Node, forever()),
    %% check they can't be joined locally
    ?assertException(error, {nolocal, _}, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, RemotePid)),
    ?assertException(error, {nolocal, _}, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, [RemotePid, RemotePid])),
    ?assertException(error, {nolocal, _}, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, [LocalPid, RemotePid])),
    %% check that non-pid also triggers error
    ?assertException(error, function_clause, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, undefined)),
    ?assertException(error, {nolocal, _}, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, [undefined])),
    %% stop the peer
    peer:stop(Peer),
    ok.

overlay_missing() ->
    [{doc, "Tests that scope process that is not a part of overlay network does not change state"}].

overlay_missing(_Config) ->
    {Peer, Node} = spawn_node(?FUNCTION_NAME),
    %% join self (sanity check)
    ?assertEqual(ok, pg:join(?FUNCTION_NAME, group, self())),
    %% remember pid from remote
    PgPid = rpc:call(Node, erlang, whereis, [?FUNCTION_NAME]),
    RemotePid = erlang:spawn(Node, forever()),
    %% stop remote scope
    gen_server:stop(PgPid),
    %% craft white-box request: ensure it's rejected
    ?FUNCTION_NAME ! {join, PgPid, group, RemotePid},
    %% rejected!
    ?assertEqual([self()], pg:get_members(?FUNCTION_NAME, group)),
    %% ... reject leave too
    ?FUNCTION_NAME ! {leave, PgPid, RemotePid, [group]},
    ?assertEqual([self()], pg:get_members(?FUNCTION_NAME, group)),
    %% join many times on remote
    %RemotePids = [erlang:spawn(TwoPeer, forever()) || _ <- lists:seq(1, 1024)],
    %?assertEqual(ok, rpc:call(TwoPeer, pg, join, [?FUNCTION_NAME, ?FUNCTION_NAME, Pid2])),
    %% check they can't be joined locally
    %?assertException(error, {nolocal, _}, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, RemotePid)),
    %?assertException(error, {nolocal, _}, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, [RemotePid, RemotePid])),
    %?assertException(error, {nolocal, _}, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, [LocalPid, RemotePid])),
    %% check that non-pid also triggers error
    %?assertException(error, function_clause, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, undefined)),
    %?assertException(error, {nolocal, _}, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, [undefined])),
    %% stop the peer
    peer:stop(Peer).


two(Config) when is_list(Config) ->
    {Peer, Node} = spawn_node(?FUNCTION_NAME),
    Pid = erlang:spawn(forever()),
    ?assertEqual(ok, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, Pid)),
    ?assertEqual([Pid], pg:get_local_members(?FUNCTION_NAME, ?FUNCTION_NAME)),
    %% first RPC must be serialised
    sync({?FUNCTION_NAME, Node}),
    ?assertEqual([Pid], rpc:call(Node, pg, get_members, [?FUNCTION_NAME, ?FUNCTION_NAME])),
    ?assertEqual([], rpc:call(Node, pg, get_local_members, [?FUNCTION_NAME, ?FUNCTION_NAME])),
    stop_proc(Pid),
    %% again, must be serialised
    sync(?FUNCTION_NAME),
    ?assertEqual([], pg:get_local_members(?FUNCTION_NAME, ?FUNCTION_NAME)),
    ?assertEqual([], rpc:call(Node, pg, get_members, [?FUNCTION_NAME, ?FUNCTION_NAME])),

    Pid2 = erlang:spawn(Node, forever()),
    Pid3 = erlang:spawn(Node, forever()),
    ?assertEqual(ok, rpc:call(Node, pg, join, [?FUNCTION_NAME, ?FUNCTION_NAME, Pid2])),
    ?assertEqual(ok, rpc:call(Node, pg, join, [?FUNCTION_NAME, ?FUNCTION_NAME, Pid3])),
    %% serialise through the *other* node
    sync({?FUNCTION_NAME, Node}),
    ?assertEqual(lists:sort([Pid2, Pid3]),
        lists:sort(pg:get_members(?FUNCTION_NAME, ?FUNCTION_NAME))),
    %% stop the peer
    peer:stop(Peer),
    %% hope that 'nodedown' comes before we route our request
    sync(?FUNCTION_NAME),
    ok.

empty_group_by_remote_leave() ->
    [{doc, "Empty group should be deleted from nodes."}].

empty_group_by_remote_leave(Config) when is_list(Config) ->
    {Peer, Node} = spawn_node(?FUNCTION_NAME),
    RemoteNode = rpc:call(Node, erlang, whereis, [?FUNCTION_NAME]),
    RemotePid = erlang:spawn(Node, forever()),
    % remote join
    ?assertEqual(ok, rpc:call(Node, pg, join, [?FUNCTION_NAME, ?FUNCTION_NAME, RemotePid])),
    sync({?FUNCTION_NAME, Node}),
    ?assertEqual([RemotePid], pg:get_members(?FUNCTION_NAME, ?FUNCTION_NAME)),
    % inspecting internal state is not best practice, but there's no other way to check if the state is correct.
    {state, _, _, #{RemoteNode := {_, RemoteMap}}} = sys:get_state(?FUNCTION_NAME),
    ?assertEqual(#{?FUNCTION_NAME => [RemotePid]}, RemoteMap),
    % remote leave
    ?assertEqual(ok, rpc:call(Node, pg, leave, [?FUNCTION_NAME, ?FUNCTION_NAME, RemotePid])),
    sync({?FUNCTION_NAME, Node}),
    ?assertEqual([], pg:get_members(?FUNCTION_NAME, ?FUNCTION_NAME)),
    {state, _, _, #{RemoteNode := {_, NewRemoteMap}}} = sys:get_state(?FUNCTION_NAME),
    % empty group should be deleted.
    ?assertEqual(#{}, NewRemoteMap),

    %% another variant of emptying a group remotely: join([Pi1, Pid2]) and leave ([Pid2, Pid1])
    RemotePid2 = erlang:spawn(Node, forever()),
    ?assertEqual(ok, rpc:call(Node, pg, join, [?FUNCTION_NAME, ?FUNCTION_NAME, [RemotePid, RemotePid2]])),
    sync({?FUNCTION_NAME, Node}),
    ?assertEqual([RemotePid, RemotePid2], pg:get_members(?FUNCTION_NAME, ?FUNCTION_NAME)),
    %% now leave
    ?assertEqual(ok, rpc:call(Node, pg, leave, [?FUNCTION_NAME, ?FUNCTION_NAME, [RemotePid2, RemotePid]])),
    sync({?FUNCTION_NAME, Node}),
    ?assertEqual([], pg:get_members(?FUNCTION_NAME, ?FUNCTION_NAME)),
    {state, _, _, #{RemoteNode := {_, NewRemoteMap}}} = sys:get_state(?FUNCTION_NAME),
    peer:stop(Peer),
    ok.

thundering_herd() ->
    [{doc, "Thousands of overlay network nodes sending sync to us, and we time out!"}, {timetrap, {seconds, 5}}].

thundering_herd(Config) when is_list(Config) ->
    GroupCount = 10000,
    SyncCount = 2000,
    %% make up a large amount of groups
    [pg:join(?FUNCTION_NAME, {group, Seq}, self()) || Seq <- lists:seq(1, GroupCount)],
    %% initiate a few syncs - and those are really slow...
    {Peer, Node} = spawn_node(?FUNCTION_NAME),
    PeerPid = erlang:spawn(Node, forever()),
    PeerPg = rpc:call(Node, erlang, whereis, [?FUNCTION_NAME], 1000),
    %% WARNING: code below acts for white-box! %% WARNING
    FakeSync = [{{group, 1}, [PeerPid, PeerPid]}],
    [gen_server:cast(?FUNCTION_NAME, {sync, PeerPg, FakeSync}) || _ <- lists:seq(1, SyncCount)],
    %% next call must not timetrap, otherwise test fails
    pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, self()),
    peer:stop(Peer).

initial(Config) when is_list(Config) ->
    Pid = erlang:spawn(forever()),
    ?assertEqual(ok, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, Pid)),
    ?assertEqual([Pid], pg:get_local_members(?FUNCTION_NAME, ?FUNCTION_NAME)),
    {Peer, Node} = spawn_node(?FUNCTION_NAME),
    %% first RPC must be serialised
    sync({?FUNCTION_NAME, Node}),
    ?assertEqual([Pid], rpc:call(Node, pg, get_members, [?FUNCTION_NAME, ?FUNCTION_NAME])),

    ?assertEqual([], rpc:call(Node, pg, get_local_members, [?FUNCTION_NAME, ?FUNCTION_NAME])),
    stop_proc(Pid),
    sync({?FUNCTION_NAME, Node}),
    ?assertEqual([], rpc:call(Node, pg, get_members, [?FUNCTION_NAME, ?FUNCTION_NAME])),
    peer:stop(Peer),
    ok.

netsplit(Config) when is_list(Config) ->
    {Peer, Node} = spawn_node(?FUNCTION_NAME),
    ?assertEqual(Node, peer:call(Peer, erlang, node, [])), %% just to test RPC
    RemoteOldPid = erlang:spawn(Node, forever()),
    ?assertEqual(ok, rpc:call(Node, pg, join, [?FUNCTION_NAME, '$invisible', RemoteOldPid])),
    %% hohoho, partition!
    disconnect_nodes([Node]),
    ?assertEqual(Node, peer:call(Peer, erlang, node, [])), %% just to ensure RPC still works
    RemotePid = peer:call(Peer, erlang, spawn, [forever()]),
    ?assertEqual([], peer:call(Peer, erlang, nodes, [])),
    ?assertNot(lists:member(Node, nodes())), %% should be no nodes in the cluster
    ?assertEqual(ok, peer:call(Peer, pg, join, [?FUNCTION_NAME, ?FUNCTION_NAME, RemotePid])), %% join - in a partition!

    ?assertEqual(ok, peer:call(Peer, pg, leave, [?FUNCTION_NAME, '$invisible', RemoteOldPid])),
    ?assertEqual(ok, peer:call(Peer, pg, join, [?FUNCTION_NAME, '$visible', RemoteOldPid])),
    ?assertEqual([RemoteOldPid], peer:call(Peer, pg, get_local_members, [?FUNCTION_NAME, '$visible'])),
    %% join locally too
    LocalPid = erlang:spawn(forever()),
    ?assertEqual(ok, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, LocalPid)),

    ?assertNot(lists:member(Node, nodes())), %% should be no nodes in the cluster

    pong = net_adm:ping(Node),
    %% now ensure sync happened
    Pids = lists:sort([RemotePid, LocalPid]),
    sync({?FUNCTION_NAME, Node}),
    ?assertEqual(Pids, lists:sort(rpc:call(Node, pg, get_members, [?FUNCTION_NAME, ?FUNCTION_NAME]))),
    ?assertEqual([RemoteOldPid], pg:get_members(?FUNCTION_NAME, '$visible')),
    peer:stop(Peer),
    ok.

trisplit(Config) when is_list(Config) ->
    {Peer, Node} = spawn_node(?FUNCTION_NAME),
    _PeerPid1 = erlang:spawn(Node, forever()),
    PeerPid2 = erlang:spawn(Node, forever()),
    ?assertEqual(ok, rpc:call(Node, pg, join, [?FUNCTION_NAME, three, PeerPid2])),
    disconnect_nodes([Node]),
    ?assertEqual(true, net_kernel:connect_node(Node)),
    ?assertEqual(ok, rpc:call(Node, pg, join, [?FUNCTION_NAME, one, PeerPid2])),
    %% now ensure sync happened
    {Peer2, Node2} = spawn_node(?FUNCTION_NAME),
    ?assertEqual(true, rpc:call(Node2, net_kernel, connect_node, [Node])),
    ?assertEqual(lists:sort([node(), Node]), lists:sort(rpc:call(Node2, erlang, nodes, []))),
    ok = rpc:call(Node2, ?MODULE, ensure_peers_info, [?FUNCTION_NAME, [node(), Node]]),
    ?assertEqual([PeerPid2], rpc:call(Node2, pg, get_members, [?FUNCTION_NAME, one])),
    peer:stop(Peer),
    peer:stop(Peer2),
    ok.

foursplit(Config) when is_list(Config) ->
    Pid = erlang:spawn(forever()),
    {Peer, Node} = spawn_node(?FUNCTION_NAME),
    ?assertEqual(ok, pg:join(?FUNCTION_NAME, one, Pid)),
    ?assertEqual(ok, pg:join(?FUNCTION_NAME, two, Pid)),
    PeerPid1 = erlang:spawn(Node, forever()),
    ?assertEqual(ok, pg:leave(?FUNCTION_NAME, one, Pid)),
    ?assertEqual(not_joined, pg:leave(?FUNCTION_NAME, three, Pid)),
    disconnect_nodes([Node]),
    ?assertEqual(ok, peer:call(Peer, ?MODULE, stop_proc, [PeerPid1])),
    ?assertEqual(not_joined, pg:leave(?FUNCTION_NAME, three, Pid)),
    ?assertEqual(true, net_kernel:connect_node(Node)),
    ?assertEqual([], pg:get_members(?FUNCTION_NAME, one)),
    ?assertEqual([], peer:call(Peer, pg, get_members, [?FUNCTION_NAME, one])),
    peer:stop(Peer),
    ok.

exchange(Config) when is_list(Config) ->
    {Peer1, Node1} = spawn_node(?FUNCTION_NAME),
    {Peer2, Node2} = spawn_node(?FUNCTION_NAME),
    Pids10 = [peer:call(Peer1, erlang, spawn, [forever()]) || _ <- lists:seq(1, 10)],
    Pids2 = [peer:call(Peer2, erlang, spawn, [forever()]) || _ <- lists:seq(1, 10)],
    Pids11 = [peer:call(Peer1, erlang, spawn, [forever()]) || _ <- lists:seq(1, 10)],
    %% kill first 3 pids from node1
    {PidsToKill, Pids1} = lists:split(3, Pids10),

    ?assertEqual(ok, peer:call(Peer1, pg, join, [?FUNCTION_NAME, ?FUNCTION_NAME, Pids10])),
    sync({?FUNCTION_NAME, Node1}), %% Join broadcast have reached local
    sync(?FUNCTION_NAME), %% Join broadcast has been processed by local
    ?assertEqual(lists:sort(Pids10), lists:sort(pg:get_members(?FUNCTION_NAME, ?FUNCTION_NAME))),
    [peer:call(Peer1, ?MODULE, stop_proc, [Pid]) || Pid <- PidsToKill],
    sync(?FUNCTION_NAME),
    sync({?FUNCTION_NAME, Node1}),

    Pids = lists:sort(Pids1 ++ Pids2 ++ Pids11),
    ?assert(lists:all(fun erlang:is_pid/1, Pids)),

    disconnect_nodes([Node1, Node2]),

    sync(?FUNCTION_NAME), %% Processed nodedowns...
    ?assertEqual([], lists:sort(pg:get_members(?FUNCTION_NAME, ?FUNCTION_NAME))),

    [?assertEqual(ok, peer:call(Peer2, pg, join, [?FUNCTION_NAME, ?FUNCTION_NAME, Pid])) || Pid <- Pids2],
    [?assertEqual(ok, peer:call(Peer1, pg, join, [?FUNCTION_NAME, second, Pid])) || Pid <- Pids11],
    ?assertEqual(ok, peer:call(Peer1, pg, join, [?FUNCTION_NAME, third, Pids11])),
    %% rejoin
    ?assertEqual(true, net_kernel:connect_node(Node1)),
    ?assertEqual(true, net_kernel:connect_node(Node2)),
    %% need to sleep longer to ensure both nodes made the exchange
    ensure_peers_info(?FUNCTION_NAME, [Node1, Node2]),
    ?assertEqual(Pids, lists:sort(pg:get_members(?FUNCTION_NAME, second) ++ pg:get_members(?FUNCTION_NAME, ?FUNCTION_NAME))),
    ?assertEqual(lists:sort(Pids11), lists:sort(pg:get_members(?FUNCTION_NAME, third))),

    {Left, Stay} = lists:split(3, Pids11),
    ?assertEqual(ok, peer:call(Peer1, pg, leave, [?FUNCTION_NAME, third, Left])),
    sync({?FUNCTION_NAME, Node1}),
    sync(?FUNCTION_NAME),
    ?assertEqual(lists:sort(Stay), lists:sort(pg:get_members(?FUNCTION_NAME, third))),
    ?assertEqual(not_joined, peer:call(Peer1, pg, leave, [?FUNCTION_NAME, left, Stay])),
    ?assertEqual(ok, peer:call(Peer1, pg, leave, [?FUNCTION_NAME, third, Stay])),
    sync({?FUNCTION_NAME, Node1}),
    sync(?FUNCTION_NAME),
    ?assertEqual([], lists:sort(pg:get_members(?FUNCTION_NAME, third))),
    sync({?FUNCTION_NAME, Node1}),
    sync(?FUNCTION_NAME),

    peer:stop(Peer1),
    peer:stop(Peer2),
    ok.

nolocal(Config) when is_list(Config) ->
    {Peer, Node} = spawn_node(?FUNCTION_NAME),
    RemotePid = erlang:spawn(Node, forever()),
    ?assertEqual(ok, rpc:call(Node, pg, join, [?FUNCTION_NAME, ?FUNCTION_NAME, RemotePid])),
    ?assertEqual(ok, rpc:call(Node, pg, join, [?FUNCTION_NAME, ?FUNCTION_NAME, RemotePid])),
    ?assertEqual([], pg:get_local_members(?FUNCTION_NAME, ?FUNCTION_NAME)),
    peer:stop(Peer),
    ok.

double(Config) when is_list(Config) ->
    Pid = erlang:spawn(forever()),
    ?assertEqual(ok, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, Pid)),
    {Peer, Node} = spawn_node(?FUNCTION_NAME),
    ?assertEqual(ok, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, [Pid])),
    ?assertEqual([Pid, Pid], pg:get_members(?FUNCTION_NAME, ?FUNCTION_NAME)),
    sync(?FUNCTION_NAME),
    sync({?FUNCTION_NAME, Node}),
    ?assertEqual([Pid, Pid], rpc:call(Node, pg, get_members, [?FUNCTION_NAME, ?FUNCTION_NAME])),
    peer:stop(Peer),
    ok.

scope_restart(Config) when is_list(Config) ->
    Pid = erlang:spawn(forever()),
    ?assertEqual(ok, pg:join(?FUNCTION_NAME, ?FUNCTION_NAME, [Pid, Pid])),
    {Peer, Node} = spawn_node(?FUNCTION_NAME),
    RemotePid = erlang:spawn(Node, forever()),
    ?assertEqual(ok, rpc:call(Node, pg, join, [?FUNCTION_NAME, ?FUNCTION_NAME, RemotePid])),
    sync({?FUNCTION_NAME, Node}),
    ?assertEqual(lists:sort([RemotePid, Pid, Pid]), lists:sort(pg:get_members(?FUNCTION_NAME, ?FUNCTION_NAME))),
    %% stop scope locally, and restart
    gen_server:stop(?FUNCTION_NAME),
    pg:start(?FUNCTION_NAME),
    %% ensure remote pids joined, local are missing
    sync(?FUNCTION_NAME),
    sync({?FUNCTION_NAME, Node}),
    sync(?FUNCTION_NAME),
    ?assertEqual([RemotePid], pg:get_members(?FUNCTION_NAME, ?FUNCTION_NAME)),
    peer:stop(Peer),
    ok.

missing_scope_join(Config) when is_list(Config) ->
    {Peer, Node} = spawn_node(?FUNCTION_NAME),
    ?assertEqual(ok, rpc:call(Node, gen_server, stop, [?FUNCTION_NAME])),
    RemotePid = erlang:spawn(Node, forever()),
    ?assertMatch({badrpc, {'EXIT', {noproc, _}}}, rpc:call(Node, pg, join, [?FUNCTION_NAME, ?FUNCTION_NAME, RemotePid])),
    ?assertMatch({badrpc, {'EXIT', {noproc, _}}}, rpc:call(Node, pg, leave, [?FUNCTION_NAME, ?FUNCTION_NAME, RemotePid])),
    peer:stop(Peer),
    ok.

disconnected_start(Config) when is_list(Config) ->
    {Peer, Node} = spawn_disconnected_node(?FUNCTION_NAME, ?FUNCTION_NAME),
    ?assertNot(lists:member(Node, nodes())),
    ?assertEqual(ok, peer:call(Peer, gen_server, stop, [?FUNCTION_NAME])),
    ?assertMatch({ok, _Pid}, peer:call(Peer, pg, start,[?FUNCTION_NAME])),
    ?assertEqual(ok, peer:call(Peer, gen_server, stop, [?FUNCTION_NAME])),
    RemotePid = peer:call(Peer, erlang, spawn, [forever()]),
    ?assert(is_pid(RemotePid)),
    peer:stop(Peer),
    ok.

forced_sync() ->
    [{doc, "This test was added when lookup_element was erroneously used instead of lookup, crashing pg with badmatch, and it tests rare out-of-order sync operations"}].

forced_sync(Config) when is_list(Config) ->
    {Peer, Node} = spawn_node(?FUNCTION_NAME),
    Pid = erlang:spawn(forever()),
    RemotePid = erlang:spawn(Node, forever()),
    Expected = lists:sort([Pid, RemotePid]),
    pg:join(?FUNCTION_NAME, one, Pid),

    ?assertEqual(ok, rpc:call(Node, pg, join, [?FUNCTION_NAME, one, RemotePid])),
    RemoteScopePid = rpc:call(Node, erlang, whereis, [?FUNCTION_NAME]),
    ?assert(is_pid(RemoteScopePid)),
    %% hohoho, partition!
    disconnect_nodes([Node]),
    ?assertEqual(true, net_kernel:connect_node(Node)),
    ensure_peers_info(?FUNCTION_NAME, [Node]),
    ?assertEqual(Expected, lists:sort(pg:get_members(?FUNCTION_NAME, one))),
    %% WARNING: this code uses pg as white-box, exploiting internals,
    %%  only to simulate broken 'sync'
    %% Fake Groups: one should disappear, one should be replaced, one stays
    %% This tests handle_sync function.
    FakeGroups = [{one, [RemotePid, RemotePid]}, {?FUNCTION_NAME, [RemotePid, RemotePid]}],
    gen_server:cast(?FUNCTION_NAME, {sync, RemoteScopePid, FakeGroups}),
    %% ensure it is broken well enough
    sync(?FUNCTION_NAME),
    ?assertEqual(lists:sort([RemotePid, RemotePid]), lists:sort(pg:get_members(?FUNCTION_NAME, ?FUNCTION_NAME))),
    ?assertEqual(lists:sort([RemotePid, RemotePid, Pid]), lists:sort(pg:get_members(?FUNCTION_NAME, one))),
    %% simulate force-sync via 'discover' - ask peer to send sync to us
    {?FUNCTION_NAME, Node} ! {discover, whereis(?FUNCTION_NAME)},
    sync({?FUNCTION_NAME, Node}),
    sync(?FUNCTION_NAME),
    ?assertEqual(Expected, lists:sort(pg:get_members(?FUNCTION_NAME, one))),
    ?assertEqual([], lists:sort(pg:get_members(?FUNCTION_NAME, ?FUNCTION_NAME))),
    %% and simulate extra sync
    sync({?FUNCTION_NAME, Node}),
    sync(?FUNCTION_NAME),
    ?assertEqual(Expected, lists:sort(pg:get_members(?FUNCTION_NAME, one))),

    peer:stop(Peer),
    ok.

group_leave(Config) when is_list(Config) ->
    {Peer, Node} = spawn_node(?FUNCTION_NAME),
    RemotePid = erlang:spawn(Node, forever()),
    Total = lists:duplicate(16, RemotePid),
    {Left, Remain} = lists:split(4, Total),
    %% join 16 times!
    ?assertEqual(ok, rpc:call(Node, pg, join, [?FUNCTION_NAME, two, Total])),
    ?assertEqual(ok, rpc:call(Node, pg, leave, [?FUNCTION_NAME, two, Left])),

    sync({?FUNCTION_NAME, Node}),
    sync(?FUNCTION_NAME),
    ?assertEqual(Remain, pg:get_members(?FUNCTION_NAME, two)),
    peer:stop(Peer),
    sync(?FUNCTION_NAME),
    ?assertEqual([], pg:get_members(?FUNCTION_NAME, two)),
    ok.

%%--------------------------------------------------------------------
%% Test Helpers - start/stop additional Erlang nodes

sync(GS) ->
    _ = sys:log(GS, get).

ensure_peers_info(Scope, Nodes) ->
    %% Ensures that pg server on local node has gotten info from
    %% pg servers on all Peer nodes passed as argument (assuming
    %% no connection failures).
    %% 
    %% This function assumes that all nodeup messages has been
    %% delivered to all local recipients (pg server) when called.
    %%
    %% Note that this relies on current ERTS implementation; not
    %% language guarantees.
    %%

    sync(Scope),
    %% Known: nodup handled and discover sent to Peer

    lists:foreach(fun (Node) -> sync({Scope, Node}) end, Nodes),
    %% Known: nodeup handled by Peers and discover sent to local
    %% Known: discover received/handled by Peers and sync sent to local
    %% Known: discover received from Peer
    %% Known: sync received from Peer

    sync(Scope),
    %% Known: discover handled from Peers and sync sent to Peers
    %% Known: sync from Peers handled
    ok.

-ifdef(CURRENTLY_UNUSED_BUT_SERVES_AS_DOC).

ensure_synced(Scope, Nodes) ->
    %% Ensures that the pg server on local node have synced
    %% with pg servers on all Peer nodes (assuming no connection
    %% failures).
    %% 
    %% This function assumes that all nodeup messages has been
    %% delivered to all local recipients (pg server) when called.
    %%
    %% Note that this relies on current ERTS implementation; not
    %% language guarantees.
    %%
    ensure_peer_info(Scope, Node),
    %% Known: local has gotten info from all Peers
    %% Known: discover from Peers handled and sync sent to Peers
    lists:foreach(fun (Node) -> sync({Scope, Node}) end, Nodes),
    %% Known: sync from local handled by Peers
    ok.

-endif.

disconnect_nodes(Nodes) ->
    %% The following is not a language guarantee, but internal
    %% knowledge about current implementation of ERTS and pg.
    %%
    %% The pg server reacts on 'DOWN's via process monitors of
    %% its peers. These are delivered before 'nodedown's from
    %% net_kernel:monitor_nodes(). That is, by waiting for
    %% 'nodedown' from net_kernel:monitor_nodes() we know that
    %% the 'DOWN' has been delivered to the pg server.
    %%
    %% We do this in a separate process to avoid stray
    %% nodeup/nodedown messages in the test process after
    %% the operation...
    F = fun () ->
                ok = net_kernel:monitor_nodes(true),
                lists:foreach(fun (Node) ->
                                      true = erlang:disconnect_node(Node)
                              end,
                              Nodes),
                lists:foreach(fun (Node) ->
                                      receive {nodedown, Node} -> ok end
                              end,
                              Nodes)
        end,
    {Pid, Mon} = spawn_monitor(F),
    receive
        {'DOWN', Mon, process, Pid, Reason} ->
            ?assertEqual(normal, Reason)
    end,
    ok.

%% @doc Kills process Pid and waits for it to exit using monitor,
%%      and yields after (for 1 ms).
-spec stop_proc(pid()) -> ok.
stop_proc(Pid) ->
    monitor(process, Pid),
    erlang:exit(Pid, kill),
    receive
        {'DOWN', _MRef, process, Pid, _Info} ->
            timer:sleep(1)
    end.

forever() ->
    Parent = self(),
    fun() ->
            %% forever() is used both locally and on a remote node,
            %% if used locally, we want to terminate when the
            %% parent terminates in order to not leak process to
            %% later test cases
            Ref = monitor(process,Parent),
            receive
                {'DOWN',Ref,_,_,_} when node() =:= node(Parent) ->
                    ok
            end
    end.

spawn_node(TestCase) ->
    {Peer, Node} = spawn_disconnected_node(TestCase, TestCase),
    true = net_kernel:connect_node(Node),
    {Peer, Node}.

spawn_disconnected_node(Scope, TestCase) ->
    {ok, Peer, Node} = ?CT_PEER(#{name => ?CT_PEER_NAME(TestCase), connection => 0,
        args => ["-connect_all", "false", "-kernel", "dist_auto_connect", "never"]}),
    {ok, _Pid} = peer:call(Peer, pg, start, [Scope]),
    {Peer, Node}.
