(texera) 01/01: test(pyamber): add unit tests for single_blocking_io (#6671)

[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-6671-69286ac90fc3b2b968f5c53833ca11f2e781a7c9
in repository https://gitbox.apache.org/repos/asf/texera.git

commit b88b93e21a5e1d0dca8de802d7e2282e889ce3ad
Author: Matthew B. <mg...@uci.edu>
AuthorDate: Tue Jul 21 18:55:57 2026 -0700

    test(pyamber): add unit tests for single_blocking_io (#6671)
    
    ### What changes were proposed in this PR?
    - Add unit tests for SingleBlockingIO.
    - Cover write() appending to the buffer and flush() completing with a
    newline and publishing the value.
    - Cover readline() returning and clearing the value, driving the
    blocking wait loop deterministically with a scripted Condition.
    ### Any related issues, documentation, discussions?
    Closes: #6670
    ### How was this PR tested?
    - Run: `cd amber && python -m pytest
    src/test/python/core/models/test_single_blocking_io.py`, expect all 5
    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
---
 .../python/core/models/test_single_blocking_io.py  | 116 +++++++++++++++++++++
 1 file changed, 116 insertions(+)

diff --git a/amber/src/test/python/core/models/test_single_blocking_io.py b/amber/src/test/python/core/models/test_single_blocking_io.py
new file mode 100644
index 0000000000..1e284b4198
--- /dev/null
+++ b/amber/src/test/python/core/models/test_single_blocking_io.py
@@ -0,0 +1,116 @@
+# 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 threading import Condition
+
+from core.models.single_blocking_io import SingleBlockingIO
+
+
+class ScriptedCondition:
+    """A Condition stand-in that lets a test drive readline()'s wait loop
+    deterministically, with no real thread and no sleep.
+
+    SingleBlockingIO only uses the context-manager protocol plus notify() and
+    wait(); this records those calls and runs an optional ``on_wait`` hook to
+    simulate a producer flushing a value while the reader is parked.
+    """
+
+    def __init__(self):
+        self.wait_calls = 0
+        self.notify_calls = 0
+        self.on_wait = None
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc):
+        return False
+
+    def notify(self, n: int = 1) -> None:
+        self.notify_calls += 1
+
+    def wait(self, timeout=None) -> bool:
+        self.wait_calls += 1
+        if self.on_wait is not None:
+            self.on_wait()
+        return True
+
+
+class TestWrite:
+    def test_write_appends_to_buffer_without_publishing_a_value(self):
+        io = SingleBlockingIO(Condition())
+
+        io.write("ab")
+        io.write("cd")
+
+        assert io.buf == "abcd"
+        # Nothing is readable until flush().
+        assert io.value is None
+
+
+class TestFlush:
+    def test_flush_completes_with_newline_and_moves_buffer_to_value(self):
+        io = SingleBlockingIO(Condition())
+        io.write("hello")
+
+        io.flush()
+
+        assert io.value == "hello\n"
+        # The buffer is reset so the next line starts empty.
+        assert io.buf == ""
+
+    def test_flush_of_empty_buffer_publishes_just_a_newline(self):
+        io = SingleBlockingIO(Condition())
+
+        io.flush()
+
+        assert io.value == "\n"
+        assert io.buf == ""
+
+
+class TestReadline:
+    def test_returns_the_flushed_value_and_clears_it(self):
+        # With a value already published, readline() must not block: it
+        # returns immediately and clears the IO for the next line.
+        io = SingleBlockingIO(Condition())
+        io.write("hello")
+        io.flush()
+
+        assert io.readline() == "hello\n"
+        assert io.value is None
+
+    def test_blocks_until_a_value_appears_then_returns_and_clears(self):
+        # value starts as None, so readline() must enter its wait loop. The
+        # scripted condition simulates a producer that writes+flushes a line
+        # on the first wait(), which unblocks the reader deterministically.
+        cond = ScriptedCondition()
+        io = SingleBlockingIO(cond)
+
+        def produce():
+            io.write("data")
+            io.flush()
+
+        cond.on_wait = produce
+
+        line = io.readline()
+
+        assert line == "data\n"
+        # It genuinely parked once and signalled the waiting producer.
+        assert cond.wait_calls == 1
+        assert cond.notify_calls >= 1
+        # The value is cleared after being handed out.
+        assert io.value is None