%%%------------------------------------------------------------------- %%% @doc Тесты WebSocket API (пользовательские и административные). %%% %%% Покрывает эндпоинты: %%% ws://localhost:8081/ws %%% ws://localhost:8446/admin/ws %%% ws://localhost:8446/admin/ws/metrics %%% %%% Проверяет: %%% - подключение с валидным пользовательским токеном %%% - подписку на календарь и получение подтверждения %%% - подключение с валидным админским токеном %%% - подписку на каналы отчётов и тикетов %%% - получение уведомления о создании отчёта %%% - Ping/Pong %%% - отписку от канала %%% - получение метрик узла через /admin/ws/metrics %%% - отклонение при использовании пользовательского токена для админского сокета %%% - отклонение при невалидном токене %%% @end %%%------------------------------------------------------------------- -module(admin_websocket_tests). -include_lib("eunit/include/eunit.hrl"). -export([test/0]). %%%=================================================================== %%% Главная тестовая функция %%%=================================================================== -spec test() -> ok. test() -> ct:pal("=== Admin WebSocket Tests ==="), application:ensure_all_started(gun), AdminToken = api_test_runner:get_admin_token(), UserToken = api_test_runner:get_user_token(), % Создаём календарь и событие для тестов #{<<"id">> := CalId} = api_test_runner:client_post( <<"/v1/calendars">>, UserToken, #{title => <<"WS Test Calendar">>, type => <<"commercial">>}), #{<<"id">> := EventId} = api_test_runner:client_post( <<"/v1/calendars/", CalId/binary, "/events">>, UserToken, #{title => <<"WS Test Event">>, start_time => api_test_runner:future_date_iso8601(), duration => 60}), WsUrl = api_test_runner:get_base_ws_url() ++ "/ws", AdminWsUrl = api_test_runner:get_admin_ws_url() ++ "/admin/ws", MetricsWsUrl = api_test_runner:get_admin_ws_url() ++ "/admin/ws/metrics", % ── Пользовательские WebSocket тесты ── {ok, UserWS} = test_user_ws_connect(WsUrl, UserToken), test_user_ws_subscribe(UserWS, CalId), test_ws_close(UserWS), ct:pal("~n✅ User WebSocket tests passed!"), % ── Административные WebSocket тесты ── {ok, AdminWS} = test_admin_ws_connect(AdminWsUrl, AdminToken), test_admin_ws_subscribe_reports(AdminWS), test_admin_ws_subscribe_tickets(AdminWS), test_admin_ws_report_notification(AdminWS, UserToken, EventId), test_admin_ws_ping(AdminWS), test_admin_ws_unsubscribe(AdminWS), test_ws_close(AdminWS), % ── Метрики узлов ── test_admin_ws_node_metrics(MetricsWsUrl, AdminToken), % ── Негативные тесты ── test_admin_ws_user_token_rejected(AdminWsUrl, UserToken), test_admin_ws_invalid_token_rejected(AdminWsUrl), ct:pal("~n=== All admin WebSocket tests passed! ==="), {?MODULE, ok}. %%%=================================================================== %%% Тестовые функции %%%=================================================================== %% @doc Подключение пользовательского WebSocket с валидным токеном. -spec test_user_ws_connect(string(), binary()) -> {ok, pid()}. test_user_ws_connect(Url, Token) -> ct:pal(" TEST: Connect user WebSocket with valid token"), case test_ws_connect_debug(Url, Token) of {ok, WS} -> {ok, WS}; Other -> error({unexpected, Other}) end. %% @doc Подписка на календарь и получение подтверждения. -spec test_user_ws_subscribe(pid(), binary()) -> ok. test_user_ws_subscribe(WS, CalId) -> ct:pal(" TEST: Subscribe to calendar"), SubMsg = #{action => <<"subscribe">>, calendar_id => CalId}, ok = test_ws_send(WS, SubMsg), case test_ws_recv(WS) of {ok, #{<<"status">> := <<"subscribed">>}} -> ok; {ok, Other} -> error({unexpected_response, Other}); {error, timeout} -> error(timeout) end. %% @doc Подключение административного WebSocket. -spec test_admin_ws_connect(string(), binary()) -> {ok, pid()}. test_admin_ws_connect(Url, Token) -> ct:pal(" TEST: Connect admin WebSocket"), case test_ws_connect_debug(Url, Token) of {ok, WS} -> {ok, WS}; Other -> error({unexpected, Other}) end. %% @doc Подписка на канал reports. -spec test_admin_ws_subscribe_reports(pid()) -> ok. test_admin_ws_subscribe_reports(WS) -> ct:pal(" TEST: Admin subscribe to reports"), ok = test_ws_send(WS, #{action => <<"subscribe">>, channel => <<"reports">>}), {ok, #{<<"status">> := <<"subscribed">>}} = test_ws_recv(WS). %% @doc Подписка на канал tickets. -spec test_admin_ws_subscribe_tickets(pid()) -> ok. test_admin_ws_subscribe_tickets(WS) -> ct:pal(" TEST: Admin subscribe to tickets"), ok = test_ws_send(WS, #{action => <<"subscribe">>, channel => <<"tickets">>}), {ok, #{<<"status">> := <<"subscribed">>}} = test_ws_recv(WS). %% @doc Получение уведомления о создании отчёта. -spec test_admin_ws_report_notification(pid(), binary(), binary()) -> ok. test_admin_ws_report_notification(WS, UserToken, EventId) -> ct:pal(" TEST: Admin receives report notification"), api_test_runner:client_post(<<"/v1/reports">>, UserToken, #{target_type => <<"event">>, target_id => EventId, reason => <<"Test">>}), {ok, #{<<"type">> := <<"report_created">>}} = test_ws_recv(WS, 5000). %% @doc Ping/Pong. -spec test_admin_ws_ping(pid()) -> ok. test_admin_ws_ping(WS) -> ct:pal(" TEST: Admin Ping/Pong"), ok = test_ws_send(WS, #{action => <<"ping">>}), {ok, #{<<"status">> := <<"pong">>}} = test_ws_recv(WS). %% @doc Отписка от канала reports. -spec test_admin_ws_unsubscribe(pid()) -> ok. test_admin_ws_unsubscribe(WS) -> ct:pal(" TEST: Admin unsubscribe from reports"), ok = test_ws_send(WS, #{action => <<"unsubscribe">>, channel => <<"reports">>}), {ok, #{<<"status">> := <<"unsubscribed">>}} = test_ws_recv(WS). %% @doc Получение метрик узла через /admin/ws/metrics. -spec test_admin_ws_node_metrics(string(), binary()) -> ok. test_admin_ws_node_metrics(Url, Token) -> ct:pal(" TEST: Admin WS receive node metrics"), {ok, WS} = test_ws_connect_debug(Url, Token), ok = test_ws_send(WS, #{action => <<"ping">>}), {ok, #{<<"status">> := <<"pong">>}} = test_ws_recv(WS, 3000), ct:pal(" Ping/Pong OK – waiting for node metric..."), case test_ws_recv_type(WS, <<"node_metric">>, 15000) of {ok, #{<<"data">> := MetricData}} -> ct:pal(" OK - Received node metric"), ?assert(is_map(MetricData)), ?assert(maps:is_key(<<"timestamp">>, MetricData)), ?assert(maps:is_key(<<"node">>, MetricData)), ?assert(maps:is_key(<<"memory_total">>, MetricData)), ct:pal(" Node: ~s, Memory: ~p MB", [maps:get(<<"node">>, MetricData), maps:get(<<"memory_total">>, MetricData) div 1048576]); {error, timeout} -> ct:pal(" WARNING: No node metric received (timeout) – check monitor logs") end, test_ws_close(WS). %% @doc Отклонение при использовании пользовательского токена для админского сокета. -spec test_admin_ws_user_token_rejected(string(), binary()) -> ok. test_admin_ws_user_token_rejected(Url, Token) -> ct:pal(" TEST: Admin WS with user token rejected"), ?assertMatch({error, {403, _}}, test_ws_connect_debug(Url, Token)). %% @doc Отклонение при невалидном токене. -spec test_admin_ws_invalid_token_rejected(string()) -> ok. test_admin_ws_invalid_token_rejected(Url) -> ct:pal(" TEST: Admin WS with invalid token rejected"), Chars = <<"abcdefghijklmnopqrstuvwxyz0123456789">>, InvalidToken = << <<(binary:at(Chars, rand:uniform(byte_size(Chars)) - 1))>> || _ <- lists:seq(1, 30) >>, ?assertMatch({error, {401, _}}, test_ws_connect_debug(Url, InvalidToken)). %% ============ WebSocket хелперы с отладкой ============ test_ws_connect_debug(Url, Token) -> Path = case string:split(Url, "://", trailing) of [_, Rest] -> case string:split(Rest, "/", leading) of [_HostPort, WsPath] -> "/" ++ WsPath ++ "?token=" ++ binary_to_list(Token); _ -> "/ws?token=" ++ binary_to_list(Token) end; _ -> "/ws?token=" ++ binary_to_list(Token) end, {ok, Port} = extract_port(Url), {ok, Host} = extract_host(Url), Opts = case Port of 443 -> #{protocols => [http], transport => tls, tls_opts => [{verify, verify_none}]}; _ -> #{protocols => [http]} end, ct:pal(" Host: ~s", [Host]), ct:pal(" Port: ~p", [Port]), ct:pal(" Path: ~s", [Path]), {ok, ConnPid} = gun:open(Host, Port, Opts), {ok, http} = gun:await_up(ConnPid, 5000), Headers = [{<<"host">>, list_to_binary(Host ++ ":" ++ integer_to_list(Port))}], StreamRef = gun:ws_upgrade(ConnPid, Path, Headers), receive {gun_upgrade, ConnPid, StreamRef, [<<"websocket">>], _} -> ct:pal(" WebSocket upgrade OK"), {ok, ConnPid}; {gun_response, ConnPid, StreamRef, fin, 401, _} -> ct:pal(" ERROR: HTTP 401 Unauthorized"), gun:close(ConnPid), {error, {401, <<"Invalid token">>}}; {gun_response, ConnPid, StreamRef, fin, 403, _} -> ct:pal(" ERROR: HTTP 403 Forbidden"), gun:close(ConnPid), {error, {403, <<"Admin access required">>}}; {gun_response, ConnPid, StreamRef, nofin, 403, _} -> ct:pal(" ERROR: HTTP 403 Forbidden (nofin)"), gun:close(ConnPid), {error, {403, <<"Admin access required">>}}; {gun_response, ConnPid, StreamRef, fin, Status, _} -> ct:pal(" ERROR: HTTP ~p", [Status]), gun:close(ConnPid), {error, {Status, <<"WebSocket upgrade failed">>}}; {gun_response, ConnPid, StreamRef, nofin, Status, _} -> ct:pal(" ERROR: HTTP ~p (nofin)", [Status]), gun:close(ConnPid), {error, {Status, <<"WebSocket upgrade failed">>}}; {gun_error, ConnPid, Reason} -> ct:pal(" ERROR: ~p", [Reason]), gun:close(ConnPid), {error, Reason} after 5000 -> ct:pal(" ERROR: Timeout"), gun:close(ConnPid), {error, timeout} end. test_ws_send(ConnPid, Data) -> Msg = jsx:encode(Data), ct:pal(" Sending: ~s", [Msg]), case catch gun:ws_send(ConnPid, {text, Msg}) of ok -> ok; {'EXIT', {undef, _}} -> gun:ws_send(ConnPid, fin, {text, Msg}); Other -> ct:pal(" ERROR sending: ~p", [Other]), error({ws_send_failed, Other}) end. test_ws_recv(ConnPid) -> test_ws_recv(ConnPid, 3000). test_ws_recv(ConnPid, Timeout) -> receive {gun_ws, ConnPid, _StreamRef, {text, Msg}} -> ct:pal(" Received (with StreamRef): ~s", [Msg]), {ok, jsx:decode(Msg, [return_maps])}; {gun_ws, ConnPid, {text, Msg}} -> ct:pal(" Received: ~s", [Msg]), {ok, jsx:decode(Msg, [return_maps])}; {gun_ws, ConnPid, _StreamRef, Frame} -> ct:pal(" Received frame: ~p", [Frame]), {ok, Frame}; {gun_ws, ConnPid, Frame} -> ct:pal(" Received: ~p", [Frame]), {ok, Frame}; {gun_error, ConnPid, Reason} -> ct:pal(" ERROR: gun_error ~p", [Reason]), {error, Reason}; Other -> ct:pal(" Received unexpected: ~p", [Other]), test_ws_recv(ConnPid, Timeout) after Timeout -> {error, timeout} end. %% @doc Ждём WS-сообщение заданного type, пропуская посторонние (шум на shared stand). test_ws_recv_type(ConnPid, ExpectedType, Timeout) -> Deadline = erlang:monotonic_time(millisecond) + Timeout, test_ws_recv_type_loop(ConnPid, ExpectedType, Deadline). test_ws_recv_type_loop(ConnPid, ExpectedType, Deadline) -> Remaining = max(0, Deadline - erlang:monotonic_time(millisecond)), case test_ws_recv(ConnPid, Remaining) of {ok, #{<<"type">> := ExpectedType} = Msg} -> {ok, Msg}; {ok, #{<<"type">> := ActualType} = Other} -> ct:pal(" Skipping WS type=~s (waiting for ~s)", [ActualType, ExpectedType]), test_ws_recv_type_loop(ConnPid, ExpectedType, Deadline); {ok, Other} -> ct:pal(" Skipping WS message ~p (waiting for type ~s)", [Other, ExpectedType]), test_ws_recv_type_loop(ConnPid, ExpectedType, Deadline); {error, timeout} = Err -> Err end. test_ws_close(ConnPid) -> gun:close(ConnPid). %% ========== URL parsing helpers ========== extract_port(Url) -> case string:split(Url, "://", trailing) of [_, Rest] -> HostPort = case string:split(Rest, "/", leading) of [H, _] -> H; [H] -> H end, case string:split(HostPort, ":", trailing) of [_, PortStr] -> {ok, list_to_integer(PortStr)}; _ -> case string:split(Rest, "://", trailing) of [_, R] -> extract_port("https://" ++ R); _ -> {ok, default_port(Url)} end end; _ -> {ok, default_port(Url)} end. default_port(Url) -> case string:prefix(Url, "wss://") of nomatch -> case string:prefix(Url, "ws://") of nomatch -> 80; _ -> 80 end; _ -> 443 end. extract_host(Url) -> case string:split(Url, "://", trailing) of [_, Rest] -> HostPort = case string:split(Rest, "/", leading) of [H, _] -> H; [H] -> H end, case string:split(HostPort, ":", trailing) of [Host, _] -> {ok, Host}; [Host] -> {ok, Host} end; _ -> {ok, "localhost"} end.