diff --git a/libs/input/tests/Android.bp b/libs/input/tests/Android.bp
index 132866b..43bc894 100644
--- a/libs/input/tests/Android.bp
+++ b/libs/input/tests/Android.bp
@@ -16,6 +16,7 @@
         "BlockingQueue_test.cpp",
         "IdGenerator_test.cpp",
         "InputChannel_test.cpp",
+        "InputConsumer_test.cpp",
         "InputDevice_test.cpp",
         "InputEvent_test.cpp",
         "InputPublisherAndConsumer_test.cpp",
@@ -25,6 +26,8 @@
         "MotionPredictorMetricsManager_test.cpp",
         "Resampler_test.cpp",
         "RingBuffer_test.cpp",
+        "TestInputChannel.cpp",
+        "TestLooper.cpp",
         "TfLiteMotionPredictor_test.cpp",
         "TouchResampling_test.cpp",
         "TouchVideoFrame_test.cpp",
diff --git a/libs/input/tests/InputConsumer_test.cpp b/libs/input/tests/InputConsumer_test.cpp
new file mode 100644
index 0000000..c30f243
--- /dev/null
+++ b/libs/input/tests/InputConsumer_test.cpp
@@ -0,0 +1,123 @@
+/**
+ * Copyright 2024 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <input/InputConsumerNoResampling.h>
+
+#include <memory>
+#include <optional>
+#include <utility>
+
+#include <TestInputChannel.h>
+#include <TestLooper.h>
+#include <android-base/logging.h>
+#include <gtest/gtest.h>
+#include <input/BlockingQueue.h>
+#include <input/InputEventBuilders.h>
+#include <utils/StrongPointer.h>
+
+namespace android {
+
+class InputConsumerTest : public testing::Test, public InputConsumerCallbacks {
+protected:
+    InputConsumerTest()
+          : mClientTestChannel{std::make_shared<TestInputChannel>("TestChannel")},
+            mTestLooper{std::make_shared<TestLooper>()} {
+        Looper::setForThread(mTestLooper->getLooper());
+        mConsumer = std::make_unique<InputConsumerNoResampling>(mClientTestChannel, mTestLooper,
+                                                                *this, /*resampler=*/nullptr);
+    }
+
+    void assertOnBatchedInputEventPendingWasCalled();
+
+    std::shared_ptr<TestInputChannel> mClientTestChannel;
+    std::shared_ptr<TestLooper> mTestLooper;
+    std::unique_ptr<InputConsumerNoResampling> mConsumer;
+
+    BlockingQueue<std::unique_ptr<KeyEvent>> mKeyEvents;
+    BlockingQueue<std::unique_ptr<MotionEvent>> mMotionEvents;
+    BlockingQueue<std::unique_ptr<FocusEvent>> mFocusEvents;
+    BlockingQueue<std::unique_ptr<CaptureEvent>> mCaptureEvents;
+    BlockingQueue<std::unique_ptr<DragEvent>> mDragEvents;
+    BlockingQueue<std::unique_ptr<TouchModeEvent>> mTouchModeEvents;
+
+private:
+    size_t onBatchedInputEventPendingInvocationCount{0};
+
+    // InputConsumerCallbacks interface
+    void onKeyEvent(std::unique_ptr<KeyEvent> event, uint32_t seq) override {
+        mKeyEvents.push(std::move(event));
+        mConsumer->finishInputEvent(seq, true);
+    }
+    void onMotionEvent(std::unique_ptr<MotionEvent> event, uint32_t seq) override {
+        mMotionEvents.push(std::move(event));
+        mConsumer->finishInputEvent(seq, true);
+    }
+    void onBatchedInputEventPending(int32_t pendingBatchSource) override {
+        if (!mConsumer->probablyHasInput()) {
+            ADD_FAILURE() << "should deterministically have input because there is a batch";
+        }
+        ++onBatchedInputEventPendingInvocationCount;
+    };
+    void onFocusEvent(std::unique_ptr<FocusEvent> event, uint32_t seq) override {
+        mFocusEvents.push(std::move(event));
+        mConsumer->finishInputEvent(seq, true);
+    };
+    void onCaptureEvent(std::unique_ptr<CaptureEvent> event, uint32_t seq) override {
+        mCaptureEvents.push(std::move(event));
+        mConsumer->finishInputEvent(seq, true);
+    };
+    void onDragEvent(std::unique_ptr<DragEvent> event, uint32_t seq) override {
+        mDragEvents.push(std::move(event));
+        mConsumer->finishInputEvent(seq, true);
+    }
+    void onTouchModeEvent(std::unique_ptr<TouchModeEvent> event, uint32_t seq) override {
+        mTouchModeEvents.push(std::move(event));
+        mConsumer->finishInputEvent(seq, true);
+    };
+};
+
+void InputConsumerTest::assertOnBatchedInputEventPendingWasCalled() {
+    ASSERT_GT(onBatchedInputEventPendingInvocationCount, 0UL)
+            << "onBatchedInputEventPending has not been called.";
+    --onBatchedInputEventPendingInvocationCount;
+}
+
+TEST_F(InputConsumerTest, MessageStreamBatchedInMotionEvent) {
+    mClientTestChannel->enqueueMessage(
+            InputMessageBuilder{InputMessage::Type::MOTION, /*seq=*/0}.build());
+    mClientTestChannel->enqueueMessage(
+            InputMessageBuilder{InputMessage::Type::MOTION, /*seq=*/1}.build());
+    mClientTestChannel->enqueueMessage(
+            InputMessageBuilder{InputMessage::Type::MOTION, /*seq=*/2}.build());
+
+    mClientTestChannel->assertNoSentMessages();
+
+    mTestLooper->invokeCallback(mClientTestChannel->getFd(), ALOOPER_EVENT_INPUT);
+
+    assertOnBatchedInputEventPendingWasCalled();
+
+    mConsumer->consumeBatchedInputEvents(std::nullopt);
+
+    std::unique_ptr<MotionEvent> batchedMotionEvent = mMotionEvents.pop();
+    ASSERT_NE(batchedMotionEvent, nullptr);
+
+    mClientTestChannel->assertFinishMessage(/*seq=*/0, /*handled=*/true);
+    mClientTestChannel->assertFinishMessage(/*seq=*/1, /*handled=*/true);
+    mClientTestChannel->assertFinishMessage(/*seq=*/2, /*handled=*/true);
+
+    EXPECT_EQ(batchedMotionEvent->getHistorySize() + 1, 3UL);
+}
+} // namespace android
diff --git a/libs/input/tests/TestInputChannel.cpp b/libs/input/tests/TestInputChannel.cpp
new file mode 100644
index 0000000..d5f00b6
--- /dev/null
+++ b/libs/input/tests/TestInputChannel.cpp
@@ -0,0 +1,85 @@
+/**
+ * Copyright 2024 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#define LOG_TAG "TestInputChannel"
+#define ATRACE_TAG ATRACE_TAG_INPUT
+
+#include <TestInputChannel.h>
+
+#include <android-base/logging.h>
+#include <android-base/unique_fd.h>
+#include <binder/IBinder.h>
+#include <utils/StrongPointer.h>
+
+namespace android {
+
+namespace {
+constexpr int FAKE_FD{-1};
+} // namespace
+
+// --- TestInputChannel ---
+
+TestInputChannel::TestInputChannel(const std::string& name)
+      : InputChannel{name, base::unique_fd(FAKE_FD), sp<BBinder>::make()} {}
+
+void TestInputChannel::enqueueMessage(const InputMessage& message) {
+    mReceivedMessages.push(message);
+}
+
+status_t TestInputChannel::sendMessage(const InputMessage* message) {
+    LOG_IF(FATAL, message == nullptr)
+            << "TestInputChannel " << getName() << ". No message was passed to sendMessage.";
+
+    mSentMessages.push(*message);
+    return OK;
+}
+
+base::Result<InputMessage> TestInputChannel::receiveMessage() {
+    if (mReceivedMessages.empty()) {
+        return base::Error(WOULD_BLOCK);
+    }
+    InputMessage message = mReceivedMessages.front();
+    mReceivedMessages.pop();
+    return message;
+}
+
+bool TestInputChannel::probablyHasInput() const {
+    return !mReceivedMessages.empty();
+}
+
+void TestInputChannel::assertFinishMessage(uint32_t seq, bool handled) {
+    ASSERT_FALSE(mSentMessages.empty())
+            << "TestInputChannel " << getName() << ". Cannot assert. mSentMessages is empty.";
+
+    const InputMessage& finishMessage = mSentMessages.front();
+
+    EXPECT_EQ(finishMessage.header.seq, seq)
+            << "TestInputChannel " << getName()
+            << ". Sequence mismatch. Message seq: " << finishMessage.header.seq
+            << " Expected seq: " << seq;
+
+    EXPECT_EQ(finishMessage.body.finished.handled, handled)
+            << "TestInputChannel " << getName()
+            << ". Handled value mismatch. Message val: " << std::boolalpha
+            << finishMessage.body.finished.handled << "Expected val: " << handled
+            << std::noboolalpha;
+    mSentMessages.pop();
+}
+
+void TestInputChannel::assertNoSentMessages() const {
+    ASSERT_TRUE(mSentMessages.empty());
+}
+} // namespace android
\ No newline at end of file
diff --git a/libs/input/tests/TestInputChannel.h b/libs/input/tests/TestInputChannel.h
new file mode 100644
index 0000000..43253ec
--- /dev/null
+++ b/libs/input/tests/TestInputChannel.h
@@ -0,0 +1,66 @@
+/**
+ * Copyright 2024 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <queue>
+#include <string>
+
+#include <android-base/result.h>
+#include <gtest/gtest.h>
+#include <input/InputTransport.h>
+#include <utils/Errors.h>
+
+namespace android {
+
+class TestInputChannel final : public InputChannel {
+public:
+    explicit TestInputChannel(const std::string& name);
+
+    /**
+     * Enqueues a message in mReceivedMessages.
+     */
+    void enqueueMessage(const InputMessage& message);
+
+    /**
+     * Pushes message to mSentMessages. In the default implementation, InputChannel sends messages
+     * through a file descriptor. TestInputChannel, on the contrary, stores sent messages in
+     * mSentMessages for assertion reasons.
+     */
+    status_t sendMessage(const InputMessage* message) override;
+
+    /**
+     * Returns an InputMessage from mReceivedMessages. This is done instead of retrieving data
+     * directly from fd.
+     */
+    base::Result<InputMessage> receiveMessage() override;
+
+    /**
+     * Returns if mReceivedMessages is not empty.
+     */
+    bool probablyHasInput() const override;
+
+    void assertFinishMessage(uint32_t seq, bool handled);
+
+    void assertNoSentMessages() const;
+
+private:
+    // InputMessages received by the endpoint.
+    std::queue<InputMessage> mReceivedMessages;
+    // InputMessages sent by the endpoint.
+    std::queue<InputMessage> mSentMessages;
+};
+} // namespace android
diff --git a/libs/input/tests/TestLooper.cpp b/libs/input/tests/TestLooper.cpp
new file mode 100644
index 0000000..e0f01ed
--- /dev/null
+++ b/libs/input/tests/TestLooper.cpp
@@ -0,0 +1,51 @@
+/**
+ * Copyright 2024 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <TestLooper.h>
+
+#include <android-base/logging.h>
+
+namespace android {
+
+TestLooper::TestLooper() : mLooper(sp<Looper>::make(/*allowNonCallbacks=*/false)) {}
+
+int TestLooper::addFd(int fd, int ident, int events, const sp<LooperCallback>& callback,
+                      void* data) {
+    mCallbacks[fd] = callback;
+    constexpr int SUCCESS{1};
+    return SUCCESS;
+}
+
+int TestLooper::removeFd(int fd) {
+    if (auto it = mCallbacks.find(fd); it != mCallbacks.cend()) {
+        mCallbacks.erase(fd);
+        constexpr int SUCCESS{1};
+        return SUCCESS;
+    }
+    constexpr int FAILURE{0};
+    return FAILURE;
+}
+
+void TestLooper::invokeCallback(int fd, int events) {
+    auto it = mCallbacks.find(fd);
+    LOG_IF(FATAL, it == mCallbacks.cend()) << "Fd does not exist in mCallbacks.";
+    mCallbacks[fd]->handleEvent(fd, events, /*data=*/nullptr);
+}
+
+sp<Looper> TestLooper::getLooper() const {
+    return mLooper;
+}
+} // namespace android
\ No newline at end of file
diff --git a/libs/input/tests/TestLooper.h b/libs/input/tests/TestLooper.h
new file mode 100644
index 0000000..3242bc7
--- /dev/null
+++ b/libs/input/tests/TestLooper.h
@@ -0,0 +1,56 @@
+/**
+ * Copyright 2024 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <map>
+
+#include <input/LooperInterface.h>
+
+namespace android {
+/**
+ * TestLooper provides a mechanism to directly trigger Looper's callback.
+ */
+class TestLooper final : public LooperInterface {
+public:
+    TestLooper();
+
+    /**
+     * Adds a file descriptor to mCallbacks. Ident, events, and data parameters are ignored. If
+     * addFd is called with an existent file descriptor and a different callback, the previous
+     * callback is overwritten.
+     */
+    int addFd(int fd, int ident, int events, const sp<LooperCallback>& callback,
+              void* data) override;
+
+    /**
+     * Removes a file descriptor from mCallbacks. If fd is not in mCallbacks, returns FAILURE.
+     */
+    int removeFd(int fd) override;
+
+    /**
+     * Calls handleEvent of the file descriptor. Fd must be in mCallbacks. Otherwise, invokeCallback
+     * fatally logs.
+     */
+    void invokeCallback(int fd, int events);
+
+    sp<Looper> getLooper() const override;
+
+private:
+    std::map<int /*fd*/, sp<LooperCallback>> mCallbacks;
+    sp<Looper> mLooper;
+};
+} // namespace android
\ No newline at end of file