← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_smart_transport.py

  • Committer: John Arbash Meinel
  • Date: 2011-05-11 11:35:28 UTC
  • mto: This revision was merged to the branch mainline in revision 5851.
  • Revision ID: john@arbash-meinel.com-20110511113528-qepibuwxicjrbb2h
Break compatibility with python <2.6.

This includes auditing the code for places where we were doing
explicit 'sys.version' checks and removing them as appropriate.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_smart_transport.py
1
 
# Copyright (C) 2006-2010 Canonical Ltd
 
1
# Copyright (C) 2006-2011 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
615
615
 
616
616
    def setUp(self):
617
617
        super(TestSmartServerStreamMedium, self).setUp()
618
 
        self._captureVar('BZR_NO_SMART_VFS', None)
 
618
        self.overrideEnv('BZR_NO_SMART_VFS', None)
619
619
 
620
620
    def portable_socket_pair(self):
621
621
        """Return a pair of TCP sockets connected to each other.
970
970
 
971
971
    def test_get_error_unexpected(self):
972
972
        """Error reported by server with no specific representation"""
973
 
        self._captureVar('BZR_NO_SMART_VFS', None)
 
973
        self.overrideEnv('BZR_NO_SMART_VFS', None)
974
974
        class FlakyTransport(object):
975
975
            base = 'a_url'
976
976
            def external_url(self):
1077
1077
 
1078
1078
    def test_smart_transport_has(self):
1079
1079
        """Checking for file existence over smart."""
1080
 
        self._captureVar('BZR_NO_SMART_VFS', None)
 
1080
        self.overrideEnv('BZR_NO_SMART_VFS', None)
1081
1081
        self.backing_transport.put_bytes("foo", "contents of foo\n")
1082
1082
        self.assertTrue(self.transport.has("foo"))
1083
1083
        self.assertFalse(self.transport.has("non-foo"))
1084
1084
 
1085
1085
    def test_smart_transport_get(self):
1086
1086
        """Read back a file over smart."""
1087
 
        self._captureVar('BZR_NO_SMART_VFS', None)
 
1087
        self.overrideEnv('BZR_NO_SMART_VFS', None)
1088
1088
        self.backing_transport.put_bytes("foo", "contents\nof\nfoo\n")
1089
1089
        fp = self.transport.get("foo")
1090
1090
        self.assertEqual('contents\nof\nfoo\n', fp.read())
1094
1094
        # The path in a raised NoSuchFile exception should be the precise path
1095
1095
        # asked for by the client. This gives meaningful and unsurprising errors
1096
1096
        # for users.
1097
 
        self._captureVar('BZR_NO_SMART_VFS', None)
 
1097
        self.overrideEnv('BZR_NO_SMART_VFS', None)
1098
1098
        err = self.assertRaises(
1099
1099
            errors.NoSuchFile, self.transport.get, 'not%20a%20file')
1100
1100
        self.assertSubset([err.path], ['not%20a%20file', './not%20a%20file'])
1119
1119
 
1120
1120
    def test_open_dir(self):
1121
1121
        """Test changing directory"""
1122
 
        self._captureVar('BZR_NO_SMART_VFS', None)
 
1122
        self.overrideEnv('BZR_NO_SMART_VFS', None)
1123
1123
        transport = self.transport
1124
1124
        self.backing_transport.mkdir('toffee')
1125
1125
        self.backing_transport.mkdir('toffee/apple')
1147
1147
 
1148
1148
    def test_mkdir_error_readonly(self):
1149
1149
        """TransportNotPossible should be preserved from the backing transport."""
1150
 
        self._captureVar('BZR_NO_SMART_VFS', None)
 
1150
        self.overrideEnv('BZR_NO_SMART_VFS', None)
1151
1151
        self.start_server(readonly=True)
1152
1152
        self.assertRaises(errors.TransportNotPossible, self.transport.mkdir,
1153
1153
            'foo')
1261
1261
 
1262
1262
    def setUp(self):
1263
1263
        super(SmartServerRequestHandlerTests, self).setUp()
1264
 
        self._captureVar('BZR_NO_SMART_VFS', None)
 
1264
        self.overrideEnv('BZR_NO_SMART_VFS', None)
1265
1265
 
1266
1266
    def build_handler(self, transport):
1267
1267
        """Returns a handler for the commands in protocol version one."""
1286
1286
        handler = vfs.HasRequest(None, '/')
1287
1287
        # set environment variable after construction to make sure it's
1288
1288
        # examined.
1289
 
        # Note that we can safely clobber BZR_NO_SMART_VFS here, because setUp
1290
 
        # has called _captureVar, so it will be restored to the right state
1291
 
        # afterwards.
1292
 
        os.environ['BZR_NO_SMART_VFS'] = ''
 
1289
        self.overrideEnv('BZR_NO_SMART_VFS', '')
1293
1290
        self.assertRaises(errors.DisabledMethod, handler.execute)
1294
1291
 
1295
1292
    def test_readonly_exception_becomes_transport_not_possible(self):
1602
1599
        self.assertTrue(self.end_received)
1603
1600
 
1604
1601
    def test_accept_request_and_body_all_at_once(self):
1605
 
        self._captureVar('BZR_NO_SMART_VFS', None)
 
1602
        self.overrideEnv('BZR_NO_SMART_VFS', None)
1606
1603
        mem_transport = memory.MemoryTransport()
1607
1604
        mem_transport.put_bytes('foo', 'abcdefghij')
1608
1605
        out_stream = StringIO()
1868
1865
        self.assertTrue(self.end_received)
1869
1866
 
1870
1867
    def test_accept_request_and_body_all_at_once(self):
1871
 
        self._captureVar('BZR_NO_SMART_VFS', None)
 
1868
        self.overrideEnv('BZR_NO_SMART_VFS', None)
1872
1869
        mem_transport = memory.MemoryTransport()
1873
1870
        mem_transport.put_bytes('foo', 'abcdefghij')
1874
1871
        out_stream = StringIO()
2415
2412
        self.assertEqual('aaa', stream.next())
2416
2413
        self.assertEqual('bbb', stream.next())
2417
2414
        exc = self.assertRaises(errors.ErrorFromSmartServer, stream.next)
2418
 
        self.assertEqual(('error', 'Boom!'), exc.error_tuple)
 
2415
        self.assertEqual(('error', 'Exception', 'Boom!'), exc.error_tuple)
2419
2416
 
2420
2417
    def test_interrupted_by_connection_lost(self):
2421
2418
        interrupted_body_stream = (
2818
2815
    'b\x00\x00\x00\x03aaa' # body part ('aaa')
2819
2816
    'b\x00\x00\x00\x03bbb' # body part ('bbb')
2820
2817
    'oE' # status flag (error)
2821
 
    's\x00\x00\x00\x10l5:error5:Boom!e' # err struct ('error', 'Boom!')
 
2818
    # err struct ('error', 'Exception', 'Boom!')
 
2819
    's\x00\x00\x00\x1bl5:error9:Exception5:Boom!e'
2822
2820
    'e' # EOM
2823
2821
    )
2824
2822
 
3517
3515
    def setUp(self):
3518
3516
        super(HTTPTunnellingSmokeTest, self).setUp()
3519
3517
        # We use the VFS layer as part of HTTP tunnelling tests.
3520
 
        self._captureVar('BZR_NO_SMART_VFS', None)
 
3518
        self.overrideEnv('BZR_NO_SMART_VFS', None)
3521
3519
 
3522
3520
    def test_smart_http_medium_request_accept_bytes(self):
3523
3521
        medium = FakeHTTPMedium()