~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_lockdir.py

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

• reverse the comparison (1959.4.3 to 3596)
• stop comparing with revision 3596

Show diffs side-by-side

added

removed

bzrlib/tests/test_lockdir.py

1

 

# Copyright (C) 2006, 2007 Canonical Ltd

 

1

# Copyright (C) 2006 Canonical Ltd

2

2

#

3

3

# This program is free software; you can redistribute it and/or modify

4

4

# it under the terms of the GNU General Public License as published by

27

27

    errors,

28

28

    osutils,

29

29

    tests,

30

 

    transport,

31

30

    )

32

31

from bzrlib.errors import (

33

 

    LockBreakMismatch,

34

 

    LockBroken,

35

 

    LockContention,

36

 

    LockError,

37

 

    LockFailed,

38

 

    LockNotHeld,

39

 

    )

40

 

from bzrlib.lockdir import LockDir

 

32

        LockBreakMismatch,

 

33

        LockContention, LockError, UnlockableTransport,

 

34

        LockNotHeld, LockBroken

 

35

        )

 

36

from bzrlib.lockdir import LockDir, _DEFAULT_TIMEOUT_SECONDS

41

37

from bzrlib.tests import TestCaseWithTransport

42

38

from bzrlib.trace import note

43

39

 

108

104

        """Fail to create lock on readonly transport"""

109

105

        t = self.get_readonly_transport()

110

106

        lf = LockDir(t, 'test_lock')

111

 

        self.assertRaises(LockFailed, lf.create)

 

107

        self.assertRaises(UnlockableTransport, lf.create)

112

108

 

113

109

    def test_12_lock_readonly_transport(self):

114

110

        """Fail to lock on readonly transport"""

115

111

        lf = LockDir(self.get_transport(), 'test_lock')

116

112

        lf.create()

117

113

        lf = LockDir(self.get_readonly_transport(), 'test_lock')

118

 

        self.assertRaises(LockFailed, lf.attempt_lock)

 

114

        self.assertRaises(UnlockableTransport, lf.attempt_lock)

119

115

 

120

116

    def test_20_lock_contested(self):

121

117

        """Contention to get a lock"""

190

186

            lf1.unlock()

191

187

        lock_base = lf2.transport.abspath(lf2.path)

192

188

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

193

 

        lock_url = lf2.transport.abspath(lf2.path)

194

189

        self.assertEqual('%s %s\n'

195

190

                         '%s\n%s\n'

196

 

                         'Will continue to try until %s, unless '

197

 

                         'you press Ctrl-C\n'

198

 

                         'If you\'re sure that it\'s not being '

199

 

                         'modified, use bzr break-lock %s',

 

191

                         'Will continue to try until %s\n',

200

192

                         self._logged_reports[0][0])

201

193

        args = self._logged_reports[0][1]

202

194

        self.assertEqual('Unable to obtain', args[0])

228

220

        One thread holds on a lock and then releases it; another 

229

221

        tries to lock it.

230

222

        """

231

 

        # This test sometimes fails like this:

232

 

        # Traceback (most recent call last):

233

 

 

234

 

        #   File "/home/pqm/bzr-pqm-workdir/home/+trunk/bzrlib/tests/

235

 

        # test_lockdir.py", line 247, in test_32_lock_wait_succeed

236

 

        #     self.assertEqual(1, len(self._logged_reports))

237

 

        # AssertionError: not equal:

238

 

        # a = 1

239

 

        # b = 0

240

 

        raise tests.TestSkipped("Test fails intermittently")

241

223

        t = self.get_transport()

242

224

        lf1 = LockDir(t, 'test_lock')

243

225

        lf1.create()

275

257

        self.assertEndsWith(args[3], ' ago')

276

258

        self.assertContainsRe(args[4], r'\d\d:\d\d:\d\d')

277

259

 

 

260

    def test_33_wait(self):

 

261

        """Succeed when waiting on a lock that gets released

 

262

 

 

263

        The difference from test_32_lock_wait_succeed is that the second 

 

264

        caller does not actually acquire the lock, but just waits for it

 

265

        to be released.  This is done over a readonly transport.

 

266

        """

 

267

        t = self.get_transport()

 

268

        lf1 = LockDir(t, 'test_lock')

 

269

        lf1.create()

 

270

        lf1.attempt_lock()

 

271

 

 

272

        def wait_and_unlock():

 

273

            time.sleep(0.1)

 

274

            lf1.unlock()

 

275

        unlocker = Thread(target=wait_and_unlock)

 

276

        unlocker.start()

 

277

        try:

 

278

            lf2 = LockDir(self.get_readonly_transport(), 'test_lock')

 

279

            before = time.time()

 

280

            # wait but don't lock

 

281

            lf2.wait(timeout=0.4, poll=0.1)

 

282

            after = time.time()

 

283

            self.assertTrue(after - before <= 1.0)

 

284

        finally:

 

285

            unlocker.join()

 

286

 

278

287

    def test_34_lock_write_waits(self):

279

288

        """LockDir.lock_write() will wait for the lock.""" 

280

 

        # the test suite sets the default to 0 to make deadlocks fail fast.

281

 

        # change it for this test, as we want to try a manual deadlock.

282

 

        raise tests.TestSkipped('Timing-sensitive test')

283

 

        bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = 300

284

289

        t = self.get_transport()

285

290

        lf1 = LockDir(t, 'test_lock')

286

291

        lf1.create()

337

342

           unlocks the lockdir, allowing Lock2 to acquire the lock.

338

343

        """

