from __future__ import absolute_import

import sys
import unittest
import zlib

from mercurial import (
    error,
    ui as uimod,
    wireprotoframing as framing,
)
from mercurial.utils import cborutil

try:
    from mercurial import zstd

    zstd.__version__
except ImportError:
    zstd = None

ffs = framing.makeframefromhumanstring

globalui = uimod.ui()


def sendframe(reactor, frame):
    """Send a frame bytearray to a reactor."""
    header = framing.parseheader(frame)
    payload = frame[framing.FRAME_HEADER_SIZE :]
    assert len(payload) == header.length

    return reactor.onframerecv(
        framing.frame(
            header.requestid,
            header.streamid,
            header.streamflags,
            header.typeid,
            header.flags,
            payload,
        )
    )


class SingleSendTests(unittest.TestCase):
    """A reactor that can only send once rejects subsequent sends."""

    if not getattr(unittest.TestCase, 'assertRaisesRegex', False):
        # Python 3.7 deprecates the regex*p* version, but 2.7 lacks
        # the regex version.
        assertRaisesRegex = (  # camelcase-required
            unittest.TestCase.assertRaisesRegexp
        )

    def testbasic(self):
        reactor = framing.clientreactor(
            globalui, hasmultiplesend=False, buffersends=True
        )

        request, action, meta = reactor.callcommand(b'foo', {})
        self.assertEqual(request.state, b'pending')
        self.assertEqual(action, b'noop')

        action, meta = reactor.flushcommands()
        self.assertEqual(action, b'sendframes')

        for frame in meta[b'framegen']:
            self.assertEqual(request.state, b'sending')

        self.assertEqual(request.state, b'sent')

        with self.assertRaisesRegex(
            error.ProgrammingError, 'cannot issue new commands'
        ):
            reactor.callcommand(b'foo', {})

        with self.assertRaisesRegex(
            error.ProgrammingError, 'cannot issue new commands'
        ):
            reactor.callcommand(b'foo', {})


class NoBufferTests(unittest.TestCase):
    """A reactor without send buffering sends requests immediately."""

    def testbasic(self):
        reactor = framing.clientreactor(
            globalui, hasmultiplesend=True, buffersends=False
        )

        request, action, meta = reactor.callcommand(b'command1', {})
        self.assertEqual(request.requestid, 1)
        self.assertEqual(action, b'sendframes')

        self.assertEqual(request.state, b'pending')

        for frame in meta[b'framegen']:
            self.assertEqual(request.state, b'sending')

        self.assertEqual(request.state, b'sent')

        action, meta = reactor.flushcommands()
        self.assertEqual(action, b'noop')

        # And we can send another command.
        request, action, meta = reactor.callcommand(b'command2', {})
        self.assertEqual(request.requestid, 3)
        self.assertEqual(action, b'sendframes')

        for frame in meta[b'framegen']:
            self.assertEqual(request.state, b'sending')

        self.assertEqual(request.state, b'sent')


class BadFrameRecvTests(unittest.TestCase):
    if not getattr(unittest.TestCase, 'assertRaisesRegex', False):
        # Python 3.7 deprecates the regex*p* version, but 2.7 lacks
        # the regex version.
        assertRaisesRegex = (  # camelcase-required
            unittest.TestCase.assertRaisesRegexp
        )

    def testoddstream(self):
        reactor = framing.clientreactor(globalui)

        action, meta = sendframe(reactor, ffs(b'1 1 0 1 0 foo'))
        self.assertEqual(action, b'error')
        self.assertEqual(
            meta[b'message'], b'received frame with odd numbered stream ID: 1'
        )

    def testunknownstream(self):
        reactor = framing.clientreactor(globalui)

        action, meta = sendframe(reactor, ffs(b'1 0 0 1 0 foo'))
        self.assertEqual(action, b'error')
        self.assertEqual(
            meta[b'message'],
            b'received frame on unknown stream without beginning '
            b'of stream flag set',
        )

    def testunhandledframetype(self):
        reactor = framing.clientreactor(globalui, buffersends=False)

        request, action, meta = reactor.callcommand(b'foo', {})
        for frame in meta[b'framegen']:
            pass

        with self.assertRaisesRegex(
            error.ProgrammingError, 'unhandled frame type'
        ):
            sendframe(reactor, ffs(b'1 0 stream-begin text-output 0 foo'))


