868
868
excName = str(excClass)
869
869
raise self.failureException, "%s not raised" % excName
871
def assertRaises(self, excClass, func, *args, **kwargs):
872
"""Assert that a callable raises a particular exception.
874
:param excClass: As for the except statement, this may be either an
875
exception class, or a tuple of classes.
877
Returns the exception so that you can examine it.
880
func(*args, **kwargs)
884
if getattr(excClass,'__name__', None) is not None:
885
excName = excClass.__name__
888
excName = str(excClass)
889
raise self.failureException, "%s not raised" % excName
871
891
def assertIs(self, left, right, message=None):
872
892
if not (left is right):
873
893
if message is not None:
1483
1503
this_tree=wt_to)
1484
1504
wt_to.add_parent_tree_id(branch_from.last_revision())
1506
def reduceLockdirTimeout(self):
1507
"""Reduce the default lock timeout for the duration of the test, so that
1508
if LockContention occurs during a test, it does so quickly.
1510
Tests that expect to provoke LockContention errors should call this.
1512
orig_timeout = bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS
1514
bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = orig_timeout
1515
self.addCleanup(resetTimeout)
1516
bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = 0
1487
1518
BzrTestBase = TestCase
1512
1543
# execution. Variables that the parameteriser sets need to be
1513
1544
# ones that are not set by setUp, or setUp will trash them.
1514
1545
super(TestCaseWithMemoryTransport, self).__init__(methodName)
1515
self.transport_server = default_transport
1546
self.vfs_transport_factory = default_transport
1547
self.transport_server = None
1516
1548
self.transport_readonly_server = None
1549
self.__vfs_server = None
1518
1551
def get_transport(self):
1519
1552
"""Return a writeable transport for the test scratch space"""
1547
1580
if self.transport_readonly_server is None:
1548
1581
# readonly decorator requested
1549
1582
# bring up the server
1551
1583
self.__readonly_server = ReadonlyServer()
1552
self.__readonly_server.setUp(self.__server)
1584
self.__readonly_server.setUp(self.get_vfs_only_server())
1554
1586
self.__readonly_server = self.create_transport_readonly_server()
1555
self.__readonly_server.setUp()
1587
self.__readonly_server.setUp(self.get_vfs_only_server())
1556
1588
self.addCleanup(self.__readonly_server.tearDown)
1557
1589
return self.__readonly_server
1580
1612
For TestCaseWithMemoryTransport this is always a MemoryServer, and there
1581
1613
is no means to override it.
1615
if self.__vfs_server is None:
1616
self.__vfs_server = MemoryServer()
1617
self.__vfs_server.setUp()
1618
self.addCleanup(self.__vfs_server.tearDown)
1619
return self.__vfs_server
1621
def get_server(self):
1622
"""Get the read/write server instance.
1624
This is useful for some tests with specific servers that need
1627
This is built from the self.transport_server factory. If that is None,
1628
then the self.get_vfs_server is returned.
1583
1630
if self.__server is None:
1584
self.__server = MemoryServer()
1585
self.__server.setUp()
1631
if self.transport_server is None or self.transport_server is self.vfs_transport_factory:
1632
return self.get_vfs_only_server()
1634
# bring up a decorated means of access to the vfs only server.
1635
self.__server = self.transport_server()
1637
self.__server.setUp(self.get_vfs_only_server())
1638
except TypeError, e:
1639
# This should never happen; the try:Except here is to assist
1640
# developers having to update code rather than seeing an
1641
# uninformative TypeError.
1642
raise Exception, "Old server API in use: %s, %s" % (self.__server, e)
1586
1643
self.addCleanup(self.__server.tearDown)
1587
1644
return self.__server
1589
def get_url(self, relpath=None):
1646
def _adjust_url(self, base, relpath):
1590
1647
"""Get a URL (or maybe a path) for the readwrite transport.
1592
1649
This will either be backed by '.' or to an equivalent non-file based
1594
1651
relpath provides for clients to get a path relative to the base url.
1595
1652
These should only be downwards relative, not upwards.
1597
base = self.get_server().get_url()
1598
1654
if relpath is not None and relpath != '.':
1599
1655
if not base.endswith('/'):
1600
1656
base = base + '/'
1608
1664
base += urlutils.escape(relpath)
1667
def get_url(self, relpath=None):
1668
"""Get a URL (or maybe a path) for the readwrite transport.
1670
This will either be backed by '.' or to an equivalent non-file based
1672
relpath provides for clients to get a path relative to the base url.
1673
These should only be downwards relative, not upwards.
1675
base = self.get_server().get_url()
1676
return self._adjust_url(base, relpath)
1678
def get_vfs_only_url(self, relpath=None):
1679
"""Get a URL (or maybe a path for the plain old vfs transport.
1681
This will never be a smart protocol.
1682
:param relpath: provides for clients to get a path relative to the base
1683
url. These should only be downwards relative, not upwards.
1685
base = self.get_vfs_only_server().get_url()
1686
return self._adjust_url(base, relpath)
1611
1688
def _make_test_root(self):
1612
1689
if TestCaseWithMemoryTransport.TEST_ROOT is not None:
1835
1913
readwrite one must both define get_url() as resolving to os.getcwd().
1838
def create_transport_server(self):
1839
"""Create a transport server from class defined at init.
1841
This is mostly a hook for daughter classes.
1843
return self.transport_server()
1845
def get_server(self):
1916
def get_vfs_only_server(self):
1846
1917
"""See TestCaseWithMemoryTransport.
1848
1919
This is useful for some tests with specific servers that need
1851
if self.__server is None:
1852
self.__server = self.create_transport_server()
1853
self.__server.setUp()
1854
self.addCleanup(self.__server.tearDown)
1855
return self.__server
1922
if self.__vfs_server is None:
1923
self.__vfs_server = self.vfs_transport_factory()
1924
self.__vfs_server.setUp()
1925
self.addCleanup(self.__vfs_server.tearDown)
1926
return self.__vfs_server
1857
1928
def make_branch_and_tree(self, relpath, format=None):
1858
1929
"""Create a branch on the transport and a tree locally.
1860
1931
If the transport is not a LocalTransport, the Tree can't be created on
1861
the transport. In that case the working tree is created in the local
1862
directory, and the returned tree's branch and repository will also be
1865
This will fail if the original default transport for this test
1866
case wasn't backed by the working directory, as the branch won't
1867
be on disk for us to open it.
1932
the transport. In that case if the vfs_transport_factory is
1933
LocalURLServer the working tree is created in the local
1934
directory backing the transport, and the returned tree's branch and
1935
repository will also be accessed locally. Otherwise a lightweight
1936
checkout is created and returned.
1869
1938
:param format: The BzrDirFormat.
1870
1939
:returns: the WorkingTree.
1878
1947
return b.bzrdir.create_workingtree()
1879
1948
except errors.NotLocalUrl:
1880
1949
# We can only make working trees locally at the moment. If the
1881
# transport can't support them, then reopen the branch on a local
1882
# transport, and create the working tree there.
1884
# Possibly we should instead keep
1885
# the non-disk-backed branch and create a local checkout?
1886
bd = bzrdir.BzrDir.open(relpath)
1887
return bd.create_workingtree()
1950
# transport can't support them, then we keep the non-disk-backed
1951
# branch and create a local checkout.
1952
if self.vfs_transport_factory is LocalURLServer:
1953
# the branch is colocated on disk, we cannot create a checkout.
1954
# hopefully callers will expect this.
1955
local_controldir= bzrdir.BzrDir.open(self.get_vfs_only_url(relpath))
1956
return local_controldir.create_workingtree()
1958
return b.create_checkout(relpath, lightweight=True)
1889
1960
def assertIsDirectory(self, relpath, transport):
1890
1961
"""Assert that relpath within transport is a directory.
1930
2001
def setUp(self):
1931
2002
super(ChrootedTestCase, self).setUp()
1932
if not self.transport_server == MemoryServer:
2003
if not self.vfs_transport_factory == MemoryServer:
1933
2004
self.transport_readonly_server = HttpServer