~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_lockdir.py

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

• reverse the comparison (2690 to 4264)
• stop comparing with revision 4264

Show diffs side-by-side

added

removed

bzrlib/tests/test_lockdir.py

1

 

# Copyright (C) 2006, 2007, 2008 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

12

12

#

13

13

# You should have received a copy of the GNU General Public License

14

14

# along with this program; if not, write to the Free Software

15

 

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

 

15

# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

16

16

 

17

17

"""Tests for LockDir"""

18

18

 

25

25

from bzrlib import (

26

26

    config,

27

27

    errors,

28

 

    lock,

29

28

    osutils,

30

29

    tests,

31

30

    transport,

32

31

    )

33

32

from bzrlib.errors import (

34

 

    LockBreakMismatch,

35

 

    LockBroken,

36

 

    LockContention,

37

 

    LockError,

38

 

    LockFailed,

39

 

    LockNotHeld,

40

 

    )

 

33

        LockBreakMismatch,

 

34

        LockContention, LockError, UnlockableTransport,

 

35

        LockNotHeld, LockBroken

 

36

        )

41

37

from bzrlib.lockdir import LockDir

42

38

from bzrlib.tests import TestCaseWithTransport

43

39

from bzrlib.trace import note

109

105

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

110

106

        t = self.get_readonly_transport()

111

107

        lf = LockDir(t, 'test_lock')

112

 

        self.assertRaises(LockFailed, lf.create)

 

108

        self.assertRaises(UnlockableTransport, lf.create)

113

109

 

114

110

    def test_12_lock_readonly_transport(self):

115

111

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

116

112

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

117

113

        lf.create()

118

114

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

119

 

        self.assertRaises(LockFailed, lf.attempt_lock)

 

115

        self.assertRaises(UnlockableTransport, lf.attempt_lock)

120

116

 

121

117

    def test_20_lock_contested(self):

122

118

        """Contention to get a lock"""

126

122

        lf1.attempt_lock()

127

123

        lf2 = LockDir(t, 'test_lock')

128

124

        try:

129

 

            # locking is between LockDir instances; aliases within

 

125

            # locking is between LockDir instances; aliases within 

130

126

            # a single process are not detected

131

127

            lf2.attempt_lock()

132

128

            self.fail('Failed to detect lock collision')

167

163

 

168

164

    def test_30_lock_wait_fail(self):

169

165

        """Wait on a lock, then fail

170

 

 

 

166

        

171

167

        We ask to wait up to 400ms; this should fail within at most one

172

168

        second.  (Longer times are more realistic but we don't want the test

173

169

        suite to take too long, and this should do for now.)

185

181

            after = time.time()

186

182

            # it should only take about 0.4 seconds, but we allow more time in

187

183

            # case the machine is heavily loaded

188

 

            self.assertTrue(after - before <= 8.0,

 

184

            self.assertTrue(after - before <= 8.0, 

189

185

                    "took %f seconds to detect lock contention" % (after - before))

190

186

        finally:

191

187

            lf1.unlock()

192

188

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

193

189

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

194

 

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

195

190

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

196

191

                         '%s\n%s\n'

197

 

                         'Will continue to try until %s, unless '

198

 

                         'you press Ctrl-C\n'

199

 

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

200

 

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

 

192

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

201

193

                         self._logged_reports[0][0])

202

194

        args = self._logged_reports[0][1]

203

195

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

226

218

    def test_32_lock_wait_succeed(self):

227

219

        """Succeed when trying to acquire a lock that gets released

228

220

 

229

 

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

 

221

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

230

222

        tries to lock it.

231

223

        """

232

224

        # This test sometimes fails like this:

277

269

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

278

270

 

279

271

    def test_34_lock_write_waits(self):

280

 

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

 

272

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

281

273

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

282

274

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

283

 

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

284

275

        bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = 300

285

276

        t = self.get_transport()

286

277

        lf1 = LockDir(t, 'test_lock')

320

311

 

321

312

    def test_35_wait_lock_changing(self):

322

313

        """LockDir.wait_lock() will report if the lock changes underneath.

323

 

 

 

314

        

324

315

        This is the stages we want to happen:

325

316

 

326

317

        0) Synchronization locks are created and locked.

327

318

        1) Lock1 obtains the lockdir, and releases the 'check' lock.

328

319

        2) Lock2 grabs the 'check' lock, and checks the lockdir.

329

 

           It sees the lockdir is already acquired, reports the fact,

 

320

           It sees the lockdir is already acquired, reports the fact, 

330

321

           and unsets the 'checked' lock.

331

322

        3) Thread1 blocks on acquiring the 'checked' lock, and then tells

332

323

           Lock1 to release and acquire the lockdir. This resets the 'check'

333

324

           lock.

334

325

        4) Lock2 acquires the 'check' lock, and checks again. It notices

335

 

           that the holder of the lock has changed, and so reports a new

 

326

           that the holder of the lock has changed, and so reports a new 

336

327

           lock holder.

337

328

        5) Thread1 blocks on the 'checked' lock, this time, it completely

338

329

           unlocks the lockdir, allowing Lock2 to acquire the lock.

339

