| hg-reviewers |
The latter is deprecated on Python 3.7 and causes our tests to fail
due to the warning.
| Automatic diff as part of commit; lint not applicable. |
| Automatic diff as part of commit; unit tests not applicable. |
| Path | Packages | |||
|---|---|---|---|---|
| M | tests/test-simplekeyvaluefile.py (14 lines) | |||
| M | tests/test-wireproto-clientreactor.py (19 lines) | |||
| M | tests/test-wireproto-framing.py (12 lines) | |||
| M | tests/test-wsgirequest.py (12 lines) |
| def testbasicwritingiandreading(self): | def testbasicwritingiandreading(self): | ||||
| dw = {'key1': 'value1', 'Key2': 'value2'} | dw = {'key1': 'value1', 'Key2': 'value2'} | ||||
| scmutil.simplekeyvaluefile(self.vfs, 'kvfile').write(dw) | scmutil.simplekeyvaluefile(self.vfs, 'kvfile').write(dw) | ||||
| self.assertEqual(sorted(self.vfs.read('kvfile').split('\n')), | self.assertEqual(sorted(self.vfs.read('kvfile').split('\n')), | ||||
| ['', 'Key2=value2', 'key1=value1']) | ['', 'Key2=value2', 'key1=value1']) | ||||
| dr = scmutil.simplekeyvaluefile(self.vfs, 'kvfile').read() | dr = scmutil.simplekeyvaluefile(self.vfs, 'kvfile').read() | ||||
| self.assertEqual(dr, dw) | self.assertEqual(dr, dw) | ||||
| 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 testinvalidkeys(self): | def testinvalidkeys(self): | ||||
| d = {'0key1': 'value1', 'Key2': 'value2'} | d = {'0key1': 'value1', 'Key2': 'value2'} | ||||
| with self.assertRaisesRegexp(error.ProgrammingError, | with self.assertRaisesRegex(error.ProgrammingError, | ||||
| 'keys must start with a letter.*'): | 'keys must start with a letter.*'): | ||||
| scmutil.simplekeyvaluefile(self.vfs, 'kvfile').write(d) | scmutil.simplekeyvaluefile(self.vfs, 'kvfile').write(d) | ||||
| d = {'key1@': 'value1', 'Key2': 'value2'} | d = {'key1@': 'value1', 'Key2': 'value2'} | ||||
| with self.assertRaisesRegexp(error.ProgrammingError, 'invalid key.*'): | with self.assertRaisesRegex(error.ProgrammingError, 'invalid key.*'): | ||||
| scmutil.simplekeyvaluefile(self.vfs, 'kvfile').write(d) | scmutil.simplekeyvaluefile(self.vfs, 'kvfile').write(d) | ||||
| def testinvalidvalues(self): | def testinvalidvalues(self): | ||||
| d = {'key1': 'value1', 'Key2': 'value2\n'} | d = {'key1': 'value1', 'Key2': 'value2\n'} | ||||
| with self.assertRaisesRegexp(error.ProgrammingError, 'invalid val.*'): | with self.assertRaisesRegex(error.ProgrammingError, 'invalid val.*'): | ||||
| scmutil.simplekeyvaluefile(self.vfs, 'kvfile').write(d) | scmutil.simplekeyvaluefile(self.vfs, 'kvfile').write(d) | ||||
| def testcorruptedfile(self): | def testcorruptedfile(self): | ||||
| self.vfs.contents['badfile'] = 'ababagalamaga\n' | self.vfs.contents['badfile'] = 'ababagalamaga\n' | ||||
| with self.assertRaisesRegexp(error.CorruptedState, | with self.assertRaisesRegex(error.CorruptedState, | ||||
| 'dictionary.*element.*'): | 'dictionary.*element.*'): | ||||
| scmutil.simplekeyvaluefile(self.vfs, 'badfile').read() | scmutil.simplekeyvaluefile(self.vfs, 'badfile').read() | ||||
| def testfirstline(self): | def testfirstline(self): | ||||
| dw = {'key1': 'value1'} | dw = {'key1': 'value1'} | ||||
| scmutil.simplekeyvaluefile(self.vfs, 'fl').write(dw, firstline='1.0') | scmutil.simplekeyvaluefile(self.vfs, 'fl').write(dw, firstline='1.0') | ||||
| self.assertEqual(self.vfs.read('fl'), '1.0\nkey1=value1\n') | self.assertEqual(self.vfs.read('fl'), '1.0\nkey1=value1\n') | ||||
| dr = scmutil.simplekeyvaluefile(self.vfs, 'fl')\ | dr = scmutil.simplekeyvaluefile(self.vfs, 'fl')\ | ||||
| .read(firstlinenonkeyval=True) | .read(firstlinenonkeyval=True) | ||||
| self.assertEqual(dr, {'__firstline': '1.0', 'key1': 'value1'}) | self.assertEqual(dr, {'__firstline': '1.0', 'key1': 'value1'}) | ||||
| if __name__ == "__main__": | if __name__ == "__main__": | ||||
| silenttestrunner.main(__name__) | silenttestrunner.main(__name__) | ||||
| header.streamid, | header.streamid, | ||||
| header.streamflags, | header.streamflags, | ||||
| header.typeid, | header.typeid, | ||||
| header.flags, | header.flags, | ||||
| payload)) | payload)) | ||||
| class SingleSendTests(unittest.TestCase): | class SingleSendTests(unittest.TestCase): | ||||
| """A reactor that can only send once rejects subsequent sends.""" | """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): | def testbasic(self): | ||||
| reactor = framing.clientreactor(hasmultiplesend=False, buffersends=True) | reactor = framing.clientreactor(hasmultiplesend=False, buffersends=True) | ||||
| request, action, meta = reactor.callcommand(b'foo', {}) | request, action, meta = reactor.callcommand(b'foo', {}) | ||||
| self.assertEqual(request.state, b'pending') | self.assertEqual(request.state, b'pending') | ||||
| self.assertEqual(action, b'noop') | self.assertEqual(action, b'noop') | ||||
| action, meta = reactor.flushcommands() | action, meta = reactor.flushcommands() | ||||
| self.assertEqual(action, b'sendframes') | self.assertEqual(action, b'sendframes') | ||||
| for frame in meta[b'framegen']: | for frame in meta[b'framegen']: | ||||
| self.assertEqual(request.state, b'sending') | self.assertEqual(request.state, b'sending') | ||||
| self.assertEqual(request.state, b'sent') | self.assertEqual(request.state, b'sent') | ||||
| with self.assertRaisesRegexp(error.ProgrammingError, | with self.assertRaisesRegex(error.ProgrammingError, | ||||
| 'cannot issue new commands'): | 'cannot issue new commands'): | ||||
| reactor.callcommand(b'foo', {}) | reactor.callcommand(b'foo', {}) | ||||
| with self.assertRaisesRegexp(error.ProgrammingError, | with self.assertRaisesRegex(error.ProgrammingError, | ||||
| 'cannot issue new commands'): | 'cannot issue new commands'): | ||||
| reactor.callcommand(b'foo', {}) | reactor.callcommand(b'foo', {}) | ||||
| class NoBufferTests(unittest.TestCase): | class NoBufferTests(unittest.TestCase): | ||||
| """A reactor without send buffering sends requests immediately.""" | """A reactor without send buffering sends requests immediately.""" | ||||
| def testbasic(self): | def testbasic(self): | ||||
| reactor = framing.clientreactor(hasmultiplesend=True, buffersends=False) | reactor = framing.clientreactor(hasmultiplesend=True, buffersends=False) | ||||
| self.assertEqual(action, b'sendframes') | self.assertEqual(action, b'sendframes') | ||||
| for frame in meta[b'framegen']: | for frame in meta[b'framegen']: | ||||
| self.assertEqual(request.state, b'sending') | self.assertEqual(request.state, b'sending') | ||||
| self.assertEqual(request.state, b'sent') | self.assertEqual(request.state, b'sent') | ||||
| class BadFrameRecvTests(unittest.TestCase): | 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): | def testoddstream(self): | ||||
| reactor = framing.clientreactor() | reactor = framing.clientreactor() | ||||
| action, meta = sendframe(reactor, ffs(b'1 1 0 1 0 foo')) | action, meta = sendframe(reactor, ffs(b'1 1 0 1 0 foo')) | ||||
| self.assertEqual(action, b'error') | self.assertEqual(action, b'error') | ||||
| self.assertEqual(meta[b'message'], | self.assertEqual(meta[b'message'], | ||||
| b'received frame with odd numbered stream ID: 1') | b'received frame with odd numbered stream ID: 1') | ||||
| def testunknownstream(self): | def testunknownstream(self): | ||||
| reactor = framing.clientreactor() | reactor = framing.clientreactor() | ||||
| action, meta = sendframe(reactor, ffs(b'1 0 0 1 0 foo')) | action, meta = sendframe(reactor, ffs(b'1 0 0 1 0 foo')) | ||||
| self.assertEqual(action, b'error') | self.assertEqual(action, b'error') | ||||
| self.assertEqual(meta[b'message'], | self.assertEqual(meta[b'message'], | ||||
| b'received frame on unknown stream without beginning ' | b'received frame on unknown stream without beginning ' | ||||
| b'of stream flag set') | b'of stream flag set') | ||||
| def testunhandledframetype(self): | def testunhandledframetype(self): | ||||
| reactor = framing.clientreactor(buffersends=False) | reactor = framing.clientreactor(buffersends=False) | ||||
| request, action, meta = reactor.callcommand(b'foo', {}) | request, action, meta = reactor.callcommand(b'foo', {}) | ||||
| for frame in meta[b'framegen']: | for frame in meta[b'framegen']: | ||||
| pass | pass | ||||
| with self.assertRaisesRegexp(error.ProgrammingError, | with self.assertRaisesRegex(error.ProgrammingError, | ||||
| 'unhandled frame type'): | 'unhandled frame type'): | ||||
| sendframe(reactor, ffs(b'1 0 stream-begin text-output 0 foo')) | sendframe(reactor, ffs(b'1 0 stream-begin text-output 0 foo')) | ||||
| class StreamTests(unittest.TestCase): | class StreamTests(unittest.TestCase): | ||||
| def testmultipleresponseframes(self): | def testmultipleresponseframes(self): | ||||
| reactor = framing.clientreactor(buffersends=False) | reactor = framing.clientreactor(buffersends=False) | ||||
| request, action, meta = reactor.callcommand(b'foo', {}) | request, action, meta = reactor.callcommand(b'foo', {}) | ||||
| self.assertEqual(frames, [ | self.assertEqual(frames, [ | ||||
| ffs(b'1 1 stream-begin command-request new|have-data ' | ffs(b'1 1 stream-begin command-request new|have-data ' | ||||
| b"cbor:{b'name': b'command', b'args': {b'key1': b'key1value', " | b"cbor:{b'name': b'command', b'args': {b'key1': b'key1value', " | ||||
| b"b'key2': b'key2value', b'key3': b'key3value'}}"), | b"b'key2': b'key2value', b'key3': b'key3value'}}"), | ||||
| ffs(b'1 1 0 command-data eos %s' % data.getvalue()), | ffs(b'1 1 0 command-data eos %s' % data.getvalue()), | ||||
| ]) | ]) | ||||
| 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 testtextoutputformattingstringtype(self): | def testtextoutputformattingstringtype(self): | ||||
| """Formatting string must be bytes.""" | """Formatting string must be bytes.""" | ||||
| with self.assertRaisesRegexp(ValueError, 'must use bytes formatting '): | with self.assertRaisesRegex(ValueError, 'must use bytes formatting '): | ||||
| list(framing.createtextoutputframe(None, 1, [ | list(framing.createtextoutputframe(None, 1, [ | ||||
| (b'foo'.decode('ascii'), [], [])])) | (b'foo'.decode('ascii'), [], [])])) | ||||
| def testtextoutputargumentbytes(self): | def testtextoutputargumentbytes(self): | ||||
| with self.assertRaisesRegexp(ValueError, 'must use bytes for argument'): | with self.assertRaisesRegex(ValueError, 'must use bytes for argument'): | ||||
| list(framing.createtextoutputframe(None, 1, [ | list(framing.createtextoutputframe(None, 1, [ | ||||
| (b'foo', [b'foo'.decode('ascii')], [])])) | (b'foo', [b'foo'.decode('ascii')], [])])) | ||||
| def testtextoutputlabelbytes(self): | def testtextoutputlabelbytes(self): | ||||
| with self.assertRaisesRegexp(ValueError, 'must use bytes for labels'): | with self.assertRaisesRegex(ValueError, 'must use bytes for labels'): | ||||
| list(framing.createtextoutputframe(None, 1, [ | list(framing.createtextoutputframe(None, 1, [ | ||||
| (b'foo', [], [b'foo'.decode('ascii')])])) | (b'foo', [], [b'foo'.decode('ascii')])])) | ||||
| def testtextoutput1simpleatom(self): | def testtextoutput1simpleatom(self): | ||||
| stream = framing.stream(1) | stream = framing.stream(1) | ||||
| val = list(framing.createtextoutputframe(stream, 1, [ | val = list(framing.createtextoutputframe(stream, 1, [ | ||||
| (b'foo', [], [])])) | (b'foo', [], [])])) | ||||
| self.assertEqual(r.url, b'http://hostserver/script/pathinfo') | self.assertEqual(r.url, b'http://hostserver/script/pathinfo') | ||||
| self.assertEqual(r.baseurl, b'http://hostserver') | self.assertEqual(r.baseurl, b'http://hostserver') | ||||
| self.assertEqual(r.advertisedurl, b'http://testserver/script/pathinfo') | self.assertEqual(r.advertisedurl, b'http://testserver/script/pathinfo') | ||||
| self.assertEqual(r.advertisedbaseurl, b'http://testserver') | self.assertEqual(r.advertisedbaseurl, b'http://testserver') | ||||
| self.assertEqual(r.apppath, b'/script') | self.assertEqual(r.apppath, b'/script') | ||||
| self.assertEqual(r.dispatchparts, [b'pathinfo']) | self.assertEqual(r.dispatchparts, [b'pathinfo']) | ||||
| self.assertEqual(r.dispatchpath, b'pathinfo') | self.assertEqual(r.dispatchpath, b'pathinfo') | ||||
| 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 testreponame(self): | def testreponame(self): | ||||
| """repository path components get stripped from URL.""" | """repository path components get stripped from URL.""" | ||||
| with self.assertRaisesRegexp(error.ProgrammingError, | with self.assertRaisesRegex(error.ProgrammingError, | ||||
| b'reponame requires PATH_INFO'): | b'reponame requires PATH_INFO'): | ||||
| parse(DEFAULT_ENV, reponame=b'repo') | parse(DEFAULT_ENV, reponame=b'repo') | ||||
| with self.assertRaisesRegexp(error.ProgrammingError, | with self.assertRaisesRegex(error.ProgrammingError, | ||||
| b'PATH_INFO does not begin with repo ' | b'PATH_INFO does not begin with repo ' | ||||
| b'name'): | b'name'): | ||||
| parse(DEFAULT_ENV, reponame=b'repo', extra={ | parse(DEFAULT_ENV, reponame=b'repo', extra={ | ||||
| r'PATH_INFO': r'/pathinfo', | r'PATH_INFO': r'/pathinfo', | ||||
| }) | }) | ||||
| with self.assertRaisesRegexp(error.ProgrammingError, | with self.assertRaisesRegex(error.ProgrammingError, | ||||
| b'reponame prefix of PATH_INFO'): | b'reponame prefix of PATH_INFO'): | ||||
| parse(DEFAULT_ENV, reponame=b'repo', extra={ | parse(DEFAULT_ENV, reponame=b'repo', extra={ | ||||
| r'PATH_INFO': r'/repoextra/path', | r'PATH_INFO': r'/repoextra/path', | ||||
| }) | }) | ||||
| r = parse(DEFAULT_ENV, reponame=b'repo', extra={ | r = parse(DEFAULT_ENV, reponame=b'repo', extra={ | ||||
| r'PATH_INFO': r'/repo/path1/path2', | r'PATH_INFO': r'/repo/path1/path2', | ||||
| }) | }) | ||||