"""
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.
"""
import random
import sys
import unittest

from pypaimon.common.delta_varint_compressor import DeltaVarintCompressor


class DeltaVarintCompressorTest(unittest.TestCase):
    """Tests for DeltaVarintCompressor following org.apache.paimon.utils.DeltaVarintCompressorTest."""

    def test_normal_case_1(self):
        """Test case for normal compression and decompression."""
        original = [80, 50, 90, 80, 70]
        compressed = DeltaVarintCompressor.compress(original)
        decompressed = DeltaVarintCompressor.decompress(compressed)

        self.assertEqual(original, decompressed)  # Verify data integrity
        self.assertEqual(6, len(compressed))  # Optimized size for small deltas

    def test_normal_case_2(self):
        """Test case for normal compression and decompression."""
        original = [100, 50, 150, 100, 200]
        compressed = DeltaVarintCompressor.compress(original)
        decompressed = DeltaVarintCompressor.decompress(compressed)

        self.assertEqual(original, decompressed)  # Verify data integrity
        self.assertEqual(8, len(compressed))  # Optimized size for small deltas

    def test_random(self):
        """Test with random data to ensure robustness."""
        # Run multiple iterations to test various random cases
        for _ in range(100):  # Reduced from 10000 for reasonable test time
            original = []
            for i in range(100):
                # Use a smaller range than Java's Long.MAX_VALUE for Python compatibility
                original.append(random.randint(-sys.maxsize, sys.maxsize))
            compressed = DeltaVarintCompressor.compress(original)
            decompressed = DeltaVarintCompressor.decompress(compressed)

            self.assertEqual(original, decompressed)  # Verify data integrity

    def test_empty_array(self):
        """Test case for empty array."""
        original = []
        compressed = DeltaVarintCompressor.compress(original)
        decompressed = DeltaVarintCompressor.decompress(compressed)

        self.assertEqual(original, decompressed)
        self.assertEqual(0, len(compressed))

    def test_single_element(self):
        """Test case for single-element array."""
        original = [42]
        compressed = DeltaVarintCompressor.compress(original)
        decompressed = DeltaVarintCompressor.decompress(compressed)

        self.assertEqual(original, decompressed)
        # Calculate expected size: Varint encoding for 42 (0x2A -> 1 byte)
        self.assertEqual(1, len(compressed))

    def test_extreme_values(self):
        """Test case for extreme values (sys.maxsize and -sys.maxsize)."""
        original = [-sys.maxsize, sys.maxsize]
        compressed = DeltaVarintCompressor.compress(original)
        decompressed = DeltaVarintCompressor.decompress(compressed)

        self.assertEqual(original, decompressed)
        # The compressed size will depend on the platform's sys.maxsize
        # but should be reasonable for the delta encoding
        self.assertGreater(len(compressed), 0)

    def test_negative_deltas(self):
        """Test case for negative deltas with ZigZag optimization."""
        original = [100, -50, -150, -100]  # Negative values
        compressed = DeltaVarintCompressor.compress(original)
        decompressed = DeltaVarintCompressor.decompress(compressed)

        self.assertEqual(original, decompressed)
        # Verify ZigZag optimization: -1 → 1 (1 byte)
        # Delta sequence: [100, -150, -100, 50] → ZigZag →
        # Each encoded in 1-2 bytes
        self.assertLessEqual(len(compressed), 8)  # Optimized size

    def test_unsorted_data(self):
        """Test case for unsorted data (worse compression ratio)."""
        original = [1000, 5, 9999, 12345, 6789]
        compressed = DeltaVarintCompressor.compress(original)
        decompressed = DeltaVarintCompressor.decompress(compressed)

        self.assertEqual(original, decompressed)
        # Larger deltas → more bytes (e.g., 9994 → 3 bytes)
        self.assertGreater(len(compressed), 5)  # Worse than sorted case

    def test_corrupted_input(self):
        """Test case for corrupted input (invalid Varint)."""
        # Invalid Varint (all continuation flags)
        corrupted = bytes([0x80, 0x80, 0x80])
        try:
            result = DeltaVarintCompressor.decompress(corrupted)
            # If it doesn't raise an exception, the result should be reasonable
            self.assertIsInstance(result, list)
        except (IndexError, ValueError, RuntimeError):
            # It's acceptable to raise an exception for corrupted data
            pass

    def test_zero_values(self):
        """Test case for arrays with zero values."""
        original = [0, 0, 0, 0, 0]
        compressed = DeltaVarintCompressor.compress(original)
        decompressed = DeltaVarintCompressor.decompress(compressed)

        self.assertEqual(original, decompressed)
        # All deltas are 0, so should compress very well
        self.assertLessEqual(len(compressed), 5)

    def test_ascending_sequence(self):
        """Test case for ascending sequence (optimal for delta compression)."""
        original = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
        compressed = DeltaVarintCompressor.compress(original)
        decompressed = DeltaVarintCompressor.decompress(compressed)

        self.assertEqual(original, decompressed)
        # All deltas are 1, so should compress very well
        self.assertLessEqual(len(compressed), 15)  # Much smaller than original

    def test_descending_sequence(self):
        """Test case for descending sequence."""
        original = [10, 9, 8, 7, 6, 5, 4, 3, 2, 1]
        compressed = DeltaVarintCompressor.compress(original)
        decompressed = DeltaVarintCompressor.decompress(compressed)

        self.assertEqual(original, decompressed)
        # All deltas are -1, should still compress well with ZigZag
        self.assertLessEqual(len(compressed), 15)

    def test_large_positive_values(self):
        """Test case for large positive values."""
        original = [1000000, 2000000, 3000000, 4000000]
        compressed = DeltaVarintCompressor.compress(original)
        decompressed = DeltaVarintCompressor.decompress(compressed)

        self.assertEqual(original, decompressed)
        # Large values but consistent deltas should still compress reasonably
        self.assertGreater(len(compressed), 4)  # Will be larger due to big numbers

    def test_mixed_positive_negative(self):
        """Test case for mixed positive and negative values."""
        original = [100, -200, 300, -400, 500]
        compressed = DeltaVarintCompressor.compress(original)
        decompressed = DeltaVarintCompressor.decompress(compressed)

        self.assertEqual(original, decompressed)
        # Mixed signs create larger deltas
        self.assertGreater(len(compressed), 5)

    def test_compression_efficiency(self):
        """Test that compression actually reduces size for suitable data."""
        # Create a sequence with small deltas
        original = []
        base = 1000
        for i in range(100):
            base += random.randint(-10, 10)  # Small deltas
            original.append(base)

        compressed = DeltaVarintCompressor.compress(original)
        decompressed = DeltaVarintCompressor.decompress(compressed)

        self.assertEqual(original, decompressed)
        # For small deltas, compression should be effective
        # Original would need 8 bytes per int (800 bytes), compressed should be much smaller
        self.assertLess(len(compressed), len(original) * 4)  # At least 50% compression

    def test_round_trip_consistency(self):
        """Test that multiple compress/decompress cycles are consistent."""
        original = [1, 10, 100, 1000, 10000]
        # First round trip
        compressed1 = DeltaVarintCompressor.compress(original)
        decompressed1 = DeltaVarintCompressor.decompress(compressed1)
        # Second round trip
        compressed2 = DeltaVarintCompressor.compress(decompressed1)
        decompressed2 = DeltaVarintCompressor.decompress(compressed2)
        # All should be identical
        self.assertEqual(original, decompressed1)
        self.assertEqual(original, decompressed2)
        self.assertEqual(compressed1, compressed2)

    def test_boundary_values(self):
        """Test boundary values for varint encoding."""
        # Test values around varint boundaries (127, 16383, etc.)
        boundary_values = [
            0, 1, 127, 128, 255, 256,
            16383, 16384, 32767, 32768,
            -1, -127, -128, -255, -256,
            -16383, -16384, -32767, -32768
        ]
        compressed = DeltaVarintCompressor.compress(boundary_values)
        decompressed = DeltaVarintCompressor.decompress(compressed)
        self.assertEqual(boundary_values, decompressed)

    def test_java_compatibility_zigzag_encoding(self):
        """Test ZigZag encoding compatibility with Java implementation."""
        # Test cases that verify ZigZag encoding matches Java's implementation
        # ZigZag mapping: 0->0, -1->1, 1->2, -2->3, 2->4, -3->5, 3->6, etc.
        zigzag_test_cases = [
            (0, 0),      # 0 -> 0
            (-1, 1),     # -1 -> 1
            (1, 2),      # 1 -> 2
            (-2, 3),     # -2 -> 3
            (2, 4),      # 2 -> 4
            (-3, 5),     # -3 -> 5
            (3, 6),      # 3 -> 6
            (-64, 127),  # -64 -> 127
            (64, 128),   # 64 -> 128
            (-65, 129),  # -65 -> 129
        ]

        for original_value, expected_zigzag in zigzag_test_cases:
            # Test single value compression to verify ZigZag encoding
            compressed = DeltaVarintCompressor.compress([original_value])
            decompressed = DeltaVarintCompressor.decompress(compressed)

            self.assertEqual([original_value], decompressed,
                             f"ZigZag encoding failed for value {original_value}")

    def test_java_compatibility_known_vectors(self):
        """Test with known test vectors that should match Java implementation."""
        # Test vectors with expected compressed output (hexadecimal)
        test_vectors = [
            # Simple cases
            ([0], "00"),                    # 0 -> ZigZag(0) = 0 -> Varint(0) = 0x00
            ([1], "02"),                    # 1 -> ZigZag(1) = 2 -> Varint(2) = 0x02
            ([-1], "01"),                   # -1 -> ZigZag(-1) = 1 -> Varint(1) = 0x01
            ([2], "04"),                    # 2 -> ZigZag(2) = 4 -> Varint(4) = 0x04
            ([-2], "03"),                   # -2 -> ZigZag(-2) = 3 -> Varint(3) = 0x03

            # Delta encoding cases
            ([0, 1], "0002"),               # [0, 1] -> [0, delta=1] -> [0x00, 0x02]
            ([1, 2], "0202"),               # [1, 2] -> [1, delta=1] -> [0x02, 0x02]
            ([0, -1], "0001"),              # [0, -1] -> [0, delta=-1] -> [0x00, 0x01]
            ([1, 0], "0201"),               # [1, 0] -> [1, delta=-1] -> [0x02, 0x01]

            # Larger values
            ([127], "fe01"),                # 127 -> ZigZag(127) = 254 -> Varint(254) = 0xfe01
            ([-127], "fd01"),               # -127 -> ZigZag(-127) = 253 -> Varint(253) = 0xfd01
            ([128], "8002"),                # 128 -> ZigZag(128) = 256 -> Varint(256) = 0x8002
            ([-128], "ff01"),               # -128 -> ZigZag(-128) = 255 -> Varint(255) = 0xff01
        ]

        for original, expected_hex in test_vectors:
            compressed = DeltaVarintCompressor.compress(original)
            actual_hex = compressed.hex()

            self.assertEqual(expected_hex, actual_hex,
                             f"Binary compatibility failed for {original}. "
                             f"Expected: {expected_hex}, Got: {actual_hex}")

            # Also verify round-trip
            decompressed = DeltaVarintCompressor.decompress(compressed)
            self.assertEqual(original, decompressed,
                             f"Round-trip failed for {original}")

    def test_java_compatibility_large_numbers(self):
        """Test compatibility with Java for large numbers (64-bit range)."""
        # Test cases covering the full 64-bit signed integer range
        large_number_cases = [
            2147483647,          # Integer.MAX_VALUE
            -2147483648,         # Integer.MIN_VALUE
            9223372036854775807,  # Long.MAX_VALUE
            -9223372036854775808 + 1,  # Long.MIN_VALUE + 1 (avoid overflow in Python)
            4294967295,          # 2^32 - 1
            -4294967296,         # -2^32
        ]

        for value in large_number_cases:
            # Test individual values
            compressed = DeltaVarintCompressor.compress([value])
            decompressed = DeltaVarintCompressor.decompress(compressed)
            self.assertEqual([value], decompressed,
                             f"Large number compatibility failed for {value}")

        # Test as a sequence to verify delta encoding with large numbers
        compressed_seq = DeltaVarintCompressor.compress(large_number_cases)
        decompressed_seq = DeltaVarintCompressor.decompress(compressed_seq)
        self.assertEqual(large_number_cases, decompressed_seq,
                         "Large number sequence compatibility failed")

    def test_java_compatibility_varint_boundaries(self):
        """Test Varint encoding boundaries that match Java implementation."""
        # Test values at Varint encoding boundaries
        varint_boundary_cases = [
            # 1-byte Varint boundary
            63,    # ZigZag(63) = 126, fits in 1 byte
            64,    # ZigZag(64) = 128, needs 2 bytes
            -64,   # ZigZag(-64) = 127, fits in 1 byte
            -65,   # ZigZag(-65) = 129, needs 2 bytes

            # 2-byte Varint boundary
            8191,   # ZigZag(8191) = 16382, fits in 2 bytes
            8192,   # ZigZag(8192) = 16384, needs 3 bytes
            -8192,  # ZigZag(-8192) = 16383, fits in 2 bytes
            -8193,  # ZigZag(-8193) = 16385, needs 3 bytes

            # 3-byte Varint boundary
            1048575,  # ZigZag(1048575) = 2097150, fits in 3 bytes
            1048576,  # ZigZag(1048576) = 2097152, needs 4 bytes
        ]

        for value in varint_boundary_cases:
            compressed = DeltaVarintCompressor.compress([value])
            decompressed = DeltaVarintCompressor.decompress(compressed)
            self.assertEqual([value], decompressed,
                             f"Varint boundary compatibility failed for {value}")

    def test_java_compatibility_delta_edge_cases(self):
        """Test delta encoding edge cases for Java compatibility."""
        # Edge cases that test delta encoding behavior
        delta_edge_cases = [
            # Maximum positive delta
            [0, sys.maxsize],
            # Maximum negative delta
            [sys.maxsize, 0],
            # Alternating large deltas
            [0, 1000000, -1000000, 2000000, -2000000],
            # Sequence with zero deltas
            [42, 42, 42, 42],
            # Mixed small and large deltas
            [0, 1, 1000000, 1000001, 0],
        ]

        for case in delta_edge_cases:
            compressed = DeltaVarintCompressor.compress(case)
            decompressed = DeltaVarintCompressor.decompress(compressed)
            self.assertEqual(case, decompressed,
                             f"Delta edge case compatibility failed for {case}")

    def test_java_compatibility_error_conditions(self):
        """Test error conditions that should match Java behavior."""
        # Test cases for error handling - our implementation gracefully handles
        # truncated data by returning empty lists, which is acceptable behavior

        # Test with various truncated/invalid byte sequences
        invalid_cases = [
            bytes([0x80]),           # Single incomplete byte
            bytes([0x80, 0x80]),     # Incomplete 3-byte varint
            bytes([0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x80]),  # Long sequence
        ]

        for invalid_data in invalid_cases:
            # Our implementation handles invalid data gracefully by returning empty list
            # This is acceptable behavior for robustness
            result = DeltaVarintCompressor.decompress(invalid_data)
            self.assertIsInstance(result, list,
                                  f"Should return a list for invalid data: {invalid_data.hex()}")
            # Empty result is acceptable for invalid/truncated data

        # Test that valid empty input returns empty list
        empty_result = DeltaVarintCompressor.decompress(b'')
        self.assertEqual([], empty_result, "Empty input should return empty list")


if __name__ == '__main__':
    unittest.main()