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
12
from bzrlib.bzrdir import BzrDir
13
from bzrlib.errors import NoSuchFile, BzrCommandError, NotBranchError
14
from bzrlib.osutils import pathjoin, isdir, file_iterator
15
from bzrlib.tests import TestCaseInTempDir
16
from bzrlib.trace import warning
17
from bzrlib.transform import TreeTransform, resolve_conflicts, cook_conflicts
18
from bzrlib.workingtree import WorkingTree
20
class ZipFileWrapper(object):
22
def __init__(self, zipfile):
23
self.zipfile = zipfile
26
for info in self.zipfile.infolist():
27
yield ZipInfoWrapper(self.zipfile, info)
29
def extractfile(self, infowrapper):
30
return StringIO(self.zipfile.read(infowrapper.name))
32
def add(self, filename):
34
self.zipfile.writestr(filename+'/', '')
36
self.zipfile.write(filename)
43
class ZipInfoWrapper(object):
45
def __init__(self, zipfile, info):
48
self.name = info.filename
49
self.zipfile = zipfile
53
return bool(self.name.endswith('/'))
57
return not self.isdir()
60
def top_directory(path):
61
"""Return the top directory given in a path."""
62
dirname = os.path.dirname(path)
63
last_dirname = dirname
65
dirname = os.path.dirname(dirname)
66
if dirname == '' or dirname == last_dirname:
68
last_dirname = dirname
71
def common_directory(names):
72
"""Determine a single directory prefix from a list of names"""
73
possible_prefix = None
75
name_top = top_directory(name)
76
if possible_prefix is None:
77
possible_prefix = name_top
79
if name_top != possible_prefix:
81
return possible_prefix
84
def do_directory(tt, trans_id, tree, relative_path, path):
85
if isdir(path) and tree.path2id(relative_path) is not None:
86
tt.cancel_deletion(trans_id)
88
tt.create_directory(trans_id)
91
def add_implied_parents(implied_parents, path):
92
"""Update the set of implied parents from a path"""
93
parent = os.path.dirname(path)
94
if parent in implied_parents:
96
implied_parents.add(parent)
97
add_implied_parents(implied_parents, parent)
100
def names_of_files(tar_file):
101
for member in tar_file.getmembers():
102
if member.type != "g":
106
def import_tar(tree, tar_input):
107
"""Replace the contents of a working directory with tarfile contents.
108
The tarfile may be a gzipped stream. File ids will be updated.
110
tar_file = tarfile.open('lala', 'r', tar_input)
111
import_archive(tree, tar_file)
113
def import_zip(tree, zip_input):
114
zip_file = ZipFileWrapper(zipfile.ZipFile(zip_input))
115
import_archive(tree, zip_file)
117
def import_archive(tree, archive_file):
118
prefix = common_directory(names_of_files(archive_file))
119
tt = TreeTransform(tree)
122
for path, entry in tree.inventory.iter_entries():
123
if entry.parent_id is None:
125
trans_id = tt.trans_id_tree_path(path)
126
tt.delete_contents(trans_id)
130
implied_parents = set()
132
for member in archive_file.getmembers():
133
if member.type == 'g':
134
# type 'g' is a header
136
relative_path = member.name
137
if prefix is not None:
138
relative_path = relative_path[len(prefix)+1:]
139
if relative_path == '':
141
add_implied_parents(implied_parents, relative_path)
142
trans_id = tt.trans_id_tree_path(relative_path)
143
added.add(relative_path.rstrip('/'))
144
path = tree.abspath(relative_path)
145
if member.name in seen:
146
tt.cancel_creation(trans_id)
147
seen.add(member.name)
149
tt.create_file(file_iterator(archive_file.extractfile(member)),
152
do_directory(tt, trans_id, tree, relative_path, path)
154
tt.create_symlink(member.linkname, trans_id)
156
for relative_path in implied_parents.difference(added):
157
if relative_path == "":
159
trans_id = tt.trans_id_tree_path(relative_path)
160
path = tree.abspath(relative_path)
161
do_directory(tt, trans_id, tree, relative_path, path)
162
added.add(relative_path)
164
for conflict in cook_conflicts(resolve_conflicts(tt), tt):
167
update_ids(tree, added, removed)
170
def update_ids(tree, added, removed):
171
"""Make sure that all present files files have file_ids.
174
new = added.difference(removed)
175
deleted = removed.difference(added)
176
tree.add(sorted(new))
177
tree.remove(sorted(deleted, reverse=True))
180
def do_import(source, tree_directory=None):
181
"""Implementation of import command. Intended for UI only"""
182
if tree_directory is not None:
184
tree = WorkingTree.open(tree_directory)
185
except NotBranchError:
186
if not os.path.exists(tree_directory):
187
os.mkdir(tree_directory)
188
branch = BzrDir.create_branch_convenience(tree_directory)
189
tree = branch.bzrdir.open_workingtree()
191
tree = WorkingTree.open_containing('.')[0]
194
if tree.changes_from(tree.basis_tree()).has_changed():
195
raise BzrCommandError("Working tree has uncommitted changes.")
197
if (source.endswith('.tar') or source.endswith('.tar.gz') or
198
source.endswith('.tar.bz2')) or source.endswith('.tgz'):
200
if source.endswith('.bz2'):
201
tar_input = BZ2File(source, 'r')
202
tar_input = StringIO(tar_input.read())
204
tar_input = file(source, 'rb')
206
if e.errno == errno.ENOENT:
207
raise NoSuchFile(source)
209
import_tar(tree, tar_input)
212
elif source.endswith('.zip'):
213
import_zip(tree, open(source, 'rb'))
215
raise BzrCommandError('Unhandled import source')
219
class TestImport(TestCaseInTempDir):
221
def make_tar(self, mode='w'):
223
return tarfile.open('project-0.1.tar', mode, fileobj)
224
return self.make_archive(maker)
226
def make_archive(self, maker):
228
archive_file = maker(result)
229
os.mkdir('project-0.1')
230
archive_file.add('project-0.1')
231
os.mkdir('project-0.1/junk')
232
archive_file.add('project-0.1/junk')
234
f = file('project-0.1/README', 'wb')
237
archive_file.add('project-0.1/README')
239
f = file('project-0.1/FEEDME', 'wb')
242
archive_file.add('project-0.1/FEEDME')
245
rmtree('project-0.1')
251
tar_file = tarfile.open('project-0.2.tar', 'w', result)
252
os.mkdir('project-0.2')
253
tar_file.add('project-0.2')
255
os.mkdir('project-0.2/junk')
256
tar_file.add('project-0.2/junk')
258
f = file('project-0.2/README', 'wb')
261
tar_file.add('project-0.2/README')
264
tar_file = tarfile.open('project-0.2.tar', 'a', result)
265
tar_file.add('project-0.2/README')
267
rmtree('project-0.2')
270
def make_messed_tar(self):
272
tar_file = tarfile.open('project-0.1.tar', 'w', result)
273
os.mkdir('project-0.1')
274
tar_file.add('project-0.1')
276
os.mkdir('project-0.2')
277
tar_file.add('project-0.2')
279
f = file('project-0.1/README', 'wb')
282
tar_file.add('project-0.1/README')
284
rmtree('project-0.1')
290
return ZipFileWrapper(zipfile.ZipFile(fileobj, 'w'))
291
return self.make_archive(maker)
293
def test_top_directory(self):
294
self.assertEqual(top_directory('ab/b/c'), 'ab')
295
self.assertEqual(top_directory('/etc'), '/')
297
def test_common_directory(self):
298
self.assertEqual(common_directory(['ab/c/d', 'ab/c/e']), 'ab')
299
self.assertIs(common_directory(['ab/c/d', 'ac/c/e']), None)
301
def test_untar(self):
302
tar_file = self.make_tar()
303
tree = BzrDir.create_standalone_workingtree('tree')
304
import_tar(tree, tar_file)
305
self.assertTrue(tree.path2id('README') is not None)
306
self.assertTrue(tree.path2id('FEEDME') is not None)
307
self.assertTrue(os.path.isfile(tree.abspath('README')))
308
self.assertEqual(tree.inventory[tree.path2id('README')].kind, 'file')
309
self.assertEqual(tree.inventory[tree.path2id('FEEDME')].kind, 'file')
311
f = file(tree.abspath('junk/food'), 'wb')
312
f.write('I like food\n')
315
tar_file = self.make_tar2()
316
import_tar(tree, tar_file)
317
self.assertTrue(tree.path2id('README') is not None)
318
self.assertTrue(not os.path.exists(tree.abspath('FEEDME')))
321
def test_untar2(self):
322
tar_file = self.make_messed_tar()
323
tree = BzrDir.create_standalone_workingtree('tree')
324
import_tar(tree, tar_file)
325
self.assertTrue(tree.path2id('project-0.1/README') is not None)
327
def test_untar_gzip(self):
328
tar_file = self.make_tar(mode='w:gz')
329
tree = BzrDir.create_standalone_workingtree('tree')
330
import_tar(tree, tar_file)
331
self.assertTrue(tree.path2id('README') is not None)
333
def test_unzip(self):
334
zip_file = self.make_zip()
335
tree = BzrDir.create_standalone_workingtree('tree')
336
import_zip(tree, zip_file)
337
self.assertTrue(tree.path2id('README') is not None)
338
self.assertTrue(tree.path2id('FEEDME') is not None)
339
self.assertTrue(os.path.isfile(tree.abspath('README')))
340
self.assertEqual(tree.inventory[tree.path2id('README')].kind, 'file')
341
self.assertEqual(tree.inventory[tree.path2id('FEEDME')].kind, 'file')
343
f = file(tree.abspath('junk/food'), 'wb')
344
f.write('I like food\n')
349
return makeSuite(TestImport)
added
removed
upstream_import.py
