Use locking and write groups properly, fixes compatibility with packs.
[jelmer/subvertpy.git] / tests / test_upgrade.py
1 # Copyright (C) 2007 Jelmer Vernooij <jelmer@samba.org>
2
3 # This program is free software; you can redistribute it and/or modify
4 # it under the terms of the GNU General Public License as published by
5 # the Free Software Foundation; either version 2 of the License, or
6 # (at your option) any later version.
7
8 # This program is distributed in the hope that it will be useful,
9 # but WITHOUT ANY WARRANTY; without even the implied warranty of
10 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11 # GNU General Public License for more details.
12
13 # You should have received a copy of the GNU General Public License
14 # along with this program; if not, write to the Free Software
15 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
17 """Mapping upgrade tests."""
18
19 from bzrlib.bzrdir import BzrDir
20 from bzrlib.errors import InvalidRevisionId
21 from bzrlib.repository import Repository
22 from bzrlib.tests import TestCase, TestCaseWithTransport, TestSkipped
23
24 from errors import RebaseNotPresent
25 from fileids import generate_svn_file_id
26 from format import get_rich_root_format
27 from repository import MAPPING_VERSION
28 from tests import TestCaseWithSubversionRepository
29 from upgrade import (upgrade_repository, upgrade_branch,
30                      upgrade_workingtree, UpgradeChangesContent, 
31                      parse_legacy_revision_id, create_upgraded_revid, 
32                      generate_upgrade_map)
33
34 class TestUpgradeChangesContent(TestCase):
35     def test_init(self):
36         x = UpgradeChangesContent("revisionx")
37         self.assertEqual("revisionx", x.revid)
38
39
40 class ParserTests(TestCase):
41     def test_current(self):
42         self.assertEqual(("uuid", "trunk", 1, "trunk0", 3), 
43                 parse_legacy_revision_id("svn-v3-trunk0:uuid:trunk:1"))
44
45     def test_current_undefined(self):
46         self.assertEqual(("uuid", "trunk", 1, None, 3), 
47                 parse_legacy_revision_id("svn-v3-undefined:uuid:trunk:1"))
48
49     def test_legacy2(self):
50         self.assertEqual(("uuid", "trunk", 1, None, 2), 
51                          parse_legacy_revision_id("svn-v2:1@uuid-trunk"))
52
53     def test_legacy(self):
54         self.assertEqual(("uuid", "trunk", 1, None, 1), 
55                          parse_legacy_revision_id("svn-v1:1@uuid-trunk"))
56
57     def test_except(self):
58         self.assertRaises(InvalidRevisionId, 
59                          parse_legacy_revision_id, "svn-v0:1@uuid-trunk")
60
61     def test_except_nonsvn(self):
62         self.assertRaises(InvalidRevisionId, 
63                          parse_legacy_revision_id, "blah")
64
65     def test_create_upgraded_revid_new(self):
66         self.assertEqual("bla-svn%d-upgrade" % MAPPING_VERSION,
67                          create_upgraded_revid("bla"))
68
69     def test_create_upgraded_revid_upgrade(self):
70         self.assertEqual("bla-svn%d-upgrade" % MAPPING_VERSION,
71                          create_upgraded_revid("bla-svn1-upgrade"))
72
73
74 def skip_no_rebase(unbound):
75     def check_error(self, *args, **kwargs):
76         try:
77             return unbound(self, *args, **kwargs)
78         except RebaseNotPresent, e:
79             raise TestSkipped(e)
80     check_error.__doc__ = unbound.__doc__
81     check_error.__name__ = unbound.__name__
82     return check_error
83
84
85 class UpgradeTests(TestCaseWithSubversionRepository):
86     @skip_no_rebase
87     def test_no_custom(self):
88         repos_url = self.make_client("a", "dc")
89         self.build_tree({'dc/a': 'b'})
90         self.client_add("dc/a")
91         self.client_commit("dc", "data")
92
93         oldrepos = Repository.open(repos_url)
94         dir = BzrDir.create("f",format=get_rich_root_format())
95         newrepos = dir.create_repository()
96         oldrepos.copy_content_into(newrepos)
97         dir.create_branch()
98         wt = dir.create_workingtree()
99         file("f/a","w").write("b")
100         wt.add("a")
101         wt.commit(message="data", rev_id="svn-v1:1@%s-" % oldrepos.uuid)
102
103         self.assertTrue(newrepos.has_revision("svn-v1:1@%s-" % oldrepos.uuid))
104
105         upgrade_repository(newrepos, oldrepos, allow_changes=True)
106
107         self.assertTrue(newrepos.has_revision(oldrepos.generate_revision_id(1, "", "none")))
108
109     @skip_no_rebase
110     def test_single_custom(self):
111         repos_url = self.make_client("a", "dc")
112         self.build_tree({'dc/a': 'b'})
113         self.client_add("dc/a")
114         self.client_commit("dc", "data")
115
116         oldrepos = Repository.open(repos_url)
117         dir = BzrDir.create("f",format=get_rich_root_format())
118         newrepos = dir.create_repository()
119         oldrepos.copy_content_into(newrepos)
120         dir.create_branch()
121         wt = dir.create_workingtree()
122         file("f/a", "w").write("b")
123         wt.add("a")
124         wt.commit(message="data", rev_id="svn-v1:1@%s-" % oldrepos.uuid)
125         file("f/a", 'w').write("moredata")
126         wt.commit(message='fix moredata', rev_id="customrev")
127
128         upgrade_repository(newrepos, oldrepos, allow_changes=True)
129
130         self.assertTrue(newrepos.has_revision(oldrepos.generate_revision_id(1, "", "none")))
131         self.assertTrue(newrepos.has_revision("customrev-svn%d-upgrade" % MAPPING_VERSION))
132         self.assertTrue((oldrepos.generate_revision_id(1, "", "none"),),
133                         tuple(newrepos.revision_parents("customrev-svn%d-upgrade" % MAPPING_VERSION)))
134
135     @skip_no_rebase
136     def test_single_keep_parent_fileid(self):
137         repos_url = self.make_client("a", "dc")
138         self.build_tree({'dc/a': 'b'})
139         self.client_add("dc/a")
140         self.client_commit("dc", "data")
141
142         oldrepos = Repository.open(repos_url)
143         dir = BzrDir.create("f", format=get_rich_root_format())
144         newrepos = dir.create_repository()
145         oldrepos.copy_content_into(newrepos)
146         dir.create_branch()
147         wt = dir.create_workingtree()
148         file("f/a", "w").write("b")
149         wt.add(["a"], ["someid"])
150         wt.commit(message="data", rev_id="svn-v1:1@%s-" % oldrepos.uuid)
151         wt.rename_one("a", "b")
152         file("f/a", 'w').write("moredata")
153         wt.add(["a"], ["specificid"])
154         wt.commit(message='fix moredata', rev_id="customrev")
155
156         upgrade_repository(newrepos, oldrepos, allow_changes=True)
157
158         tree = newrepos.revision_tree("customrev-svn%d-upgrade" % MAPPING_VERSION)
159         self.assertEqual("specificid", tree.inventory.path2id("a"))
160         self.assertEqual(generate_svn_file_id(oldrepos.uuid, 1, "", "a"), 
161                          tree.inventory.path2id("b"))
162
163     @skip_no_rebase
164     def test_single_custom_continue(self):
165         repos_url = self.make_client("a", "dc")
166         self.build_tree({'dc/a': 'b', 'dc/b': 'c'})
167         self.client_add("dc/a")
168         self.client_add("dc/b")
169         self.client_commit("dc", "data")
170
171         oldrepos = Repository.open(repos_url)
172         dir = BzrDir.create("f",format=get_rich_root_format())
173         newrepos = dir.create_repository()
174         oldrepos.copy_content_into(newrepos)
175         dir.create_branch()
176         wt = dir.create_workingtree()
177         file("f/a", "w").write("b")
178         file("f/b", "w").write("c")
179         wt.add("a")
180         wt.add("b")
181         wt.commit(message="data", rev_id="svn-v1:1@%s-" % oldrepos.uuid)
182         file("f/a", 'w').write("moredata")
183         file("f/b", 'w').write("moredata")
184         wt.commit(message='fix moredata', rev_id="customrev")
185
186         tree = newrepos.revision_tree("svn-v1:1@%s-" % oldrepos.uuid)
187
188         newrepos.lock_write()
189         newrepos.start_write_group()
190
191         vf = newrepos.weave_store.get_weave_or_empty(tree.inventory.path2id("a"), newrepos.get_transaction())
192         vf.clone_text("customrev-svn%d-upgrade" % MAPPING_VERSION,
193                 "svn-v1:1@%s-" % oldrepos.uuid, ["svn-v1:1@%s-" % oldrepos.uuid])
194
195         newrepos.commit_write_group()
196         newrepos.unlock()
197
198         upgrade_repository(newrepos, oldrepos, allow_changes=True)
199
200         self.assertTrue(newrepos.has_revision(oldrepos.generate_revision_id(1, "", "none")))
201         self.assertTrue(newrepos.has_revision("customrev-svn%d-upgrade" % MAPPING_VERSION))
202         self.assertTrue((oldrepos.generate_revision_id(1, "", "none"),),
203                         tuple(newrepos.revision_parents("customrev-svn%d-upgrade" % MAPPING_VERSION)))
204
205     @skip_no_rebase
206     def test_more_custom(self):
207         repos_url = self.make_client("a", "dc")
208         self.build_tree({'dc/a': 'b'})
209         self.client_add("dc/a")
210         self.client_commit("dc", "data")
211
212         oldrepos = Repository.open(repos_url)
213         dir = BzrDir.create("f",format=get_rich_root_format())
214         newrepos = dir.create_repository()
215         dir.create_branch()
216         wt = dir.create_workingtree()
217         file("f/a", "w").write("b")
218         wt.add("a")
219         wt.commit(message="data", rev_id="svn-v1:1@%s-" % oldrepos.uuid)
220         file("f/a", 'w').write("moredata")
221         wt.commit(message='fix moredata', rev_id="customrev")
222         file("f/a", 'w').write("blackfield")
223         wt.commit(message='fix it again', rev_id="anotherrev")
224
225         renames = upgrade_repository(newrepos, oldrepos, allow_changes=True)
226         self.assertEqual({
227             'svn-v1:1@%s-' % oldrepos.uuid: 'svn-v3-none:%s::1' % oldrepos.uuid,
228             "customrev": "customrev-svn%d-upgrade" % MAPPING_VERSION,
229             "anotherrev": "anotherrev-svn%d-upgrade" % MAPPING_VERSION},
230             renames)
231
232         self.assertTrue(newrepos.has_revision(oldrepos.generate_revision_id(1, "", "none")))
233         self.assertTrue(newrepos.has_revision("customrev-svn%d-upgrade" % MAPPING_VERSION))
234         self.assertTrue(newrepos.has_revision("anotherrev-svn%d-upgrade" % MAPPING_VERSION))
235         self.assertTrue((oldrepos.generate_revision_id(1, "", "none"),),
236                         tuple(newrepos.revision_parents("customrev-svn%d-upgrade" % MAPPING_VERSION)))
237         self.assertTrue(("customrev-svn%d-upgrade" % MAPPING_VERSION,),
238                         tuple(newrepos.revision_parents("anotherrev-svn%d-upgrade" % MAPPING_VERSION)))
239
240     @skip_no_rebase
241     def test_more_custom_branch(self):
242         repos_url = self.make_client("a", "dc")
243         self.build_tree({'dc/a': 'b'})
244         self.client_add("dc/a")
245         self.client_commit("dc", "data")
246
247         oldrepos = Repository.open(repos_url)
248         dir = BzrDir.create("f",format=get_rich_root_format())
249         newrepos = dir.create_repository()
250         b = dir.create_branch()
251         wt = dir.create_workingtree()
252         file("f/a", "w").write("b")
253         wt.add("a")
254         wt.commit(message="data", rev_id="svn-v1:1@%s-" % oldrepos.uuid)
255         file("f/a", 'w').write("moredata")
256         wt.commit(message='fix moredata', rev_id="customrev")
257         file("f/a", 'w').write("blackfield")
258         wt.commit(message='fix it again', rev_id="anotherrev")
259
260         upgrade_branch(b, oldrepos, allow_changes=True)
261         self.assertEqual([oldrepos.generate_revision_id(0, "", "none"),
262                           oldrepos.generate_revision_id(1, "", "none"),
263                           "customrev-svn%d-upgrade" % MAPPING_VERSION,
264                           "anotherrev-svn%d-upgrade" % MAPPING_VERSION
265                           ], b.revision_history())
266
267     @skip_no_rebase
268     def test_workingtree(self):
269         repos_url = self.make_client("a", "dc")
270         self.build_tree({'dc/a': 'b'})
271         self.client_add("dc/a")
272         self.client_commit("dc", "data")
273
274         oldrepos = Repository.open(repos_url)
275         dir = BzrDir.create("f",format=get_rich_root_format())
276         newrepos = dir.create_repository()
277         b = dir.create_branch()
278         wt = dir.create_workingtree()
279         file("f/a", "w").write("b")
280         wt.add("a")
281         wt.commit(message="data", rev_id="svn-v1:1@%s-" % oldrepos.uuid)
282         file("f/a", 'w').write("moredata")
283         wt.commit(message='fix moredata', rev_id="customrev")
284         file("f/a", 'w').write("blackfield")
285         wt.commit(message='fix it again', rev_id="anotherrev")
286
287         upgrade_workingtree(wt, oldrepos, allow_changes=True)
288         self.assertEquals(wt.last_revision(), b.last_revision())
289         self.assertEqual([oldrepos.generate_revision_id(0, "", "none"),
290                           oldrepos.generate_revision_id(1, "", "none"),
291                           "customrev-svn%d-upgrade" % MAPPING_VERSION,
292                           "anotherrev-svn%d-upgrade" % MAPPING_VERSION
293                           ], b.revision_history())
294
295     @skip_no_rebase
296     def test_branch_none(self):
297         repos_url = self.make_client("a", "dc")
298         self.build_tree({'dc/a': 'b'})
299         self.client_add("dc/a")
300         self.client_commit("dc", "data")
301
302         oldrepos = Repository.open(repos_url)
303         dir = BzrDir.create("f",format=get_rich_root_format())
304         dir.create_repository()
305         b = dir.create_branch()
306         wt = dir.create_workingtree()
307         file("f/a", "w").write("b")
308         wt.add("a")
309         wt.commit(message="data", rev_id="blarev")
310         file("f/a", 'w').write("moredata")
311         wt.commit(message='fix moredata', rev_id="customrev")
312         file("f/a", 'w').write("blackfield")
313         wt.commit(message='fix it again', rev_id="anotherrev")
314
315         upgrade_branch(b, oldrepos)
316         self.assertEqual(["blarev", "customrev", "anotherrev"],
317                 b.revision_history())
318
319     @skip_no_rebase
320     def test_raise_incompat(self):
321         repos_url = self.make_client("a", "dc")
322         self.build_tree({'dc/d': 'e'})
323         self.client_add("dc/d")
324         self.client_commit("dc", "data")
325
326         oldrepos = Repository.open(repos_url)
327         dir = BzrDir.create("f",format=get_rich_root_format())
328         dir.create_repository()
329         b = dir.create_branch()
330         wt = dir.create_workingtree()
331         file("f/a", "w").write("c")
332         wt.add("a")
333         wt.commit(message="data", rev_id="svn-v1:1@%s-" % oldrepos.uuid)
334
335         self.assertRaises(UpgradeChangesContent, upgrade_branch, b, oldrepos)
336
337
338 class TestGenerateUpdateMapTests(TestCase):
339     def test_nothing(self):
340         self.assertEquals({}, generate_upgrade_map(["bla", "bloe"]))
341
342     def test_v2(self):
343         self.assertEquals({"svn-v2:12@65390229-12b7-0310-b90b-f21a5aa7ec8e-trunk": "svn-v3-trunk0:65390229-12b7-0310-b90b-f21a5aa7ec8e:trunk:12"}, generate_upgrade_map(["svn-v2:12@65390229-12b7-0310-b90b-f21a5aa7ec8e-trunk", "bloe", "blaaa"]))