from __future__ import absolute_import, print_function

import unittest

from mercurial import (

    util,

    wireprotoframing as framing,

)

ffs = framing.makeframefromhumanstring

def makereactor(deferoutput=False):

    return framing.serverreactor(deferoutput=deferoutput)

def sendframes(reactor, gen):

    """Send a generator of frame bytearray to a reactor.

    Emits a generator of results from ``onframerecv()`` calls.

    """

    for frame in gen:

        payload = frame[framing.FRAME_HEADER_SIZE:]

        assert len(payload) == framelength

    """Generate frames to run a command and send them to a reactor."""

class FrameTests(unittest.TestCase):

    def testdataexactframesize(self):

        data = util.bytesio(b'x' * framing.DEFAULT_MAX_FRAME_SIZE)

        self.assertEqual(frames, [

        ])

    def testdatamultipleframes(self):

        data = util.bytesio(b'x' * (framing.DEFAULT_MAX_FRAME_SIZE + 1))

        self.assertEqual(frames, [

                b'x' * framing.DEFAULT_MAX_FRAME_SIZE)),

        ])

    def testargsanddata(self):

        data = util.bytesio(b'x' * 100)

            b'key1': b'key1value',

            b'key2': b'key2value',

            b'key3': b'key3value',

        }, data))

        self.assertEqual(frames, [

        ])

