-module(dream_test@parallel). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]). -define(FILEPATH, "src/dream_test/parallel.gleam"). -export([default_config/0, run_root_parallel/2, run_root_parallel_with_reporter/1]). -export_type([parallel_config/0, run_root_parallel_with_reporter_config/1, run_root_parallel_with_reporter_result/0, execute_node_config/1, indexed_result/0, running_test/0, run_parallel_loop_config/1, start_workers_up_to_limit_config/1, wait_for_event_or_timeout_config/1, worker_message/0, group_hooks/1]). -if(?OTP_RELEASE >= 27). -define(MODULEDOC(Str), -moduledoc(Str)). -define(DOC(Str), -doc(Str)). -else. -define(MODULEDOC(Str), -compile([])). -define(DOC(Str), -compile([])). -endif. ?MODULEDOC( " Unified Root/Node execution engine (parallel tests, sequential groups).\n" "\n" " This module executes `types.TestSuite(context)` which is an alias of\n" " `types.Root(context)` in the unified tree model.\n" "\n" " NOTE: This module is intentionally event-agnostic; `runner` composes it\n" " with reporters for live output.\n" "\n" " Most users should not call this module directly—prefer `dream_test/runner`.\n" " This module is public so advanced users can embed the executor in other\n" " tooling (custom runners, IDE integrations, etc.).\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_test/matchers.{have_length, or_fail_with, should, succeed}\n" " import dream_test/parallel\n" " import dream_test/unit.{describe, it}\n" "\n" " pub fn tests() {\n" " describe(\"Parallel executor\", [\n" " it(\"can run a suite and return a list of results\", fn() {\n" " let suite =\n" " describe(\"Suite\", [\n" " it(\"a\", fn() { Ok(succeed()) }),\n" " it(\"b\", fn() { Ok(succeed()) }),\n" " ])\n" "\n" " parallel.run_root_parallel(parallel.default_config(), suite)\n" " |> should\n" " |> have_length(2)\n" " |> or_fail_with(\"expected two results\")\n" " }),\n" " ])\n" " }\n" " ```\n" ). -type parallel_config() :: {parallel_config, integer(), integer()}. -type run_root_parallel_with_reporter_config(ICI) :: {run_root_parallel_with_reporter_config, parallel_config(), dream_test@types:root(ICI), gleam@option:option(dream_test@reporters@progress:progress_reporter()), fun((binary()) -> nil), integer(), integer()}. -type run_root_parallel_with_reporter_result() :: {run_root_parallel_with_reporter_result, list(dream_test@types:test_result()), integer(), gleam@option:option(dream_test@reporters@progress:progress_reporter())}. -type execute_node_config(ICJ) :: {execute_node_config, parallel_config(), list(binary()), list(binary()), ICJ, list(fun((ICJ) -> {ok, ICJ} | {error, binary()})), list(fun((ICJ) -> {ok, nil} | {error, binary()})), dream_test@types:node_(ICJ), gleam@option:option(dream_test@reporters@progress:progress_reporter()), fun((binary()) -> nil), integer(), integer(), list(dream_test@types:test_result())}. -type indexed_result() :: {indexed_result, integer(), dream_test@types:test_result()}. -type running_test() :: {running_test, integer(), gleam@erlang@process:pid_(), integer(), binary(), list(binary()), list(binary()), dream_test@types:test_kind()}. -type run_parallel_loop_config(ICK) :: {run_parallel_loop_config, parallel_config(), gleam@erlang@process:subject(worker_message()), list(binary()), list(binary()), ICK, list(fun((ICK) -> {ok, ICK} | {error, binary()})), list(fun((ICK) -> {ok, nil} | {error, binary()})), list(dream_test@types:assertion_failure()), list({integer(), dream_test@types:node_(ICK)}), list(running_test()), list(indexed_result()), list(dream_test@types:test_result()), integer(), gleam@option:option(dream_test@reporters@progress:progress_reporter()), fun((binary()) -> nil), integer(), integer(), integer()}. -type start_workers_up_to_limit_config(ICL) :: {start_workers_up_to_limit_config, parallel_config(), gleam@erlang@process:subject(worker_message()), list(binary()), list(binary()), ICL, list(fun((ICL) -> {ok, ICL} | {error, binary()})), list(fun((ICL) -> {ok, nil} | {error, binary()})), list(dream_test@types:assertion_failure()), list({integer(), dream_test@types:node_(ICL)}), list(running_test()), integer()}. -type wait_for_event_or_timeout_config(ICM) :: {wait_for_event_or_timeout_config, parallel_config(), gleam@erlang@process:subject(worker_message()), list(binary()), list({integer(), dream_test@types:node_(ICM)}), list(running_test()), list(indexed_result()), gleam@option:option(dream_test@reporters@progress:progress_reporter()), fun((binary()) -> nil), integer(), integer()}. -type worker_message() :: {worker_completed, integer(), dream_test@types:test_result()} | {worker_crashed, integer(), binary()}. -type group_hooks(ICN) :: {group_hooks, list(fun((ICN) -> {ok, ICN} | {error, binary()})), list(fun((ICN) -> {ok, ICN} | {error, binary()})), list(fun((ICN) -> {ok, nil} | {error, binary()})), list(fun((ICN) -> {ok, nil} | {error, binary()}))}. -file("src/dream_test/parallel.gleam", 259). ?DOC( " Default executor configuration.\n" "\n" " Prefer configuring these values via `dream_test/runner` unless you are using\n" " the executor directly.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_test/matchers.{have_length, or_fail_with, should, succeed}\n" " import dream_test/parallel\n" " import dream_test/unit.{describe, it}\n" "\n" " pub fn tests() {\n" " describe(\"Parallel executor\", [\n" " it(\"can run a suite and return a list of results\", fn() {\n" " let suite =\n" " describe(\"Suite\", [\n" " it(\"a\", fn() { Ok(succeed()) }),\n" " it(\"b\", fn() { Ok(succeed()) }),\n" " ])\n" "\n" " parallel.run_root_parallel(parallel.default_config(), suite)\n" " |> should\n" " |> have_length(2)\n" " |> or_fail_with(\"expected two results\")\n" " }),\n" " ])\n" " }\n" " ```\n" "\n" " ## Parameters\n" "\n" " None.\n" "\n" " ## Returns\n" "\n" " A `ParallelConfig` with sensible defaults.\n" ). -spec default_config() -> parallel_config(). default_config() -> {parallel_config, 4, 5000}. -file("src/dream_test/parallel.gleam", 325). -spec discard_write(binary()) -> nil. discard_write(_) -> nil. -file("src/dream_test/parallel.gleam", 582). -spec fail_test_nodes( list(binary()), list(binary()), list(dream_test@types:node_(any())), list(dream_test@types:assertion_failure()), list(dream_test@types:test_result()) ) -> list(dream_test@types:test_result()). fail_test_nodes(Scope, Inherited_tags, Nodes, Failures_rev, Acc_rev) -> case Nodes of [] -> Acc_rev; [{test, Name, Tags, Kind, _, _} | Rest] -> Full_name = lists:append(Scope, [Name]), All_tags = lists:append(Inherited_tags, Tags), Failures = lists:reverse(Failures_rev), Result = {test_result, Name, Full_name, failed, 0, All_tags, Failures, Kind}, fail_test_nodes( Scope, Inherited_tags, Rest, Failures_rev, [Result | Acc_rev] ); [_ | Rest@1] -> fail_test_nodes( Scope, Inherited_tags, Rest@1, Failures_rev, Acc_rev ) end. -file("src/dream_test/parallel.gleam", 661). -spec collect_children( list(dream_test@types:node_(IDZ)), group_hooks(IDZ), list(dream_test@types:node_(IDZ)), list(dream_test@types:node_(IDZ)) ) -> {group_hooks(IDZ), list(dream_test@types:node_(IDZ)), list(dream_test@types:node_(IDZ))}. collect_children(Children, Hooks, Tests_rev, Groups_rev) -> Hooks0 = case Hooks of {group_hooks, _, _, _, _} -> Hooks end, case Children of [] -> {Hooks0, lists:reverse(Tests_rev), lists:reverse(Groups_rev)}; [Child | Rest] -> case Child of {before_all, Run} -> collect_children( Rest, {group_hooks, lists:append(erlang:element(2, Hooks0), [Run]), erlang:element(3, Hooks0), erlang:element(4, Hooks0), erlang:element(5, Hooks0)}, Tests_rev, Groups_rev ); {before_each, Run@1} -> collect_children( Rest, {group_hooks, erlang:element(2, Hooks0), lists:append(erlang:element(3, Hooks0), [Run@1]), erlang:element(4, Hooks0), erlang:element(5, Hooks0)}, Tests_rev, Groups_rev ); {after_each, Run@2} -> collect_children( Rest, {group_hooks, erlang:element(2, Hooks0), erlang:element(3, Hooks0), lists:append(erlang:element(4, Hooks0), [Run@2]), erlang:element(5, Hooks0)}, Tests_rev, Groups_rev ); {after_all, Run@3} -> collect_children( Rest, {group_hooks, erlang:element(2, Hooks0), erlang:element(3, Hooks0), erlang:element(4, Hooks0), lists:append(erlang:element(5, Hooks0), [Run@3])}, Tests_rev, Groups_rev ); {test, _, _, _, _, _} -> collect_children( Rest, Hooks0, [Child | Tests_rev], Groups_rev ); {group, _, _, _} -> collect_children( Rest, Hooks0, Tests_rev, [Child | Groups_rev] ) end end. -file("src/dream_test/parallel.gleam", 858). -spec index_tests( list(dream_test@types:node_(IGH)), integer(), list({integer(), dream_test@types:node_(IGH)}) ) -> list({integer(), dream_test@types:node_(IGH)}). index_tests(Nodes, Next_index, Acc_rev) -> case Nodes of [] -> lists:reverse(Acc_rev); [{test, _, _, _, _, _} = T | Rest] -> index_tests(Rest, Next_index + 1, [{Next_index, T} | Acc_rev]); [_ | Rest@1] -> index_tests(Rest@1, Next_index, Acc_rev) end. -file("src/dream_test/parallel.gleam", 1036). -spec running_test_metadata( list(binary()), list(binary()), dream_test@types:node_(any()) ) -> {binary(), list(binary()), list(binary()), dream_test@types:test_kind()}. running_test_metadata(Scope, Inherited_tags, Node) -> case Node of {test, Name, Tags, Kind, _, _} -> {Name, lists:append(Scope, [Name]), lists:append(Inherited_tags, Tags), Kind}; _ -> {<<""/utf8>>, lists:append(Scope, [<<""/utf8>>]), Inherited_tags, unit} end. -file("src/dream_test/parallel.gleam", 1092). -spec test_timeout_ms(parallel_config(), dream_test@types:node_(any())) -> integer(). test_timeout_ms(Config, Node) -> {parallel_config, _, Default_timeout_ms} = Config, case Node of {test, _, _, _, _, {some, Ms}} -> Ms; _ -> Default_timeout_ms end. -file("src/dream_test/parallel.gleam", 1216). -spec get_running_by_index(list(running_test()), integer()) -> gleam@option:option(running_test()). get_running_by_index(Running, Index) -> case Running of [] -> none; [R | Rest] -> case erlang:element(2, R) =:= Index of true -> {some, R}; false -> get_running_by_index(Rest, Index) end end. -file("src/dream_test/parallel.gleam", 1284). -spec emit_test_finished_progress( gleam@option:option(dream_test@reporters@progress:progress_reporter()), fun((binary()) -> nil), integer(), integer(), dream_test@types:test_result() ) -> integer(). emit_test_finished_progress(Progress_reporter, Write, Completed, Total, Result) -> Next_completed = Completed + 1, case Progress_reporter of none -> Next_completed; {some, Reporter} -> case dream_test@reporters@progress:handle_event( Reporter, {test_finished, Next_completed, Total, Result} ) of none -> nil; {some, Text} -> Write(Text) end, Next_completed end. -file("src/dream_test/parallel.gleam", 1313). -spec emit_test_finished_progress_list( gleam@option:option(dream_test@reporters@progress:progress_reporter()), fun((binary()) -> nil), integer(), integer(), list(indexed_result()) ) -> integer(). emit_test_finished_progress_list( Progress_reporter, Write, Completed, Total, Results ) -> case Results of [] -> Completed; [{indexed_result, _, Result} | Rest] -> Next_completed = emit_test_finished_progress( Progress_reporter, Write, Completed, Total, Result ), emit_test_finished_progress_list( Progress_reporter, Write, Next_completed, Total, Rest ) end. -file("src/dream_test/parallel.gleam", 1342). -spec emit_test_finished_progress_results( gleam@option:option(dream_test@reporters@progress:progress_reporter()), fun((binary()) -> nil), integer(), integer(), list(dream_test@types:test_result()) ) -> integer(). emit_test_finished_progress_results( Progress_reporter, Write, Completed, Total, Results ) -> case Results of [] -> Completed; [Result | Rest] -> Next_completed = emit_test_finished_progress( Progress_reporter, Write, Completed, Total, Result ), emit_test_finished_progress_results( Progress_reporter, Write, Next_completed, Total, Rest ) end. -file("src/dream_test/parallel.gleam", 1371). -spec partition_timeouts( list(running_test()), integer(), list(running_test()), list(running_test()) ) -> {list(running_test()), list(running_test())}. partition_timeouts(Running, Now, Timed_out_rev, Still_rev) -> case Running of [] -> {lists:reverse(Timed_out_rev), lists:reverse(Still_rev)}; [R | Rest] -> case erlang:element(4, R) =< Now of true -> partition_timeouts( Rest, Now, [R | Timed_out_rev], Still_rev ); false -> partition_timeouts( Rest, Now, Timed_out_rev, [R | Still_rev] ) end end. -file("src/dream_test/parallel.gleam", 1398). -spec min_deadline_timeout(list(running_test()), integer(), integer()) -> integer(). min_deadline_timeout(Running, Now, Current) -> case Running of [] -> case Current < 0 of true -> 0; false -> Current end; [R | Rest] -> D = erlang:element(4, R) - Now, Next = case D < Current of true -> D; false -> Current end, min_deadline_timeout(Rest, Now, Next) end. -file("src/dream_test/parallel.gleam", 1387). -spec next_deadline_timeout(list(running_test()), integer(), integer()) -> integer(). next_deadline_timeout(Running, Now, Fallback) -> case Running of [] -> Fallback; [R | Rest] -> min_deadline_timeout(Rest, Now, erlang:element(4, R) - Now) end. -file("src/dream_test/parallel.gleam", 1420). -spec remove_running_by_index(list(running_test()), integer()) -> list(running_test()). remove_running_by_index(Running, Index) -> gleam@list:filter(Running, fun(R) -> erlang:element(2, R) /= Index end). -file("src/dream_test/parallel.gleam", 1427). -spec take_indexed_result( integer(), list(indexed_result()), list(indexed_result()) ) -> {gleam@option:option(indexed_result()), list(indexed_result())}. take_indexed_result(Index, Items, Acc_rev) -> case Items of [] -> {none, lists:reverse(Acc_rev)}; [Item | Rest] -> case erlang:element(2, Item) =:= Index of true -> {{some, Item}, lists:append(lists:reverse(Acc_rev), Rest)}; false -> take_indexed_result(Index, Rest, [Item | Acc_rev]) end end. -file("src/dream_test/parallel.gleam", 964). -spec emit_all_results( list(indexed_result()), integer(), list(dream_test@types:test_result()) ) -> list(dream_test@types:test_result()). emit_all_results(Pending_emit, Next_index, Acc_rev) -> {Maybe_ready, Remaining} = take_indexed_result(Next_index, Pending_emit, []), case Maybe_ready of none -> lists:reverse(Acc_rev); {some, {indexed_result, _, Result}} -> emit_all_results(Remaining, Next_index + 1, [Result | Acc_rev]) end. -file("src/dream_test/parallel.gleam", 1709). -spec run_hook_transform( parallel_config(), list(binary()), ILK, fun((ILK) -> {ok, ILK} | {error, binary()}) ) -> {ok, ILK} | {error, binary()}. run_hook_transform(Config, Scope, Context, Hook) -> {parallel_config, _, Default_timeout_ms} = Config, Sandbox_config = {sandbox_config, Default_timeout_ms, false}, case dream_test@sandbox:run_isolated( Sandbox_config, fun() -> Hook(Context) end ) of {sandbox_completed, Result} -> Result; sandbox_timed_out -> {error, <<"hook timed out in "/utf8, (gleam@string:join(Scope, <<" > "/utf8>>))/binary>>}; {sandbox_crashed, Reason} -> {error, <<<<<<"hook crashed in "/utf8, (gleam@string:join(Scope, <<" > "/utf8>>))/binary>>/binary, ": "/utf8>>/binary, Reason/binary>>} end. -file("src/dream_test/parallel.gleam", 1732). -spec run_hook_teardown( parallel_config(), list(binary()), ILQ, fun((ILQ) -> {ok, nil} | {error, binary()}) ) -> {ok, nil} | {error, binary()}. run_hook_teardown(Config, Scope, Context, Hook) -> {parallel_config, _, Default_timeout_ms} = Config, Sandbox_config = {sandbox_config, Default_timeout_ms, false}, case dream_test@sandbox:run_isolated( Sandbox_config, fun() -> Hook(Context) end ) of {sandbox_completed, Result} -> Result; sandbox_timed_out -> {error, <<"hook timed out in "/utf8, (gleam@string:join(Scope, <<" > "/utf8>>))/binary>>}; {sandbox_crashed, Reason} -> {error, <<<<<<"hook crashed in "/utf8, (gleam@string:join(Scope, <<" > "/utf8>>))/binary>>/binary, ": "/utf8>>/binary, Reason/binary>>} end. -file("src/dream_test/parallel.gleam", 1806). -spec hook_failure(binary(), binary()) -> dream_test@types:assertion_failure(). hook_failure(Operator, Message) -> {assertion_failure, Operator, Message, none}. -file("src/dream_test/parallel.gleam", 723). -spec run_before_all_chain( parallel_config(), list(binary()), IEN, list(fun((IEN) -> {ok, IEN} | {error, binary()})), list(fun((IEN) -> {ok, IEN} | {error, binary()})), list(fun((IEN) -> {ok, nil} | {error, binary()})), list(dream_test@types:assertion_failure()) ) -> {IEN, list(fun((IEN) -> {ok, IEN} | {error, binary()})), list(fun((IEN) -> {ok, nil} | {error, binary()})), list(dream_test@types:assertion_failure())}. run_before_all_chain( Config, Scope, Context, Hooks, Inherited_before_each, Inherited_after_each, Failures_rev ) -> case Hooks of [] -> {Context, lists:append(Inherited_before_each, []), lists:append([], Inherited_after_each), Failures_rev}; [Hook | Rest] -> case run_hook_transform(Config, Scope, Context, Hook) of {ok, Next} -> run_before_all_chain( Config, Scope, Next, Rest, Inherited_before_each, Inherited_after_each, Failures_rev ); {error, Message} -> {Context, Inherited_before_each, Inherited_after_each, [hook_failure(<<"before_all"/utf8>>, Message) | Failures_rev]} end end. -file("src/dream_test/parallel.gleam", 1148). -spec handle_worker_message( list(binary()), worker_message(), list({integer(), dream_test@types:node_(IIA)}), list(running_test()), list(indexed_result()), gleam@option:option(dream_test@reporters@progress:progress_reporter()), fun((binary()) -> nil), integer(), integer() ) -> {list({integer(), dream_test@types:node_(IIA)}), list(running_test()), list(indexed_result()), integer()}. handle_worker_message( Scope, Message, Pending, Running, Pending_emit, Progress_reporter, Write, Total, Completed ) -> case Message of {worker_completed, Index, Result} -> Next_completed = emit_test_finished_progress( Progress_reporter, Write, Completed, Total, Result ), {Pending, remove_running_by_index(Running, Index), [{indexed_result, Index, Result} | Pending_emit], Next_completed}; {worker_crashed, Index@1, Reason} -> Failure = hook_failure( <<"crash"/utf8>>, <<<<<<"worker crashed in "/utf8, (gleam@string:join(Scope, <<" > "/utf8>>))/binary>>/binary, ": "/utf8>>/binary, Reason/binary>> ), {Name, Full_name, Tags, Kind} = case get_running_by_index( Running, Index@1 ) of {some, R} -> {erlang:element(5, R), erlang:element(6, R), erlang:element(7, R), erlang:element(8, R)}; none -> {<<""/utf8>>, lists:append(Scope, [<<""/utf8>>]), [], unit} end, Result@1 = {test_result, Name, Full_name, failed, 0, Tags, [Failure], Kind}, Next_completed@1 = emit_test_finished_progress( Progress_reporter, Write, Completed, Total, Result@1 ), {Pending, remove_running_by_index(Running, Index@1), [{indexed_result, Index@1, Result@1} | Pending_emit], Next_completed@1} end. -file("src/dream_test/parallel.gleam", 1260). -spec timeout_result(running_test()) -> indexed_result(). timeout_result(Running) -> Failure = hook_failure(<<"timeout"/utf8>>, <<"test timed out"/utf8>>), {running_test, Index, _, _, Name, Full_name, Tags, Kind} = Running, Result = {test_result, Name, Full_name, timed_out, 0, Tags, [Failure], Kind}, {indexed_result, Index, Result}. -file("src/dream_test/parallel.gleam", 1230). -spec handle_timeouts( parallel_config(), list({integer(), dream_test@types:node_(IIM)}), list(running_test()), list(indexed_result()), gleam@option:option(dream_test@reporters@progress:progress_reporter()), fun((binary()) -> nil), integer(), integer() ) -> {list({integer(), dream_test@types:node_(IIM)}), list(running_test()), list(indexed_result()), integer()}. handle_timeouts( _, Pending, Running, Pending_emit, Progress_reporter, Write, Total, Completed ) -> Now = dream_test@timing:now_ms(), {Timed_out, Still_running} = partition_timeouts(Running, Now, [], []), gleam@list:each( Timed_out, fun(R) -> gleam@erlang@process:kill(erlang:element(3, R)) end ), Timeout_results = gleam@list:map( Timed_out, fun(R@1) -> timeout_result(R@1) end ), Next_completed = emit_test_finished_progress_list( Progress_reporter, Write, Completed, Total, Timeout_results ), {Pending, Still_running, lists:append(Timeout_results, Pending_emit), Next_completed}. -file("src/dream_test/parallel.gleam", 1101). -spec wait_for_event_or_timeout(wait_for_event_or_timeout_config(IHT)) -> {list({integer(), dream_test@types:node_(IHT)}), list(running_test()), list(indexed_result()), integer()}. wait_for_event_or_timeout(Input) -> {wait_for_event_or_timeout_config, Config, Subject, Scope, Pending, Running, Pending_emit, Progress_reporter, Write, Total, Completed} = Input, Selector = begin _pipe = gleam_erlang_ffi:new_selector(), gleam@erlang@process:select(_pipe, Subject) end, Now = dream_test@timing:now_ms(), Next_timeout = next_deadline_timeout(Running, Now, 1000), case gleam_erlang_ffi:select(Selector, Next_timeout) of {ok, Message} -> handle_worker_message( Scope, Message, Pending, Running, Pending_emit, Progress_reporter, Write, Total, Completed ); {error, nil} -> handle_timeouts( Config, Pending, Running, Pending_emit, Progress_reporter, Write, Total, Completed ) end. -file("src/dream_test/parallel.gleam", 1676). -spec run_after_all_chain( parallel_config(), list(binary()), ILD, list(fun((ILD) -> {ok, nil} | {error, binary()})), list(dream_test@types:test_result()) ) -> list(dream_test@types:test_result()). run_after_all_chain(Config, Scope, Context, Hooks, Acc_rev) -> case Hooks of [] -> Acc_rev; [Hook | Rest] -> case run_hook_teardown(Config, Scope, Context, Hook) of {ok, _} -> run_after_all_chain(Config, Scope, Context, Rest, Acc_rev); {error, Message} -> Result = {test_result, <<""/utf8>>, lists:append(Scope, [<<""/utf8>>]), failed, 0, [], [hook_failure(<<"after_all"/utf8>>, Message)], unit}, [Result | Acc_rev] end end. -file("src/dream_test/parallel.gleam", 1755). -spec run_before_each_list( parallel_config(), list(binary()), ILW, list(fun((ILW) -> {ok, ILW} | {error, binary()})), list(dream_test@types:assertion_failure()) ) -> {ILW, dream_test@types:status(), list(dream_test@types:assertion_failure())}. run_before_each_list(Config, Scope, Context, Hooks, Failures_rev) -> case Hooks of [] -> {Context, passed, Failures_rev}; [Hook | Rest] -> case run_hook_transform(Config, Scope, Context, Hook) of {ok, Next} -> run_before_each_list( Config, Scope, Next, Rest, Failures_rev ); {error, Message} -> {Context, setup_failed, [hook_failure(<<"before_each"/utf8>>, Message) | Failures_rev]} end end. -file("src/dream_test/parallel.gleam", 1776). -spec run_after_each_list( parallel_config(), list(binary()), IMD, list(fun((IMD) -> {ok, nil} | {error, binary()})), dream_test@types:status(), list(dream_test@types:assertion_failure()) ) -> {dream_test@types:status(), list(dream_test@types:assertion_failure())}. run_after_each_list(Config, Scope, Context, Hooks, Status, Failures_rev) -> case Hooks of [] -> {Status, Failures_rev}; [Hook | Rest] -> case run_hook_teardown(Config, Scope, Context, Hook) of {ok, _} -> run_after_each_list( Config, Scope, Context, Rest, Status, Failures_rev ); {error, Message} -> run_after_each_list( Config, Scope, Context, Rest, failed, [hook_failure(<<"after_each"/utf8>>, Message) | Failures_rev] ) end end. -file("src/dream_test/parallel.gleam", 1810). -spec head_failure_or_unknown(list(dream_test@types:assertion_failure())) -> dream_test@types:assertion_failure(). head_failure_or_unknown(Failures_rev) -> case Failures_rev of [F | _] -> F; [] -> hook_failure(<<"before_each"/utf8>>, <<"setup failed"/utf8>>) end. -file("src/dream_test/parallel.gleam", 1819). -spec assertion_to_status_and_failures( dream_test@types:assertion_result(), list(dream_test@types:assertion_failure()), list(dream_test@types:assertion_failure()) ) -> {dream_test@types:status(), list(dream_test@types:assertion_failure())}. assertion_to_status_and_failures( Result, Inherited_failures_rev, Setup_failures_rev ) -> case Result of assertion_ok -> {passed, lists:append(Setup_failures_rev, Inherited_failures_rev)}; {assertion_failed, Failure} -> {failed, [Failure | lists:append(Setup_failures_rev, Inherited_failures_rev)]}; assertion_skipped -> {skipped, lists:append(Setup_failures_rev, Inherited_failures_rev)} end. -file("src/dream_test/parallel.gleam", 1840). -spec run_in_sandbox( parallel_config(), gleam@option:option(integer()), fun(() -> dream_test@types:assertion_result()) ) -> dream_test@types:assertion_result(). run_in_sandbox(Config, Timeout_override, Test_function) -> {parallel_config, _, Default_timeout_ms} = Config, Timeout = case Timeout_override of {some, Ms} -> Ms; none -> Default_timeout_ms end, Sandbox_config = {sandbox_config, Timeout, false}, case dream_test@sandbox:run_isolated(Sandbox_config, Test_function) of {sandbox_completed, Assertion} -> Assertion; sandbox_timed_out -> {assertion_failed, hook_failure(<<"timeout"/utf8>>, <<"test timed out"/utf8>>)}; {sandbox_crashed, Reason} -> {assertion_failed, hook_failure(<<"crash"/utf8>>, Reason)} end. -file("src/dream_test/parallel.gleam", 1442). -spec execute_one_test_node( parallel_config(), list(binary()), list(binary()), IJQ, list(fun((IJQ) -> {ok, IJQ} | {error, binary()})), list(fun((IJQ) -> {ok, nil} | {error, binary()})), list(dream_test@types:assertion_failure()), dream_test@types:node_(IJQ) ) -> dream_test@types:test_result(). execute_one_test_node( Config, Scope, Inherited_tags, Context, Before_each_hooks, After_each_hooks, Failures_rev, Node ) -> case Node of {test, Name, Tags, Kind, Run, Timeout_ms} -> Full_name = lists:append(Scope, [Name]), All_tags = lists:append(Inherited_tags, Tags), Start = dream_test@timing:now_ms(), {Ctx_after_setup, Setup_status, Setup_failures} = run_before_each_list( Config, Scope, Context, Before_each_hooks, [] ), Assertion = case Setup_status of setup_failed -> {assertion_failed, head_failure_or_unknown(Setup_failures)}; _ -> run_in_sandbox( Config, Timeout_ms, fun() -> case Run(Ctx_after_setup) of {ok, A} -> A; {error, Message} -> {assertion_failed, hook_failure(<<"error"/utf8>>, Message)} end end ) end, {Status, Failures} = assertion_to_status_and_failures( Assertion, Failures_rev, Setup_failures ), {Final_status, Final_failures} = run_after_each_list( Config, Scope, Ctx_after_setup, After_each_hooks, Status, Failures ), Duration = dream_test@timing:now_ms() - Start, {test_result, Name, Full_name, Final_status, Duration, All_tags, lists:reverse(Final_failures), Kind}; _ -> {test_result, <<""/utf8>>, lists:append(Scope, [<<""/utf8>>]), failed, 0, [], [hook_failure(<<"internal"/utf8>>, <<"non-test node"/utf8>>)], unit} end. -file("src/dream_test/parallel.gleam", 1052). -spec spawn_test_worker( parallel_config(), gleam@erlang@process:subject(worker_message()), list(binary()), list(binary()), IHI, list(fun((IHI) -> {ok, IHI} | {error, binary()})), list(fun((IHI) -> {ok, nil} | {error, binary()})), list(dream_test@types:assertion_failure()), integer(), dream_test@types:node_(IHI) ) -> {gleam@erlang@process:pid_(), integer()}. spawn_test_worker( Config, Subject, Scope, Inherited_tags, Context, Before_each_hooks, After_each_hooks, Failures_rev, Index, Node ) -> Start = dream_test@timing:now_ms(), Timeout_ms = test_timeout_ms(Config, Node), Pid = proc_lib:spawn( fun() -> case sandbox_ffi:run_catching( fun() -> execute_one_test_node( Config, Scope, Inherited_tags, Context, Before_each_hooks, After_each_hooks, Failures_rev, Node ) end ) of {ok, Result} -> gleam@erlang@process:send( Subject, {worker_completed, Index, Result} ); {error, Reason} -> gleam@erlang@process:send( Subject, {worker_crashed, Index, Reason} ) end end ), Deadline_ms = Start + Timeout_ms, {Pid, Deadline_ms}. -file("src/dream_test/parallel.gleam", 978). -spec start_workers_up_to_limit(start_workers_up_to_limit_config(IGU)) -> {list({integer(), dream_test@types:node_(IGU)}), list(running_test())}. start_workers_up_to_limit(Input) -> {start_workers_up_to_limit_config, Config, Subject, Scope, Inherited_tags, Context, Before_each_hooks, After_each_hooks, Failures_rev, Pending, Running, Max_concurrency} = Input, Slots = Max_concurrency - erlang:length(Running), case (Slots > 0) andalso not gleam@list:is_empty(Pending) of false -> {Pending, Running}; true -> case Pending of [] -> {Pending, Running}; [{Index, Test_node} | Rest] -> {Pid, Deadline_ms} = spawn_test_worker( Config, Subject, Scope, Inherited_tags, Context, Before_each_hooks, After_each_hooks, Failures_rev, Index, Test_node ), {Name, Full_name, Tags, Kind} = running_test_metadata( Scope, Inherited_tags, Test_node ), start_workers_up_to_limit( {start_workers_up_to_limit_config, erlang:element(2, Input), erlang:element(3, Input), erlang:element(4, Input), erlang:element(5, Input), erlang:element(6, Input), erlang:element(7, Input), erlang:element(8, Input), erlang:element(9, Input), Rest, [{running_test, Index, Pid, Deadline_ms, Name, Full_name, Tags, Kind} | Running], erlang:element(12, Input)} ) end end. -file("src/dream_test/parallel.gleam", 871). -spec run_parallel_loop(run_parallel_loop_config(any())) -> {list(dream_test@types:test_result()), integer()}. run_parallel_loop(Loop) -> {run_parallel_loop_config, Config, Subject, Scope, Inherited_tags, Context, Before_each_hooks, After_each_hooks, Failures_rev, Pending, Running, Pending_emit, Emitted_rev, Next_emit_index, Progress_reporter, Write, Total, Completed, Max_concurrency} = Loop, {Pending2, Running2} = start_workers_up_to_limit( {start_workers_up_to_limit_config, Config, Subject, Scope, Inherited_tags, Context, Before_each_hooks, After_each_hooks, Failures_rev, Pending, Running, Max_concurrency} ), case gleam@list:is_empty(Pending2) andalso gleam@list:is_empty(Running2) of true -> {lists:append( lists:reverse( emit_all_results(Pending_emit, Next_emit_index, []) ), Emitted_rev ), Completed}; false -> {Pending3, Running3, Pending_emit2, Completed2} = wait_for_event_or_timeout( {wait_for_event_or_timeout_config, Config, Subject, Scope, Pending2, Running2, Pending_emit, Progress_reporter, Write, Total, Completed} ), {Maybe_ready, Remaining_emit} = take_indexed_result( Next_emit_index, Pending_emit2, [] ), case Maybe_ready of none -> run_parallel_loop( {run_parallel_loop_config, erlang:element(2, Loop), erlang:element(3, Loop), erlang:element(4, Loop), erlang:element(5, Loop), erlang:element(6, Loop), erlang:element(7, Loop), erlang:element(8, Loop), erlang:element(9, Loop), Pending3, Running3, Remaining_emit, erlang:element(13, Loop), erlang:element(14, Loop), erlang:element(15, Loop), erlang:element(16, Loop), erlang:element(17, Loop), Completed2, erlang:element(19, Loop)} ); {some, {indexed_result, _, Result}} -> run_parallel_loop( {run_parallel_loop_config, erlang:element(2, Loop), erlang:element(3, Loop), erlang:element(4, Loop), erlang:element(5, Loop), erlang:element(6, Loop), erlang:element(7, Loop), erlang:element(8, Loop), erlang:element(9, Loop), Pending3, Running3, Remaining_emit, [Result | Emitted_rev], Next_emit_index + 1, erlang:element(15, Loop), erlang:element(16, Loop), erlang:element(17, Loop), Completed2, erlang:element(19, Loop)} ) end end. -file("src/dream_test/parallel.gleam", 817). -spec run_tests_parallel( parallel_config(), list(binary()), list(binary()), IFV, list(fun((IFV) -> {ok, IFV} | {error, binary()})), list(fun((IFV) -> {ok, nil} | {error, binary()})), list(dream_test@types:node_(IFV)), list(dream_test@types:assertion_failure()), gleam@option:option(dream_test@reporters@progress:progress_reporter()), fun((binary()) -> nil), integer(), integer() ) -> {list(dream_test@types:test_result()), integer()}. run_tests_parallel( Config, Scope, Inherited_tags, Context, Before_each_hooks, After_each_hooks, Tests, Failures_rev, Progress_reporter, Write, Total, Completed ) -> Subject = gleam@erlang@process:new_subject(), Indexed = index_tests(Tests, 0, []), {parallel_config, Max_concurrency, _} = Config, run_parallel_loop( {run_parallel_loop_config, Config, Subject, Scope, Inherited_tags, Context, Before_each_hooks, After_each_hooks, Failures_rev, Indexed, [], [], [], 0, Progress_reporter, Write, Total, Completed, Max_concurrency} ). -file("src/dream_test/parallel.gleam", 1515). -spec run_tests_sequentially( parallel_config(), list(binary()), list(binary()), IKB, list(fun((IKB) -> {ok, IKB} | {error, binary()})), list(fun((IKB) -> {ok, nil} | {error, binary()})), list(dream_test@types:node_(IKB)), list(dream_test@types:assertion_failure()), gleam@option:option(dream_test@reporters@progress:progress_reporter()), fun((binary()) -> nil), integer(), integer(), list(dream_test@types:test_result()) ) -> {list(dream_test@types:test_result()), integer()}. run_tests_sequentially( Config, Scope, Inherited_tags, Context, Before_each_hooks, After_each_hooks, Tests, Failures_rev, Progress_reporter, Write, Total, Completed, Acc_rev ) -> case Tests of [] -> {Acc_rev, Completed}; [{test, Name, Tags, Kind, Run, Timeout_ms} | Rest] -> Full_name = lists:append(Scope, [Name]), All_tags = lists:append(Inherited_tags, Tags), Start = dream_test@timing:now_ms(), {Ctx_after_setup, Setup_status, Setup_failures} = run_before_each_list( Config, Scope, Context, Before_each_hooks, [] ), Assertion = case Setup_status of setup_failed -> {assertion_failed, head_failure_or_unknown(Setup_failures)}; _ -> run_in_sandbox( Config, Timeout_ms, fun() -> case Run(Ctx_after_setup) of {ok, A} -> A; {error, Message} -> {assertion_failed, hook_failure(<<"error"/utf8>>, Message)} end end ) end, {Status, Failures} = assertion_to_status_and_failures( Assertion, Failures_rev, Setup_failures ), {Final_status, Final_failures} = run_after_each_list( Config, Scope, Ctx_after_setup, After_each_hooks, Status, Failures ), Duration = dream_test@timing:now_ms() - Start, Result = {test_result, Name, Full_name, Final_status, Duration, All_tags, lists:reverse(Final_failures), Kind}, Next_completed = emit_test_finished_progress( Progress_reporter, Write, Completed, Total, Result ), run_tests_sequentially( Config, Scope, Inherited_tags, Context, Before_each_hooks, After_each_hooks, Rest, Failures_rev, Progress_reporter, Write, Total, Next_completed, [Result | Acc_rev] ); [_ | Rest@1] -> run_tests_sequentially( Config, Scope, Inherited_tags, Context, Before_each_hooks, After_each_hooks, Rest@1, Failures_rev, Progress_reporter, Write, Total, Completed, Acc_rev ) end. -file("src/dream_test/parallel.gleam", 766). -spec run_tests_in_group( parallel_config(), list(binary()), list(binary()), IFH, list(fun((IFH) -> {ok, IFH} | {error, binary()})), list(fun((IFH) -> {ok, nil} | {error, binary()})), list(dream_test@types:node_(IFH)), list(dream_test@types:assertion_failure()), gleam@option:option(dream_test@reporters@progress:progress_reporter()), fun((binary()) -> nil), integer(), integer() ) -> {list(dream_test@types:test_result()), integer()}. run_tests_in_group( Config, Scope, Inherited_tags, Context, Before_each_hooks, After_each_hooks, Tests, Failures_rev, Progress_reporter, Write, Total, Completed ) -> {parallel_config, Max_concurrency, _} = Config, case Max_concurrency =< 1 of true -> run_tests_sequentially( Config, Scope, Inherited_tags, Context, Before_each_hooks, After_each_hooks, Tests, Failures_rev, Progress_reporter, Write, Total, Completed, [] ); false -> run_tests_parallel( Config, Scope, Inherited_tags, Context, Before_each_hooks, After_each_hooks, Tests, Failures_rev, Progress_reporter, Write, Total, Completed ) end. -file("src/dream_test/parallel.gleam", 1626). -spec run_child_groups_sequentially( parallel_config(), list(binary()), list(binary()), IKQ, list(fun((IKQ) -> {ok, IKQ} | {error, binary()})), list(fun((IKQ) -> {ok, nil} | {error, binary()})), list(dream_test@types:node_(IKQ)), gleam@option:option(dream_test@reporters@progress:progress_reporter()), fun((binary()) -> nil), integer(), integer(), list(dream_test@types:test_result()) ) -> {list(dream_test@types:test_result()), integer()}. run_child_groups_sequentially( Config, Scope, Inherited_tags, Context, Before_each_hooks, After_each_hooks, Groups, Progress_reporter, Write, Total, Completed, Acc_rev ) -> case Groups of [] -> {Acc_rev, Completed}; [Group_node | Rest] -> {Next_rev, Next_completed} = execute_node( {execute_node_config, Config, Scope, Inherited_tags, Context, Before_each_hooks, After_each_hooks, Group_node, Progress_reporter, Write, Total, Completed, Acc_rev} ), run_child_groups_sequentially( Config, Scope, Inherited_tags, Context, Before_each_hooks, After_each_hooks, Rest, Progress_reporter, Write, Total, Next_completed, Next_rev ) end. -file("src/dream_test/parallel.gleam", 434). -spec execute_node(execute_node_config(any())) -> {list(dream_test@types:test_result()), integer()}. execute_node(Executor_input) -> {execute_node_config, Config, Scope, Inherited_tags, Context, Inherited_before_each, Inherited_after_each, Node, Progress_reporter, Write, Total, Completed, Base_results_rev} = Executor_input, case Node of {group, Name, Tags, Children} -> Group_scope = lists:append(Scope, [Name]), Combined_tags = lists:append(Inherited_tags, Tags), Empty_hooks = {group_hooks, [], [], [], []}, {Hooks, Tests, Groups} = collect_children( Children, Empty_hooks, [], [] ), {Ctx2, _, _, Failures_rev} = run_before_all_chain( Config, Group_scope, Context, erlang:element(2, Hooks), Inherited_before_each, Inherited_after_each, [] ), case gleam@list:is_empty(Failures_rev) of false -> Group_results_rev = fail_tests_due_to_before_all( Group_scope, Combined_tags, Tests, Groups, Failures_rev, [] ), Completed_after_failures = emit_test_finished_progress_results( Progress_reporter, Write, Completed, Total, lists:reverse(Group_results_rev) ), Final_rev = run_after_all_chain( Config, Group_scope, Ctx2, erlang:element(5, Hooks), Group_results_rev ), {lists:append(Final_rev, Base_results_rev), Completed_after_failures}; true -> Combined_before_each = lists:append( Inherited_before_each, erlang:element(3, Hooks) ), Combined_after_each = lists:append( erlang:element(4, Hooks), Inherited_after_each ), {Group_results_rev@1, Completed_after_tests} = run_tests_in_group( Config, Group_scope, Combined_tags, Ctx2, Combined_before_each, Combined_after_each, Tests, Failures_rev, Progress_reporter, Write, Total, Completed ), {After_group_rev, Completed_after_groups} = run_child_groups_sequentially( Config, Group_scope, Combined_tags, Ctx2, Combined_before_each, Combined_after_each, Groups, Progress_reporter, Write, Total, Completed_after_tests, Group_results_rev@1 ), Final_rev@1 = run_after_all_chain( Config, Group_scope, Ctx2, erlang:element(5, Hooks), After_group_rev ), {lists:append(Final_rev@1, Base_results_rev), Completed_after_groups} end; _ -> {Base_results_rev, Completed} end. -file("src/dream_test/parallel.gleam", 300). ?DOC( " Run a single suite and return results.\n" "\n" " This does **not** drive a reporter.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_test/matchers.{have_length, or_fail_with, should, succeed}\n" " import dream_test/parallel\n" " import dream_test/unit.{describe, it}\n" "\n" " pub fn tests() {\n" " describe(\"Parallel executor\", [\n" " it(\"can run a suite and return a list of results\", fn() {\n" " let suite =\n" " describe(\"Suite\", [\n" " it(\"a\", fn() { Ok(succeed()) }),\n" " it(\"b\", fn() { Ok(succeed()) }),\n" " ])\n" "\n" " parallel.run_root_parallel(parallel.default_config(), suite)\n" " |> should\n" " |> have_length(2)\n" " |> or_fail_with(\"expected two results\")\n" " }),\n" " ])\n" " }\n" " ```\n" "\n" " ## Parameters\n" "\n" " - `config`: execution settings (concurrency + default timeout)\n" " - `suite`: the suite to execute\n" "\n" " ## Returns\n" "\n" " A list of `TestResult` values, in deterministic (declaration) order.\n" ). -spec run_root_parallel(parallel_config(), dream_test@types:root(any())) -> list(dream_test@types:test_result()). run_root_parallel(Config, Suite) -> {root, Seed, Tree} = Suite, Executor_input = {execute_node_config, Config, [], [], Seed, [], [], Tree, none, fun discard_write/1, 0, 0, []}, {Results_rev, _} = execute_node(Executor_input), lists:reverse(Results_rev). -file("src/dream_test/parallel.gleam", 393). ?DOC( " Run a single suite while driving a reporter.\n" "\n" " This is used by `dream_test/runner` internally.\n" "\n" " - `total` is the total number of tests across all suites in the run\n" " - `completed` is how many tests have already completed before this suite\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_test/matchers.{succeed}\n" " import dream_test/parallel\n" " import dream_test/types.{type TestSuite}\n" " import dream_test/unit.{describe, it}\n" " import gleam/option.{None}\n" " \n" " pub fn suite() -> TestSuite(Nil) {\n" " describe(\"suite\", [\n" " it(\"passes\", fn() { Ok(succeed()) }),\n" " ])\n" " }\n" " \n" " fn ignore(_text: String) {\n" " Nil\n" " }\n" " \n" " pub fn main() {\n" " let total = 1\n" " let completed = 0\n" " let parallel_result =\n" " parallel.run_root_parallel_with_reporter(\n" " parallel.RunRootParallelWithReporterConfig(\n" " config: parallel.default_config(),\n" " suite: suite(),\n" " progress_reporter: None,\n" " write: ignore,\n" " total: total,\n" " completed: completed,\n" " ),\n" " )\n" " let parallel.RunRootParallelWithReporterResult(\n" " results: results,\n" " completed: completed_after_suite,\n" " progress_reporter: _progress_reporter,\n" " ) = parallel_result\n" " results\n" " }\n" " ```\n" "\n" " ## Parameters\n" "\n" " - `config`: execution settings for this suite\n" " - `suite`: the suite to execute\n" " - `progress_reporter`: optional progress reporter state (typically `Some(progress.new())`)\n" " - `write`: output sink for any progress output (typically `io.print`)\n" " - `total`: total number of tests in the overall run (across all suites)\n" " - `completed`: number of tests already completed before this suite starts\n" "\n" " ## Returns\n" "\n" " A `RunRootParallelWithReporterResult` containing:\n" " - `results`: this suite’s results, in deterministic (declaration) order\n" " - `completed`: updated completed count after driving `TestFinished` events\n" " - `progress_reporter`: progress reporter state (unchanged for the built-in progress reporter)\n" ). -spec run_root_parallel_with_reporter( run_root_parallel_with_reporter_config(any()) ) -> run_root_parallel_with_reporter_result(). run_root_parallel_with_reporter(Config) -> {run_root_parallel_with_reporter_config, Executor_config, Suite, Progress_reporter, Write, Total, Completed} = Config, {root, Seed, Tree} = Suite, Executor_input = {execute_node_config, Executor_config, [], [], Seed, [], [], Tree, Progress_reporter, Write, Total, Completed, []}, {Results_rev, Completed_after_suite} = execute_node(Executor_input), {run_root_parallel_with_reporter_result, lists:reverse(Results_rev), Completed_after_suite, Progress_reporter}. -file("src/dream_test/parallel.gleam", 615). -spec fail_group_nodes( list(binary()), list(binary()), list(dream_test@types:node_(any())), list(dream_test@types:assertion_failure()), list(dream_test@types:test_result()) ) -> list(dream_test@types:test_result()). fail_group_nodes(Scope, Inherited_tags, Nodes, Failures_rev, Acc_rev) -> case Nodes of [] -> Acc_rev; [{group, Name, Tags, Children} | Rest] -> Group_scope = lists:append(Scope, [Name]), Combined_tags = lists:append(Inherited_tags, Tags), Empty_hooks = {group_hooks, [], [], [], []}, {_, Tests, Groups} = collect_children(Children, Empty_hooks, [], []), Next = fail_tests_due_to_before_all( Group_scope, Combined_tags, Tests, Groups, Failures_rev, Acc_rev ), fail_group_nodes(Scope, Inherited_tags, Rest, Failures_rev, Next); [_ | Rest@1] -> fail_group_nodes( Scope, Inherited_tags, Rest@1, Failures_rev, Acc_rev ) end. -file("src/dream_test/parallel.gleam", 569). -spec fail_tests_due_to_before_all( list(binary()), list(binary()), list(dream_test@types:node_(IDB)), list(dream_test@types:node_(IDB)), list(dream_test@types:assertion_failure()), list(dream_test@types:test_result()) ) -> list(dream_test@types:test_result()). fail_tests_due_to_before_all( Scope, Inherited_tags, Tests, Groups, Failures_rev, Acc_rev ) -> After_tests = fail_test_nodes( Scope, Inherited_tags, Tests, Failures_rev, Acc_rev ), fail_group_nodes(Scope, Inherited_tags, Groups, Failures_rev, After_tests).