# storage.py - Testing of storage primitives.

#

# Copyright 2018 Gregory Szorc <gregory.szorc@gmail.com>

#

# This software may be used and distributed according to the terms of the

# GNU General Public License version 2 or any later version.

from __future__ import absolute_import

import unittest

from ..node import (

    hex,

    nullid,

    nullrev,

)

from .. import (

    error,

    mdiff,

    revlog,

)

class basetestcase(unittest.TestCase):

    if not getattr(unittest.TestCase, r'assertRaisesRegex', False):

        assertRaisesRegex = (# camelcase-required

            unittest.TestCase.assertRaisesRegexp)

class ifileindextests(basetestcase):

    """Generic tests for the ifileindex interface.

    All file storage backends for index data should conform to the tests in this

    class.

    Use ``makeifileindextests()`` to create an instance of this type.

    """

    def testempty(self):

        f = self._makefilefn()

        self.assertEqual(len(f), 0, 'new file store has 0 length by default')

        self.assertEqual(list(f), [], 'iter yields nothing by default')

        gen = iter(f)

        with self.assertRaises(StopIteration):

            next(gen)

        # revs() should evaluate to an empty list.

        self.assertEqual(list(f.revs()), [])

        revs = iter(f.revs())

        with self.assertRaises(StopIteration):

            next(revs)

        self.assertEqual(list(f.revs(start=20)), [])

        # parents() and parentrevs() work with nullid/nullrev.

        self.assertEqual(f.parents(nullid), (nullid, nullid))

        self.assertEqual(f.parentrevs(nullrev), (nullrev, nullrev))

        with self.assertRaises(error.LookupError):

            f.parents(b'\x01' * 20)

        for i in range(-5, 5):

            if i == nullrev:

                continue

            with self.assertRaises(IndexError):

                f.parentrevs(i)

        # nullid/nullrev lookup always works.

        self.assertEqual(f.rev(nullid), nullrev)

        self.assertEqual(f.node(nullrev), nullid)

        with self.assertRaises(error.LookupError):

            f.rev(b'\x01' * 20)

        for i in range(-5, 5):

            if i == nullrev:

                continue

            with self.assertRaises(IndexError):

                f.node(i)

        self.assertEqual(f.lookup(nullid), nullid)

        self.assertEqual(f.lookup(nullrev), nullid)

        self.assertEqual(f.lookup(hex(nullid)), nullid)

        # String converted to integer doesn't work for nullrev.

        with self.assertRaises(error.LookupError):

            f.lookup(b'%d' % nullrev)

        self.assertEqual(f.linkrev(nullrev), nullrev)

        for i in range(-5, 5):

            if i == nullrev:

                continue

            with self.assertRaises(IndexError):

                f.linkrev(i)

        self.assertFalse(f.iscensored(nullrev))

        for i in range(-5, 5):

            if i == nullrev:

                continue

            with self.assertRaises(IndexError):

                f.iscensored(i)

        self.assertEqual(list(f.commonancestorsheads(nullid, nullid)), [])

        with self.assertRaises(ValueError):

            self.assertEqual(list(f.descendants([])), [])

        self.assertEqual(list(f.descendants([nullrev])), [])

        self.assertEqual(f.heads(), [nullid])

        self.assertEqual(f.heads(nullid), [nullid])

        self.assertEqual(f.heads(None, [nullid]), [nullid])

        self.assertEqual(f.heads(nullid, [nullid]), [nullid])

        self.assertEqual(f.children(nullid), [])

        with self.assertRaises(error.LookupError):

            f.children(b'\x01' * 20)

    def testsinglerevision(self):

        f = self._makefilefn()

        with self._maketransactionfn() as tr:

            node = f.add(b'initial', None, tr, 0, nullid, nullid)

        self.assertEqual(len(f), 1)

        self.assertEqual(list(f), [0])

        gen = iter(f)

        self.assertEqual(next(gen), 0)

        with self.assertRaises(StopIteration):

            next(gen)

        self.assertEqual(list(f.revs()), [0])

        self.assertEqual(list(f.revs(start=1)), [])

        self.assertEqual(list(f.revs(start=0)), [0])

        self.assertEqual(list(f.revs(stop=0)), [0])

        self.assertEqual(list(f.revs(stop=1)), [0])

        self.assertEqual(list(f.revs(1, 1)), [])

        # TODO buggy

        self.assertEqual(list(f.revs(1, 0)), [1, 0])

        self.assertEqual(list(f.revs(2, 0)), [2, 1, 0])

        self.assertEqual(f.parents(node), (nullid, nullid))

        self.assertEqual(f.parentrevs(0), (nullrev, nullrev))

        with self.assertRaises(error.LookupError):

            f.parents(b'\x01' * 20)

        with self.assertRaises(IndexError):

            f.parentrevs(1)

        self.assertEqual(f.rev(node), 0)

        with self.assertRaises(error.LookupError):

            f.rev(b'\x01' * 20)

        self.assertEqual(f.node(0), node)

        with self.assertRaises(IndexError):

            f.node(1)

        self.assertEqual(f.lookup(node), node)

        self.assertEqual(f.lookup(0), node)

        self.assertEqual(f.lookup(b'0'), node)

        self.assertEqual(f.lookup(hex(node)), node)

        self.assertEqual(f.linkrev(0), 0)

        with self.assertRaises(IndexError):

            f.linkrev(1)

        self.assertFalse(f.iscensored(0))

        with self.assertRaises(IndexError):

            f.iscensored(1)

        self.assertEqual(list(f.descendants([0])), [])

        self.assertEqual(f.heads(), [node])

        self.assertEqual(f.heads(node), [node])

        self.assertEqual(f.heads(stop=[node]), [node])

        with self.assertRaises(error.LookupError):

            f.heads(stop=[b'\x01' * 20])

        self.assertEqual(f.children(node), [])

    def testmultiplerevisions(self):

        fulltext0 = b'x' * 1024

        fulltext1 = fulltext0 + b'y'

        fulltext2 = b'y' + fulltext0 + b'z'

        f = self._makefilefn()

        with self._maketransactionfn() as tr:

            node0 = f.add(fulltext0, None, tr, 0, nullid, nullid)

            node1 = f.add(fulltext1, None, tr, 1, node0, nullid)

            node2 = f.add(fulltext2, None, tr, 3, node1, nullid)

        self.assertEqual(len(f), 3)

        self.assertEqual(list(f), [0, 1, 2])

        gen = iter(f)

        self.assertEqual(next(gen), 0)

        self.assertEqual(next(gen), 1)

        self.assertEqual(next(gen), 2)

        with self.assertRaises(StopIteration):

            next(gen)

        self.assertEqual(list(f.revs()), [0, 1, 2])

        self.assertEqual(list(f.revs(0)), [0, 1, 2])

        self.assertEqual(list(f.revs(1)), [1, 2])

        self.assertEqual(list(f.revs(2)), [2])

        self.assertEqual(list(f.revs(3)), [])

        self.assertEqual(list(f.revs(stop=1)), [0, 1])

        self.assertEqual(list(f.revs(stop=2)), [0, 1, 2])

        self.assertEqual(list(f.revs(stop=3)), [0, 1, 2])

        self.assertEqual(list(f.revs(2, 0)), [2, 1, 0])

        self.assertEqual(list(f.revs(2, 1)), [2, 1])

        # TODO this is wrong

        self.assertEqual(list(f.revs(3, 2)), [3, 2])

        self.assertEqual(f.parents(node0), (nullid, nullid))

        self.assertEqual(f.parents(node1), (node0, nullid))

        self.assertEqual(f.parents(node2), (node1, nullid))

        self.assertEqual(f.parentrevs(0), (nullrev, nullrev))

        self.assertEqual(f.parentrevs(1), (0, nullrev))

        self.assertEqual(f.parentrevs(2), (1, nullrev))

        self.assertEqual(f.rev(node0), 0)

        self.assertEqual(f.rev(node1), 1)

        self.assertEqual(f.rev(node2), 2)

        with self.assertRaises(error.LookupError):

            f.rev(b'\x01' * 20)

        self.assertEqual(f.node(0), node0)

        self.assertEqual(f.node(1), node1)

        self.assertEqual(f.node(2), node2)

        with self.assertRaises(IndexError):

            f.node(3)

        self.assertEqual(f.lookup(node0), node0)

        self.assertEqual(f.lookup(0), node0)

        self.assertEqual(f.lookup(b'0'), node0)

        self.assertEqual(f.lookup(hex(node0)), node0)

        self.assertEqual(f.lookup(node1), node1)

        self.assertEqual(f.lookup(1), node1)

        self.assertEqual(f.lookup(b'1'), node1)

        self.assertEqual(f.lookup(hex(node1)), node1)

        self.assertEqual(f.linkrev(0), 0)

        self.assertEqual(f.linkrev(1), 1)

        self.assertEqual(f.linkrev(2), 3)

        with self.assertRaises(IndexError):

            f.linkrev(3)

        self.assertFalse(f.iscensored(0))

        self.assertFalse(f.iscensored(1))

        self.assertFalse(f.iscensored(2))

        with self.assertRaises(IndexError):

            f.iscensored(3)

        self.assertEqual(f.commonancestorsheads(node1, nullid), [])

        self.assertEqual(f.commonancestorsheads(node1, node0), [node0])

        self.assertEqual(f.commonancestorsheads(node1, node1), [node1])

        self.assertEqual(f.commonancestorsheads(node0, node1), [node0])

        self.assertEqual(f.commonancestorsheads(node1, node2), [node1])

        self.assertEqual(f.commonancestorsheads(node2, node1), [node1])

        self.assertEqual(list(f.descendants([0])), [1, 2])

        self.assertEqual(list(f.descendants([1])), [2])

        self.assertEqual(list(f.descendants([0, 1])), [1, 2])

        self.assertEqual(f.heads(), [node2])

        self.assertEqual(f.heads(node0), [node2])

        self.assertEqual(f.heads(node1), [node2])

        self.assertEqual(f.heads(node2), [node2])

        # TODO this behavior seems wonky. Is it correct? If so, the

        # docstring for heads() should be updated to reflect desired

        # behavior.

        self.assertEqual(f.heads(stop=[node1]), [node1, node2])

        self.assertEqual(f.heads(stop=[node0]), [node0, node2])

        self.assertEqual(f.heads(stop=[node1, node2]), [node1, node2])

        with self.assertRaises(error.LookupError):

            f.heads(stop=[b'\x01' * 20])

        self.assertEqual(f.children(node0), [node1])

        self.assertEqual(f.children(node1), [node2])

        self.assertEqual(f.children(node2), [])

    def testmultipleheads(self):

        f = self._makefilefn()

        with self._maketransactionfn() as tr:

            node0 = f.add(b'0', None, tr, 0, nullid, nullid)

            node1 = f.add(b'1', None, tr, 1, node0, nullid)

            node2 = f.add(b'2', None, tr, 2, node1, nullid)

            node3 = f.add(b'3', None, tr, 3, node0, nullid)

            node4 = f.add(b'4', None, tr, 4, node3, nullid)

            node5 = f.add(b'5', None, tr, 5, node0, nullid)

        self.assertEqual(len(f), 6)

        self.assertEqual(list(f.descendants([0])), [1, 2, 3, 4, 5])

        self.assertEqual(list(f.descendants([1])), [2])

        self.assertEqual(list(f.descendants([2])), [])

        self.assertEqual(list(f.descendants([3])), [4])

        self.assertEqual(list(f.descendants([0, 1])), [1, 2, 3, 4, 5])

        self.assertEqual(list(f.descendants([1, 3])), [2, 4])

        self.assertEqual(f.heads(), [node2, node4, node5])

        self.assertEqual(f.heads(node0), [node2, node4, node5])

        self.assertEqual(f.heads(node1), [node2])

        self.assertEqual(f.heads(node2), [node2])

        self.assertEqual(f.heads(node3), [node4])

        self.assertEqual(f.heads(node4), [node4])

        self.assertEqual(f.heads(node5), [node5])

        # TODO this seems wrong.

        self.assertEqual(f.heads(stop=[node0]), [node0, node2, node4, node5])

        self.assertEqual(f.heads(stop=[node1]), [node1, node2, node4, node5])

        self.assertEqual(f.children(node0), [node1, node3, node5])

        self.assertEqual(f.children(node1), [node2])

        self.assertEqual(f.children(node2), [])

        self.assertEqual(f.children(node3), [node4])

        self.assertEqual(f.children(node4), [])

        self.assertEqual(f.children(node5), [])

