1
"""Import upstream source into a branch"""
3
from bz2 import BZ2File
6
from shutil import rmtree
7
from StringIO import StringIO
9
from unittest import makeSuite
11
from bzrlib.bzrdir import BzrDir
12
from bzrlib.errors import NoSuchFile, BzrCommandError, NotBranchError
13
from bzrlib.osutils import pathjoin, isdir, file_iterator
14
from bzrlib.tests import TestCaseInTempDir
15
from bzrlib.trace import warning
16
from bzrlib.transform import TreeTransform, resolve_conflicts, cook_conflicts
17
from bzrlib.workingtree import WorkingTree
20
def top_directory(path):
21
"""Return the top directory given in a path."""
22
dirname = os.path.dirname(path)
23
last_dirname = dirname
25
dirname = os.path.dirname(dirname)
26
if dirname == '' or dirname == last_dirname:
28
last_dirname = dirname
31
def common_directory(names):
32
"""Determine a single directory prefix from a list of names"""
33
possible_prefix = None
35
name_top = top_directory(name)
36
if possible_prefix is None:
37
possible_prefix = name_top
39
if name_top != possible_prefix:
41
return possible_prefix
44
def do_directory(tt, trans_id, tree, relative_path, path):
45
if isdir(path) and tree.path2id(relative_path) is not None:
46
tt.cancel_deletion(trans_id)
48
tt.create_directory(trans_id)
51
def add_implied_parents(implied_parents, path):
52
"""Update the set of implied parents from a path"""
53
parent = os.path.dirname(path)
54
if parent in implied_parents:
56
implied_parents.add(parent)
57
add_implied_parents(implied_parents, parent)
60
def names_of_files(tar_file):
61
for member in tar_file.getmembers():
62
if member.type != "g":
66
def import_tar(tree, tar_input):
67
"""Replace the contents of a working directory with tarfile contents.
68
The tarfile may be a gzipped stream. File ids will be updated.
70
tar_file = tarfile.open('lala', 'r', tar_input)
71
prefix = common_directory(names_of_files(tar_file))
72
tt = TreeTransform(tree)
75
for path, entry in tree.inventory.iter_entries():
76
trans_id = tt.trans_id_tree_path(path)
77
tt.delete_contents(trans_id)
81
implied_parents = set()
83
for member in tar_file.getmembers():
84
if member.type == 'g':
85
# type 'g' is a header
87
relative_path = member.name
88
if prefix is not None:
89
relative_path = relative_path[len(prefix)+1:]
90
if relative_path == '':
92
add_implied_parents(implied_parents, relative_path)
93
trans_id = tt.trans_id_tree_path(relative_path)
94
added.add(relative_path.rstrip('/'))
95
path = tree.abspath(relative_path)
96
if member.name in seen:
97
tt.cancel_creation(trans_id)
100
tt.create_file(file_iterator(tar_file.extractfile(member)),
103
do_directory(tt, trans_id, tree, relative_path, path)
105
tt.create_symlink(member.linkname, trans_id)
107
for relative_path in implied_parents.difference(added):
108
if relative_path == "":
110
trans_id = tt.trans_id_tree_path(relative_path)
111
path = tree.abspath(relative_path)
112
do_directory(tt, trans_id, tree, relative_path, path)
113
added.add(relative_path)
115
for conflict in cook_conflicts(resolve_conflicts(tt), tt):
118
update_ids(tree, added, removed)
121
def update_ids(tree, added, removed):
122
"""Make sure that all present files files have file_ids.
125
new = added.difference(removed)
126
deleted = removed.difference(added)
127
tree.add(sorted(new))
128
tree.remove(sorted(deleted, reverse=True))
131
def do_import(source, tree_directory=None):
132
"""Implementation of import command. Intended for UI only"""
133
if tree_directory is not None:
135
tree = WorkingTree.open(tree_directory)
136
except NotBranchError:
137
if not os.path.exists(tree_directory):
138
os.mkdir(tree_directory)
139
branch = BzrDir.create_branch_convenience(tree_directory)
140
tree = branch.bzrdir.open_workingtree()
142
tree = WorkingTree.open_containing('.')[0]
145
if tree.changes_from(tree.basis_tree()).has_changed():
146
raise BzrCommandError("Working tree has uncommitted changes.")
148
if (source.endswith('.tar') or source.endswith('.tar.gz') or
149
source.endswith('.tar.bz2')) or source.endswith('.tgz'):
151
if source.endswith('.bz2'):
152
tar_input = BZ2File(source, 'r')
153
tar_input = StringIO(tar_input.read())
155
tar_input = file(source, 'rb')
157
if e.errno == errno.ENOENT:
158
raise NoSuchFile(source)
160
import_tar(tree, tar_input)
166
class TestImport(TestCaseInTempDir):
168
def make_tar(self, mode='w'):
170
tar_file = tarfile.open('project-0.1.tar', mode, result)
171
os.mkdir('project-0.1')
172
tar_file.add('project-0.1')
173
os.mkdir('project-0.1/junk')
174
tar_file.add('project-0.1/junk')
176
f = file('project-0.1/README', 'wb')
179
tar_file.add('project-0.1/README')
181
f = file('project-0.1/FEEDME', 'wb')
184
tar_file.add('project-0.1/FEEDME')
187
rmtree('project-0.1')
193
tar_file = tarfile.open('project-0.2.tar', 'w', result)
194
os.mkdir('project-0.2')
195
tar_file.add('project-0.2')
197
os.mkdir('project-0.2/junk')
198
tar_file.add('project-0.2/junk')
200
f = file('project-0.2/README', 'wb')
203
tar_file.add('project-0.2/README')
206
tar_file = tarfile.open('project-0.2.tar', 'a', result)
207
tar_file.add('project-0.2/README')
209
rmtree('project-0.2')
212
def make_messed_tar(self):
214
tar_file = tarfile.open('project-0.1.tar', 'w', result)
215
os.mkdir('project-0.1')
216
tar_file.add('project-0.1')
218
os.mkdir('project-0.2')
219
tar_file.add('project-0.2')
221
f = file('project-0.1/README', 'wb')
224
tar_file.add('project-0.1/README')
226
rmtree('project-0.1')
230
def test_top_directory(self):
231
self.assertEqual(top_directory('ab/b/c'), 'ab')
232
self.assertEqual(top_directory('/etc'), '/')
234
def test_common_directory(self):
235
self.assertEqual(common_directory(['ab/c/d', 'ab/c/e']), 'ab')
236
self.assertIs(common_directory(['ab/c/d', 'ac/c/e']), None)
238
def test_untar(self):
239
tar_file = self.make_tar()
240
tree = BzrDir.create_standalone_workingtree('tree')
241
import_tar(tree, tar_file)
242
self.assertTrue(tree.path2id('README') is not None)
243
self.assertTrue(tree.path2id('FEEDME') is not None)
244
self.assertTrue(os.path.isfile(tree.abspath('README')))
245
self.assertEqual(tree.inventory[tree.path2id('README')].kind, 'file')
246
self.assertEqual(tree.inventory[tree.path2id('FEEDME')].kind, 'file')
248
f = file(tree.abspath('junk/food'), 'wb')
249
f.write('I like food\n')
252
tar_file = self.make_tar2()
253
import_tar(tree, tar_file)
254
self.assertTrue(tree.path2id('README') is not None)
255
self.assertTrue(not os.path.exists(tree.abspath('FEEDME')))
258
def test_untar2(self):
259
tar_file = self.make_messed_tar()
260
tree = BzrDir.create_standalone_workingtree('tree')
261
import_tar(tree, tar_file)
262
self.assertTrue(tree.path2id('project-0.1/README') is not None)
264
def test_untar_gzip(self):
265
tar_file = self.make_tar(mode='w:gz')
266
tree = BzrDir.create_standalone_workingtree('tree')
267
import_tar(tree, tar_file)
268
self.assertTrue(tree.path2id('README') is not None)
272
return makeSuite(TestImport)
added
removed
upstream_import.py
