1 from __future__ import absolute_import

     2 

     3 import unittest

     4 

     5 from mercurial import (

     6     error,

     7     wireprotoframing as framing,

     8 )

     9 

    10 class SingleSendTests(unittest.TestCase):

    11     """A reactor that can only send once rejects subsequent sends."""

    12     def testbasic(self):

    13         reactor = framing.clientreactor(hasmultiplesend=False, buffersends=True)

    14 

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

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

    17         self.assertEqual(action, 'noop')

    18 

    19         action, meta = reactor.flushcommands()

    20         self.assertEqual(action, 'sendframes')

    21 

    22         for frame in meta['framegen']:

    23             self.assertEqual(request.state, 'sending')

    24 

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

    26 

    27         with self.assertRaisesRegexp(error.ProgrammingError,

    28                                      'cannot issue new commands'):

    29             reactor.callcommand(b'foo', {})

    30 

    31         with self.assertRaisesRegexp(error.ProgrammingError,

    32                                      'cannot issue new commands'):

    33             reactor.callcommand(b'foo', {})

    34 

    35 class NoBufferTests(unittest.TestCase):

    36     """A reactor without send buffering sends requests immediately."""

    37     def testbasic(self):

    38         reactor = framing.clientreactor(hasmultiplesend=True, buffersends=False)

    39 

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

    41         self.assertEqual(request.requestid, 1)

    42         self.assertEqual(action, 'sendframes')

    43 

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

    45 

    46         for frame in meta['framegen']:

    47             self.assertEqual(request.state, 'sending')

    48 

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

    50 

    51         action, meta = reactor.flushcommands()

    52         self.assertEqual(action, 'noop')

    53 

    54         # And we can send another command.

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

    56         self.assertEqual(request.requestid, 3)

    57         self.assertEqual(action, 'sendframes')

    58 

    59         for frame in meta['framegen']:

    60             self.assertEqual(request.state, 'sending')

    61 

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

    63 

    64 if __name__ == '__main__':

    65     import silenttestrunner

    66     silenttestrunner.main(__name__)