Mercurial > hg

comparison tests/test-wireproto-clientreactor.py @ 37715:1859b9a7ddef

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
cleanup: polyfill assertRaisesRegex so we can avoid assertRaisesRegexp The latter is deprecated on Python 3.7 and causes our tests to fail due to the warning. Differential Revision: https://phab.mercurial-scm.org/D3375
author Augie Fackler <augie@google.com>
date Sat, 14 Apr 2018 11:20:38 -0400
parents cb71e0f9ac6f
children deff7cf7eefd
comparison
equal deleted inserted replaced
37714:5dd71e9ae68a 37715:1859b9a7ddef
22 header.flags, 22 header.flags,
23 payload)) 23 payload))
24 24
25 class SingleSendTests(unittest.TestCase): 25 class SingleSendTests(unittest.TestCase):
26 """A reactor that can only send once rejects subsequent sends.""" 26 """A reactor that can only send once rejects subsequent sends."""
27
28 if not getattr(unittest.TestCase, 'assertRaisesRegex', False):
29 # Python 3.7 deprecates the regex*p* version, but 2.7 lacks
30 # the regex version.
31 assertRaisesRegex = (# camelcase-required
32 unittest.TestCase.assertRaisesRegexp)
33
27 def testbasic(self): 34 def testbasic(self):
28 reactor = framing.clientreactor(hasmultiplesend=False, buffersends=True) 35 reactor = framing.clientreactor(hasmultiplesend=False, buffersends=True)
29 36
30 request, action, meta = reactor.callcommand(b'foo', {}) 37 request, action, meta = reactor.callcommand(b'foo', {})
31 self.assertEqual(request.state, b'pending') 38 self.assertEqual(request.state, b'pending')
37 for frame in meta[b'framegen']: 44 for frame in meta[b'framegen']:
38 self.assertEqual(request.state, b'sending') 45 self.assertEqual(request.state, b'sending')
39 46
40 self.assertEqual(request.state, b'sent') 47 self.assertEqual(request.state, b'sent')
41 48
42 with self.assertRaisesRegexp(error.ProgrammingError, 49 with self.assertRaisesRegex(error.ProgrammingError,
43 'cannot issue new commands'): 50 'cannot issue new commands'):
44 reactor.callcommand(b'foo', {}) 51 reactor.callcommand(b'foo', {})
45 52
46 with self.assertRaisesRegexp(error.ProgrammingError, 53 with self.assertRaisesRegex(error.ProgrammingError,
47 'cannot issue new commands'): 54 'cannot issue new commands'):
48 reactor.callcommand(b'foo', {}) 55 reactor.callcommand(b'foo', {})
49 56
50 class NoBufferTests(unittest.TestCase): 57 class NoBufferTests(unittest.TestCase):
51 """A reactor without send buffering sends requests immediately.""" 58 """A reactor without send buffering sends requests immediately."""
75 self.assertEqual(request.state, b'sending') 82 self.assertEqual(request.state, b'sending')
76 83
77 self.assertEqual(request.state, b'sent') 84 self.assertEqual(request.state, b'sent')
78 85
79 class BadFrameRecvTests(unittest.TestCase): 86 class BadFrameRecvTests(unittest.TestCase):
87 if not getattr(unittest.TestCase, 'assertRaisesRegex', False):
88 # Python 3.7 deprecates the regex*p* version, but 2.7 lacks
89 # the regex version.
90 assertRaisesRegex = (# camelcase-required
91 unittest.TestCase.assertRaisesRegexp)
92
80 def testoddstream(self): 93 def testoddstream(self):
81 reactor = framing.clientreactor() 94 reactor = framing.clientreactor()
82 95
83 action, meta = sendframe(reactor, ffs(b'1 1 0 1 0 foo')) 96 action, meta = sendframe(reactor, ffs(b'1 1 0 1 0 foo'))
84 self.assertEqual(action, b'error') 97 self.assertEqual(action, b'error')
99 112
100 request, action, meta = reactor.callcommand(b'foo', {}) 113 request, action, meta = reactor.callcommand(b'foo', {})
101 for frame in meta[b'framegen']: 114 for frame in meta[b'framegen']:
102 pass 115 pass
103 116
104 with self.assertRaisesRegexp(error.ProgrammingError, 117 with self.assertRaisesRegex(error.ProgrammingError,
105 'unhandled frame type'): 118 'unhandled frame type'):
106 sendframe(reactor, ffs(b'1 0 stream-begin text-output 0 foo')) 119 sendframe(reactor, ffs(b'1 0 stream-begin text-output 0 foo'))
107 120
108 class StreamTests(unittest.TestCase): 121 class StreamTests(unittest.TestCase):
109 def testmultipleresponseframes(self): 122 def testmultipleresponseframes(self):