~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/teststore.py

• browse files at revision 1442.1.66
• compare with another revision
• download diff
• download tarball
• view history from revision 1442.1.66

• reverse the comparison (1442.1.66 to 1692.7.3)
• stop comparing with revision 1692.7.3

Show diffs side-by-side

added

removed

bzrlib/selftest/teststore.py

18

18

 

19

19

from cStringIO import StringIO

20

20

import os

21

 

import gzip

22

21

 

23

 

import bzrlib.errors as errors

24

 

from bzrlib.errors import BzrError, UnlistableStore, NoSuchFile

 

22

from bzrlib.errors import BzrError, UnlistableStore

 

23

from bzrlib.store import copy_all

25

24

from bzrlib.transport.local import LocalTransport

 

25

from bzrlib.transport import NoSuchFile

 

26

from bzrlib.store.compressed_text import CompressedTextStore

26

27

from bzrlib.store.text import TextStore

27

 

from bzrlib.tests import TestCase, TestCaseInTempDir, TestCaseWithTransport

 

28

from bzrlib.selftest import TestCase, TestCaseInTempDir

28

29

import bzrlib.store as store

29

 

import bzrlib.transactions as transactions

30

30

import bzrlib.transport as transport

31

31

from bzrlib.transport.memory import MemoryTransport

32

32

 

33

33

 

34

34

class TestStores(object):

35

 

    """Mixin template class that provides some common tests for stores"""

36

35

 

37

36

    def check_content(self, store, fileid, value):

38

37

        f = store.get(fileid)

51

50

        store_a.add('foo', '1')

52

51

        os.mkdir('b')

53

52

        store_b = self.get_store('b')

54

 

        store_b.copy_all_ids(store_a)

 

53

        copy_all(store_a, store_b)

55

54

        self.assertEqual(store_a.get('1').read(), 'foo')

56

55

        self.assertEqual(store_b.get('1').read(), 'foo')

57

56

        # TODO: Switch the exception form UnlistableStore to

80

79

 

81

80

class TestCompressedTextStore(TestCaseInTempDir, TestStores):

82

81

 

83

 

    def get_store(self, path=u'.'):

 

82

    def get_store(self, path='.'):

84

83

        t = LocalTransport(path)

85

 

        return TextStore(t, compressed=True)

 

84

        return CompressedTextStore(t)

86

85

 

87

86

    def test_total_size(self):

88

 

        store = self.get_store(u'.')

 

87

        store = self.get_store('.')

89

88

        store.register_suffix('dsc')

90

89

        store.add(StringIO('goodbye'), '123123')

91

90

        store.add(StringIO('goodbye2'), '123123', 'dsc')

93

92

        self.assertEqual(store.total_size(), (2, 55))

94

93

        

95

94

    def test__relpath_suffixed(self):

96

 

        my_store = TextStore(MockTransport(),

97

 

                             prefixed=True, compressed=True)

 

95

        my_store = CompressedTextStore(MockTransport(), True)

98

96

        my_store.register_suffix('dsc')

99

 

        self.assertEqual('45/foo.dsc', my_store._relpath('foo', ['dsc']))

 

97

        self.assertEqual('45/foo.dsc.gz', my_store._relpath('foo', ['dsc']))

100

98

 

101

99

 

102

100

class TestMemoryStore(TestCase):

103

101

    

104

102

    def get_store(self):

105

 

        return TextStore(MemoryTransport())

 

103

        return store.ImmutableMemoryStore()

106

104

    

 

105

    def test_imports(self):

 

106

        from bzrlib.store import ImmutableMemoryStore

 

107

 

107

108

    def test_add_and_retrieve(self):

108

109

        store = self.get_store()

109

110

        store.add(StringIO('hello'), 'aa')

137

138

 

138

139

class TestTextStore(TestCaseInTempDir, TestStores):

139

140

 

140

 

    def get_store(self, path=u'.'):

 

141

    def get_store(self, path='.'):

141

142

        t = LocalTransport(path)

142

 

        return TextStore(t, compressed=False)

 

143

        return TextStore(t)

143

144

 

144

145

    def test_total_size(self):

145

146

        store = self.get_store()

153

