clevercloud-storage/tests/test_stream.py

"""
Stream Utilities Tests
====================

This module contains tests for the stream utility classes.
"""

import io
import pytest

from clevercloud_storage_framework.utils.stream import ChunkedReader, ChunkedWriter


class TestChunkedReader:
    """Tests for the ChunkedReader class."""
    
    def test_read_chunks(self):
        """Test reading data in chunks."""
        # Create a file-like object with test data
        data = b'0123456789' * 100  # 1000 bytes
        file_obj = io.BytesIO(data)
        
        # Create a ChunkedReader with a chunk size of 100 bytes
        reader = ChunkedReader(file_obj, chunk_size=100)
        
        # Read all chunks
        chunks = list(reader)
        
        # Check that we got the expected number of chunks
        assert len(chunks) == 10
        
        # Check that each chunk has the expected size
        for chunk in chunks[:-1]:  # All but the last chunk
            assert len(chunk) == 100
        
        # Check that the chunks contain the expected data
        assert b''.join(chunks) == data
    
    def test_empty_file(self):
        """Test reading from an empty file."""
        # Create an empty file-like object
        file_obj = io.BytesIO(b'')
        
        # Create a ChunkedReader
        reader = ChunkedReader(file_obj)
        
        # Try to read chunks
        chunks = list(reader)
        
        # Check that we got no chunks
        assert len(chunks) == 0


class TestChunkedWriter:
    """Tests for the ChunkedWriter class."""
    
    def test_write_chunks(self):
        """Test writing data in chunks."""
        # Create an iterator of chunks
        chunks = [b'chunk1', b'chunk2', b'chunk3']
        
        # Create a ChunkedWriter with a chunk size larger than any individual chunk
        writer = ChunkedWriter(iter(chunks), chunk_size=10)
        
        # Get all chunks from the writer
        result_chunks = list(writer)
        
        # Check that we got the expected number of chunks (combined into 2 chunks due to chunk_size=10)
        assert len(result_chunks) == 2
        
        # Check that the chunks contain the expected combined data
        assert b''.join(result_chunks) == b''.join(chunks)
    
    def test_combine_small_chunks(self):
        """Test combining small chunks into larger ones."""
        # Create an iterator of small chunks
        chunks = [b'a', b'b', b'c', b'd', b'e', b'f', b'g', b'h', b'i', b'j']
        
        # Create a ChunkedWriter with a chunk size of 3 bytes
        writer = ChunkedWriter(iter(chunks), chunk_size=3)
        
        # Get all chunks from the writer
        result_chunks = list(writer)
        
        # Check that we got the expected number of chunks
        assert len(result_chunks) == 4  # 3 chunks of 3 bytes + 1 chunk of 1 byte
        
        # Check that the chunks contain the expected data
        assert result_chunks == [b'abc', b'def', b'ghi', b'j']
    
    def test_split_large_chunks(self):
        """Test splitting large chunks into smaller ones."""
        # Create an iterator with one large chunk
        chunks = [b'0123456789' * 10]  # 100 bytes
        
        # Create a ChunkedWriter with a chunk size of 20 bytes
        writer = ChunkedWriter(iter(chunks), chunk_size=20)
        
        # Get all chunks from the writer
        result_chunks = list(writer)
        
        # Check that we got the expected number of chunks
        assert len(result_chunks) == 5  # 100 bytes / 20 bytes per chunk = 5 chunks
        
        # Check that each chunk has the expected size
        for chunk in result_chunks:
            assert len(chunk) == 20
        
        # Check that the chunks contain the expected data
        assert b''.join(result_chunks) == chunks[0]
    
    def test_empty_iterator(self):
        """Test writing from an empty iterator."""
        # Create an empty iterator
        chunks = []
        
        # Create a ChunkedWriter
        writer = ChunkedWriter(iter(chunks))
        
        # Try to get chunks from the writer
        result_chunks = list(writer)
        
        # Check that we got no chunks
        assert len(result_chunks) == 0
        
    def test_mixed_chunk_sizes(self):
        """Test handling mixed chunk sizes."""
        # Create an iterator with mixed chunk sizes
        chunks = [b'a', b'bcdefghij', b'k']  # 1 byte, 10 bytes, 1 byte
        
        # Create a ChunkedWriter with a chunk size of 5 bytes
        writer = ChunkedWriter(iter(chunks), chunk_size=5)
        
        # Get all chunks from the writer
        result_chunks = list(writer)
        
        # Check that we got the expected chunks
        assert result_chunks == [b'abcde', b'fghij', b'k']
        
        # Check that the chunks contain the expected data
        assert b''.join(result_chunks) == b'abcdefghijk'
    
    def test_chunked_reader_iteration(self):
        """Test ChunkedReader iteration protocol."""
        # Create a file-like object with test data
        data = b'test data'
        file_obj = io.BytesIO(data)
        
        # Create a ChunkedReader
        reader = ChunkedReader(file_obj, chunk_size=4)
        
        # Test that the reader is an iterator
        assert iter(reader) is reader
        
        # Test reading chunks
        assert next(reader) == b'test'
        assert next(reader) == b' dat'
        assert next(reader) == b'a'
        
        # Test StopIteration
        with pytest.raises(StopIteration):
            next(reader)
    
    def test_chunked_writer_iteration(self):
        """Test ChunkedWriter iteration protocol."""
        # Create an iterator of chunks
        chunks = [b'chunk1', b'chunk2']
        
        # Create a ChunkedWriter
        writer = ChunkedWriter(iter(chunks), chunk_size=100)
        
        # Test that the writer is an iterator
        assert iter(writer) is writer
        
        # Test getting chunks - with chunk_size=100, all input chunks are combined
        assert next(writer) == b'chunk1chunk2'
        
        # Test StopIteration
        with pytest.raises(StopIteration):
            next(writer)