339

344

 

340

 

        raise tests.KnownFailure(

341

 

            "timing dependency in lock tests (#213182)")

342

 

 

343

345

        wait_to_check_lock = Lock()

344

346

        wait_until_checked_lock = Lock()

345

347

 

405

407

        unlocker.start()

406

408

        try:

407

409

            # Wait and play against the other thread

408

 

            lf2.wait_lock(timeout=20.0, poll=0.01)

 

410

            lf2.wait_lock(timeout=1.0, poll=0.01)

409

411

        finally:

410

412

            unlocker.join()

411

413

        lf2.unlock()

413

415

        # There should be 2 reports, because the lock changed

414

416

        lock_base = lf2.transport.abspath(lf2.path)

415

417

        self.assertEqual(2, len(self._logged_reports))

416

 

        lock_url = lf2.transport.abspath(lf2.path)

 

418

 

417

419

        self.assertEqual('%s %s\n'

418

420

                         '%s\n%s\n'

419

 

                         'Will continue to try until %s, unless '

420

 

                         'you press Ctrl-C\n'

421

 

                         'If you\'re sure that it\'s not being '

422

 

                         'modified, use bzr break-lock %s',

 

421

                         'Will continue to try until %s\n',

423

422

                         self._logged_reports[0][0])

424

423

        args = self._logged_reports[0][1]

425

424

        self.assertEqual('Unable to obtain', args[0])

431

430

 

432

431

        self.assertEqual('%s %s\n'

433

432

                         '%s\n%s\n'

434

 

                         'Will continue to try until %s, unless '

435

 

                         'you press Ctrl-C\n'

436

 

                         'If you\'re sure that it\'s not being '

437

 

                         'modified, use bzr break-lock %s',

 

433

                         'Will continue to try until %s\n',

438

434

                         self._logged_reports[1][0])

439

435

        args = self._logged_reports[1][1]

440

436

        self.assertEqual('Lock owner changed for', args[0])

615

611

        lock_path = ld1.transport.local_abspath('test_lock')

616

612

        os.mkdir(lock_path)

617

613

        osutils.make_readonly(lock_path)

618

 

        self.assertRaises(errors.LockFailed, ld1.attempt_lock)

619

 

 

620

 

    def test_lock_by_token(self):

621

 

        ld1 = self.get_lock()

622

 

        token = ld1.lock_write()

623

 

        self.assertNotEqual(None, token)

624

 

        ld2 = self.get_lock()

625

 

        t2 = ld2.lock_write(token)

626

 

        self.assertEqual(token, t2)

627

 

 

628

 

    def test_lock_with_buggy_rename(self):

629

 

        # test that lock acquisition handles servers which pretend they

630

 

        # renamed correctly but that actually fail

631

 

        t = transport.get_transport('brokenrename+' + self.get_url())

632

 

        ld1 = LockDir(t, 'test_lock')

633

 

        ld1.create()

634

 

        ld1.attempt_lock()

635

 

        ld2 = LockDir(t, 'test_lock')

636

 

        # we should fail to lock

637

 

        e = self.assertRaises(errors.LockContention, ld2.attempt_lock)

638

 

        # now the original caller should succeed in unlocking

639

 

        ld1.unlock()

640

 

        # and there should be nothing left over

641

 

        self.assertEquals([], t.list_dir('test_lock'))

642

 

 

643

 

    def test_failed_lock_leaves_no_trash(self):

644

 

        # if we fail to acquire the lock, we don't leave pending directories

645

 

        # behind -- https://bugs.launchpad.net/bzr/+bug/109169

646

 

        ld1 = self.get_lock()

647

 

        ld2 = self.get_lock()

648

 

        # should be nothing before we start

649

 

        ld1.create()

650

 

        t = self.get_transport().clone('test_lock')

651

 

        def check_dir(a):

652

 

            self.assertEquals(a, t.list_dir('.'))

653

 

        check_dir([])

654

 

        # when held, that's all we see

655

 

        ld1.attempt_lock()

656

 

        check_dir(['held'])

657

 

        # second guy should fail

658

 

        self.assertRaises(errors.LockContention, ld2.attempt_lock)

659

 

        # no kibble

660

 

        check_dir(['held'])

 

614

        self.assertRaises(errors.PermissionDenied, ld1.attempt_lock)

• Older »

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