154

        # self.assertRaises(UnlistableStore, copy_all, store_c, store_b)

154

155

 

155

156

 

156

 

class TestMixedTextStore(TestCaseInTempDir, TestStores):

157

 

 

158

 

    def get_store(self, path=u'.', compressed=True):

159

 

        t = LocalTransport(path)

160

 

        return TextStore(t, compressed=compressed)

161

 

 

162

 

    def test_get_mixed(self):

163

 

        cs = self.get_store(u'.', compressed=True)

164

 

        s = self.get_store(u'.', compressed=False)

165

 

        cs.add(StringIO('hello there'), 'a')

166

 

 

167

 

        self.failUnlessExists('a.gz')

168

 

        self.failIf(os.path.lexists('a'))

169

 

 

170

 

        self.assertEquals(gzip.GzipFile('a.gz').read(), 'hello there')

171

 

 

172

 

        self.assertEquals(cs.has_id('a'), True)

173

 

        self.assertEquals(s.has_id('a'), True)

174

 

        self.assertEquals(cs.get('a').read(), 'hello there')

175

 

        self.assertEquals(s.get('a').read(), 'hello there')

176

 

        

177

 

        self.assertRaises(BzrError, s.add, StringIO('goodbye'), 'a')

178

 

 

179

 

        s.add(StringIO('goodbye'), 'b')

180

 

        self.failUnlessExists('b')

181

 

        self.failIf(os.path.lexists('b.gz'))

182

 

        self.assertEquals(open('b').read(), 'goodbye')

183

 

 

184

 

        self.assertEquals(cs.has_id('b'), True)

185

 

        self.assertEquals(s.has_id('b'), True)

186

 

        self.assertEquals(cs.get('b').read(), 'goodbye')

187

 

        self.assertEquals(s.get('b').read(), 'goodbye')

188

 

        

189

 

        self.assertRaises(BzrError, cs.add, StringIO('again'), 'b')

190

 

 

191

157

class MockTransport(transport.Transport):

192

158

    """A fake transport for testing with."""

193

159

 

265

231

 

266

232

    def test__relpath_simple_suffixed(self):

267

233

        my_store = store.TransportStore(MockTransport())

 

234

        my_store.register_suffix('gz')

268

235

        my_store.register_suffix('bar')

269

 

        my_store.register_suffix('baz')

270

 

        self.assertEqual('foo.baz', my_store._relpath('foo', ['baz']))

271

 

        self.assertEqual('foo.bar.baz', my_store._relpath('foo', ['bar', 'baz']))

 

236

        self.assertEqual('foo.gz', my_store._relpath('foo', ['gz']))

 

237

        self.assertEqual('foo.gz.bar', my_store._relpath('foo', ['gz', 'bar']))

272

238

 

273

239

    def test__relpath_prefixed_suffixed(self):

274

240

        my_store = store.TransportStore(MockTransport(), True)

 

241

        my_store.register_suffix('gz')

275

242

        my_store.register_suffix('bar')

276

 

        my_store.register_suffix('baz')

277

 

        self.assertEqual('45/foo.baz', my_store._relpath('foo', ['baz']))

278

 

        self.assertEqual('45/foo.bar.baz',

279

 

                         my_store._relpath('foo', ['bar', 'baz']))

 

243

        self.assertEqual('45/foo.gz', my_store._relpath('foo', ['gz']))

 

244

        self.assertEqual('45/foo.gz.bar',

 

245

                         my_store._relpath('foo', ['gz', 'bar']))

280

246

 

281

247

    def test_add_simple(self):

282

248

        stream = StringIO("content")

304

270

        my_store.add(stream, "foo", 'dsc')

305

271

        self.assertEqual([("_add", "45/foo.dsc", stream)], my_store._calls)

306

272

 

307

 

    def get_populated_store(self, prefixed=False,

308

 

            store_class=TextStore, compressed=False):

309

 

        my_store = store_class(MemoryTransport(), prefixed,

310

 

                               compressed=compressed)

 

273

    def get_populated_store(self, prefixed=False, store_class=TextStore):

 

274

        my_store = store_class(MemoryTransport(), prefixed)

311

275

        my_store.register_suffix('sig')