class ifiledatatests(basetestcase):

    """Generic tests for the ifiledata interface.

    All file storage backends for data should conform to the tests in this

    class.

    Use ``makeifiledatatests()`` to create an instance of this type.

    """

    def testempty(self):

        f = self._makefilefn()

        self.assertEqual(f.storageinfo(), {})

        self.assertEqual(f.storageinfo(revisionscount=True, trackedsize=True),

                         {'revisionscount': 0, 'trackedsize': 0})

        self.assertEqual(f.size(nullrev), 0)

        for i in range(-5, 5):

            if i == nullrev:

                continue

            with self.assertRaises(IndexError):

                f.size(i)

        with self.assertRaises(error.StorageError):

            f.checkhash(b'', nullid)

        with self.assertRaises(error.LookupError):

            f.checkhash(b'', b'\x01' * 20)

        self.assertEqual(f.revision(nullid), b'')

        self.assertEqual(f.revision(nullid, raw=True), b'')

        with self.assertRaises(error.LookupError):

            f.revision(b'\x01' * 20)

        self.assertEqual(f.read(nullid), b'')

        with self.assertRaises(error.LookupError):

            f.read(b'\x01' * 20)

        self.assertFalse(f.renamed(nullid))

        with self.assertRaises(error.LookupError):

            f.read(b'\x01' * 20)

        self.assertTrue(f.cmp(nullid, b''))

        self.assertTrue(f.cmp(nullid, b'foo'))

        with self.assertRaises(error.LookupError):

            f.cmp(b'\x01' * 20, b'irrelevant')

        self.assertEqual(f.revdiff(nullrev, nullrev), b'')

        with self.assertRaises(IndexError):

            f.revdiff(0, nullrev)

        with self.assertRaises(IndexError):

            f.revdiff(nullrev, 0)

        with self.assertRaises(IndexError):

            f.revdiff(0, 0)

        # Emitting empty list is an empty generator.

        gen = f.emitrevisions([])

        with self.assertRaises(StopIteration):

            next(gen)

        # Emitting null node yields nothing.

        gen = f.emitrevisions([nullid])

        with self.assertRaises(StopIteration):

            next(gen)

        # Requesting unknown node fails.

        with self.assertRaises(error.LookupError):

            list(f.emitrevisions([b'\x01' * 20]))

    def testsinglerevision(self):

        fulltext = b'initial'

        f = self._makefilefn()

        with self._maketransactionfn() as tr:

            node = f.add(fulltext, None, tr, 0, nullid, nullid)

        self.assertEqual(f.storageinfo(), {})

        self.assertEqual(f.storageinfo(revisionscount=True, trackedsize=True),

                         {'revisionscount': 1, 'trackedsize': len(fulltext)})

        self.assertEqual(f.size(0), len(fulltext))

        with self.assertRaises(IndexError):

            f.size(1)

        f.checkhash(fulltext, node)

        f.checkhash(fulltext, node, nullid, nullid)

        with self.assertRaises(error.StorageError):

            f.checkhash(fulltext + b'extra', node)

        with self.assertRaises(error.StorageError):

            f.checkhash(fulltext, node, b'\x01' * 20, nullid)

        with self.assertRaises(error.StorageError):

            f.checkhash(fulltext, node, nullid, b'\x01' * 20)

        self.assertEqual(f.revision(node), fulltext)

        self.assertEqual(f.revision(node, raw=True), fulltext)

        self.assertEqual(f.read(node), fulltext)

        self.assertFalse(f.renamed(node))

        self.assertFalse(f.cmp(node, fulltext))

        self.assertTrue(f.cmp(node, fulltext + b'extra'))

        self.assertEqual(f.revdiff(0, 0), b'')

        self.assertEqual(f.revdiff(nullrev, 0),

                         b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x07%s' %

                         fulltext)

        self.assertEqual(f.revdiff(0, nullrev),

                         b'\x00\x00\x00\x00\x00\x00\x00\x07\x00\x00\x00\x00')

        # Emitting a single revision works.

        gen = f.emitrevisions([node])

        rev = next(gen)

        self.assertEqual(rev.node, node)

        self.assertEqual(rev.p1node, nullid)

        self.assertEqual(rev.p2node, nullid)

        self.assertIsNone(rev.linknode)

        self.assertEqual(rev.basenode, nullid)

        self.assertIsNone(rev.baserevisionsize)

        self.assertIsNone(rev.revision)

        self.assertIsNone(rev.delta)

        with self.assertRaises(StopIteration):

            next(gen)

        # Requesting revision data works.

        gen = f.emitrevisions([node], revisiondata=True)

        rev = next(gen)

        self.assertEqual(rev.node, node)

        self.assertEqual(rev.p1node, nullid)

        self.assertEqual(rev.p2node, nullid)

        self.assertIsNone(rev.linknode)

        self.assertEqual(rev.basenode, nullid)

        self.assertIsNone(rev.baserevisionsize)

        self.assertEqual(rev.revision, fulltext)

        self.assertIsNone(rev.delta)

        with self.assertRaises(StopIteration):

            next(gen)

        # Emitting an unknown node after a known revision results in error.

        with self.assertRaises(error.LookupError):

            list(f.emitrevisions([node, b'\x01' * 20]))

    def testmultiplerevisions(self):

        fulltext0 = b'x' * 1024

        fulltext1 = fulltext0 + b'y'

        fulltext2 = b'y' + fulltext0 + b'z'

        f = self._makefilefn()

        with self._maketransactionfn() as tr:

            node0 = f.add(fulltext0, None, tr, 0, nullid, nullid)

            node1 = f.add(fulltext1, None, tr, 1, node0, nullid)

            node2 = f.add(fulltext2, None, tr, 3, node1, nullid)

        self.assertEqual(f.storageinfo(), {})

        self.assertEqual(

            f.storageinfo(revisionscount=True, trackedsize=True),

            {

                'revisionscount': 3,

                'trackedsize': len(fulltext0) + len(fulltext1) + len(fulltext2),

            })

        self.assertEqual(f.size(0), len(fulltext0))