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
if entry.parent_id is None:
78
trans_id = tt.trans_id_tree_path(path)
79
tt.delete_contents(trans_id)
83
implied_parents = set()
85
for member in tar_file.getmembers():
86
if member.type == 'g':
87
# type 'g' is a header
89
relative_path = member.name
90
if prefix is not None:
91
relative_path = relative_path[len(prefix)+1:]
92
if relative_path == '':
94
add_implied_parents(implied_parents, relative_path)
95
trans_id = tt.trans_id_tree_path(relative_path)
96
added.add(relative_path.rstrip('/'))
97
path = tree.abspath(relative_path)
98
if member.name in seen:
99
tt.cancel_creation(trans_id)
100
seen.add(member.name)
102
tt.create_file(file_iterator(tar_file.extractfile(member)),
105
do_directory(tt, trans_id, tree, relative_path, path)
107
tt.create_symlink(member.linkname, trans_id)
109
for relative_path in implied_parents.difference(added):
110
if relative_path == "":
112
trans_id = tt.trans_id_tree_path(relative_path)
113
path = tree.abspath(relative_path)
114
do_directory(tt, trans_id, tree, relative_path, path)
115
added.add(relative_path)
117
for conflict in cook_conflicts(resolve_conflicts(tt), tt):
120
update_ids(tree, added, removed)
123
def update_ids(tree, added, removed):
124
"""Make sure that all present files files have file_ids.
127
new = added.difference(removed)
128
deleted = removed.difference(added)
129
tree.add(sorted(new))
130
tree.remove(sorted(deleted, reverse=True))
133
def do_import(source, tree_directory=None):
134
"""Implementation of import command. Intended for UI only"""
135
if tree_directory is not None:
137
tree = WorkingTree.open(tree_directory)
138
except NotBranchError:
139
if not os.path.exists(tree_directory):
140
os.mkdir(tree_directory)
141
branch = BzrDir.create_branch_convenience(tree_directory)
142
tree = branch.bzrdir.open_workingtree()
144
tree = WorkingTree.open_containing('.')[0]
147
if tree.changes_from(tree.basis_tree()).has_changed():
148
raise BzrCommandError("Working tree has uncommitted changes.")
150
if (source.endswith('.tar') or source.endswith('.tar.gz') or
151
source.endswith('.tar.bz2')) or source.endswith('.tgz'):
153
if source.endswith('.bz2'):
154
tar_input = BZ2File(source, 'r')
155
tar_input = StringIO(tar_input.read())
157
tar_input = file(source, 'rb')
159
if e.errno == errno.ENOENT:
160
raise NoSuchFile(source)
162
import_tar(tree, tar_input)
168
class TestImport(TestCaseInTempDir):
170
def make_tar(self, mode='w'):
172
tar_file = tarfile.open('project-0.1.tar', mode, result)
173
os.mkdir('project-0.1')
174
tar_file.add('project-0.1')
175
os.mkdir('project-0.1/junk')
176
tar_file.add('project-0.1/junk')
178
f = file('project-0.1/README', 'wb')
181
tar_file.add('project-0.1/README')
183
f = file('project-0.1/FEEDME', 'wb')
186
tar_file.add('project-0.1/FEEDME')
189
rmtree('project-0.1')
195
tar_file = tarfile.open('project-0.2.tar', 'w', result)
196
os.mkdir('project-0.2')
197
tar_file.add('project-0.2')
199
os.mkdir('project-0.2/junk')
200
tar_file.add('project-0.2/junk')
202
f = file('project-0.2/README', 'wb')
205
tar_file.add('project-0.2/README')
208
tar_file = tarfile.open('project-0.2.tar', 'a', result)
209
tar_file.add('project-0.2/README')
211
rmtree('project-0.2')
214
def make_messed_tar(self):
216
tar_file = tarfile.open('project-0.1.tar', 'w', result)
217
os.mkdir('project-0.1')
218
tar_file.add('project-0.1')
220
os.mkdir('project-0.2')
221
tar_file.add('project-0.2')
223
f = file('project-0.1/README', 'wb')
226
tar_file.add('project-0.1/README')
228
rmtree('project-0.1')
232
def test_top_directory(self):
233
self.assertEqual(top_directory('ab/b/c'), 'ab')
234
self.assertEqual(top_directory('/etc'), '/')
236
def test_common_directory(self):
237
self.assertEqual(common_directory(['ab/c/d', 'ab/c/e']), 'ab')
238
self.assertIs(common_directory(['ab/c/d', 'ac/c/e']), None)
240
def test_untar(self):
241
tar_file = self.make_tar()
242
tree = BzrDir.create_standalone_workingtree('tree')
243
import_tar(tree, tar_file)
244
self.assertTrue(tree.path2id('README') is not None)
245
self.assertTrue(tree.path2id('FEEDME') is not None)
246
self.assertTrue(os.path.isfile(tree.abspath('README')))
247
self.assertEqual(tree.inventory[tree.path2id('README')].kind, 'file')
248
self.assertEqual(tree.inventory[tree.path2id('FEEDME')].kind, 'file')
250
f = file(tree.abspath('junk/food'), 'wb')
251
f.write('I like food\n')
254
tar_file = self.make_tar2()
255
import_tar(tree, tar_file)
256
self.assertTrue(tree.path2id('README') is not None)
257
self.assertTrue(not os.path.exists(tree.abspath('FEEDME')))
260
def test_untar2(self):
261
tar_file = self.make_messed_tar()
262
tree = BzrDir.create_standalone_workingtree('tree')
263
import_tar(tree, tar_file)
264
self.assertTrue(tree.path2id('project-0.1/README') is not None)
266
def test_untar_gzip(self):
267
tar_file = self.make_tar(mode='w:gz')
268
tree = BzrDir.create_standalone_workingtree('tree')
269
import_tar(tree, tar_file)
270
self.assertTrue(tree.path2id('README') is not None)
274
return makeSuite(TestImport)
added
removed
upstream_import.py