class StreamTests(unittest.TestCase):
    def testmultipleresponseframes(self):
        reactor = framing.clientreactor(globalui, buffersends=False)

        request, action, meta = reactor.callcommand(b'foo', {})

        self.assertEqual(action, b'sendframes')
        for f in meta[b'framegen']:
            pass

        action, meta = sendframe(
            reactor,
            ffs(
                b'%d 0 stream-begin command-response 0 foo' % request.requestid
            ),
        )
        self.assertEqual(action, b'responsedata')

        action, meta = sendframe(
            reactor, ffs(b'%d 0 0 command-response eos bar' % request.requestid)
        )
        self.assertEqual(action, b'responsedata')


class RedirectTests(unittest.TestCase):
    def testredirect(self):
        reactor = framing.clientreactor(globalui, buffersends=False)

        redirect = {
            b'targets': [b'a', b'b'],
            b'hashes': [b'sha256'],
        }

        request, action, meta = reactor.callcommand(
            b'foo', {}, redirect=redirect
        )

        self.assertEqual(action, b'sendframes')

        frames = list(meta[b'framegen'])
        self.assertEqual(len(frames), 1)

        self.assertEqual(
            frames[0],
            ffs(
                b'1 1 stream-begin command-request new '
                b"cbor:{b'name': b'foo', "
                b"b'redirect': {b'targets': [b'a', b'b'], "
                b"b'hashes': [b'sha256']}}"
            ),
        )


