(texera) branch gh-readonly-queue/main/pr-6669-6cb4a0487e2f7e8483611dd9809d154ec450d3fd created (now ff82157989)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6669-6cb4a0487e2f7e8483611dd9809d154ec450d3fd
in repository https://gitbox.apache.org/repos/asf/texera.git


      at ff82157989 test(pyamber): add unit tests for stoppable_queue_blocking_thread (#6669)

This branch includes the following new commits:

     new ff82157989 test(pyamber): add unit tests for stoppable_queue_blocking_thread (#6669)

The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.

(texera) 01/01: test(pyamber): add unit tests for stoppable_queue_blocking_thread (#6669)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6669-6cb4a0487e2f7e8483611dd9809d154ec450d3fd
in repository https://gitbox.apache.org/repos/asf/texera.git

commit ff82157989fffae8f28fd5d0de1dd822b298bd1e
Author: Matthew B. <mg...@uci.edu>
AuthorDate: Wed Jul 29 17:52:07 2026 -0700

    test(pyamber): add unit tests for stoppable_queue_blocking_thread (#6669)
    
    ### What changes were proposed in this PR?
    - Add unit tests for StoppableQueueBlockingRunnable.
    - Cover stop() enqueuing the RUNNABLE_STOP sentinel and
    interruptible_get() raising InterruptRunnable only on that exact
    control.
    - Cover run() processing entries until interrupted and always calling
    post_stop(), including on unexpected errors.
    ### Any related issues, documentation, discussions?
    Closes: #6668
    ### How was this PR tested?
    - Run: `cd amber && python -m pytest
    src/test/python/core/util/stoppable/test_stoppable_queue_blocking_thread.py`,
    expect all 7 tests passing.
    - Test-only change; no production code is modified.
    ### Was this PR authored or co-authored using generative AI tooling?
    Co-authored with Claude Opus 4.8 in compliance with ASF
    
    ---------
    
    Signed-off-by: Xinyuan Lin <xi...@uci.edu>
    Co-authored-by: Xinyuan Lin <xi...@uci.edu>
    Co-authored-by: Copilot Autofix powered by AI <17...@users.noreply.github.com>
    Co-authored-by: Claude Opus 4.8 <no...@anthropic.com>
---
 .../test_stoppable_queue_blocking_thread.py        | 159 +++++++++++++++++++++
 1 file changed, 159 insertions(+)

diff --git a/amber/src/test/python/core/util/stoppable/test_stoppable_queue_blocking_thread.py b/amber/src/test/python/core/util/stoppable/test_stoppable_queue_blocking_thread.py
new file mode 100644
index 0000000000..c6558ca084
--- /dev/null
+++ b/amber/src/test/python/core/util/stoppable/test_stoppable_queue_blocking_thread.py
@@ -0,0 +1,159 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+
+from collections import deque
+
+import pytest
+
+from core.util.customized_queue.queue_base import QueueControl, QueueElement
+from core.util.stoppable.stoppable_queue_blocking_thread import (
+    StoppableQueueBlockingRunnable,
+)
+
+
+class FakeQueue:
+    """An in-memory stand-in for IQueue.
+
+    ``get`` pops entries in FIFO order so ``run`` can be driven to completion
+    without real threads or blocking: once the stop sentinel is popped the
+    loop is interrupted before the queue is ever exhausted.
+    """
+
+    def __init__(self):
+        self.items: deque = deque()
+
+    def put(self, item) -> None:
+        self.items.append(item)
+
+    def get(self):
+        return self.items.popleft()
+
+    def is_empty(self) -> bool:
+        return not self.items
+
+
+class RecordingRunnable(StoppableQueueBlockingRunnable):
+    """Records the lifecycle callbacks so their order can be asserted."""
+
+    def __init__(self, name: str, queue: FakeQueue):
+        super().__init__(name=name, queue=queue)
+        self.events: list = []
+
+    def pre_start(self) -> None:
+        self.events.append(("pre_start",))
+
+    def receive(self, next_entry) -> None:
+        self.events.append(("receive", next_entry))
+
+    def post_stop(self) -> None:
+        self.events.append(("post_stop",))
+
+
+class RaisingRunnable(StoppableQueueBlockingRunnable):
+    """receive() raises an unexpected error to probe the finally: clause."""
+
+    def __init__(self, name: str, queue: FakeQueue):
+        super().__init__(name=name, queue=queue)
+        self.post_stop_called = 0
+
+    def receive(self, next_entry) -> None:
+        raise ValueError("boom")
+
+    def post_stop(self) -> None:
+        self.post_stop_called += 1
+
+
+class TestStop:
+    def test_stop_enqueues_the_runnable_stop_sentinel(self):
+        queue = FakeQueue()
+        runnable = StoppableQueueBlockingRunnable(name="r", queue=queue)
+
+        runnable.stop()
+
+        assert list(queue.items) == [StoppableQueueBlockingRunnable.RUNNABLE_STOP]
+        # The sentinel is exactly the marker control, not merely truthy.
+        assert queue.items[0] is StoppableQueueBlockingRunnable.RUNNABLE_STOP
+
+
+class TestInterruptibleGet:
+    def test_returns_a_normal_entry_unchanged(self):
+        queue = FakeQueue()
+        entry = QueueElement()
+        queue.put(entry)
+        runnable = StoppableQueueBlockingRunnable(name="r", queue=queue)
+
+        assert runnable.interruptible_get() is entry
+
+    def test_raises_interrupt_on_the_stop_sentinel(self):
+        queue = FakeQueue()
+        queue.put(StoppableQueueBlockingRunnable.RUNNABLE_STOP)
+        runnable = StoppableQueueBlockingRunnable(name="r", queue=queue)
+
+        with pytest.raises(StoppableQueueBlockingRunnable.InterruptRunnable):
+            runnable.interruptible_get()
+
+    def test_only_the_exact_sentinel_message_interrupts(self):
+        # A control whose message differs from the marker is a normal entry
+        # and must flow through, so unrelated controls do not stop the loop.
+        queue = FakeQueue()
+        other = QueueControl(msg="something-else")
+        queue.put(other)
+        runnable = StoppableQueueBlockingRunnable(name="r", queue=queue)
+
+        assert runnable.interruptible_get() is other
+
+
+class TestRun:
+    def test_processes_entries_until_interrupted_then_posts_stop(self):
+        queue = FakeQueue()
+        first, second = QueueElement(), QueueElement()
+        queue.put(first)
+        queue.put(second)
+        queue.put(StoppableQueueBlockingRunnable.RUNNABLE_STOP)
+        runnable = RecordingRunnable(name="r", queue=queue)
+
+        runnable.run()
+
+        # pre_start first, both entries received in FIFO order (by identity), post_stop last.
+        assert len(runnable.events) == 4
+        assert runnable.events[0] == ("pre_start",)
+        assert runnable.events[1][0] == "receive" and runnable.events[1][1] is first
+        assert runnable.events[2][0] == "receive" and runnable.events[2][1] is second
+        assert runnable.events[3] == ("post_stop",)
+
+    def test_post_stop_runs_even_when_receive_raises(self):
+        queue = FakeQueue()
+        queue.put(QueueElement())
+        runnable = RaisingRunnable(name="r", queue=queue)
+
+        # An unexpected error is not swallowed as an interruption; it
+        # propagates, but the finally: clause still tears down via post_stop.
+        with pytest.raises(ValueError, match="boom"):
+            runnable.run()
+
+        assert runnable.post_stop_called == 1
+
+    def test_stop_then_run_terminates_immediately(self):
+        # Enqueuing the sentinel via stop() must be enough to end a run with
+        # no entries processed in between.
+        queue = FakeQueue()
+        runnable = RecordingRunnable(name="r", queue=queue)
+        runnable.stop()
+
+        runnable.run()
+
+        assert runnable.events == [("pre_start",), ("post_stop",)]