← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

Merge cleanup into first-try

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/__init__.py
29
29
 
30
30
import atexit
31
31
import codecs
32
 
from copy import copy
 
32
import copy
33
33
from cStringIO import StringIO
34
34
import difflib
35
35
import doctest
37
37
import logging
38
38
import math
39
39
import os
40
 
from pprint import pformat
 
40
import pprint
41
41
import random
42
42
import re
43
43
import shlex
44
44
import stat
45
 
from subprocess import Popen, PIPE, STDOUT
 
45
import subprocess
46
46
import sys
47
47
import tempfile
48
48
import threading
74
74
    ui,
75
75
    urlutils,
76
76
    registry,
 
77
    transport as _mod_transport,
77
78
    workingtree,
78
79
    )
79
80
import bzrlib.branch
103
104
    )
104
105
import bzrlib.trace
105
106
from bzrlib.transport import (
106
 
    get_transport,
107
107
    memory,
108
108
    pathfilter,
109
109
    )
110
 
import bzrlib.transport
111
110
from bzrlib.trace import mutter, note
112
111
from bzrlib.tests import (
113
112
    test_server,
940
939
 
941
940
    def permit_dir(self, name):
942
941
        """Permit a directory to be used by this test. See permit_url."""
943
 
        name_transport = get_transport(name)
 
942
        name_transport = _mod_transport.get_transport(name)
944
943
        self.permit_url(name)
945
944
        self.permit_url(name_transport.base)
946
945
 
1025
1024
        self.addCleanup(transport_server.stop_server)
1026
1025
        # Obtain a real transport because if the server supplies a password, it
1027
1026
        # will be hidden from the base on the client side.
1028
 
        t = get_transport(transport_server.get_url())
 
1027
        t = _mod_transport.get_transport(transport_server.get_url())
1029
1028
        # Some transport servers effectively chroot the backing transport;
1030
1029
        # others like SFTPServer don't - users of the transport can walk up the
1031
1030
        # transport to read the entire backing transport. This wouldn't matter
1092
1091
            message += '\n'
1093
1092
        raise AssertionError("%snot equal:\na = %s\nb = %s\n"
1094
1093
            % (message,
1095
 
               pformat(a), pformat(b)))
 
1094
               pprint.pformat(a), pprint.pformat(b)))
1096
1095
 
1097
1096
    assertEquals = assertEqual
1098
1097
 
1983
1982
            if not allow_plugins:
1984
1983
                command.append('--no-plugins')
1985
1984
            command.extend(process_args)
1986
 
            process = self._popen(command, stdin=PIPE, stdout=PIPE, stderr=PIPE)
 
1985
            process = self._popen(command, stdin=subprocess.PIPE,
 
1986
                                  stdout=subprocess.PIPE,
 
1987
                                  stderr=subprocess.PIPE)
1987
1988
        finally:
1988
1989
            restore_environment()
1989
1990
            if cwd is not None:
1997
1998
        Allows tests to override this method to intercept the calls made to
1998
1999
        Popen for introspection.
1999
2000
        """
2000
 
        return Popen(*args, **kwargs)
 
2001
        return subprocess.Popen(*args, **kwargs)
2001
2002
 
2002
2003
    def get_source_path(self):
2003
2004
        """Return the path of the directory containing bzrlib."""
2183
2184
 
2184
2185
        :param relpath: a path relative to the base url.
2185
2186
        """
2186
 
        t = get_transport(self.get_url(relpath))
 
2187
        t = _mod_transport.get_transport(self.get_url(relpath))
2187
2188
        self.assertFalse(t.is_readonly())
2188
2189
        return t
2189
2190
 
2195
2196
 
2196
2197
        :param relpath: a path relative to the base url.
2197
2198
        """
2198
 
        t = get_transport(self.get_readonly_url(relpath))
 
2199
        t = _mod_transport.get_transport(self.get_readonly_url(relpath))
2199
2200
        self.assertTrue(t.is_readonly())
2200
2201
        return t
2201
2202
 
2331
2332
        propagating. This method ensures than a test did not leaked.
2332
2333
        """
2333
2334
        root = TestCaseWithMemoryTransport.TEST_ROOT
2334
 
        self.permit_url(get_transport(root).base)
 
2335
        self.permit_url(_mod_transport.get_transport(root).base)
2335
2336
        wt = workingtree.WorkingTree.open(root)
2336
2337
        last_rev = wt.last_revision()
2337
2338
        if last_rev != 'null:':
2382
2383
            # might be a relative or absolute path
2383
2384
            maybe_a_url = self.get_url(relpath)
2384
2385
            segments = maybe_a_url.rsplit('/', 1)
2385
 
            t = get_transport(maybe_a_url)
 
2386
            t = _mod_transport.get_transport(maybe_a_url)
2386
2387
            if len(segments) > 1 and segments[-1] not in ('', '.'):
2387
2388
                t.ensure_base()
2388
2389
            if format is None:
2408
2409
    def make_smart_server(self, path):
2409
2410
        smart_server = test_server.SmartTCPServer_for_testing()
2410
2411
        self.start_server(smart_server, self.get_server())
2411
 
        remote_transport = get_transport(smart_server.get_url()).clone(path)
 
2412
        remote_transport = _mod_transport.get_transport(smart_server.get_url()
 
2413
                                                   ).clone(path)
2412
2414
        return remote_transport
2413
2415
 
2414
2416
    def make_branch_and_memory_tree(self, relpath, format=None):
2479
2481
 
2480
2482
    def check_file_contents(self, filename, expect):
2481
2483
        self.log("check contents of file %s" % filename)
2482
 
        contents = file(filename, 'r').read()
 
2484
        f = file(filename)
 
2485
        try:
 
2486
            contents = f.read()
 
2487
        finally:
 
2488
            f.close()
2483
2489
        if contents != expect:
2484
2490
            self.log("expected: %r" % expect)
2485
2491
            self.log("actually: %r" % contents)
2559
2565
                "a list or a tuple. Got %r instead" % (shape,))
2560
2566
        # It's OK to just create them using forward slashes on windows.
2561
2567
        if transport is None or transport.is_readonly():
2562
 
            transport = get_transport(".")
 
2568
            transport = _mod_transport.get_transport(".")
2563
2569
        for name in shape:
2564
2570
            self.assertIsInstance(name, basestring)
2565
2571
            if name[-1] == '/':
3307
3313
                '--subunit']
3308
3314
            if '--no-plugins' in sys.argv:
3309
3315
                argv.append('--no-plugins')
3310
 
            # stderr=STDOUT would be ideal, but until we prevent noise on
3311
 
            # stderr it can interrupt the subunit protocol.
3312
 
            process = Popen(argv, stdin=PIPE, stdout=PIPE, stderr=PIPE,
3313
 
                bufsize=1)
 
3316
            # stderr=subrocess.STDOUT would be ideal, but until we prevent
 
3317
            # noise on stderr it can interrupt the subunit protocol.
 
3318
            process = subrocess.Popen(argv, stdin=subrocess.PIPE,
 
3319
                                      stdout=subrocess.PIPE,
 
3320
                                      stderr=subrocess.PIPE,
 
3321
                                      bufsize=1)
3314
3322
            test = TestInSubprocess(process, test_list_file_name)
3315
3323
            result.append(test)
3316
3324
        except:
4034
4042
    :param new_id: The id to assign to it.
4035
4043
    :return: The new test.
4036
4044
    """
4037
 
    new_test = copy(test)
 
4045
    new_test = copy.copy(test)
4038
4046
    new_test.id = lambda: new_id
4039
4047
    return new_test
4040
4048