/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/per_branch/test_locking.py

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

Show diffs side-by-side

added

removed

bzrlib/tests/per_branch/test_locking.py

1

 

# Copyright (C) 2006 Canonical Ltd

 

1

# Copyright (C) 2006-2010 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

16

16

 

17

17

"""Test locks across all branch implemenations"""

18

18

 

19

 

from bzrlib import errors

20

 

from bzrlib.branch import BzrBranchFormat4

21

 

from bzrlib.bzrdir import RemoteBzrDirFormat

22

 

from bzrlib.tests import TestSkipped

23

 

from bzrlib.tests.lock_helpers import TestPreventLocking, LockWrapper

24

 

from bzrlib.tests.per_branch.test_branch import TestCaseWithBranch

25

 

 

26

 

 

27

 

class TestBranchLocking(TestCaseWithBranch):

 

19

from bzrlib import (

 

20

    branch as _mod_branch,

 

21

    errors,

 

22

    tests,

 

23

    )

 

24

from bzrlib.tests import (

 

25

    lock_helpers,

 

26

    per_branch,

 

27

    )

 

28

from bzrlib.tests.matchers import *

 

29

 

 

30

 

 

31

class TestBranchLocking(per_branch.TestCaseWithBranch):

28

32

 

29

33

    def setUp(self):

30

 

        TestCaseWithBranch.setUp(self)

 

34

        super(TestBranchLocking, self).setUp()

31

35

        self.reduceLockdirTimeout()

32

36

 

33

37

    def get_instrumented_branch(self):

37

41

        # not there. But assuming it has them lets us test the exact

38

42

        # lock/unlock order.

39

43

        self.locks = []

40

 

        b = LockWrapper(self.locks, self.get_branch(), 'b')

41

 

        b.repository = LockWrapper(self.locks, b.repository, 'r')

 

44

        b = lock_helpers.LockWrapper(self.locks, self.get_branch(), 'b')

 

45

        b.repository = lock_helpers.LockWrapper(self.locks, b.repository, 'r')

42

46

        bcf = b.control_files

43

47

        rcf = getattr(b.repository, 'control_files', None)

44

48

        if rcf is None:

47

51

            # Look out for branch types that reuse their control files

48

52

            self.combined_control = bcf is rcf

49

53

        try:

50

 

            b.control_files = LockWrapper(self.locks, b.control_files, 'bc')

 

54

            b.control_files = lock_helpers.LockWrapper(

 

55

                self.locks, b.control_files, 'bc')

51

56

        except AttributeError:

52

57

            # RemoteBranch seems to trigger this.

53

 

            raise TestSkipped("Could not instrument branch control files.")

 

58

            raise tests.TestSkipped(

 

59

                'Could not instrument branch control files.')

54

60

        if self.combined_control:

55

61

            # instrument the repository control files too to ensure its worked

56

62

            # with correctly. When they are not shared, we trust the repository

57

63

            # API and only instrument the repository itself.

58

 

            b.repository.control_files = \

59

 

                LockWrapper(self.locks, b.repository.control_files, 'rc')

 

64

            b.repository.control_files = lock_helpers.LockWrapper(

 

65

                self.locks, b.repository.control_files, 'rc')

60

66

        return b

61

67

 

62

68

    def test_01_lock_read(self):

139

145

        try:

140

146

            self.assertTrue(b.is_locked())

141

147

            self.assertTrue(b.repository.is_locked())

142

 

            self.assertRaises(TestPreventLocking, b.unlock)

 

148

            self.assertLogsError(lock_helpers.TestPreventLocking, b.unlock)

143

149

            if self.combined_control:

144

150

                self.assertTrue(b.is_locked())

145

151

            else:

183

189

        try:

184

190

            self.assertTrue(b.is_locked())

185

191

            self.assertTrue(b.repository.is_locked())

186

 

            self.assertRaises(TestPreventLocking, b.unlock)

 

192

            self.assertLogsError(lock_helpers.TestPreventLocking, b.unlock)

187

193

            self.assertTrue(b.is_locked())

188

194

            self.assertTrue(b.repository.is_locked())

189

195

 

216

222

        b = self.get_instrumented_branch()

217

223

        b.repository.disable_lock_read()

218

224

 

219

 

        self.assertRaises(TestPreventLocking, b.lock_read)

 

225

        self.assertRaises(lock_helpers.TestPreventLocking, b.lock_read)

220

226

        self.assertFalse(b.is_locked())

221

227

        self.assertFalse(b.repository.is_locked())

222

228

 

229

235

        b = self.get_instrumented_branch()

230

236

        b.repository.disable_lock_write()

231

237

 

232

 

        self.assertRaises(TestPreventLocking, b.lock_write)

 

238

        self.assertRaises(lock_helpers.TestPreventLocking, b.lock_write)

233

239

        self.assertFalse(b.is_locked())

234

240

        self.assertFalse(b.repository.is_locked())

235

241

 

242

248

        b = self.get_instrumented_branch()

243

249

        b.control_files.disable_lock_read()

244

250

 

245

 

        self.assertRaises(TestPreventLocking, b.lock_read)

 

251

        self.assertRaises(lock_helpers.TestPreventLocking, b.lock_read)

246

