Mercurial > hgsubversion

comparison tests/comprehensive/test_rebuildmeta.py @ 1106:5cb6c95e0283 stable

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Merge default and stable so I can do stable releases again.
author Augie Fackler <raf@durin42.com>
date Tue, 11 Feb 2014 12:48:49 -0500
parents 6e1dbf6cbc92
children ff4e102932ed
comparison
equal deleted inserted replaced
1020:b5b1fce26f1f 1106:5cb6c95e0283
1 import os
2 import pickle
3 import unittest
4 import sys
5
6 # wrapped in a try/except because of weirdness in how
7 # run.py works as compared to nose.
8 try:
9 import test_util
10 except ImportError:
11 sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
12 import test_util
13
14 from mercurial import context
15 from mercurial import extensions
16 from mercurial import hg
17 from mercurial import ui
18
19 from hgsubversion import compathacks
20 from hgsubversion import svncommands
21 from hgsubversion import svnmeta
22
23 # These test repositories have harmless skew in rebuildmeta for the
24 # last-pulled-rev because the last rev in svn causes absolutely no
25 # changes in hg.
26 expect_youngest_skew = [('file_mixed_with_branches.svndump', False, False),
27 ('file_mixed_with_branches.svndump', True, False),
28 ('unrelatedbranch.svndump', False, False),
29 ('unrelatedbranch.svndump', True, False),
30 ]
31
32
33
34 def _do_case(self, name, layout):
35 subdir = test_util.subdir.get(name, '')
36 single = layout == 'single'
37 u = ui.ui()
38 config = {}
39 if layout == 'custom':
40 for branch, path in test_util.custom.get(name, {}).iteritems():
41 config['hgsubversionbranch.%s' % branch] = path
42 u.setconfig('hgsubversionbranch', branch, path)
43 repo, repo_path = self.load_and_fetch(name, subdir=subdir, layout=layout)
44 assert test_util.repolen(self.repo) > 0
45 wc2_path = self.wc_path + '_clone'
46 src, dest = test_util.hgclone(u, self.wc_path, wc2_path, update=False)
47 src = test_util.getlocalpeer(src)
48 dest = test_util.getlocalpeer(dest)
49
50 # insert a wrapper that prevents calling changectx.children()
51 def failfn(orig, ctx):
52 self.fail('calling %s is forbidden; it can cause massive slowdowns '
53 'when rebuilding large repositories' % orig)
54
55 origchildren = getattr(context.changectx, 'children')
56 extensions.wrapfunction(context.changectx, 'children', failfn)
57
58 try:
59 svncommands.rebuildmeta(u, dest,
60 args=[test_util.fileurl(repo_path +
61 subdir), ])
62 finally:
63 # remove the wrapper
64 context.changectx.children = origchildren
65
66 self._run_assertions(name, single, src, dest, u)
67
68 wc3_path = self.wc_path + '_partial'
69 src, dest = test_util.hgclone(u,
70 self.wc_path,
71 wc3_path,
72 update=False,
73 rev=[0])
74 srcrepo = test_util.getlocalpeer(src)
75 dest = test_util.getlocalpeer(dest)
76
77 # insert a wrapper that prevents calling changectx.children()
78 extensions.wrapfunction(context.changectx, 'children', failfn)
79
80 try:
81 svncommands.rebuildmeta(u, dest,
82 args=[test_util.fileurl(repo_path +
83 subdir), ])
84 finally:
85 # remove the wrapper
86 context.changectx.children = origchildren
87
88 dest.pull(src)
89
90 # insert a wrapper that prevents calling changectx.children()
91 extensions.wrapfunction(context.changectx, 'children', failfn)
92 try:
93 svncommands.updatemeta(u, dest,
94 args=[test_util.fileurl(repo_path +
95 subdir), ])
96 finally:
97 # remove the wrapper
98 context.changectx.children = origchildren
99
100 self._run_assertions(name, single, srcrepo, dest, u)
101
102
103 def _run_assertions(self, name, single, src, dest, u):
104
105 self.assertTrue(os.path.isdir(os.path.join(src.path, 'svn')),
106 'no .hg/svn directory in the source!')
107 self.assertTrue(os.path.isdir(os.path.join(dest.path, 'svn')),
108 'no .hg/svn directory in the destination!')
109 dest = hg.repository(u, os.path.dirname(dest.path))
110 for tf in ('lastpulled', 'rev_map', 'uuid', 'tagmap', 'layout', 'subdir',):
111
112 stf = os.path.join(src.path, 'svn', tf)
113 self.assertTrue(os.path.isfile(stf), '%r is missing!' % stf)
114 dtf = os.path.join(dest.path, 'svn', tf)
115 self.assertTrue(os.path.isfile(dtf), '%r is missing!' % tf)
116 old, new = open(stf).read(), open(dtf).read()
117 if tf == 'lastpulled' and (name,
118 self.stupid, single) in expect_youngest_skew:
119 self.assertNotEqual(old, new,
120 'rebuildmeta unexpected match on youngest rev!')
121 continue
122 self.assertMultiLineEqual(old, new, tf + ' differs')
123 try:
124 self.assertEqual(src.branchmap(), dest.branchmap())
125 except AttributeError:
126 # hg 2.8 and earlier
127 self.assertEqual(src.branchtags(), dest.branchtags())
128 srcbi = pickle.load(open(os.path.join(src.path, 'svn', 'branch_info')))
129 destbi = pickle.load(open(os.path.join(dest.path, 'svn', 'branch_info')))
130 self.assertEqual(sorted(srcbi.keys()), sorted(destbi.keys()))
131 revkeys = svnmeta.SVNMeta(dest).revmap.keys()
132 for branch in destbi:
133 srcinfo = srcbi[branch]
134 destinfo = destbi[branch]
135 if srcinfo[:2] == (None, 0) or destinfo[:2] == (None, 0):
136 self.assertTrue(srcinfo[2] <= destinfo[2],
137 'Latest revision for %s decreased from %d to %d!'
138 % (branch or 'default', srcinfo[2], destinfo[2]))
139 self.assertEqual(srcinfo[0], destinfo[0])
140 else:
141 pr = sorted(filter(lambda x: x[1] == srcinfo[0] and x[0] <= srcinfo[1],
142 revkeys), reverse=True)[0][0]
143 self.assertEqual(pr, destinfo[1])
144 self.assertEqual(srcinfo[2], destinfo[2])
145
146
147 def buildmethod(case, name, layout):
148 m = lambda self: self._do_case(case, layout)
149 m.__name__ = name
150 m.__doc__ = ('Test rebuildmeta on %s (%s)' %
151 (case, layout))
152 return m
153
154
155 skip = set([
156 'project_root_not_repo_root.svndump',
157 'corrupt.svndump',
158 ])
159
160 attrs = {'_do_case': _do_case,
161 '_run_assertions': _run_assertions,
162 'stupid_mode_tests': True,
163 }
164 for case in [f for f in os.listdir(test_util.FIXTURES) if f.endswith('.svndump')]:
165 # this fixture results in an empty repository, don't use it
166 if case in skip:
167 continue
168 bname = 'test_' + case[:-len('.svndump')]
169 attrs[bname] = buildmethod(case, bname, 'auto')
170 attrs[bname + '_single'] = buildmethod(case, bname + '_single', 'single')
171 if case in test_util.custom:
172 attrs[bname + '_custom'] = buildmethod(case,
173 bname + '_custom',
174 'single')
175
176 RebuildMetaTests = type('RebuildMetaTests', (test_util.TestBase,), attrs)