Merge pull request #61 from alphatwirl/dev

Update tests on `Stream`

tests/stream/st.py

@@ -0,0 +1,6 @@
+from hypothesis import strategies as st
+
+
+def st_text() -> st.SearchStrategy[str]:
+    '''A strategy for text without control characters.'''
+    return st.text(alphabet=st.characters(blacklist_categories=['Cc', 'Cs']))

tests/stream/test_example.py

@@ -0,0 +1,78 @@
+import sys
+from unittest.mock import Mock, call
+
+from pytest import MonkeyPatch, fixture
+
+from atpbar.stream import FD, Stream
+
+
+@fixture()
+def mock_queue() -> Mock:
+    return Mock()
+
+
+@fixture()
+def obj(mock_queue: Mock) -> Stream:
+    return Stream(mock_queue, FD.STDOUT)
+
+
+def test_print(mock_queue: Mock, obj: Stream, monkeypatch: MonkeyPatch) -> None:
+    with monkeypatch.context() as m:
+        m.setattr(sys, 'stdout', obj)
+
+        print('abc')
+        assert [call(('abc\n', FD.STDOUT))] == mock_queue.put.call_args_list
+
+        mock_queue.reset_mock()
+
+        print('abc', 456)
+        assert [call(('abc 456\n', FD.STDOUT))] == mock_queue.put.call_args_list
+
+        mock_queue.reset_mock()
+
+        print('abc', 456, flush=True)
+        assert [call(('abc 456\n', FD.STDOUT))] == mock_queue.put.call_args_list
+
+        mock_queue.reset_mock()
+
+        print('abc', 456, end='')
+        assert [] == mock_queue.put.call_args_list
+        print()
+        assert [call(('abc 456\n', FD.STDOUT))] == mock_queue.put.call_args_list
+
+        mock_queue.reset_mock()
+
+        print('abc', 456, end='zzz', flush=True)
+        assert [call(('abc 456zzz', FD.STDOUT))] == mock_queue.put.call_args_list
+
+        mock_queue.reset_mock()
+
+        print('abc', end='')
+        assert [] == mock_queue.put.call_args_list
+        print(end='', flush=True)
+        assert [call(('abc', FD.STDOUT))] == mock_queue.put.call_args_list
+
+
+def test_print_bytes(mock_queue: Mock, obj: Stream, monkeypatch: MonkeyPatch) -> None:
+    with monkeypatch.context() as m:
+        m.setattr(sys, 'stdout', obj)
+
+        print(b'abc')
+        assert [call(("b'abc'\n", FD.STDOUT))] == mock_queue.put.call_args_list
+
+
+def test_stdout(mock_queue: Mock, obj: Stream, monkeypatch: MonkeyPatch) -> None:
+    with monkeypatch.context() as m:
+        m.setattr(sys, 'stdout', obj)
+
+        sys.stdout.write('abc')
+        sys.stdout.flush()
+        assert [call(('abc', FD.STDOUT))] == mock_queue.put.call_args_list
+
+        mock_queue.reset_mock()
+
+        sys.stdout.write('abc')
+        sys.stdout.flush()
+        assert [call(('abc', FD.STDOUT))] == mock_queue.put.call_args_list
+
+    mock_queue.reset_mock()

tests/stream/test_print.py

@@ -0,0 +1,68 @@
+import sys
+from typing import Literal
+from unittest.mock import Mock, call
+
+from hypothesis import given, settings
+from hypothesis import strategies as st
+from pytest import MonkeyPatch
+
+from atpbar.stream import FD, Stream
+
+from .st import st_text
+
+
+def st_end() -> st.SearchStrategy[str | None]:
+    '''A strategy for the `end` parameter of the `print` function.'''
+    return st.one_of(st.none(), st.just('\n'), st_text())
+
+
+@settings(max_examples=500)
+@given(
+    texts=st.lists(st_text()),
+    end=st_end(),
+    flush=st.booleans(),
+    fd_name=st.sampled_from(['stdout', 'stderr']),
+)
+def test_print(
+    texts: list[str], end: str | None, flush: bool, fd_name: Literal['stdout', 'stderr']
+) -> None:
+    queue = Mock()
+    stdout = Stream(queue, fd=FD.STDOUT)
+    stderr = Stream(queue, fd=FD.STDERR)
+
+    with MonkeyPatch.context() as m:
+        m.setattr(sys, 'stdout', stdout)
+        m.setattr(sys, 'stderr', stderr)
+
+        file = {'stdout': sys.stdout, 'stderr': sys.stderr}[fd_name]
+        fd = {'stdout': FD.STDOUT, 'stderr': FD.STDERR}[fd_name]
+
+        print(*texts, end=end, flush=flush, file=file)
+        joined = ' '.join(texts)
+        match joined, end, flush:
+            case '', None | '\n', _:
+                expected = '\n'
+                assert [call((expected, fd))] == queue.put.call_args_list
+            case '', '', _:
+                assert [] == queue.put.call_args_list
+            case '', str(), False if end:
+                assert [] == queue.put.call_args_list
+                print(file=file)
+                expected = f'{end}\n'
+                assert [call((expected, fd))] == queue.put.call_args_list
+            case '', str(), True if end:
+                expected = f'{end}'
+                assert [call((expected, fd))] == queue.put.call_args_list
+            case str(), None | '\n', _ if joined:
+                expected = f'{joined}\n'
+                assert [call((expected, fd))] == queue.put.call_args_list
+            case str(), str(), False if joined:
+                assert [] == queue.put.call_args_list
+                print(file=file)
+                expected = f'{joined}{end}\n'
+                assert [call((expected, fd))] == queue.put.call_args_list
+            case str(), str(), True if joined:
+                expected = f'{joined}{end}'
+                assert [call((expected, fd))] == queue.put.call_args_list
+            case _:  # pragma: no cover
+                assert False, (joined, end, flush)

tests/stream/test_st.py

@@ -0,0 +1,9 @@
+from hypothesis import given, settings
+
+from .st import st_text
+
+
+@settings(max_examples=500)
+@given(text=st_text())
+def test_st_text(text: str) -> None:
+    assert not text.endswith('\n')

tests/stream/test_stream.py

@@ -0,0 +1,49 @@
+from unittest.mock import sentinel
+
+from hypothesis import given, settings
+from hypothesis import strategies as st
+
+from atpbar.stream import Queue, Stream, StreamQueue
+
+from .st import st_text
+
+
+class StatefulTest:
+    def __init__(self, data: st.DataObject) -> None:
+        self.draw = data.draw
+        self.queue: StreamQueue = Queue()
+        self.fd = sentinel.fd
+        self.stream = Stream(self.queue, self.fd)
+        self.written = list[str]()
+
+    def write(self) -> None:
+        text = self.draw(st_text())
+        self.stream.write(text)
+        self.written.append(text)
+
+    def write_with_newline(self) -> None:
+        text = self.draw(st_text())
+        text += '\n'
+        self.stream.write(text)
+        self.written.append(text)
+        expected = ''.join(self.written)
+        assert self.queue.get() == (expected, self.fd)
+        self.written.clear()
+
+    def flush(self) -> None:
+        self.stream.flush()
+        expected = ''.join(self.written)
+        if not expected:
+            return
+        assert self.queue.get() == (expected, self.fd)
+        self.written.clear()
+
+
+@settings(max_examples=200)
+@given(data=st.data())
+def test_stream(data: st.DataObject) -> None:
+    test = StatefulTest(data=data)
+    METHODS = [test.write, test.write_with_newline, test.flush]
+    methods = data.draw(st.lists(st.sampled_from(METHODS)))
+    for method in methods:
+        method()