252

        self.assertFalse(b.is_locked())

247

253

        self.assertFalse(b.repository.is_locked())

248

254

 

266

272

        b = self.get_instrumented_branch()

267

273

        b.control_files.disable_lock_write()

268

274

 

269

 

        self.assertRaises(TestPreventLocking, b.lock_write)

 

275

        self.assertRaises(lock_helpers.TestPreventLocking, b.lock_write)

270

276

        self.assertFalse(b.is_locked())

271

277

        self.assertFalse(b.repository.is_locked())

272

278

        if self.combined_control:

286

292

 

287

293

    def test_lock_write_returns_None_refuses_token(self):

288

294

        branch = self.make_branch('b')

289

 

        token = branch.lock_write()

 

295

        token = branch.lock_write().branch_token

290

296

        try:

291

297

            if token is not None:

292

298

                # This test does not apply, because this lockable supports

299

305

 

300

306

    def test_reentering_lock_write_raises_on_token_mismatch(self):

301

307

        branch = self.make_branch('b')

302

 

        token = branch.lock_write()

 

308

        token = branch.lock_write().branch_token

303

309

        try:

304

310

            if token is None:

305

311

                # This test does not apply, because this lockable refuses

316

322

 

317

323

    def test_lock_write_with_nonmatching_token(self):

318

324

        branch = self.make_branch('b')

319

 

        token = branch.lock_write()

 

325

        token = branch.lock_write().branch_token

320

326

        try:

321

327

            if token is None:

322

328

                # This test does not apply, because this branch refuses

339

345

        """Test that a branch can be locked with a token, if it is already

340

346

        locked by that token."""

341

347

        branch = self.make_branch('b')

342

 

        token = branch.lock_write()

 

348

        token = branch.lock_write().branch_token

343

349

        try:

344

350

            if token is None:

345

351

                # This test does not apply, because this branch refuses tokens.

363

369

        # If lock_write did not physically acquire the lock (because it was

364

370

        # passed some tokens), then unlock should not physically release it.

365

371

        branch = self.make_branch('b')

366

 

        token = branch.lock_write()

 

372

        token = branch.lock_write().branch_token

367

373

        try:

368

374

            if token is None:

369

375

                # This test does not apply, because this lockable refuses

385

391

        # to lock with a token that is no longer valid (because the original

386

392

        # lock was released).

387

393

        branch = self.make_branch('b')

388

 

        token = branch.lock_write()

 

394

        token = branch.lock_write().branch_token

389

395

        branch.unlock()

390

396

        if token is None:

391

397

            # This test does not apply, because this lockable refuses

392

398

            # tokens.

393

399

            return

394

 

 

395

 

        self.assertRaises(errors.TokenMismatch,

396

 

                          branch.lock_write, token=token)

 

400

        self.assertRaises(errors.TokenMismatch, branch.lock_write, token=token)

397

401

 

398

402

    def test_lock_write_reenter_with_token(self):

399

403

        branch = self.make_branch('b')

400

 

        token = branch.lock_write()

 

404

        token = branch.lock_write().branch_token

401

405

        try:

402

406

            if token is None:

403

407

                # This test does not apply, because this lockable refuses

419

423

        branch = self.make_branch('b')

420

424

        # Lock the branch, then use leave_lock_in_place so that when we

421

425

        # unlock the branch the lock is still held on disk.

422

 

        token = branch.lock_write()

 

426

        token = branch.lock_write().branch_token

423

427

        try:

424

428

            if token is None:

425

429

                # This test does not apply, because this repository refuses lock

440

444

    def test_dont_leave_lock_in_place(self):

441

445

        branch = self.make_branch('b')

442

446

        # Create a lock on disk.

443

 

        token = branch.lock_write()

 

447

        token = branch.lock_write().branch_token

444

448

        try:

445

449

            if token is None:

446

450

                # This test does not apply, because this branch refuses lock

492

496

        branch.lock_read()

493

497

        branch.unlock()

494

498

 

 

499

    def test_lock_read_returns_unlockable(self):

 

500

        branch = self.make_branch('b')

 

501

        self.assertThat(branch.lock_read, ReturnsUnlockable(branch))

 

502

 

495

503

    def test_lock_write_locks_repo_too(self):

496

 

        if isinstance(self.branch_format, BzrBranchFormat4):

 

504

        if isinstance(self.branch_format, _mod_branch.BzrBranchFormat4):

497

505

            # Branch format 4 is combined with the repository, so this test

498

506

            # doesn't apply.

499

507

            return

518

526

        finally:

519

527

            branch.unlock()

520

528

 

 

529

    def test_lock_write_returns_unlockable(self):

 

530

        branch = self.make_branch('b')

 

531

        self.assertThat(branch.lock_write, ReturnsUnlockable(branch))

 

532

 

 

533

    def test_lock_write_raises_in_lock_read(self):

 

534

        branch = self.make_branch('b')

 

535

        branch.lock_read()

 

536

        err = self.assertRaises(errors.ReadOnlyError, branch.lock_write)

 

537

 

521

538

    def test_lock_and_unlock_leaves_repo_unlocked(self):

522

539

        branch = self.make_branch('b')

523

540

        branch.lock_write()

• Older »

Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0