Trixy7/tests/test_network_compression.py

# -*- coding: utf-8 -*-
"""
Tests für Network Compression.
"""

import pytest

from trixy_core.network.compression import (
    Compressor,
    CompressionLevel,
    CompressionStats,
    ZlibCompressor,
    CompressionRegistry,
    get_default_registry,
    get_compressor,
)
from trixy_core.network.compression.compressor import (
    CompressionError,
    DecompressionError,
)
from trixy_core.network.compression.zlib import GzipCompressor


class TestCompressionLevel:
    """Tests für CompressionLevel."""

    def test_level_values(self):
        """Testet Level-Werte."""
        assert CompressionLevel.NONE == 0
        assert CompressionLevel.FASTEST == 1
        assert CompressionLevel.DEFAULT == 6
        assert CompressionLevel.BEST == 9

    def test_level_ordering(self):
        """Testet Level-Reihenfolge."""
        assert CompressionLevel.FASTEST < CompressionLevel.DEFAULT
        assert CompressionLevel.DEFAULT < CompressionLevel.BEST


class TestCompressionStats:
    """Tests für CompressionStats."""

    def test_initial_stats(self):
        """Testet initiale Statistiken."""
        stats = CompressionStats()
        assert stats.total_compressed == 0
        assert stats.total_decompressed == 0
        assert stats.bytes_in == 0
        assert stats.bytes_out == 0

    def test_update_compression(self):
        """Testet Komprimierungs-Update."""
        stats = CompressionStats()
        stats.update_compression(1000, 100, 5.0)

        assert stats.total_compressed == 1
        assert stats.bytes_in == 1000
        assert stats.bytes_out == 100
        assert stats.compression_ratio == 0.1
        assert stats.total_time_ms == 5.0

    def test_update_decompression(self):
        """Testet Dekomprimierungs-Update."""
        stats = CompressionStats()
        stats.update_decompression(100, 1000, 3.0)

        assert stats.total_decompressed == 1
        assert stats.total_time_ms == 3.0

    def test_to_dict(self):
        """Testet Dictionary-Konvertierung."""
        stats = CompressionStats()
        stats.update_compression(1000, 100, 5.0)

        d = stats.to_dict()
        assert "total_compressed" in d
        assert "compression_ratio" in d
        assert "space_saved_percent" in d
        assert d["space_saved_percent"] == 90.0


class TestZlibCompressor:
    """Tests für ZlibCompressor."""

    @pytest.fixture
    def compressor(self):
        """Erstellt ZlibCompressor."""
        return ZlibCompressor(level=CompressionLevel.DEFAULT)

    @pytest.fixture
    def test_data(self):
        """Test-Daten."""
        return b"Hello World! " * 100

    def test_compression_basic(self, compressor, test_data):
        """Testet grundlegende Komprimierung."""
        compressed = compressor.compress(test_data)
        assert len(compressed) < len(test_data)
        assert compressed != test_data

    def test_decompression_basic(self, compressor, test_data):
        """Testet grundlegende Dekomprimierung."""
        compressed = compressor.compress(test_data)
        decompressed = compressor.decompress(compressed)
        assert decompressed == test_data

    def test_compression_empty_data(self, compressor):
        """Testet Komprimierung leerer Daten."""
        result = compressor.compress(b"")
        assert result == b""

    def test_decompression_empty_data(self, compressor):
        """Testet Dekomprimierung leerer Daten."""
        result = compressor.decompress(b"")
        assert result == b""

    def test_compression_levels(self, test_data):
        """Testet verschiedene Komprimierungsstufen."""
        fast = ZlibCompressor(level=CompressionLevel.FASTEST)
        best = ZlibCompressor(level=CompressionLevel.BEST)

        fast_compressed = fast.compress(test_data)
        best_compressed = best.compress(test_data)

        # BEST sollte kleinere Ausgabe produzieren
        assert len(best_compressed) <= len(fast_compressed)

        # Beide sollten korrekt dekomprimieren
        assert fast.decompress(fast_compressed) == test_data
        assert best.decompress(best_compressed) == test_data

    def test_statistics_tracking(self, compressor, test_data):
        """Testet Statistik-Tracking."""
        compressor.compress(test_data)
        compressor.compress(test_data)

        assert compressor.stats.total_compressed == 2
        assert compressor.stats.bytes_in == len(test_data) * 2

    def test_compress_with_header(self, compressor, test_data):
        """Testet Komprimierung mit Header."""
        compressed = compressor.compress_with_header(test_data)
        # Header: 4 Bytes Größe
        assert len(compressed) > 4

    def test_decompress_with_header(self, compressor, test_data):
        """Testet Dekomprimierung mit Header."""
        compressed = compressor.compress_with_header(test_data)
        decompressed = compressor.decompress_with_header(compressed)
        assert decompressed == test_data

    def test_should_compress(self, compressor):
        """Testet should_compress."""
        small_data = b"Hi"
        large_data = b"Hello World! " * 100

        assert not compressor.should_compress(small_data)
        assert compressor.should_compress(large_data)

    def test_compress_if_beneficial(self, compressor):
        """Testet compress_if_beneficial."""
        # Komprimierbare Daten
        large_data = b"Hello World! " * 100
        result, was_compressed = compressor.compress_if_beneficial(large_data)
        assert was_compressed
        assert len(result) < len(large_data)

        # Zu kleine Daten
        small_data = b"Hi"
        result, was_compressed = compressor.compress_if_beneficial(small_data)
        assert not was_compressed
        assert result == small_data

    def test_invalid_decompression(self, compressor):
        """Testet ungültige Dekomprimierung."""
        with pytest.raises(DecompressionError):
            compressor.decompress(b"invalid data that is not compressed")

    def test_properties(self, compressor):
        """Testet Compressor-Properties."""
        assert compressor.name == "zlib"
        assert compressor.content_type == "application/zlib"
        assert compressor.min_size > 0

    def test_reset_stats(self, compressor, test_data):
        """Testet Stats-Reset."""
        compressor.compress(test_data)
        assert compressor.stats.total_compressed == 1

        compressor.reset_stats()
        assert compressor.stats.total_compressed == 0


