Current section
Files
Jump to
Current section
Files
src/dream_test@parallel.erl
-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_suite_parallel/2, run_parallel/2]).
-export_type([parallel_config/0, execution_state/0, indexed_test/0, running_test/0, indexed_result/0, worker_result/0, test_run_result/0]).
-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.
-type parallel_config() :: {parallel_config, integer(), integer()}.
-type execution_state() :: {execution_state,
list(indexed_test()),
list(running_test()),
list(indexed_result()),
gleam@erlang@process:subject(worker_result()),
parallel_config()}.
-type indexed_test() :: {indexed_test, integer(), dream_test@types:test_case()}.
-type running_test() :: {running_test,
integer(),
dream_test@types:test_case(),
gleam@erlang@process:pid_(),
gleam@erlang@process:monitor()}.
-type indexed_result() :: {indexed_result,
integer(),
dream_test@types:test_result()}.
-type worker_result() :: {worker_completed,
integer(),
test_run_result(),
integer()} |
{worker_down, integer(), binary()}.
-type test_run_result() :: test_passed |
test_skipped |
{setup_failure, dream_test@types:assertion_failure()} |
{test_failure, dream_test@types:assertion_failure()} |
{teardown_failure, dream_test@types:assertion_failure()}.
-file("src/dream_test/parallel.gleam", 28).
?DOC(" Default parallel configuration: 4 concurrent tests, 5 second timeout.\n").
-spec default_config() -> parallel_config().
default_config() ->
{parallel_config, 4, 5000}.
-file("src/dream_test/parallel.gleam", 94).
?DOC(" Add indices to tests for ordering.\n").
-spec index_tests(
list(dream_test@types:test_case()),
integer(),
list(indexed_test())
) -> list(indexed_test()).
index_tests(Tests, Index, Accumulated) ->
case Tests of
[] ->
lists:reverse(Accumulated);
[Head | Tail] ->
Indexed = {indexed_test, Index, Head},
index_tests(Tail, Index + 1, [Indexed | Accumulated])
end.
-file("src/dream_test/parallel.gleam", 122).
?DOC(" Check if all tests are complete.\n").
-spec is_execution_complete(execution_state()) -> boolean().
is_execution_complete(State) ->
gleam@list:is_empty(erlang:element(2, State)) andalso gleam@list:is_empty(
erlang:element(3, State)
).
-file("src/dream_test/parallel.gleam", 248).
-spec run_hooks_from_list(list(fun(() -> dream_test@types:assertion_result()))) -> dream_test@types:assertion_result().
run_hooks_from_list(Remaining) ->
case Remaining of
[] ->
assertion_ok;
[Hook | Rest] ->
Result = Hook(),
case Result of
assertion_ok ->
run_hooks_from_list(Rest);
assertion_skipped ->
run_hooks_from_list(Rest);
{assertion_failed, _} ->
Result
end
end.
-file("src/dream_test/parallel.gleam", 244).
?DOC(" Run a list of hooks sequentially, stopping on first failure.\n").
-spec run_hooks(list(fun(() -> dream_test@types:assertion_result()))) -> dream_test@types:assertion_result().
run_hooks(Hooks) ->
run_hooks_from_list(Hooks).
-file("src/dream_test/parallel.gleam", 225).
-spec run_test_and_after(dream_test@types:single_test_config()) -> test_run_result().
run_test_and_after(Config) ->
Test_result = (erlang:element(6, Config))(),
After_result = run_hooks(erlang:element(9, Config)),
case {Test_result, After_result} of
{assertion_skipped, _} ->
test_skipped;
{{assertion_failed, Failure}, _} ->
{test_failure, Failure};
{assertion_ok, {assertion_failed, Failure@1}} ->
{teardown_failure, Failure@1};
{assertion_ok, assertion_ok} ->
test_passed;
{assertion_ok, assertion_skipped} ->
test_passed
end.
-file("src/dream_test/parallel.gleam", 214).
?DOC(" Run a test with its before_each and after_each hooks.\n").
-spec run_with_hooks(dream_test@types:single_test_config()) -> test_run_result().
run_with_hooks(Config) ->
Before_result = run_hooks(erlang:element(8, Config)),
case Before_result of
{assertion_failed, Failure} ->
{setup_failure, Failure};
assertion_ok ->
run_test_and_after(Config);
assertion_skipped ->
run_test_and_after(Config)
end.
-file("src/dream_test/parallel.gleam", 207).
?DOC(
" Run a test case directly in the current process.\n"
"\n"
" Executes lifecycle hooks in the correct order:\n"
" 1. Run before_each hooks (outer to inner)\n"
" 2. If all hooks pass, run the test\n"
" 3. Run after_each hooks (inner to outer), even if test failed\n"
).
-spec run_test_directly(dream_test@types:test_case()) -> test_run_result().
run_test_directly(Test_case) ->
case Test_case of
{test_case, Config} ->
run_with_hooks(Config)
end.
-file("src/dream_test/parallel.gleam", 157).
?DOC(" Spawn a worker process for a test.\n").
-spec spawn_test_worker(
gleam@erlang@process:subject(worker_result()),
indexed_test(),
parallel_config()
) -> running_test().
spawn_test_worker(Results_subject, Indexed_test, _) ->
Test_index = erlang:element(2, Indexed_test),
Test_case = erlang:element(3, Indexed_test),
Worker_pid = proc_lib:spawn(
fun() ->
Start_time = dream_test@timing:now_ms(),
Test_run_result = run_test_directly(Test_case),
Duration_ms = dream_test@timing:now_ms() - Start_time,
gleam@erlang@process:send(
Results_subject,
{worker_completed, Test_index, Test_run_result, Duration_ms}
)
end
),
Worker_monitor = gleam@erlang@process:monitor(Worker_pid),
{running_test, Test_index, Test_case, Worker_pid, Worker_monitor}.
-file("src/dream_test/parallel.gleam", 141).
?DOC(" Start the next pending test as a worker.\n").
-spec start_next_worker(execution_state()) -> execution_state().
start_next_worker(State) ->
case erlang:element(2, State) of
[] ->
State;
[Indexed_test | Remaining_pending] ->
Running_test = spawn_test_worker(
erlang:element(5, State),
Indexed_test,
erlang:element(6, State)
),
{execution_state,
Remaining_pending,
[Running_test | erlang:element(3, State)],
erlang:element(4, State),
erlang:element(5, State),
erlang:element(6, State)}
end.
-file("src/dream_test/parallel.gleam", 127).
?DOC(" Start workers up to the concurrency limit.\n").
-spec start_workers_up_to_limit(execution_state()) -> execution_state().
start_workers_up_to_limit(State) ->
Current_running = erlang:length(erlang:element(3, State)),
Slots_available = erlang:element(2, erlang:element(6, State)) - Current_running,
case (Slots_available > 0) andalso not gleam@list:is_empty(
erlang:element(2, State)
) of
false ->
State;
true ->
State_with_worker = start_next_worker(State),
start_workers_up_to_limit(State_with_worker)
end.
-file("src/dream_test/parallel.gleam", 302).
?DOC(" Format a down message reason as a string.\n").
-spec format_down_reason(gleam@erlang@process:down()) -> binary().
format_down_reason(Down) ->
case erlang:element(4, Down) of
normal ->
<<"normal"/utf8>>;
killed ->
<<"killed"/utf8>>;
{abnormal, Reason} ->
gleam@string:inspect(Reason)
end.
-file("src/dream_test/parallel.gleam", 283).
?DOC(" Add monitor handlers for all running workers.\n").
-spec add_monitor_handlers(
gleam@erlang@process:selector(worker_result()),
list(running_test())
) -> gleam@erlang@process:selector(worker_result()).
add_monitor_handlers(Selector, Running) ->
case Running of
[] ->
Selector;
[Running_test | Rest] ->
Updated_selector = gleam@erlang@process:select_specific_monitor(
Selector,
erlang:element(5, Running_test),
fun(Down) ->
{worker_down,
erlang:element(2, Running_test),
format_down_reason(Down)}
end
),
add_monitor_handlers(Updated_selector, Rest)
end.
-file("src/dream_test/parallel.gleam", 276).
?DOC(" Build a selector for worker results and monitor events.\n").
-spec build_results_selector(execution_state()) -> gleam@erlang@process:selector(worker_result()).
build_results_selector(State) ->
_pipe = gleam_erlang_ffi:new_selector(),
_pipe@1 = gleam@erlang@process:select(_pipe, erlang:element(5, State)),
add_monitor_handlers(_pipe@1, erlang:element(3, State)).
-file("src/dream_test/parallel.gleam", 335).
?DOC(" Extract the index from a worker result.\n").
-spec get_worker_result_index(worker_result()) -> integer().
get_worker_result_index(Result) ->
case Result of
{worker_completed, Index, _, _} ->
Index;
{worker_down, Index@1, _} ->
Index@1
end.
-file("src/dream_test/parallel.gleam", 365).
?DOC(" Remove a running test by index.\n").
-spec remove_running_test(list(running_test()), integer()) -> list(running_test()).
remove_running_test(Running, Index) ->
gleam@list:filter(
Running,
fun(Running_test) -> erlang:element(2, Running_test) /= Index end
).
-file("src/dream_test/parallel.gleam", 409).
-spec make_passed_result(dream_test@types:single_test_config(), integer()) -> dream_test@types:test_result().
make_passed_result(Config, Duration_ms) ->
{test_result,
erlang:element(2, Config),
erlang:element(3, Config),
passed,
Duration_ms,
erlang:element(4, Config),
[],
erlang:element(5, Config)}.
-file("src/dream_test/parallel.gleam", 421).
-spec make_skipped_result(dream_test@types:single_test_config(), integer()) -> dream_test@types:test_result().
make_skipped_result(Config, Duration_ms) ->
{test_result,
erlang:element(2, Config),
erlang:element(3, Config),
skipped,
Duration_ms,
erlang:element(4, Config),
[],
erlang:element(5, Config)}.
-file("src/dream_test/parallel.gleam", 433).
-spec make_setup_failed_result(
dream_test@types:single_test_config(),
dream_test@types:assertion_failure(),
integer()
) -> dream_test@types:test_result().
make_setup_failed_result(Config, Failure, Duration_ms) ->
{test_result,
erlang:element(2, Config),
erlang:element(3, Config),
setup_failed,
Duration_ms,
erlang:element(4, Config),
[Failure],
erlang:element(5, Config)}.
-file("src/dream_test/parallel.gleam", 449).
-spec make_failed_result(
dream_test@types:single_test_config(),
dream_test@types:assertion_failure(),
integer()
) -> dream_test@types:test_result().
make_failed_result(Config, Failure, Duration_ms) ->
{test_result,
erlang:element(2, Config),
erlang:element(3, Config),
failed,
Duration_ms,
erlang:element(4, Config),
[Failure],
erlang:element(5, Config)}.
-file("src/dream_test/parallel.gleam", 394).
?DOC(" Convert a TestRunResult to a TestResult.\n").
-spec test_run_result_to_test_result(
dream_test@types:single_test_config(),
test_run_result(),
integer()
) -> dream_test@types:test_result().
test_run_result_to_test_result(Config, Test_run_result, Duration_ms) ->
case Test_run_result of
test_passed ->
make_passed_result(Config, Duration_ms);
test_skipped ->
make_skipped_result(Config, Duration_ms);
{setup_failure, Failure} ->
make_setup_failed_result(Config, Failure, Duration_ms);
{test_failure, Failure@1} ->
make_failed_result(Config, Failure@1, Duration_ms);
{teardown_failure, Failure@2} ->
make_failed_result(Config, Failure@2, Duration_ms)
end.
-file("src/dream_test/parallel.gleam", 466).
?DOC(" Create a TestResult for a timed-out test.\n").
-spec make_timeout_result(dream_test@types:single_test_config(), integer()) -> dream_test@types:test_result().
make_timeout_result(Config, Duration_ms) ->
{test_result,
erlang:element(2, Config),
erlang:element(3, Config),
timed_out,
Duration_ms,
erlang:element(4, Config),
[],
erlang:element(5, Config)}.
-file("src/dream_test/parallel.gleam", 479).
?DOC(" Create a TestResult for a crashed test.\n").
-spec make_crashed_result(
dream_test@types:single_test_config(),
binary(),
integer()
) -> dream_test@types:test_result().
make_crashed_result(Config, Reason, Duration_ms) ->
Failure = {assertion_failure,
<<"crash"/utf8>>,
<<"Test process crashed: "/utf8, Reason/binary>>,
none},
{test_result,
erlang:element(2, Config),
erlang:element(3, Config),
failed,
Duration_ms,
erlang:element(4, Config),
[Failure],
erlang:element(5, Config)}.
-file("src/dream_test/parallel.gleam", 382).
-spec convert_worker_result_with_config(
dream_test@types:single_test_config(),
worker_result()
) -> dream_test@types:test_result().
convert_worker_result_with_config(Config, Worker_result) ->
case Worker_result of
{worker_completed, _, Test_run_result, Duration_ms} ->
test_run_result_to_test_result(Config, Test_run_result, Duration_ms);
{worker_down, _, Reason} ->
make_crashed_result(Config, Reason, 0)
end.
-file("src/dream_test/parallel.gleam", 373).
?DOC(" Convert a worker result to a TestResult.\n").
-spec convert_worker_result(running_test(), worker_result()) -> dream_test@types:test_result().
convert_worker_result(Running_test, Worker_result) ->
case erlang:element(3, Running_test) of
{test_case, Config} ->
convert_worker_result_with_config(Config, Worker_result)
end.
-file("src/dream_test/parallel.gleam", 528).
?DOC(" Kill all running worker processes.\n").
-spec kill_all_running(list(running_test())) -> nil.
kill_all_running(Running) ->
case Running of
[] ->
nil;
[Running_test | Rest] ->
gleam@erlang@process:kill(erlang:element(4, Running_test)),
kill_all_running(Rest)
end.
-file("src/dream_test/parallel.gleam", 503).
?DOC(" Handle selector timeout (should not happen in normal operation).\n").
-spec handle_selector_timeout(execution_state()) -> execution_state().
handle_selector_timeout(State) ->
kill_all_running(erlang:element(3, State)),
Timeout_results = gleam@list:map(
erlang:element(3, State),
fun(Running_test) -> case erlang:element(3, Running_test) of
{test_case, Config} ->
{indexed_result,
erlang:element(2, Running_test),
make_timeout_result(
Config,
erlang:element(3, erlang:element(6, State))
)}
end end
),
{execution_state,
erlang:element(2, State),
[],
lists:append(Timeout_results, erlang:element(4, State)),
erlang:element(5, State),
erlang:element(6, State)}.
-file("src/dream_test/parallel.gleam", 553).
-spec compare_indices_not_less(integer(), integer()) -> gleam@order:order().
compare_indices_not_less(A, B) ->
case A > B of
true ->
gt;
false ->
eq
end.
-file("src/dream_test/parallel.gleam", 546).
?DOC(" Compare two indices for sorting.\n").
-spec compare_indices(integer(), integer()) -> gleam@order:order().
compare_indices(A, B) ->
case A < B of
true ->
lt;
false ->
compare_indices_not_less(A, B)
end.
-file("src/dream_test/parallel.gleam", 539).
?DOC(" Sort completed results by their original index.\n").
-spec sort_results(list(indexed_result())) -> list(dream_test@types:test_result()).
sort_results(Completed) ->
_pipe = Completed,
_pipe@1 = gleam@list:sort(
_pipe,
fun(A, B) ->
compare_indices(erlang:element(2, A), erlang:element(2, B))
end
),
gleam@list:map(_pipe@1, fun(Indexed) -> erlang:element(3, Indexed) end).
-file("src/dream_test/parallel.gleam", 657).
-spec collect_tests_from_items(
list(dream_test@types:test_suite_item()),
list(dream_test@types:test_case())
) -> list(dream_test@types:test_case()).
collect_tests_from_items(Remaining, Accumulated) ->
case Remaining of
[] ->
lists:reverse(Accumulated);
[Item | Rest] ->
Updated = case Item of
{suite_test, Test_case} ->
[Test_case | Accumulated];
{suite_group, _} ->
Accumulated
end,
collect_tests_from_items(Rest, Updated)
end.
-file("src/dream_test/parallel.gleam", 673).
-spec collect_groups_from_items(
list(dream_test@types:test_suite_item()),
list(dream_test@types:test_suite())
) -> list(dream_test@types:test_suite()).
collect_groups_from_items(Remaining, Accumulated) ->
case Remaining of
[] ->
lists:reverse(Accumulated);
[Item | Rest] ->
Updated = case Item of
{suite_test, _} ->
Accumulated;
{suite_group, Suite} ->
[Suite | Accumulated]
end,
collect_groups_from_items(Rest, Updated)
end.
-file("src/dream_test/parallel.gleam", 689).
-spec run_groups_sequentially(
parallel_config(),
list(dream_test@types:test_suite()),
list(dream_test@types:test_result())
) -> list(dream_test@types:test_result()).
run_groups_sequentially(Config, Remaining, Accumulated) ->
case Remaining of
[] ->
lists:reverse(Accumulated);
[Suite | Rest] ->
Results = run_suite_group(Config, Suite),
Updated = lists:append(lists:reverse(Results), Accumulated),
run_groups_sequentially(Config, Rest, Updated)
end.
-file("src/dream_test/parallel.gleam", 581).
-spec run_suite_group(parallel_config(), dream_test@types:test_suite()) -> list(dream_test@types:test_result()).
run_suite_group(Config, Suite) ->
Before_all_result = run_hooks(erlang:element(4, Suite)),
case Before_all_result of
{assertion_failed, Failure} ->
mark_all_items_as_setup_failed(erlang:element(6, Suite), Failure);
assertion_ok ->
Results = run_suite_items(Config, erlang:element(6, Suite)),
_ = run_hooks(erlang:element(5, Suite)),
Results;
assertion_skipped ->
Results = run_suite_items(Config, erlang:element(6, Suite)),
_ = run_hooks(erlang:element(5, Suite)),
Results
end.
-file("src/dream_test/parallel.gleam", 639).
-spec run_suite_items(
parallel_config(),
list(dream_test@types:test_suite_item())
) -> list(dream_test@types:test_result()).
run_suite_items(Config, Items) ->
Tests = collect_tests_from_items(Items, []),
Groups = collect_groups_from_items(Items, []),
Test_results = run_parallel(Config, Tests),
Group_results = run_groups_sequentially(Config, Groups, []),
lists:append(Test_results, Group_results).
-file("src/dream_test/parallel.gleam", 574).
?DOC(
" Run a test suite with before_all/after_all semantics.\n"
"\n"
" Execution flow for each group:\n"
" 1. Run before_all hooks sequentially\n"
" 2. If any fail, mark all tests in group as SetupFailed\n"
" 3. Run tests in parallel (with their before_each/after_each)\n"
" 4. Wait for all tests to complete\n"
" 5. Run after_all hooks sequentially\n"
" 6. Recurse for nested groups\n"
).
-spec run_suite_parallel(parallel_config(), dream_test@types:test_suite()) -> list(dream_test@types:test_result()).
run_suite_parallel(Config, Suite) ->
run_suite_group(Config, Suite).
-file("src/dream_test/parallel.gleam", 624).
-spec mark_item_as_setup_failed(
dream_test@types:test_suite_item(),
dream_test@types:assertion_failure()
) -> list(dream_test@types:test_result()).
mark_item_as_setup_failed(Item, Failure) ->
case Item of
{suite_test, Test_case} ->
{test_case, Config} = Test_case,
[make_setup_failed_result(Config, Failure, 0)];
{suite_group, Nested_suite} ->
mark_all_items_as_setup_failed(
erlang:element(6, Nested_suite),
Failure
)
end.
-file("src/dream_test/parallel.gleam", 602).
-spec mark_all_items_as_setup_failed(
list(dream_test@types:test_suite_item()),
dream_test@types:assertion_failure()
) -> list(dream_test@types:test_result()).
mark_all_items_as_setup_failed(Items, Failure) ->
mark_items_failed_from_list(Items, Failure, []).
-file("src/dream_test/parallel.gleam", 609).
-spec mark_items_failed_from_list(
list(dream_test@types:test_suite_item()),
dream_test@types:assertion_failure(),
list(dream_test@types:test_result())
) -> list(dream_test@types:test_result()).
mark_items_failed_from_list(Remaining, Failure, Accumulated) ->
case Remaining of
[] ->
lists:reverse(Accumulated);
[Item | Rest] ->
Results = mark_item_as_setup_failed(Item, Failure),
Updated = lists:append(lists:reverse(Results), Accumulated),
mark_items_failed_from_list(Rest, Failure, Updated)
end.
-file("src/dream_test/parallel.gleam", 353).
-spec find_running_test_check(running_test(), list(running_test()), integer()) -> gleam@option:option(running_test()).
find_running_test_check(Running_test, Rest, Index) ->
case erlang:element(2, Running_test) =:= Index of
true ->
{some, Running_test};
false ->
find_running_test(Rest, Index)
end.
-file("src/dream_test/parallel.gleam", 343).
?DOC(" Find a running test by index.\n").
-spec find_running_test(list(running_test()), integer()) -> gleam@option:option(running_test()).
find_running_test(Running, Index) ->
case Running of
[] ->
none;
[Running_test | Rest] ->
find_running_test_check(Running_test, Rest, Index)
end.
-file("src/dream_test/parallel.gleam", 311).
?DOC(" Handle a result from a worker.\n").
-spec handle_worker_result(execution_state(), worker_result()) -> execution_state().
handle_worker_result(State, Worker_result) ->
Index = get_worker_result_index(Worker_result),
Running_test = find_running_test(erlang:element(3, State), Index),
case Running_test of
none ->
State;
{some, Running} ->
Test_result = convert_worker_result(Running, Worker_result),
Indexed_result = {indexed_result, Index, Test_result},
{execution_state,
erlang:element(2, State),
remove_running_test(erlang:element(3, State), Index),
[Indexed_result | erlang:element(4, State)],
erlang:element(5, State),
erlang:element(6, State)}
end.
-file("src/dream_test/parallel.gleam", 265).
?DOC(" Wait for a result from any running worker.\n").
-spec wait_for_result(execution_state()) -> execution_state().
wait_for_result(State) ->
Selector = build_results_selector(State),
case gleam_erlang_ffi:select(Selector, 60000) of
{ok, Worker_result} ->
handle_worker_result(State, Worker_result);
{error, nil} ->
handle_selector_timeout(State)
end.
-file("src/dream_test/parallel.gleam", 109).
?DOC(" Main execution loop.\n").
-spec execute_loop(execution_state()) -> execution_state().
execute_loop(State) ->
State_with_workers = start_workers_up_to_limit(State),
case is_execution_complete(State_with_workers) of
true ->
State_with_workers;
false ->
Updated_state = wait_for_result(State_with_workers),
execute_loop(Updated_state)
end.
-file("src/dream_test/parallel.gleam", 73).
?DOC(
" Run tests in parallel with the given configuration.\n"
"\n"
" Tests are executed concurrently up to max_concurrency.\n"
" Results are returned in the same order as the input tests.\n"
).
-spec run_parallel(parallel_config(), list(dream_test@types:test_case())) -> list(dream_test@types:test_result()).
run_parallel(Config, Test_cases) ->
Indexed_tests = index_tests(Test_cases, 0, []),
Results_subject = gleam@erlang@process:new_subject(),
Initial_state = {execution_state,
Indexed_tests,
[],
[],
Results_subject,
Config},
Final_state = execute_loop(Initial_state),
sort_results(erlang:element(4, Final_state)).