330

        """

340

331

 

341

 

        raise tests.KnownFailure(

342

 

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

343

 

 

344

332

        wait_to_check_lock = Lock()

345

333

        wait_until_checked_lock = Lock()

346

334

 

406

394

        unlocker.start()

407

395

        try:

408

396

            # Wait and play against the other thread

409

 

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

 

397

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

410

398

        finally:

411

399

            unlocker.join()

412

400

        lf2.unlock()

414

402

        # There should be 2 reports, because the lock changed

415

403

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

416

404

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

417

 

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

 

405

 

418

406

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

419

407

                         '%s\n%s\n'

420

 

                         'Will continue to try until %s, unless '

421

 

                         'you press Ctrl-C\n'

422

 

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

423

 

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

 

408

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

424

409

                         self._logged_reports[0][0])

425

410

        args = self._logged_reports[0][1]

426

411

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

432

417

 

433

418

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

434

419

                         '%s\n%s\n'

435

 

                         'Will continue to try until %s, unless '

436

 

                         'you press Ctrl-C\n'

437

 

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

438

 

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

 

420

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

439

421

                         self._logged_reports[1][0])

440

422

        args = self._logged_reports[1][1]

441

423

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

532

514

        """Check the on-disk representation of LockDirs is as expected.

533

515

 

534

516

        There should always be a top-level directory named by the lock.

535

 

        When the lock is held, there should be a lockname/held directory

 

517

        When the lock is held, there should be a lockname/held directory 

536

518

        containing an info file.

537

519

        """

538

520

        t = self.get_transport()

616

598

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

617

599

        os.mkdir(lock_path)

618

600

        osutils.make_readonly(lock_path)

619

 

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

 

601

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

620

602

 

621

603

    def test_lock_by_token(self):

622

604

        ld1 = self.get_lock()

659

641

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

660

642

        # no kibble

661

643

        check_dir(['held'])

662

 

 

663

 

    def record_hook(self, result):

664

 

        self._calls.append(result)

665

 

 

666

 

    def reset_hooks(self):

667

 

        self._old_hooks = lock.Lock.hooks

668

 

        self.addCleanup(self.restore_hooks)

669

 

        lock.Lock.hooks = lock.LockHooks()

670

 

 

671

 

    def restore_hooks(self):

672

 

        lock.Lock.hooks = self._old_hooks

673

 

 

674

 

    def test_LockDir_acquired_success(self):

675

 

        # the LockDir.lock_acquired hook fires when a lock is acquired.

676

 

        self._calls = []

677

 

        self.reset_hooks()

678

 

        LockDir.hooks.install_named_hook('lock_acquired',

679

 

            self.record_hook, 'record_hook')

680

 

        ld = self.get_lock()

681

 

        ld.create()

682

 

        self.assertEqual([], self._calls)

683

 

        result = ld.attempt_lock()

684

 

        lock_path = ld.transport.abspath(ld.path)

685

 

        self.assertEqual([lock.LockResult(lock_path, result)], self._calls)

686

 

        ld.unlock()

687

 

        self.assertEqual([lock.LockResult(lock_path, result)], self._calls)

688

 

 

689

 

    def test_LockDir_acquired_fail(self):

690

 

        # the LockDir.lock_acquired hook does not fire on failure.

691

 

        self._calls = []

692

 

        self.reset_hooks()

693

 

        ld = self.get_lock()

694

 

        ld.create()

695

 

        ld2 = self.get_lock()

696

 

        ld2.attempt_lock()

697

 

        # install a lock hook now, when the disk lock is locked

698

 

        LockDir.hooks.install_named_hook('lock_acquired',

699

 

            self.record_hook, 'record_hook')

700

 

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

701

 

        self.assertEqual([], self._calls)

702

 

        ld2.unlock()

703

 

        self.assertEqual([], self._calls)

704

 

 

705

 

    def test_LockDir_released_success(self):

706

 

        # the LockDir.lock_released hook fires when a lock is acquired.

707

 

        self._calls = []

708

 

        self.reset_hooks()

709

 

        LockDir.hooks.install_named_hook('lock_released',

710

 

            self.record_hook, 'record_hook')

711

 

        ld = self.get_lock()

712

 

        ld.create()

713

 

        self.assertEqual([], self._calls)

714

 

        result = ld.attempt_lock()

715

 

        self.assertEqual([], self._calls)

716

 

        ld.unlock()

717

 

        lock_path = ld.transport.abspath(ld.path)

718

 

        self.assertEqual([lock.LockResult(lock_path, result)], self._calls)

719

 

 

720

 

    def test_LockDir_released_fail(self):

721

 

        # the LockDir.lock_released hook does not fire on failure.

722

 

        self._calls = []

723

 

        self.reset_hooks()

724

 

        ld = self.get_lock()

725

 

        ld.create()

726

 

        ld2 = self.get_lock()

727

 

        ld.attempt_lock()

728

 

        ld2.force_break(ld2.peek())

729

 

        LockDir.hooks.install_named_hook('lock_released',

730

 

            self.record_hook, 'record_hook')

731

 

        self.assertRaises(LockBroken, ld.unlock)

732

 

        self.assertEqual([], self._calls)

• Older »

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