class StreamSettingsTests(unittest.TestCase):
    def testnoflags(self):
        reactor = framing.clientreactor(globalui, buffersends=False)

        request, action, meta = reactor.callcommand(b'foo', {})
        for f in meta[b'framegen']:
            pass

        action, meta = sendframe(
            reactor, ffs(b'1 2 stream-begin stream-settings 0 ')
        )

        self.assertEqual(action, b'error')
        self.assertEqual(
            meta,
            {
                b'message': b'stream encoding settings frame must have '
                b'continuation or end of stream flag set',
            },
        )

    def testconflictflags(self):
        reactor = framing.clientreactor(globalui, buffersends=False)

        request, action, meta = reactor.callcommand(b'foo', {})
        for f in meta[b'framegen']:
            pass

        action, meta = sendframe(
            reactor, ffs(b'1 2 stream-begin stream-settings continuation|eos ')
        )

        self.assertEqual(action, b'error')
        self.assertEqual(
            meta,
            {
                b'message': b'stream encoding settings frame cannot have both '
                b'continuation and end of stream flags set',
            },
        )

    def testemptypayload(self):
        reactor = framing.clientreactor(globalui, buffersends=False)

        request, action, meta = reactor.callcommand(b'foo', {})
        for f in meta[b'framegen']:
            pass

        action, meta = sendframe(
            reactor, ffs(b'1 2 stream-begin stream-settings eos ')
        )

        self.assertEqual(action, b'error')
        self.assertEqual(
            meta,
            {
                b'message': b'stream encoding settings frame did not contain '
                b'CBOR data'
            },
        )

    def testbadcbor(self):
        reactor = framing.clientreactor(globalui, buffersends=False)

        request, action, meta = reactor.callcommand(b'foo', {})
        for f in meta[b'framegen']:
            pass

        action, meta = sendframe(
            reactor, ffs(b'1 2 stream-begin stream-settings eos badvalue')
        )

        self.assertEqual(action, b'error')

    def testsingleobject(self):
        reactor = framing.clientreactor(globalui, buffersends=False)

        request, action, meta = reactor.callcommand(b'foo', {})
        for f in meta[b'framegen']:
            pass

        action, meta = sendframe(
            reactor,
            ffs(b'1 2 stream-begin stream-settings eos cbor:b"identity"'),
        )

        self.assertEqual(action, b'noop')
        self.assertEqual(meta, {})

    def testmultipleobjects(self):
        reactor = framing.clientreactor(globalui, buffersends=False)

        request, action, meta = reactor.callcommand(b'foo', {})
        for f in meta[b'framegen']:
            pass

        data = b''.join(
            [
                b''.join(cborutil.streamencode(b'identity')),
                b''.join(cborutil.streamencode({b'foo', b'bar'})),
            ]
        )

        action, meta = sendframe(
            reactor, ffs(b'1 2 stream-begin stream-settings eos %s' % data)
        )

        self.assertEqual(action, b'error')
        self.assertEqual(
            meta,
            {
                b'message': b'error setting stream decoder: identity decoder '
                b'received unexpected additional values',
            },
        )

    def testmultipleframes(self):
        reactor = framing.clientreactor(globalui, buffersends=False)

        request, action, meta = reactor.callcommand(b'foo', {})
        for f in meta[b'framegen']:
            pass

        data = b''.join(cborutil.streamencode(b'identity'))

        action, meta = sendframe(
            reactor,
            ffs(
                b'1 2 stream-begin stream-settings continuation %s' % data[0:3]
            ),
        )

        self.assertEqual(action, b'noop')
        self.assertEqual(meta, {})

        action, meta = sendframe(
            reactor, ffs(b'1 2 0 stream-settings eos %s' % data[3:])
        )

        self.assertEqual(action, b'noop')
        self.assertEqual(meta, {})

    def testinvalidencoder(self):
        reactor = framing.clientreactor(globalui, buffersends=False)

        request, action, meta = reactor.callcommand(b'foo', {})
        for f in meta[b'framegen']:
            pass

        action, meta = sendframe(
            reactor,
            ffs(b'1 2 stream-begin stream-settings eos cbor:b"badvalue"'),
        )

        self.assertEqual(action, b'error')
        self.assertEqual(
            meta,
            {
                b'message': b'error setting stream decoder: unknown stream '
                b'decoder: badvalue',
            },
        )

    def testzlibencoding(self):
        reactor = framing.clientreactor(globalui, buffersends=False)

        request, action, meta = reactor.callcommand(b'foo', {})
        for f in meta[b'framegen']:
            pass

        action, meta = sendframe(
            reactor,
            ffs(
                b'%d 2 stream-begin stream-settings eos cbor:b"zlib"'
                % request.requestid
            ),
        )

        self.assertEqual(action, b'noop')
        self.assertEqual(meta, {})

        result = {
            b'status': b'ok',
        }
        encoded = b''.join(cborutil.streamencode(result))

        compressed = zlib.compress(encoded)
        self.assertEqual(zlib.decompress(compressed), encoded)

        action, meta = sendframe(
            reactor,
            ffs(
                b'%d 2 encoded command-response eos %s'
                % (request.requestid, compressed)
            ),
        )

        self.assertEqual(action, b'responsedata')
        self.assertEqual(meta[b'data'], encoded)

    def testzlibencodingsinglebyteframes(self):
        reactor = framing.clientreactor(globalui, buffersends=False)

        request, action, meta = reactor.callcommand(b'foo', {})
        for f in meta[b'framegen']:
            pass

        action, meta = sendframe(
            reactor,
            ffs(
                b'%d 2 stream-begin stream-settings eos cbor:b"zlib"'
                % request.requestid
            ),
        )

        self.assertEqual(action, b'noop')
        self.assertEqual(meta, {})

        result = {
            b'status': b'ok',
        }
        encoded = b''.join(cborutil.streamencode(result))

        compressed = zlib.compress(encoded)
        self.assertEqual(zlib.decompress(compressed), encoded)

        chunks = []

        for i in range(len(compressed)):
            char = compressed[i : i + 1]
            if char == b'\\':
                char = b'\\\\'
            action, meta = sendframe(
                reactor,
                ffs(
                    b'%d 2 encoded command-response continuation %s'
                    % (request.requestid, char)
                ),
            )

            self.assertEqual(action, b'responsedata')
            chunks.append(meta[b'data'])
            self.assertTrue(meta[b'expectmore'])
            self.assertFalse(meta[b'eos'])

        # zlib will have the full data decoded at this point, even though
        # we haven't flushed.
        self.assertEqual(b''.join(chunks), encoded)

        # End the stream for good measure.
        action, meta = sendframe(
            reactor,
            ffs(b'%d 2 stream-end command-response eos ' % request.requestid),
        )

        self.assertEqual(action, b'responsedata')
        self.assertEqual(meta[b'data'], b'')
        self.assertFalse(meta[b'expectmore'])
        self.assertTrue(meta[b'eos'])

    def testzlibmultipleresponses(self):
        # We feed in zlib compressed data on the same stream but belonging to
        # 2 different requests. This tests our flushing behavior.
        reactor = framing.clientreactor(
            globalui, buffersends=False, hasmultiplesend=True
        )

        request1, action, meta = reactor.callcommand(b'foo', {})
        for f in meta[b'framegen']:
            pass

        request2, action, meta = reactor.callcommand(b'foo', {})
        for f in meta[b'framegen']:
            pass

        outstream = framing.outputstream(2)
        outstream.setencoder(globalui, b'zlib')

        response1 = b''.join(
            cborutil.streamencode(
                {b'status': b'ok', b'extra': b'response1' * 10,}
            )
        )

        response2 = b''.join(
            cborutil.streamencode(
                {b'status': b'error', b'extra': b'response2' * 10,}
            )
        )

        action, meta = sendframe(
            reactor,
            ffs(
                b'%d 2 stream-begin stream-settings eos cbor:b"zlib"'
                % request1.requestid
            ),
        )

        self.assertEqual(action, b'noop')
        self.assertEqual(meta, {})

        # Feeding partial data in won't get anything useful out.
        action, meta = sendframe(
            reactor,
            ffs(
                b'%d 2 encoded command-response continuation %s'
                % (request1.requestid, outstream.encode(response1))
            ),
        )
        self.assertEqual(action, b'responsedata')
        self.assertEqual(meta[b'data'], b'')

        # But flushing data at both ends will get our original data.
        action, meta = sendframe(
            reactor,
            ffs(
                b'%d 2 encoded command-response eos %s'
                % (request1.requestid, outstream.flush())
            ),
        )
        self.assertEqual(action, b'responsedata')
        self.assertEqual(meta[b'data'], response1)

        # We should be able to reuse the compressor/decompressor for the
        # 2nd response.
        action, meta = sendframe(
            reactor,
            ffs(
                b'%d 2 encoded command-response continuation %s'
                % (request2.requestid, outstream.encode(response2))
            ),
        )
        self.assertEqual(action, b'responsedata')
        self.assertEqual(meta[b'data'], b'')

        action, meta = sendframe(
            reactor,
            ffs(
                b'%d 2 encoded command-response eos %s'
                % (request2.requestid, outstream.flush())
            ),
        )
        self.assertEqual(action, b'responsedata')
        self.assertEqual(meta[b'data'], response2)

    @unittest.skipUnless(zstd, 'zstd not available')
    def testzstd8mbencoding(self):
        reactor = framing.clientreactor(globalui, buffersends=False)

        request, action, meta = reactor.callcommand(b'foo', {})
        for f in meta[b'framegen']:
            pass

        action, meta = sendframe(
            reactor,
            ffs(
                b'%d 2 stream-begin stream-settings eos cbor:b"zstd-8mb"'
                % request.requestid
            ),
        )

        self.assertEqual(action, b'noop')
        self.assertEqual(meta, {})

        result = {
            b'status': b'ok',
        }
        encoded = b''.join(cborutil.streamencode(result))

        encoder = framing.zstd8mbencoder(globalui)
        compressed = encoder.encode(encoded) + encoder.finish()
        self.assertEqual(
            zstd.ZstdDecompressor().decompress(
                compressed, max_output_size=len(encoded)
            ),
            encoded,
        )

        action, meta = sendframe(
            reactor,
            ffs(
                b'%d 2 encoded command-response eos %s'
                % (request.requestid, compressed)
            ),
        )

        self.assertEqual(action, b'responsedata')
        self.assertEqual(meta[b'data'], encoded)

    @unittest.skipUnless(zstd, 'zstd not available')
    def testzstd8mbencodingsinglebyteframes(self):
        reactor = framing.clientreactor(globalui, buffersends=False)

        request, action, meta = reactor.callcommand(b'foo', {})
        for f in meta[b'framegen']:
            pass

        action, meta = sendframe(
            reactor,
            ffs(
                b'%d 2 stream-begin stream-settings eos cbor:b"zstd-8mb"'
                % request.requestid
            ),
        )

        self.assertEqual(action, b'noop')
        self.assertEqual(meta, {})

        result = {
            b'status': b'ok',
        }
        encoded = b''.join(cborutil.streamencode(result))

        compressed = zstd.ZstdCompressor().compress(encoded)
        self.assertEqual(
            zstd.ZstdDecompressor().decompress(compressed), encoded
        )

        chunks = []

        for i in range(len(compressed)):
            char = compressed[i : i + 1]
            if char == b'\\':
                char = b'\\\\'
            action, meta = sendframe(
                reactor,
                ffs(
                    b'%d 2 encoded command-response continuation %s'
                    % (request.requestid, char)
                ),
            )

            self.assertEqual(action, b'responsedata')
            chunks.append(meta[b'data'])
            self.assertTrue(meta[b'expectmore'])
            self.assertFalse(meta[b'eos'])

        # zstd decompressor will flush at frame boundaries.
        self.assertEqual(b''.join(chunks), encoded)

        # End the stream for good measure.
        action, meta = sendframe(
            reactor,
            ffs(b'%d 2 stream-end command-response eos ' % request.requestid),
        )

        self.assertEqual(action, b'responsedata')
        self.assertEqual(meta[b'data'], b'')
        self.assertFalse(meta[b'expectmore'])
        self.assertTrue(meta[b'eos'])

    @unittest.skipUnless(zstd, 'zstd not available')
    def testzstd8mbmultipleresponses(self):
        # We feed in zstd compressed data on the same stream but belonging to
        # 2 different requests. This tests our flushing behavior.
        reactor = framing.clientreactor(
            globalui, buffersends=False, hasmultiplesend=True
        )

        request1, action, meta = reactor.callcommand(b'foo', {})
        for f in meta[b'framegen']:
            pass

        request2, action, meta = reactor.callcommand(b'foo', {})
        for f in meta[b'framegen']:
            pass

        outstream = framing.outputstream(2)
        outstream.setencoder(globalui, b'zstd-8mb')

        response1 = b''.join(
            cborutil.streamencode(
                {b'status': b'ok', b'extra': b'response1' * 10,}
            )
        )

        response2 = b''.join(
            cborutil.streamencode(
                {b'status': b'error', b'extra': b'response2' * 10,}
            )
        )

        action, meta = sendframe(
            reactor,
            ffs(
                b'%d 2 stream-begin stream-settings eos cbor:b"zstd-8mb"'
                % request1.requestid
            ),
        )

        self.assertEqual(action, b'noop')
        self.assertEqual(meta, {})

        # Feeding partial data in won't get anything useful out.
        action, meta = sendframe(
            reactor,
            ffs(
                b'%d 2 encoded command-response continuation %s'
                % (request1.requestid, outstream.encode(response1))
            ),
        )
        self.assertEqual(action, b'responsedata')
        self.assertEqual(meta[b'data'], b'')

        # But flushing data at both ends will get our original data.
        action, meta = sendframe(
            reactor,
            ffs(
                b'%d 2 encoded command-response eos %s'
                % (request1.requestid, outstream.flush())
            ),
        )
        self.assertEqual(action, b'responsedata')
        self.assertEqual(meta[b'data'], response1)

        # We should be able to reuse the compressor/decompressor for the
        # 2nd response.
        action, meta = sendframe(
            reactor,
            ffs(
                b'%d 2 encoded command-response continuation %s'
                % (request2.requestid, outstream.encode(response2))
            ),
        )
        self.assertEqual(action, b'responsedata')
        self.assertEqual(meta[b'data'], b'')

        action, meta = sendframe(
            reactor,
            ffs(
                b'%d 2 encoded command-response eos %s'
                % (request2.requestid, outstream.flush())
            ),
        )
        self.assertEqual(action, b'responsedata')
        self.assertEqual(meta[b'data'], response2)


if __name__ == '__main__':
    if (3, 6, 0) <= sys.version_info < (3, 6, 4):
        # Python 3.6.0 through 3.6.3 inclusive shipped with
        # https://bugs.python.org/issue31825 and we can't run these
        # tests on those specific versions of Python. Sigh.
        sys.exit(80)
    import silenttestrunner

    silenttestrunner.main(__name__)