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,4 +1,5 @@
 set(TEST_SOURCES
+    TestAsyncTestStreams.cpp
     TestNoCrash.cpp
     TestGenerator.cpp
 )

Tests/LibTest/TestAsyncTestStreams.cpp

@@ -0,0 +1,214 @@
+/*
+ * Copyright (c) 2024, Dan Klishch <danilklishch@gmail.com>
+ *
+ * SPDX-License-Identifier: BSD-2-Clause
+ */
+
+#include <LibTest/AsyncTestCase.h>
+#include <LibTest/AsyncTestStreams.h>
+
+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<void> {
+            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<void> {
+            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());
+}

Userland/Libraries/LibTest/AsyncTestStreams.cpp

@@ -0,0 +1,223 @@
+/*
+ * Copyright (c) 2024, Dan Klishch <danilklishch@gmail.com>
+ *
+ * SPDX-License-Identifier: BSD-2-Clause
+ */
+
+#include <AK/Random.h>
+#include <LibTest/AsyncTestStreams.h>
+
+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<Core::DeferredInvocationEvent>(s_deferred_context, [&] {
+                m_awaiter.resume();
+            }));
+    }
+
+    void await_resume() { m_awaiter = {}; }
+
+    std::coroutine_handle<>& m_awaiter;
+};
+
+AsyncMemoryInputStream::AsyncMemoryInputStream(StringView data, StreamCloseExpectation expectation, Vector<size_t>&& 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<ErrorOr<void>> 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<ErrorOr<bool>> AsyncMemoryInputStream::enqueue_some(Badge<AsyncInputStream>)
+{
+    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<AsyncInputStream>) const
+{
+    return m_data.bytes().slice(m_read_head, m_peek_head - m_read_head);
+}
+
+void AsyncMemoryInputStream::dequeue(Badge<AsyncInputStream>, 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<ErrorOr<void>> AsyncMemoryOutputStream::close()
+{
+    VERIFY(is_open());
+    m_is_closed = true;
+    co_return {};
+}
+
+bool AsyncMemoryOutputStream::is_open() const
+{
+    return !m_is_closed && !m_is_reset;
+}
+
+Coroutine<ErrorOr<size_t>> AsyncMemoryOutputStream::write_some(ReadonlyBytes data)
+{
+    VERIFY(is_open());
+    m_buffer.append(data);
+    co_return data.size();
+}
+
+Coroutine<ErrorOr<ReadonlyBytes>> 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<size_t> randomly_partition_input(u32 partition_probability_numerator, u32 partition_probability_denominator, size_t length)
+{
+    Vector<size_t> 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;
+}
+
+}

Userland/Libraries/LibTest/AsyncTestStreams.h

@@ -0,0 +1,79 @@
+/*
+ * Copyright (c) 2024, Dan Klishch <danilklishch@gmail.com>
+ *
+ * SPDX-License-Identifier: BSD-2-Clause
+ */
+
+#pragma once
+
+#include <AK/AsyncStream.h>
+#include <LibCore/Event.h>
+#include <LibCore/ThreadEventQueue.h>
+#include <LibTest/Macros.h>
+
+namespace Test {
+
+enum class StreamCloseExpectation {
+    Reset,
+    Close,
+};
+
+class AsyncMemoryInputStream final : public AsyncInputStream {
+public:
+    AsyncMemoryInputStream(StringView data, StreamCloseExpectation expectation, Vector<size_t>&& chunks);
+    ~AsyncMemoryInputStream();
+
+    void reset() override;
+    Coroutine<ErrorOr<void>> close() override;
+    bool is_open() const override;
+
+    Coroutine<ErrorOr<bool>> enqueue_some(Badge<AsyncInputStream>) override;
+    ReadonlyBytes buffered_data_unchecked(Badge<AsyncInputStream>) const override;
+    void dequeue(Badge<AsyncInputStream>, size_t bytes) override;
+
+private:
+    StringView m_data;
+    StreamCloseExpectation m_expectation;
+    Vector<size_t> 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<ErrorOr<void>> close() override;
+    bool is_open() const override;
+
+    Coroutine<ErrorOr<size_t>> 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<ErrorOr<ReadonlyBytes>> read_until_eof(AsyncInputStream& stream);
+Vector<size_t> randomly_partition_input(u32 partition_probability_numerator, u32 partition_probability_denominator, size_t length);
+
+}

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)