# Note: these tests are part of the self test, do not use or import pytest functionality here.
#       See borg.selftest for details. If you add/remove test methods, update SELFTEST_COUNT

from io import BytesIO

from ..chunker import ChunkerFixed, Chunker, get_chunker, buzhash, buzhash_update
from ..constants import *  # NOQA
from . import BaseTestCase


def cf(chunks):
    """chunk filter"""
    # this is to simplify testing: either return the data piece (bytes) or the hole length (int).
    def _cf(chunk):
        if chunk.meta['allocation'] == CH_DATA:
            assert len(chunk.data) == chunk.meta['size']
            return bytes(chunk.data)  # make sure we have bytes, not memoryview
        if chunk.meta['allocation'] in (CH_HOLE, CH_ALLOC):
            assert chunk.data is None
            return chunk.meta['size']
        assert False, "unexpected allocation value"
    return [_cf(chunk) for chunk in chunks]


class ChunkerFixedTestCase(BaseTestCase):

    def test_chunkify_just_blocks(self):
        data = b'foobar' * 1500
        chunker = ChunkerFixed(4096)
        parts = cf(chunker.chunkify(BytesIO(data)))
        self.assert_equal(parts, [data[0:4096], data[4096:8192], data[8192:]])

    def test_chunkify_header_and_blocks(self):
        data = b'foobar' * 1500
        chunker = ChunkerFixed(4096, 123)
        parts = cf(chunker.chunkify(BytesIO(data)))
        self.assert_equal(parts, [data[0:123], data[123:123+4096], data[123+4096:123+8192], data[123+8192:]])

    def test_chunkify_just_blocks_fmap_complete(self):
        data = b'foobar' * 1500
        chunker = ChunkerFixed(4096)
        fmap = [
            (0, 4096, True),
            (4096, 8192, True),
            (8192, 99999999, True),
        ]
        parts = cf(chunker.chunkify(BytesIO(data), fmap=fmap))
        self.assert_equal(parts, [data[0:4096], data[4096:8192], data[8192:]])

    def test_chunkify_header_and_blocks_fmap_complete(self):
        data = b'foobar' * 1500
        chunker = ChunkerFixed(4096, 123)
        fmap = [
            (0, 123, True),
            (123, 4096, True),
            (123+4096, 4096, True),
            (123+8192, 4096, True),
        ]
        parts = cf(chunker.chunkify(BytesIO(data), fmap=fmap))
        self.assert_equal(parts, [data[0:123], data[123:123+4096], data[123+4096:123+8192], data[123+8192:]])

    def test_chunkify_header_and_blocks_fmap_zeros(self):
        data = b'H' * 123 + b'_' * 4096 + b'X' * 4096 + b'_' * 4096
        chunker = ChunkerFixed(4096, 123)
        fmap = [
            (0, 123, True),
            (123, 4096, False),
            (123+4096, 4096, True),
            (123+8192, 4096, False),
        ]
        parts = cf(chunker.chunkify(BytesIO(data), fmap=fmap))
        # because we marked the '_' ranges as holes, we will get hole ranges instead!
        self.assert_equal(parts, [data[0:123], 4096, data[123+4096:123+8192], 4096])

    def test_chunkify_header_and_blocks_fmap_partial(self):
        data = b'H' * 123 + b'_' * 4096 + b'X' * 4096 + b'_' * 4096
        chunker = ChunkerFixed(4096, 123)
        fmap = [
            (0, 123, True),
            # (123, 4096, False),
            (123+4096, 4096, True),
            # (123+8192, 4096, False),
        ]
        parts = cf(chunker.chunkify(BytesIO(data), fmap=fmap))
        # because we left out the '_' ranges from the fmap, we will not get them at all!
        self.assert_equal(parts, [data[0:123], data[123+4096:123+8192]])


class ChunkerTestCase(BaseTestCase):

    def test_chunkify(self):
        data = b'0' * int(1.5 * (1 << CHUNK_MAX_EXP)) + b'Y'
        parts = cf(Chunker(0, 1, CHUNK_MAX_EXP, 2, 2).chunkify(BytesIO(data)))
        self.assert_equal(len(parts), 2)
        self.assert_equal(b''.join(parts), data)
        self.assert_equal(cf(Chunker(0, 1, CHUNK_MAX_EXP, 2, 2).chunkify(BytesIO(b''))), [])
        self.assert_equal(cf(Chunker(0, 1, CHUNK_MAX_EXP, 2, 2).chunkify(BytesIO(b'foobarboobaz' * 3))), [b'fooba', b'rboobaz', b'fooba', b'rboobaz', b'fooba', b'rboobaz'])
        self.assert_equal(cf(Chunker(1, 1, CHUNK_MAX_EXP, 2, 2).chunkify(BytesIO(b'foobarboobaz' * 3))), [b'fo', b'obarb', b'oob', b'azf', b'oobarb', b'oob', b'azf', b'oobarb', b'oobaz'])
        self.assert_equal(cf(Chunker(2, 1, CHUNK_MAX_EXP, 2, 2).chunkify(BytesIO(b'foobarboobaz' * 3))), [b'foob', b'ar', b'boobazfoob', b'ar', b'boobazfoob', b'ar', b'boobaz'])
        self.assert_equal(cf(Chunker(0, 2, CHUNK_MAX_EXP, 2, 3).chunkify(BytesIO(b'foobarboobaz' * 3))), [b'foobarboobaz' * 3])
        self.assert_equal(cf(Chunker(1, 2, CHUNK_MAX_EXP, 2, 3).chunkify(BytesIO(b'foobarboobaz' * 3))), [b'foobar', b'boobazfo', b'obar', b'boobazfo', b'obar', b'boobaz'])
        self.assert_equal(cf(Chunker(2, 2, CHUNK_MAX_EXP, 2, 3).chunkify(BytesIO(b'foobarboobaz' * 3))), [b'foob', b'arboobaz', b'foob', b'arboobaz', b'foob', b'arboobaz'])
        self.assert_equal(cf(Chunker(0, 3, CHUNK_MAX_EXP, 2, 3).chunkify(BytesIO(b'foobarboobaz' * 3))), [b'foobarboobaz' * 3])
        self.assert_equal(cf(Chunker(1, 3, CHUNK_MAX_EXP, 2, 3).chunkify(BytesIO(b'foobarboobaz' * 3))), [b'foobarbo', b'obazfoobar', b'boobazfo', b'obarboobaz'])
        self.assert_equal(cf(Chunker(2, 3, CHUNK_MAX_EXP, 2, 3).chunkify(BytesIO(b'foobarboobaz' * 3))), [b'foobarboobaz', b'foobarboobaz', b'foobarboobaz'])

    def test_buzhash(self):
        self.assert_equal(buzhash(b'abcdefghijklmnop', 0), 3795437769)
        self.assert_equal(buzhash(b'abcdefghijklmnop', 1), 3795400502)
        self.assert_equal(buzhash(b'abcdefghijklmnop', 1), buzhash_update(buzhash(b'Xabcdefghijklmno', 1), ord('X'), ord('p'), 16, 1))
        # Test with more than 31 bytes to make sure our barrel_shift macro works correctly
        self.assert_equal(buzhash(b'abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz', 0), 566521248)

    def test_small_reads(self):
        class SmallReadFile:
            input = b'a' * (20 + 1)

            def read(self, nbytes):
                self.input = self.input[:-1]
                return self.input[:1]

        chunker = get_chunker(*CHUNKER_PARAMS, seed=0)
        reconstructed = b''.join(cf(chunker.chunkify(SmallReadFile())))
        assert reconstructed == b'a' * 20