~bzr-pqm/bzr/bzr.dev : revision 4524

15

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

16

17

18

from bzrlib import errors, chk_map, inventory, osutils

18

from bzrlib import (

19

chk_map,

20

bzrdir,

21

errors,

22

inventory,

23

osutils,

24

repository,

25

revision,

26

)

19

27

from bzrlib.inventory import (CHKInventory, Inventory, ROOT_ID, InventoryFile,

20

28

InventoryDirectory, InventoryEntry, TreeReference)

21

from bzrlib.tests import TestCase, TestCaseWithTransport

29

from bzrlib.tests import (

30

TestCase,

31

TestCaseWithTransport,

32

condition_isinstance,

33

multiply_tests,

34

split_suite_by_condition,

35

)

36

from bzrlib.tests.workingtree_implementations import workingtree_formats

37

38

39

def load_tests(standard_tests, module, loader):

40

"""Parameterise some inventory tests."""

41

to_adapt, result = split_suite_by_condition(standard_tests,

42

condition_isinstance(TestDeltaApplication))

43

scenarios = [

44

('Inventory', {'apply_delta':apply_inventory_Inventory}),

45

]

46

# Working tree basis delta application

47

# Repository add_inv_by_delta.

48

# Reduce form of the per_repository test logic - that logic needs to be

49

# be able to get /just/ repositories whereas these tests are fine with

50

# just creating trees.

51

formats = set()

52

for _, format in repository.format_registry.iteritems():

53

scenarios.append((str(format.__name__), {

54

'apply_delta':apply_inventory_Repository_add_inventory_by_delta,

55

'format':format}))

56

for format in workingtree_formats():

57

scenarios.append((str(format.__class__.__name__), {

58

'apply_delta':apply_inventory_WT_basis,

59

'format':format}))

60

return multiply_tests(to_adapt, scenarios, result)

61

62

63

def apply_inventory_Inventory(self, basis, delta):

64

"""Apply delta to basis and return the result.

65

66

:param basis: An inventory to be used as the basis.

67

:param delta: The inventory delta to apply:

68

:return: An inventory resulting from the application.

69

"""

70

basis.apply_delta(delta)

71

return basis

72

73

74

def apply_inventory_WT_basis(self, basis, delta):

75

"""Apply delta to basis and return the result.

76

77

This sets the parent and then calls update_basis_by_delta.

78

It also puts the basis in the repository under both 'basis' and 'result' to

79

allow safety checks made by the WT to succeed, and finally ensures that all

80

items in the delta with a new path are present in the WT before calling

81

update_basis_by_delta.

82

83

:param basis: An inventory to be used as the basis.

84

:param delta: The inventory delta to apply:

85

:return: An inventory resulting from the application.

86

"""

87

control = self.make_bzrdir('tree', format=self.format._matchingbzrdir)

88

control.create_repository()

89

control.create_branch()

90

tree = self.format.initialize(control)

91

tree.lock_write()

92

try:

93

repo = tree.branch.repository

94

repo.start_write_group()

95

try:

96

rev = revision.Revision('basis', timestamp=0, timezone=None,

97

message="", committer="foo@example.com")

98

basis.revision_id = 'basis'

99

repo.add_revision('basis', rev, basis)

100

# Add a revision for the result, with the basis content -

101

# update_basis_by_delta doesn't check that the delta results in

102

# result, and we want inconsistent deltas to get called on the

103

# tree, or else the code isn't actually checked.

104

rev = revision.Revision('result', timestamp=0, timezone=None,

105

message="", committer="foo@example.com")

106

basis.revision_id = 'result'

107

repo.add_revision('result', rev, basis)

108

except:

109

repo.abort_write_group()

110

raise

111

else:

112

repo.commit_write_group()

113

# Set the basis state as the trees current state

114

tree._write_inventory(basis)

115

# This reads basis from the repo and puts it into the tree's local

116

# cache, if it has one.