312

276

        stream = StringIO("signature")

313

277

        my_store.add(stream, "foo", 'sig')

356

320

                         my_store.get('missing', 'sig').read())

357

321

 

358

322

    def test___iter__no_suffix(self):

359

 

        my_store = TextStore(MemoryTransport(),

360

 

                             prefixed=False, compressed=False)

 

323

        my_store = TextStore(MemoryTransport(), False)

361

324

        stream = StringIO("content")

362

325

        my_store.add(stream, "foo")

363

326

        self.assertEqual(set(['foo']),

372

335

    def test___iter__compressed(self):

373

336

        self.assertEqual(set(['foo']),

374

337

                         set(self.get_populated_store(

375

 

                             compressed=True).__iter__()))

 

338

                             store_class=CompressedTextStore).__iter__()))

376

339

        self.assertEqual(set(['foo']),

377

340

                         set(self.get_populated_store(

378

 

                             True, compressed=True).__iter__()))

 

341

                             True, CompressedTextStore).__iter__()))

379

342

 

380

343

    def test___len__(self):

381

344

        self.assertEqual(1, len(self.get_populated_store()))

382

345

 

383

346

    def test_copy_suffixes(self):

384

347

        from_store = self.get_populated_store()

385

 

        to_store = TextStore(MemoryTransport(),

386

 

                             prefixed=True, compressed=True)

 

348

        to_store = CompressedTextStore(MemoryTransport(), True)

387

349

        to_store.register_suffix('sig')

388

 

        to_store.copy_all_ids(from_store)

 

350

        copy_all(from_store, to_store)

389

351

        self.assertEqual(1, len(to_store))

390

352

        self.assertEqual(set(['foo']), set(to_store.__iter__()))

391

353

        self.assertEqual('content', to_store.get('foo').read())

392

354

        self.assertEqual('signature', to_store.get('foo', 'sig').read())

393

355

        self.assertRaises(KeyError, to_store.get, 'missing', 'sig')

394

 

 

395

 

    def test_relpath_escaped(self):

396

 

        my_store = store.TransportStore(MemoryTransport())

397

 

        self.assertEqual('%25', my_store._relpath('%'))

398

 

 

399

 

    def test_escaped_uppercase(self):

400

 

        """Uppercase letters are escaped for safety on Windows"""

401

 

        my_store = store.TransportStore(MemoryTransport(), escaped=True)

402

 

        # a particularly perverse file-id! :-)

403

 

        self.assertEquals(my_store._escape_file_id('C:<>'), '%43%3a%3c%3e')

404

 

 

405

 

 

406

 

class TestVersionFileStore(TestCaseWithTransport):

407

 

 

408

 

    def setUp(self):

409

 

        super(TestVersionFileStore, self).setUp()

410

 

        self.vfstore = store.versioned.VersionedFileStore(MemoryTransport())

411

 

 

412

 

    def test_get_weave_registers_dirty_in_write(self):

413

 

        transaction = transactions.WriteTransaction()

414

 

        vf = self.vfstore.get_weave_or_empty('id', transaction)

415

 

        transaction.finish()

416

 

        self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])

417

 

        transaction = transactions.WriteTransaction()

418

 

        vf = self.vfstore.get_weave('id', transaction)

419

 

        transaction.finish()

420

 

        self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])

421

 

 

422

 

    def test_get_weave_or_empty_readonly_fails(self):

423

 

        transaction = transactions.ReadOnlyTransaction()

424

 

        vf = self.assertRaises(errors.ReadOnlyError,

425

 

                               self.vfstore.get_weave_or_empty,

426

 

                               'id',

427

 

                               transaction)

428

 

 

429

 

    def test_get_weave_readonly_cant_write(self):

430

 

        transaction = transactions.WriteTransaction()

431

 

        vf = self.vfstore.get_weave_or_empty('id', transaction)

432

 

        transaction.finish()

433

 

        transaction = transactions.ReadOnlyTransaction()

434

 

        vf = self.vfstore.get_weave_or_empty('id', transaction)

435

 

        self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])

436

 

 

• Older »

Loggerhead is a web-based interface for Breezy
Version: 2.0.1