Class PartialWriteStream

python/pyfory/tests/test_buffer.py:68–81

class PartialWriteStream:
    def __init__(self):
        self._data = bytearray()

    def write(self, payload):
        if not payload:
            return 0
        view = memoryview(payload).cast("B")
        wrote = min(2, len(view))
        self._data.extend(view[:wrote])
        return wrote

    def to_bytes(self):
        return bytes(self._data)


def test_buffer():