class ServerReactorTests(unittest.TestCase):

    def _sendsingleframe(self, reactor, s):

        results = list(sendframes(reactor, [ffs(s)]))

        self.assertEqual(len(results), 1)

        return results[0]

    def assertaction(self, res, expected):

        self.assertIsInstance(res, tuple)

        self.assertEqual(len(res), 2)

        self.assertIsInstance(res[1], dict)

        self.assertEqual(res[0], expected)

    def assertframesequal(self, frames, framestrings):

        expected = [ffs(s) for s in framestrings]

        self.assertEqual(list(frames), expected)

    def test1framecommand(self):

        """Receiving a command in a single frame yields request to run it."""

        reactor = makereactor()

        self.assertEqual(len(results), 1)

        self.assertaction(results[0], 'runcommand')

        self.assertEqual(results[0][1], {

            'command': b'mycommand',

            'args': {},

            'data': None,

        })

        result = reactor.oninputeof()

        self.assertaction(result, 'noop')

    def test1argument(self):

        reactor = makereactor()

                                         {b'foo': b'bar'}))

        self.assertEqual(len(results), 2)

        self.assertaction(results[0], 'wantframe')

        self.assertaction(results[1], 'runcommand')

        self.assertEqual(results[1][1], {

            'command': b'mycommand',

            'args': {b'foo': b'bar'},

            'data': None,

        })

    def testmultiarguments(self):

        reactor = makereactor()

                                         {b'foo': b'bar', b'biz': b'baz'}))

        self.assertEqual(len(results), 3)

        self.assertaction(results[0], 'wantframe')

        self.assertaction(results[1], 'wantframe')

        self.assertaction(results[2], 'runcommand')

        self.assertEqual(results[2][1], {

            'command': b'mycommand',

            'args': {b'foo': b'bar', b'biz': b'baz'},

            'data': None,

        })

    def testsimplecommanddata(self):

        reactor = makereactor()

                                         util.bytesio(b'data!')))

        self.assertEqual(len(results), 2)

        self.assertaction(results[0], 'wantframe')

        self.assertaction(results[1], 'runcommand')

        self.assertEqual(results[1][1], {

            'command': b'mycommand',

            'args': {},

            'data': b'data!',

        })

    def testmultipledataframes(self):

        frames = [

        ]

        reactor = makereactor()

        results = list(sendframes(reactor, frames))

        self.assertEqual(len(results), 4)

        for i in range(3):

            self.assertaction(results[i], 'wantframe')

        self.assertaction(results[3], 'runcommand')

        self.assertEqual(results[3][1], {

            'command': b'mycommand',

            'args': {},

            'data': b'data1data2data3',

        })

    def testargumentanddata(self):

        frames = [

        ]

        reactor = makereactor()

        results = list(sendframes(reactor, frames))

        self.assertaction(results[-1], 'runcommand')

        self.assertEqual(results[-1][1], {

            'command': b'command',

            'args': {

                b'key': b'val',

                b'foo': b'bar',

            },

            'data': b'value1value2',

        })

    def testunexpectedcommandargument(self):

        """Command argument frame when not running a command is an error."""

        result = self._sendsingleframe(makereactor(),

        self.assertaction(result, 'error')

        self.assertEqual(result[1], {

            'message': b'expected command frame; got 2',

        })

    def testunexpectedcommanddata(self):

        """Command argument frame when not running a command is an error."""

        result = self._sendsingleframe(makereactor(),

        self.assertaction(result, 'error')

        self.assertEqual(result[1], {

            'message': b'expected command frame; got 3',

        })

    def testmissingcommandframeflags(self):

        """Command name frame must have flags set."""

        result = self._sendsingleframe(makereactor(),

        self.assertaction(result, 'error')

        self.assertEqual(result[1], {

            'message': b'missing frame flags on command frame',

        })

    def testmissingargumentframe(self):

        frames = [

        ]

        results = list(sendframes(makereactor(), frames))

        self.assertEqual(len(results), 2)

        self.assertaction(results[0], 'wantframe')

        self.assertaction(results[1], 'error')

        self.assertEqual(results[1][1], {

            'message': b'expected command argument frame; got 1',

        })

    def testincompleteargumentname(self):

        """Argument frame with incomplete name."""

        frames = [

        ]

        results = list(sendframes(makereactor(), frames))

        self.assertEqual(len(results), 2)

        self.assertaction(results[0], 'wantframe')

        self.assertaction(results[1], 'error')

        self.assertEqual(results[1][1], {

            'message': b'malformed argument frame: partial argument name',

        })

    def testincompleteargumentvalue(self):

        """Argument frame with incomplete value."""

        frames = [

        ]

        results = list(sendframes(makereactor(), frames))

        self.assertEqual(len(results), 2)

        self.assertaction(results[0], 'wantframe')

        self.assertaction(results[1], 'error')

        self.assertEqual(results[1][1], {

            'message': b'malformed argument frame: partial argument value',

        })

    def testmissingcommanddataframe(self):

        frames = [

        ]

        results = list(sendframes(makereactor(), frames))

        self.assertEqual(len(results), 2)

        self.assertaction(results[0], 'wantframe')

        self.assertaction(results[1], 'error')

        self.assertEqual(results[1][1], {

            'message': b'expected command data frame; got 1',

        })

    def testmissingcommanddataframeflags(self):

        frames = [

        ]

        results = list(sendframes(makereactor(), frames))

        self.assertEqual(len(results), 2)

        self.assertaction(results[0], 'wantframe')

        self.assertaction(results[1], 'error')

        self.assertEqual(results[1][1], {

            'message': b'command data frame without flags',

        })

    def testsimpleresponse(self):

        """Bytes response to command sends result frames."""

        reactor = makereactor()

        self.assertaction(result, 'sendframes')

        self.assertframesequal(result[1]['framegen'], [

        ])

    def testmultiframeresponse(self):

        """Bytes response spanning multiple frames is handled."""

        first = b'x' * framing.DEFAULT_MAX_FRAME_SIZE

        second = b'y' * 100

        reactor = makereactor()

        self.assertaction(result, 'sendframes')

        self.assertframesequal(result[1]['framegen'], [

        ])

    def testapplicationerror(self):

        reactor = makereactor()

        self.assertaction(result, 'sendframes')

        self.assertframesequal(result[1]['framegen'], [

        ])

    def test1commanddeferresponse(self):

        """Responses when in deferred output mode are delayed until EOF."""

        reactor = makereactor(deferoutput=True)

        self.assertEqual(len(results), 1)

        self.assertaction(results[0], 'runcommand')

        self.assertaction(result, 'noop')

        result = reactor.oninputeof()

        self.assertaction(result, 'sendframes')

        self.assertframesequal(result[1]['framegen'], [

        ])

    def testmultiplecommanddeferresponse(self):

        reactor = makereactor(deferoutput=True)

        self.assertaction(result, 'noop')

        self.assertaction(result, 'noop')

        result = reactor.oninputeof()

        self.assertaction(result, 'sendframes')

        self.assertframesequal(result[1]['framegen'], [

        ])

if __name__ == '__main__':