117

tree.set_parent_ids(['basis'])

118

paths = {}

119

parents = set()

120

for old, new, id, entry in delta:

121

if entry is None:

122

continue

123

paths[new] = (entry.file_id, entry.kind)

124

parents.add(osutils.dirname(new))

125

parents = osutils.minimum_path_selection(parents)

126

parents.discard('')

127

# Put place holders in the tree to permit adding the other entries.

128

for pos, parent in enumerate(parents):

129

if not tree.path2id(parent):

130

# add a synthetic directory in the tree so we can can put the

131

# tree0 entries in place for dirstate.

132

tree.add([parent], ["id%d" % pos], ["directory"])

133

if paths:

134

# Many deltas may cause this mini-apply to fail, but we want to see what

135

# the delta application code says, not the prep that we do to deal with

136

# limitations of dirstate's update_basis code.

137

for path, (file_id, kind) in sorted(paths.items()):

138

try:

139

tree.add([path], [file_id], [kind])

140

except (KeyboardInterrupt, SystemExit):

141

raise

142

except:

143

pass

144

finally:

145

tree.unlock()

146

# Fresh lock, reads disk again.

147

tree.lock_write()

148

try:

149

tree.update_basis_by_delta('result', delta)

150

finally:

151

tree.unlock()

152

# reload tree - ensure we get what was written.

153

tree = tree.bzrdir.open_workingtree()

154

basis_tree = tree.basis_tree()

155

basis_tree.lock_read()

156

self.addCleanup(basis_tree.unlock)

157

# Note, that if the tree does not have a local cache, the trick above of

158

# setting the result as the basis, will come back to bite us. That said,

159

# all the implementations in bzr do have a local cache.

160

return basis_tree.inventory

161

162

163

def apply_inventory_Repository_add_inventory_by_delta(self, basis, delta):

164

"""Apply delta to basis and return the result.

165

166

This inserts basis as a whole inventory and then uses

167

add_inventory_by_delta to add delta.

168

169

:param basis: An inventory to be used as the basis.

170

:param delta: The inventory delta to apply:

171

:return: An inventory resulting from the application.

172

"""

173

format = self.format()

174

control = self.make_bzrdir('tree', format=format._matchingbzrdir)

175

repo = format.initialize(control)

176

repo.lock_write()

177

try:

178

repo.start_write_group()

179

try:

180

rev = revision.Revision('basis', timestamp=0, timezone=None,

181

message="", committer="foo@example.com")

182

basis.revision_id = 'basis'

183

repo.add_revision('basis', rev, basis)

184

except:

185

repo.abort_write_group()

186

raise

187

else:

188

repo.commit_write_group()

189

finally:

190

repo.unlock()

191

repo.lock_write()

192

try:

193

repo.start_write_group()

194

try:

195

inv_sha1 = repo.add_inventory_by_delta('basis', delta,

196

'result', ['basis'])

197

except:

198

repo.abort_write_group()

199

raise

200

else:

201

repo.commit_write_group()

202

finally:

203

repo.unlock()

204

# Fresh lock, reads disk again.

205

repo = repo.bzrdir.open_repository()

206

repo.lock_read()

207

self.addCleanup(repo.unlock)

208

return repo.get_inventory('result')

209

210

211

class TestDeltaApplication(TestCaseWithTransport):

212

213

def get_empty_inventory(self, reference_inv=None):

214

"""Get an empty inventory.

215

216

Note that tests should not depend on the revision of the root for

217

setting up test conditions, as it has to be flexible to accomodate non

218

rich root repositories.

219

220

:param reference_inv: If not None, get the revision for the root from

221

this inventory. This is useful for dealing with older repositories

222

that routinely discarded the root entry data. If None, the root's

223

revision is set to 'basis'.

224

"""

225

inv = inventory.Inventory()

226

if reference_inv is not None:

227

inv.root.revision = reference_inv.root.revision

228

else:

229

inv.root.revision = 'basis'

230

return inv

231

232

def test_empty_delta(self):

233

inv = self.get_empty_inventory()

234

delta = []

235

inv = self.apply_delta(self, inv, delta)

236

inv2 = self.get_empty_inventory(inv)

237

self.assertEqual([], inv2._make_delta(inv))

238

239

def test_repeated_file_id(self):

240

inv = self.get_empty_inventory()

241

file1 = inventory.InventoryFile('id', 'path1', inv.root.file_id)

242

file1.revision = 'result'

243

file1.text_size = 0

244

file1.text_sha1 = ""

245

file2 = inventory.InventoryFile('id', 'path2', inv.root.file_id)

246

file2.revision = 'result'

247

file2.text_size = 0

248

file2.text_sha1 = ""

249

delta = [(None, u'path1', 'id', file1), (None, u'path2', 'id', file2)]

250

self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,

251

inv, delta)

252

253

def test_repeated_new_path(self):

254

inv = self.get_empty_inventory()

255

file1 = inventory.InventoryFile('id1', 'path', inv.root.file_id)

256

file1.revision = 'result'

257

file1.text_size = 0

258

file1.text_sha1 = ""

259

file2 = inventory.InventoryFile('id2', 'path', inv.root.file_id)

260

file2.revision = 'result'

261

file2.text_size = 0

262

file2.text_sha1 = ""

263

delta = [(None, u'path', 'id1', file1), (None, u'path', 'id2', file2)]

264

self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,

265

inv, delta)

266

267

def test_repeated_old_path(self):

268

inv = self.get_empty_inventory()

269

file1 = inventory.InventoryFile('id1', 'path', inv.root.file_id)

270

file1.revision = 'result'

271

file1.text_size = 0

272

file1.text_sha1 = ""

273

# We can't *create* a source inventory with the same path, but

274

# a badly generated partial delta might claim the same source twice.

275

# This would be buggy in two ways: the path is repeated in the delta,

276

# And the path for one of the file ids doesn't match the source

277

# location. Alternatively, we could have a repeated fileid, but that

278

# is separately checked for.

279

file2 = inventory.InventoryFile('id2', 'path2', inv.root.file_id)

280

file2.revision = 'result'

281

file2.text_size = 0

282

file2.text_sha1 = ""

283

inv.add(file1)

284

inv.add(file2)

285

delta = [(u'path', None, 'id1', None), (u'path', None, 'id2', None)]

286

self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,

287

inv, delta)

288

289

def test_mismatched_id_entry_id(self):

290

inv = self.get_empty_inventory()

291

file1 = inventory.InventoryFile('id1', 'path', inv.root.file_id)

292

file1.revision = 'result'

293

file1.text_size = 0

294

file1.text_sha1 = ""

295

delta = [(None, u'path', 'id', file1)]

296

self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,

297

inv, delta)

298

299

def test_parent_is_not_directory(self):

300

inv = self.get_empty_inventory()

301

file1 = inventory.InventoryFile('id1', 'path', inv.root.file_id)

302

file1.revision = 'result'

303

file1.text_size = 0

304

file1.text_sha1 = ""

305

file2 = inventory.InventoryFile('id2', 'path2', 'id1')

306

file2.revision = 'result'

307

file2.text_size = 0

308

file2.text_sha1 = ""

309

inv.add(file1)

310

delta = [(None, u'path/path2', 'id2', file2)]

311

self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,

312

inv, delta)

313

314

def test_parent_is_missing(self):

315

inv = self.get_empty_inventory()

316

file2 = inventory.InventoryFile('id2', 'path2', 'missingparent')

317

file2.revision = 'result'

318

file2.text_size = 0

319

file2.text_sha1 = ""

320

delta = [(None, u'path/path2', 'id2', file2)]

321

self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,

322

inv, delta)

22

323

23

324

24

325

class TestInventoryEntry(TestCase):