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.delta import compare_trees
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
21
def top_directory(path):
22
"""Return the top directory given in a path."""
23
dirname = os.path.dirname(path)
24
last_dirname = dirname
26
dirname = os.path.dirname(dirname)
27
if dirname == '' or dirname == last_dirname:
29
last_dirname = dirname
32
def common_directory(names):
33
"""Determine a single directory prefix from a list of names"""
34
possible_prefix = None
36
name_top = top_directory(name)
37
if possible_prefix is None:
38
possible_prefix = name_top
40
if name_top != possible_prefix:
42
return possible_prefix
45
def do_directory(tt, trans_id, tree, relative_path, path):
46
if isdir(path) and tree.path2id(relative_path) is not None:
47
tt.cancel_deletion(trans_id)
49
tt.create_directory(trans_id)
52
def add_implied_parents(implied_parents, path):
53
"""Update the set of implied parents from a path"""
54
parent = os.path.dirname(path)
55
if parent in implied_parents:
57
implied_parents.add(parent)
58
add_implied_parents(implied_parents, parent)
61
def names_of_files(tar_file):
62
for member in tar_file.getmembers():
63
if member.type != "g":
67
def import_tar(tree, tar_input):
68
"""Replace the contents of a working directory with tarfile contents.
69
The tarfile may be a gzipped stream. File ids will be updated.
71
tar_file = tarfile.open('lala', 'r', tar_input)
72
prefix = common_directory(names_of_files(tar_file))
73
tt = TreeTransform(tree)
76
for path, entry in tree.inventory.iter_entries():
77
trans_id = tt.trans_id_tree_path(path)
78
tt.delete_contents(trans_id)
82
implied_parents = set()
84
for member in tar_file.getmembers():
85
if member.type == 'g':
86
# type 'g' is a header
88
relative_path = member.name
89
if prefix is not None:
90
relative_path = relative_path[len(prefix)+1:]
91
if relative_path == '':
93
add_implied_parents(implied_parents, relative_path)
94
trans_id = tt.trans_id_tree_path(relative_path)
95
added.add(relative_path.rstrip('/'))
96
path = tree.abspath(relative_path)
97
if member.name in seen:
98
tt.cancel_creation(trans_id)
101
tt.create_file(file_iterator(tar_file.extractfile(member)),
104
do_directory(tt, trans_id, tree, relative_path, path)
106
tt.create_symlink(member.linkname, trans_id)
108
for relative_path in implied_parents.difference(added):
109
if relative_path == "":
111
trans_id = tt.trans_id_tree_path(relative_path)
112
path = tree.abspath(relative_path)
113
do_directory(tt, trans_id, tree, relative_path, path)
114
added.add(relative_path)
116
for conflict in cook_conflicts(resolve_conflicts(tt), tt):
119
update_ids(tree, added, removed)
122
def update_ids(tree, added, removed):
123
"""Make sure that all present files files have file_ids.
126
new = added.difference(removed)
127
deleted = removed.difference(added)
128
tree.add(sorted(new))
129
tree.remove(sorted(deleted, reverse=True))
132
def do_import(source, tree_directory=None):
133
"""Implementation of import command. Intended for UI only"""
134
if tree_directory is not None:
136
tree = WorkingTree.open(tree_directory)
137
except NotBranchError:
138
if not os.path.exists(tree_directory):
139
os.mkdir(tree_directory)
140
branch = BzrDir.create_branch_convenience(tree_directory)
141
tree = branch.bzrdir.open_workingtree()
143
tree = WorkingTree.open_containing('.')[0]
146
if compare_trees(tree, tree.basis_tree()).has_changed():
147
raise BzrCommandError("Working tree has uncommitted changes.")
149
if (source.endswith('.tar') or source.endswith('.tar.gz') or
150
source.endswith('.tar.bz2')) or source.endswith('.tgz'):
152
if source.endswith('.bz2'):
153
tar_input = BZ2File(source, 'r')
154
tar_input = StringIO(tar_input.read())
156
tar_input = file(source, 'rb')
158
if e.errno == errno.ENOENT:
159
raise NoSuchFile(source)
161
import_tar(tree, tar_input)
167
class TestImport(TestCaseInTempDir):
169
def make_tar(self, mode='w'):
171
tar_file = tarfile.open('project-0.1.tar', mode, result)
172
os.mkdir('project-0.1')
173
tar_file.add('project-0.1')
174
os.mkdir('project-0.1/junk')
175
tar_file.add('project-0.1/junk')
177
f = file('project-0.1/README', 'wb')
180
tar_file.add('project-0.1/README')
182
f = file('project-0.1/FEEDME', 'wb')
185
tar_file.add('project-0.1/FEEDME')
188
rmtree('project-0.1')
194
tar_file = tarfile.open('project-0.2.tar', 'w', result)
195
os.mkdir('project-0.2')
196
tar_file.add('project-0.2')
198
os.mkdir('project-0.2/junk')
199
tar_file.add('project-0.2/junk')
201
f = file('project-0.2/README', 'wb')
204
tar_file.add('project-0.2/README')
207
tar_file = tarfile.open('project-0.2.tar', 'a', result)
208
tar_file.add('project-0.2/README')
210
rmtree('project-0.2')
213
def make_messed_tar(self):
215
tar_file = tarfile.open('project-0.1.tar', 'w', result)
216
os.mkdir('project-0.1')
217
tar_file.add('project-0.1')
219
os.mkdir('project-0.2')
220
tar_file.add('project-0.2')
222
f = file('project-0.1/README', 'wb')
225
tar_file.add('project-0.1/README')
227
rmtree('project-0.1')
231
def test_top_directory(self):
232
self.assertEqual(top_directory('ab/b/c'), 'ab')
233
self.assertEqual(top_directory('/etc'), '/')
235
def test_common_directory(self):
236
self.assertEqual(common_directory(['ab/c/d', 'ab/c/e']), 'ab')
237
self.assertIs(common_directory(['ab/c/d', 'ac/c/e']), None)
239
def test_untar(self):
240
tar_file = self.make_tar()
241
tree = BzrDir.create_standalone_workingtree('tree')
242
import_tar(tree, tar_file)
243
self.assertTrue(tree.path2id('README') is not None)
244
self.assertTrue(tree.path2id('FEEDME') is not None)
245
self.assertTrue(os.path.isfile(tree.abspath('README')))
246
self.assertEqual(tree.inventory[tree.path2id('README')].kind, 'file')
247
self.assertEqual(tree.inventory[tree.path2id('FEEDME')].kind, 'file')
249
f = file(tree.abspath('junk/food'), 'wb')
250
f.write('I like food\n')
253
tar_file = self.make_tar2()
254
import_tar(tree, tar_file)
255
self.assertTrue(tree.path2id('README') is not None)
256
self.assertTrue(not os.path.exists(tree.abspath('FEEDME')))
259
def test_untar2(self):
260
tar_file = self.make_messed_tar()
261
tree = BzrDir.create_standalone_workingtree('tree')
262
import_tar(tree, tar_file)
263
self.assertTrue(tree.path2id('project-0.1/README') is not None)
265
def test_untar_gzip(self):
266
tar_file = self.make_tar(mode='w:gz')
267
tree = BzrDir.create_standalone_workingtree('tree')
268
import_tar(tree, tar_file)
269
self.assertTrue(tree.path2id('README') is not None)
273
return makeSuite(TestImport)
added
removed
upstream_import.py
