From 205bfcc6c82fcef28ca457121a7f4334d74a1725 Mon Sep 17 00:00:00 2001 From: Dan Klishch Date: Sat, 11 May 2024 11:23:03 -0400 Subject: [PATCH] LibTest: Add Test::AsyncMemory{Input,Output}Stream They are useful for unit testing other asynchronous streams. They have to be tested themselves in the first place though. --- Tests/LibTest/CMakeLists.txt | 1 + Tests/LibTest/TestAsyncTestStreams.cpp | 214 +++++++++++++++++ .../Libraries/LibTest/AsyncTestStreams.cpp | 223 ++++++++++++++++++ Userland/Libraries/LibTest/AsyncTestStreams.h | 79 +++++++ Userland/Libraries/LibTest/CMakeLists.txt | 2 + 5 files changed, 519 insertions(+) create mode 100644 Tests/LibTest/TestAsyncTestStreams.cpp create mode 100644 Userland/Libraries/LibTest/AsyncTestStreams.cpp create mode 100644 Userland/Libraries/LibTest/AsyncTestStreams.h diff --git a/Tests/LibTest/CMakeLists.txt b/Tests/LibTest/CMakeLists.txt index 84610cb211f..2e5ff264734 100644 --- a/Tests/LibTest/CMakeLists.txt +++ b/Tests/LibTest/CMakeLists.txt @@ -1,4 +1,5 @@ set(TEST_SOURCES + TestAsyncTestStreams.cpp TestNoCrash.cpp TestGenerator.cpp ) diff --git a/Tests/LibTest/TestAsyncTestStreams.cpp b/Tests/LibTest/TestAsyncTestStreams.cpp new file mode 100644 index 00000000000..c7bcbbd10be --- /dev/null +++ b/Tests/LibTest/TestAsyncTestStreams.cpp @@ -0,0 +1,214 @@ +/* + * Copyright (c) 2024, Dan Klishch + * + * SPDX-License-Identifier: BSD-2-Clause + */ + +#include +#include + +ASYNC_TEST_CASE(input_basic) +{ + Test::AsyncMemoryInputStream stream("hello"sv, Test::StreamCloseExpectation::Close, { 1, 2, 2 }); + + auto first_letter_coro = stream.read(1); + EXPECT(first_letter_coro.await_ready()); + + auto first_letter_view = CO_TRY_OR_FAIL(co_await first_letter_coro); + EXPECT_EQ(StringView { first_letter_view }, "h"sv); + + auto next_letters_coro = stream.read(3); + EXPECT(!next_letters_coro.await_ready()); + + auto next_letters_view = CO_TRY_OR_FAIL(co_await next_letters_coro); + EXPECT_EQ(StringView { next_letters_view }, "ell"sv); + + auto last_letter_coro = stream.peek_or_eof(); + EXPECT(last_letter_coro.await_ready()); + + auto [last_letter_view, is_eof_1] = CO_TRY_OR_FAIL(co_await last_letter_coro); + EXPECT_EQ(StringView { last_letter_view }, "o"sv); + EXPECT(!is_eof_1); + + auto o_again = CO_TRY_OR_FAIL(co_await stream.read(1)); + EXPECT_EQ(StringView { o_again }, "o"sv); + + auto [empty_view, is_eof_2] = CO_TRY_OR_FAIL(co_await stream.peek_or_eof()); + EXPECT(empty_view.is_empty()); + EXPECT(is_eof_2); + + CO_TRY_OR_FAIL(co_await stream.close()); +} + +ASYNC_TEST_CASE(input_eof_read) +{ + Test::AsyncMemoryInputStream stream("hello"sv, Test::StreamCloseExpectation::Close, { 5 }); + + auto [hello, is_eof_1] = CO_TRY_OR_FAIL(co_await stream.peek_or_eof()); + EXPECT_EQ(StringView { hello }, "hello"sv); + EXPECT(!is_eof_1); + + auto [also_hello, is_eof_2] = CO_TRY_OR_FAIL(co_await stream.peek_or_eof()); + EXPECT_EQ(StringView { also_hello }, "hello"sv); + EXPECT(is_eof_2); + + auto [hello_one_more_time, is_eof_3] = CO_TRY_OR_FAIL(co_await stream.peek_or_eof()); + EXPECT_EQ(StringView { hello_one_more_time }, "hello"sv); + EXPECT(is_eof_3); + + auto hello_for_the_final_time = CO_TRY_OR_FAIL(co_await stream.read(5)); + EXPECT_EQ(StringView { hello_for_the_final_time }, "hello"sv); + + CO_TRY_OR_FAIL(co_await stream.close()); +} + +ASYNC_TEST_CASE(input_eof_read_2) +{ + Test::AsyncMemoryInputStream stream("hello"sv, Test::StreamCloseExpectation::Close, { 0, 5 }); + + auto [hello, is_eof_1] = CO_TRY_OR_FAIL(co_await stream.peek_or_eof()); + EXPECT_EQ(StringView { hello }, "hello"sv); + EXPECT(!is_eof_1); + + auto hello_again = must_sync(stream.read(5)); + EXPECT_EQ(StringView { hello_again }, "hello"sv); + + auto [not_hello, is_eof_2] = CO_TRY_OR_FAIL(co_await stream.peek_or_eof()); + EXPECT(not_hello.is_empty()); + EXPECT(is_eof_2); + + must_sync(stream.read(0)); + + for (int i = 0; i < 2; ++i) { + auto [not_hello_2, is_eof_3] = CO_TRY_OR_FAIL(co_await stream.peek_or_eof()); + EXPECT(not_hello_2.is_empty()); + EXPECT(is_eof_3); + } + + CO_TRY_OR_FAIL(co_await stream.close()); +} + +// FIXME: Maybe add some kind of intentionally failing tests? +ASYNC_TEST_CASE(input_unexpected_operations) +{ + { + Test::AsyncMemoryInputStream stream("hello"sv, Test::StreamCloseExpectation::Close, { 5 }); + } + VERIFY(Test::current_test_result() == Test::TestResult::Failed); + Test::set_current_test_result(Test::TestResult::NotRun); + + { + Test::AsyncMemoryInputStream stream("hello"sv, Test::StreamCloseExpectation::Reset, { 5 }); + } + VERIFY(Test::current_test_result() == Test::TestResult::NotRun); + + { + Test::AsyncMemoryInputStream stream("hello"sv, Test::StreamCloseExpectation::Close, { 1, 1, 1, 1, 1 }); + auto view = CO_TRY_OR_FAIL(co_await stream.read(5)); + EXPECT_EQ(StringView { view }, "hello"sv); + CO_TRY_OR_FAIL(co_await stream.close()); + } + VERIFY(Test::current_test_result() == Test::TestResult::NotRun); + + { + Test::AsyncMemoryInputStream stream("hello"sv, Test::StreamCloseExpectation::Close, { 1, 1, 1, 1, 1 }); + (void)co_await stream.close(); + } + VERIFY(Test::current_test_result() == Test::TestResult::Failed); + Test::set_current_test_result(Test::TestResult::NotRun); + + { + Test::AsyncMemoryInputStream stream("hello"sv, Test::StreamCloseExpectation::Reset, { 1, 1, 1, 1, 1 }); + stream.reset(); + } + VERIFY(Test::current_test_result() == Test::TestResult::NotRun); +} + +ASYNC_TEST_CASE(input_reset_during_wait) +{ + Test::AsyncMemoryInputStream stream("hello"sv, Test::StreamCloseExpectation::Reset, { 0, 5 }); + + auto read_coro = stream.read(5); + EXPECT(!read_coro.await_ready()); + + stream.reset(); + + auto error = co_await read_coro; + EXPECT_EQ(error.error().code(), ECANCELED); +} + +TEST_CASE(input_crash) +{ + EXPECT_CRASH("input_concurrent_reads", [] -> Test::Crash::Failure { + Test::AsyncMemoryInputStream stream("hello"sv, Test::StreamCloseExpectation::Reset, { 0, 5 }); + + auto read_coro1 = stream.read(2); + if (read_coro1.await_ready()) { + // NOTE: Intentionally don't run destructors. + _exit(0); + } + + auto read_coro2 = stream.read(3); + _exit(0); + }); + + EXPECT_CRASH("input_peek_read_condition_violation", [] { + auto coro = [] -> Coroutine { + Test::AsyncMemoryInputStream stream("hello"sv, Test::StreamCloseExpectation::Reset, { 2, 1, 1, 1 }); + + auto peek_view_1 = CO_TRY_OR_FAIL(co_await stream.peek()); + if (peek_view_1.size() != 2) + co_return; + + auto peek_view_2 = CO_TRY_OR_FAIL(co_await stream.peek()); + if (peek_view_2.size() != 3) + co_return; + + auto peek_view_3 = CO_TRY_OR_FAIL(co_await stream.peek()); + if (peek_view_3.size() != 4) + co_return; + + (void)co_await stream.read(2); + }; + Core::run_async_in_new_event_loop(coro); + return Test::Crash::Failure::DidNotCrash; + }); + + EXPECT_CRASH("one_less_than_eof", ([] { + auto coro = [] -> Coroutine { + Test::AsyncMemoryInputStream stream("hello"sv, Test::StreamCloseExpectation::Reset, { 5 }); + + CO_TRY_OR_FAIL(co_await stream.peek_or_eof()); + CO_TRY_OR_FAIL(co_await stream.peek_or_eof()); + CO_TRY_OR_FAIL(co_await stream.read(4)); + _exit(0); + }; + Core::run_async_in_new_event_loop(coro); + return Test::Crash::Failure::DidNotCrash; + })); +} + +ASYNC_TEST_CASE(input_close_ebusy) +{ + Test::AsyncMemoryInputStream stream("hello"sv, Test::StreamCloseExpectation::Reset, { 5 }); + auto error = co_await stream.close(); + EXPECT_EQ(error.error().code(), EBUSY); +} + +ASYNC_TEST_CASE(output_basic) +{ + Test::AsyncMemoryOutputStream stream(Test::StreamCloseExpectation::Close); + + CO_TRY_OR_FAIL(co_await stream.write({ { + "Consider a non-trivial loop $\\alpha$ in $\\R P^2$ "sv.bytes(), + "and $f \\circ \\alpha$. $[f \\circ \\alpha]$ maps to some integer "sv.bytes(), + } })); + + size_t nwritten = CO_TRY_OR_FAIL(co_await stream.write_some("$n$ from a fundamental group $\\pi_1(S^1)$."sv.bytes())); + EXPECT_EQ(nwritten, 42uz); + + auto sentence = "Consider a non-trivial loop $\\alpha$ in $\\R P^2$ and $f \\circ \\alpha$. $[f \\circ \\alpha]$ maps to some integer $n$ from a fundamental group $\\pi_1(S^1)$."sv; + EXPECT_EQ(sentence, StringView { stream.view() }); + + CO_TRY_OR_FAIL(co_await stream.close()); +} diff --git a/Userland/Libraries/LibTest/AsyncTestStreams.cpp b/Userland/Libraries/LibTest/AsyncTestStreams.cpp new file mode 100644 index 00000000000..b34f847b7f0 --- /dev/null +++ b/Userland/Libraries/LibTest/AsyncTestStreams.cpp @@ -0,0 +1,223 @@ +/* + * Copyright (c) 2024, Dan Klishch + * + * SPDX-License-Identifier: BSD-2-Clause + */ + +#include +#include + +namespace Test { + +static auto s_deferred_context = Core::DeferredInvocationContext::construct(); + +struct Spinner { + Spinner(std::coroutine_handle<>& awaiter) + : m_awaiter(awaiter) + { + VERIFY(!m_awaiter); + } + + bool await_ready() const { return false; } + + void await_suspend(std::coroutine_handle<> awaiter) + { + m_awaiter = awaiter; + Core::ThreadEventQueue::current().post_event( + s_deferred_context, + make(s_deferred_context, [&] { + m_awaiter.resume(); + })); + } + + void await_resume() { m_awaiter = {}; } + + std::coroutine_handle<>& m_awaiter; +}; + +AsyncMemoryInputStream::AsyncMemoryInputStream(StringView data, StreamCloseExpectation expectation, Vector&& chunks) + : m_data(data) + , m_expectation(expectation) + , m_chunks(move(chunks)) + , m_peek_head(m_chunks[0]) +{ + size_t accumulator = 0; + for (auto& value : m_chunks) { + accumulator += value; + value = accumulator; + } + VERIFY(accumulator == m_data.length()); +} + +AsyncMemoryInputStream::~AsyncMemoryInputStream() +{ + // 1. Assert that nobody is awaiting on the resource. + VERIFY(!m_awaiter); + + // 2. If resource is open, perform Reset AO. + if (is_open()) + reset(); + + if (m_expectation == StreamCloseExpectation::Reset) { + EXPECT(m_is_reset); + } else if (m_expectation == StreamCloseExpectation::Close) { + EXPECT(m_is_closed); + } else { + VERIFY_NOT_REACHED(); + } +} + +void AsyncMemoryInputStream::reset() +{ + // 1. Assert that the resource is open. + VERIFY(is_open()); + + // 2. Perform Reset AO. + // 1. Schedule returning an error (preferably, ECANCELED) from the current resource awaiters. + // 2. Ensure that further attempts to wait on a resource will assert. + m_is_reset = true; + + // 3. Free synchronously the associated low-level resource. + + // 4. Return synchronously. +} + +Coroutine> AsyncMemoryInputStream::close() +{ + // 1. Assert that the object is fully constructed. + // 2. Assert that the resource is open. + VERIFY(is_open()); + + // 3. Perform Close AO, await and return its result. + // 1. Assert that nobody is awaiting on a resource. + VERIFY(!m_awaiter); + + // 3. Shutdown (possibly asynchronously) the associated low-level resource. + + // 4. Check if the state of the resource is clean. If it is not, call Reset AO and return an + // error (preferably, EBUSY). + if (m_read_head != m_data.length()) { + reset(); + co_return Error::from_errno(EBUSY); + } + + // 2. Ensure that further attempts to wait on a resource will assert. + m_is_closed = true; + + // 5. Free (possibly asynchronously) the associated low-level resource. + // 6. Return success. + co_return {}; +} + +bool AsyncMemoryInputStream::is_open() const +{ + return !m_is_closed && !m_is_reset; +} + +Coroutine> AsyncMemoryInputStream::enqueue_some(Badge) +{ + if (m_next_chunk_index == m_chunks.size()) { + m_last_enqueue = m_peek_head; + co_return false; + } + + co_await Spinner { m_awaiter }; + if (m_is_reset) + co_return Error::from_errno(ECANCELED); + + m_last_enqueue = m_peek_head; + m_peek_head = m_chunks[m_next_chunk_index++]; + co_return true; +} + +ReadonlyBytes AsyncMemoryInputStream::buffered_data_unchecked(Badge) const +{ + return m_data.bytes().slice(m_read_head, m_peek_head - m_read_head); +} + +void AsyncMemoryInputStream::dequeue(Badge, size_t bytes) +{ + m_read_head += bytes; + VERIFY(m_last_enqueue <= m_read_head && m_read_head <= m_peek_head); +} + +AsyncMemoryOutputStream::AsyncMemoryOutputStream(StreamCloseExpectation expectation) + : m_expectation(expectation) +{ +} + +AsyncMemoryOutputStream::~AsyncMemoryOutputStream() +{ + if (is_open()) + reset(); + + if (m_expectation == StreamCloseExpectation::Reset) { + EXPECT(m_is_reset); + } else if (m_expectation == StreamCloseExpectation::Close) { + EXPECT(m_is_closed); + } else { + VERIFY_NOT_REACHED(); + } +} + +void AsyncMemoryOutputStream::reset() +{ + VERIFY(is_open()); + m_is_reset = true; +} + +Coroutine> AsyncMemoryOutputStream::close() +{ + VERIFY(is_open()); + m_is_closed = true; + co_return {}; +} + +bool AsyncMemoryOutputStream::is_open() const +{ + return !m_is_closed && !m_is_reset; +} + +Coroutine> AsyncMemoryOutputStream::write_some(ReadonlyBytes data) +{ + VERIFY(is_open()); + m_buffer.append(data); + co_return data.size(); +} + +Coroutine> read_until_eof(AsyncInputStream& stream) +{ + size_t previously_returned_size = 0; + while (true) { + auto [data, is_eof] = CO_TRY(co_await stream.peek_or_eof()); + + EXPECT(is_eof || previously_returned_size < data.size()); + previously_returned_size = data.size(); + + if (is_eof) { + auto result = must_sync(stream.read(data.size())); + + // Poke stream one more time just to be sure :^) + auto [empty_data, set_eof_flag] = CO_TRY(co_await stream.peek_or_eof()); + EXPECT(empty_data.is_empty()); + EXPECT(set_eof_flag); + + co_return result; + } + } +} + +Vector randomly_partition_input(u32 partition_probability_numerator, u32 partition_probability_denominator, size_t length) +{ + Vector result { 0 }; + for (size_t i = 0; i < length; ++i) { + if (AK::get_random_uniform(partition_probability_denominator) < partition_probability_numerator) { + result.append(1); + } else { + ++result.last(); + } + } + return result; +} + +} diff --git a/Userland/Libraries/LibTest/AsyncTestStreams.h b/Userland/Libraries/LibTest/AsyncTestStreams.h new file mode 100644 index 00000000000..261b8fd417d --- /dev/null +++ b/Userland/Libraries/LibTest/AsyncTestStreams.h @@ -0,0 +1,79 @@ +/* + * Copyright (c) 2024, Dan Klishch + * + * SPDX-License-Identifier: BSD-2-Clause + */ + +#pragma once + +#include +#include +#include +#include + +namespace Test { + +enum class StreamCloseExpectation { + Reset, + Close, +}; + +class AsyncMemoryInputStream final : public AsyncInputStream { +public: + AsyncMemoryInputStream(StringView data, StreamCloseExpectation expectation, Vector&& chunks); + ~AsyncMemoryInputStream(); + + void reset() override; + Coroutine> close() override; + bool is_open() const override; + + Coroutine> enqueue_some(Badge) override; + ReadonlyBytes buffered_data_unchecked(Badge) const override; + void dequeue(Badge, size_t bytes) override; + +private: + StringView m_data; + StreamCloseExpectation m_expectation; + Vector m_chunks; + + bool m_is_closed { false }; + bool m_is_reset { false }; + + bool m_encountered_eof { false }; + size_t m_read_head { 0 }; + size_t m_peek_head { 0 }; + size_t m_next_chunk_index { 1 }; + + size_t m_last_enqueue { 0 }; + + std::coroutine_handle<> m_awaiter; +}; + +class AsyncMemoryOutputStream final : public AsyncOutputStream { +public: + // FIXME: Support artificial atomic write limits similar to `chunks` parameter in + // AsyncMemoryInputStream. + AsyncMemoryOutputStream(StreamCloseExpectation expectation); + ~AsyncMemoryOutputStream(); + + void reset() override; + Coroutine> close() override; + bool is_open() const override; + + Coroutine> write_some(ReadonlyBytes data) override; + + ReadonlyBytes view() const { return m_buffer; } + +private: + StreamCloseExpectation m_expectation; + + bool m_is_closed { false }; + bool m_is_reset { false }; + + ByteBuffer m_buffer; +}; + +Coroutine> read_until_eof(AsyncInputStream& stream); +Vector randomly_partition_input(u32 partition_probability_numerator, u32 partition_probability_denominator, size_t length); + +} diff --git a/Userland/Libraries/LibTest/CMakeLists.txt b/Userland/Libraries/LibTest/CMakeLists.txt index a1ca2eea8f9..de22ff79a9d 100644 --- a/Userland/Libraries/LibTest/CMakeLists.txt +++ b/Userland/Libraries/LibTest/CMakeLists.txt @@ -1,11 +1,13 @@ serenity_install_sources("Userland/Libraries/LibTest") set(SOURCES + AsyncTestStreams.cpp TestSuite.cpp CrashTest.cpp ) serenity_lib(LibTest test) +target_link_libraries(LibTest PRIVATE LibCore) add_library(LibTestMain OBJECT TestMain.cpp) add_library(JavaScriptTestRunnerMain OBJECT JavaScriptTestRunnerMain.cpp)