Mercurial > hgsubversion

comparison tests/comprehensive/test_rebuildmeta.py @ 1042:af84ef787d93

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
tests: move updatemeta & rebuildmeta tests into comprehensive this decreases normal the test count from 689 to 269, as both iterate over all our fixture, and test_utility_commands already tests both rudimentarily.
author Dan Villiom Podlaski Christiansen <danchr@gmail.com>
date Wed, 07 Aug 2013 09:57:54 +0200
parents tests/test_rebuildmeta.py@3b43b1c45e2e
children d741f536f23a
comparison
equal deleted inserted replaced
1041:70090e2ee262 1042:af84ef787d93
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 svncommands
20 from hgsubversion import svnmeta
21
22 # These test repositories have harmless skew in rebuildmeta for the
23 # last-pulled-rev because the last rev in svn causes absolutely no
24 # changes in hg.
25 expect_youngest_skew = [('file_mixed_with_branches.svndump', False, False),
26 ('file_mixed_with_branches.svndump', True, False),
27 ('unrelatedbranch.svndump', False, False),
28 ('unrelatedbranch.svndump', True, False),
29 ]
30
31
32
33 def _do_case(self, name, stupid, single):
34 subdir = test_util.subdir.get(name, '')
35 layout = 'auto'
36 if single:
37 layout = 'single'
38 repo, repo_path = self.load_and_fetch(name, subdir=subdir, stupid=stupid,
39 layout=layout)
40 assert len(self.repo) > 0
41 wc2_path = self.wc_path + '_clone'
42 u = ui.ui()
43 src, dest = test_util.hgclone(u, self.wc_path, wc2_path, update=False)
44 src = test_util.getlocalpeer(src)
45 dest = test_util.getlocalpeer(dest)
46
47 # insert a wrapper that prevents calling changectx.children()
48 def failfn(orig, ctx):
49 self.fail('calling %s is forbidden; it can cause massive slowdowns '
50 'when rebuilding large repositories' % orig)
51
52 origchildren = getattr(context.changectx, 'children')
53 extensions.wrapfunction(context.changectx, 'children', failfn)
54
55 try:
56 svncommands.rebuildmeta(u, dest,
57 args=[test_util.fileurl(repo_path +
58 subdir), ])
59 finally:
60 # remove the wrapper
61 context.changectx.children = origchildren
62
63 self._run_assertions(name, stupid, single, src, dest, u)
64
65 wc3_path = self.wc_path + '_partial'
66 src, dest = test_util.hgclone(u,
67 self.wc_path,
68 wc3_path,
69 update=False,
70 rev=[0])
71 srcrepo = test_util.getlocalpeer(src)
72 dest = test_util.getlocalpeer(dest)
73
74 # insert a wrapper that prevents calling changectx.children()
75 extensions.wrapfunction(context.changectx, 'children', failfn)
76
77 try:
78 svncommands.rebuildmeta(u, dest,
79 args=[test_util.fileurl(repo_path +
80 subdir), ])
81 finally:
82 # remove the wrapper
83 context.changectx.children = origchildren
84
85 dest.pull(src)
86
87 # insert a wrapper that prevents calling changectx.children()
88 extensions.wrapfunction(context.changectx, 'children', failfn)
89 try:
90 svncommands.updatemeta(u, dest,
91 args=[test_util.fileurl(repo_path +
92 subdir), ])
93 finally:
94 # remove the wrapper
95 context.changectx.children = origchildren
96
97 self._run_assertions(name, stupid, single, srcrepo, dest, u)
98
99
100 def _run_assertions(self, name, stupid, single, src, dest, u):
101
102 self.assertTrue(os.path.isdir(os.path.join(src.path, 'svn')),
103 'no .hg/svn directory in the source!')
104 self.assertTrue(os.path.isdir(os.path.join(dest.path, 'svn')),
105 'no .hg/svn directory in the destination!')
106 dest = hg.repository(u, os.path.dirname(dest.path))
107 for tf in ('lastpulled', 'rev_map', 'uuid', 'tagmap', 'layout', 'subdir',):
108
109 stf = os.path.join(src.path, 'svn', tf)
110 self.assertTrue(os.path.isfile(stf), '%r is missing!' % stf)
111 dtf = os.path.join(dest.path, 'svn', tf)
112 self.assertTrue(os.path.isfile(dtf), '%r is missing!' % tf)
113 old, new = open(stf).read(), open(dtf).read()
114 if tf == 'lastpulled' and (name,
115 stupid, single) in expect_youngest_skew:
116 self.assertNotEqual(old, new,
117 'rebuildmeta unexpected match on youngest rev!')
118 continue
119 self.assertMultiLineEqual(old, new, tf + ' differs')
120 self.assertEqual(src.branchtags(), dest.branchtags())
121 srcbi = pickle.load(open(os.path.join(src.path, 'svn', 'branch_info')))
122 destbi = pickle.load(open(os.path.join(dest.path, 'svn', 'branch_info')))
123 self.assertEqual(sorted(srcbi.keys()), sorted(destbi.keys()))
124 revkeys = svnmeta.SVNMeta(dest).revmap.keys()
125 for branch in destbi:
126 srcinfo = srcbi[branch]
127 destinfo = destbi[branch]
128 if srcinfo[:2] == (None, 0) or destinfo[:2] == (None, 0):
129 self.assertTrue(srcinfo[2] <= destinfo[2],
130 'Latest revision for %s decreased from %d to %d!'
131 % (branch or 'default', srcinfo[2], destinfo[2]))
132 self.assertEqual(srcinfo[0], destinfo[0])
133 else:
134 pr = sorted(filter(lambda x: x[1] == srcinfo[0] and x[0] <= srcinfo[1],
135 revkeys), reverse=True)[0][0]
136 self.assertEqual(pr, destinfo[1])
137 self.assertEqual(srcinfo[2], destinfo[2])
138
139
140 def buildmethod(case, name, stupid, single):
141 m = lambda self: self._do_case(case, stupid, single)
142 m.__name__ = name
143 m.__doc__ = ('Test rebuildmeta on %s with %s replay. (%s)' %
144 (case,
145 (stupid and 'stupid') or 'real',
146 (single and 'single') or 'standard',
147 )
148 )
149 return m
150
151
152 skip = set([
153 'project_root_not_repo_root.svndump',
154 'corrupt.svndump',
155 ])
156
157 attrs = {'_do_case': _do_case,
158 '_run_assertions': _run_assertions,
159 }
160 for case in [f for f in os.listdir(test_util.FIXTURES) if f.endswith('.svndump')]:
161 # this fixture results in an empty repository, don't use it
162 if case in skip:
163 continue
164 bname = 'test_' + case[:-len('.svndump')]
165 attrs[bname] = buildmethod(case, bname, False, False)
166 name = bname + '_stupid'
167 attrs[name] = buildmethod(case, name, True, False)
168 name = bname + '_single'
169 attrs[name] = buildmethod(case, name, False, True)
170
171 RebuildMetaTests = type('RebuildMetaTests', (test_util.TestBase,), attrs)
172
173
174 def suite():
175 all_tests = [unittest.TestLoader().loadTestsFromTestCase(RebuildMetaTests),
176 ]
177 return unittest.TestSuite(all_tests)