class TestGzipCompressor:
    """Tests für GzipCompressor."""

    @pytest.fixture
    def compressor(self):
        """Erstellt GzipCompressor."""
        return GzipCompressor(level=CompressionLevel.DEFAULT)

    @pytest.fixture
    def test_data(self):
        """Test-Daten."""
        return b"Test data for gzip compression " * 50

    def test_compression_basic(self, compressor, test_data):
        """Testet grundlegende Gzip-Komprimierung."""
        compressed = compressor.compress(test_data)
        assert len(compressed) < len(test_data)

    def test_decompression_basic(self, compressor, test_data):
        """Testet grundlegende Gzip-Dekomprimierung."""
        compressed = compressor.compress(test_data)
        decompressed = compressor.decompress(compressed)
        assert decompressed == test_data

    def test_properties(self, compressor):
        """Testet Gzip-Properties."""
        assert compressor.name == "gzip"
        assert compressor.content_type == "application/gzip"


class TestCompressionRegistry:
    """Tests für CompressionRegistry."""

    @pytest.fixture
    def registry(self):
        """Erstellt neue Registry."""
        return CompressionRegistry()

    def test_register_compressor(self, registry):
        """Testet Compressor-Registrierung."""
        registry.register("test", ZlibCompressor, "application/test")

        assert "test" in registry
        assert registry.is_available("test")

    def test_get_compressor(self, registry):
        """Testet Compressor-Abruf."""
        registry.register("test", ZlibCompressor)

        compressor = registry.get("test")
        assert compressor is not None
        assert isinstance(compressor, ZlibCompressor)

    def test_get_nonexistent(self, registry):
        """Testet Abruf nicht existierender Compressor."""
        result = registry.get("nonexistent")
        assert result is None

    def test_get_by_content_type(self, registry):
        """Testet Abruf nach Content-Type."""
        registry.register("test", ZlibCompressor, "application/test")

        compressor = registry.get_by_content_type("application/test")
        assert compressor is not None

    def test_unregister(self, registry):
        """Testet Compressor-Entfernung."""
        registry.register("test", ZlibCompressor)
        assert registry.unregister("test")
        assert "test" not in registry

    def test_list_available(self, registry):
        """Testet Auflistung."""
        registry.register("comp1", ZlibCompressor)
        registry.register("comp2", ZlibCompressor)

        available = registry.list_available()
        assert "comp1" in available
        assert "comp2" in available


class TestDefaultRegistry:
    """Tests für Default-Registry."""

    def test_get_default_registry(self):
        """Testet Default-Registry."""
        registry = get_default_registry()
        assert registry is not None
        assert len(registry) >= 2  # zlib und gzip

    def test_default_compressors(self):
        """Testet Standard-Compressors."""
        registry = get_default_registry()

        assert "zlib" in registry
        assert "gzip" in registry

    def test_get_compressor_convenience(self):
        """Testet get_compressor Convenience-Funktion."""
        compressor = get_compressor("zlib")
        assert compressor is not None
        assert isinstance(compressor, ZlibCompressor)

    def test_get_compressor_with_level(self):
        """Testet get_compressor mit Level."""
        compressor = get_compressor("zlib", CompressionLevel.BEST)
        assert compressor is not None
        assert compressor.level == CompressionLevel.BEST


class TestCompressionRoundTrip:
    """Integrationstests für Komprimierung."""

    @pytest.mark.parametrize("size", [100, 1000, 10000, 100000])
    def test_various_sizes(self, size):
        """Testet verschiedene Datengrößen."""
        compressor = ZlibCompressor()
        data = b"x" * size

        compressed = compressor.compress(data)
        decompressed = compressor.decompress(compressed)

        assert decompressed == data

    @pytest.mark.parametrize("pattern", [
        b"AAAAAAAAAA" * 100,  # Repetitiv
        bytes(range(256)) * 10,  # Sequentiell
        b"\x00" * 1000,  # Nullen
    ])
    def test_various_patterns(self, pattern):
        """Testet verschiedene Datenmuster."""
        compressor = ZlibCompressor()

        compressed = compressor.compress(pattern)
        decompressed = compressor.decompress(compressed)

        assert decompressed == pattern

    def test_binary_data(self):
        """Testet binäre Daten."""
        import os
        compressor = ZlibCompressor()
        data = os.urandom(10000)  # Zufällige Bytes

        compressed = compressor.compress(data)
        decompressed = compressor.decompress(compressed)

        assert decompressed == data