[MERGE][0.8] break-lock command

[Martin Pool]

On  5 May 2006, Robert Collins <robertc at robertcollins.net> wrote:
> This is also available at http://bazaar-vcs.org/bzr/break-lock.

+1 for 0.8 with some bugs and some comments.  Sterling work.

> === added file 'b/bzrlib/tests/branch_implementations/test_break_lock.py'

> +from cStringIO import StringIO

Given the danger of cStringIO with Unicode string I'd like to make some
effort in the future to remove or reduce use of it.  Perhaps we could go
through a decorator class that will prevent any attempt to put Unicode
in.

> +
> +import bzrlib
> +import bzrlib.errors as errors
> +from bzrlib.tests import TestCase, TestCaseWithTransport, TestSkipped
> +from bzrlib.tests.branch_implementations.test_branch import TestCaseWithBranch
> +
> +
> +class TestBreakLock(TestCaseWithBranch):
> +
> +    def setUp(self):
> +        super(TestBreakLock, self).setUp()
> +        self.unused_branch = self.make_branch('branch')
> +        self.branch = self.unused_branch.bzrdir.open_branch()
> +        # we want a UI factory that accepts canned input for the tests:
> +        # while SilentUIFactory still accepts stdin, we need to customise
> +        # ours
> +        self.old_factory = bzrlib.ui.ui_factory
> +        bzrlib.ui.ui_factory = bzrlib.ui.SilentUIFactory()
> +
> +    def tearDown(self):
> +        bzrlib.ui.ui_factory = self.old_factory

Shouldn't this be calling the super implementation?

To avoid this kind of bug I'd rather have call addCleanup from the
setUp method - it puts Same Things Together. 

> +
> +    def test_unlocked(self):
> +        # break lock when nothing is locked should just return
> +        try:
> +            self.branch.break_lock()
> +        except NotImplementedError:
> +            pass
> +
> +    def test_unlocked_repo_locked(self):
> +        # break lock on the branch should try on the repository even
> +        # if the branch isn't locked
> +        self.branch.repository.lock_write()
> +        bzrlib.ui.ui_factory.stdin = StringIO("y\n")
> +        try:
> +            self.unused_branch.break_lock()
> +        except NotImplementedError:
> +            # branch does not support break_lock
> +            self.branch.repository.unlock()
> +            return
> +        self.assertRaises(errors.LockBroken, self.branch.repository.unlock)

q: would it be better to say 'skipped' if it's unimplemented, rather
than just return?

> +class TestBreakLock(TestCaseWithRepository):
> +
> +    def setUp(self):
> +        super(TestBreakLock, self).setUp()
> +        self.unused_repo = self.make_repository('.')
> +        self.repo = self.unused_repo.bzrdir.open_repository()
> +        # we want a UI factory that accepts canned input for the tests:
> +        # while SilentUIFactory still accepts stdin, we need to customise
> +        # ours
> +        self.old_factory = bzrlib.ui.ui_factory
> +        bzrlib.ui.ui_factory = bzrlib.ui.SilentUIFactory()
> +        bzrlib.ui.ui_factory.stdin = StringIO("y\n")
> +
> +    def tearDown(self):
> +        bzrlib.ui.ui_factory = self.old_factory

Should call super - see below.

> +class TestBreakLock(TestCaseWithWorkingTree):
> +
> +    def setUp(self):
> +        super(TestBreakLock, self).setUp()
> +        self.unused_workingtree = self.make_branch_and_tree('.')
> +        self.workingtree = self.unused_workingtree.bzrdir.open_workingtree()
> +        # we want a UI factory that accepts canned input for the tests:
> +        # while SilentUIFactory still accepts stdin, we need to customise
> +        # ours
> +        self.old_factory = bzrlib.ui.ui_factory
> +        bzrlib.ui.ui_factory = bzrlib.ui.SilentUIFactory()
> +
> +    def tearDown(self):
> +        bzrlib.ui.ui_factory = self.old_factory

Again

> +    def break_lock(self):
> +        """Break a lock not held by this instance of LockDir.
> +
> +        This is a UI centric function: it uses the bzrlib.ui.ui_factory to
> +        prompt for input if a lock is detected and there is any doubt about
> +        it possibly being still active.
> +        """
> +        self._check_not_locked()
> +        holder_info = self.peek()
> +        if holder_info is not None:
> +            if bzrlib.ui.ui_factory.get_boolean(
> +                "Break lock %s held by %s@%s" % (
> +                    self.transport,
> +                    holder_info["user"],
> +                    holder_info["hostname"])):
> +                self.force_break(holder_info)
> +        

It would be good to include the pid too to help the user work out if
it's still running, but it need not be done now.

> +
> +
> +class TestRunBzr(ExternalBase):
> +
> +    def run_bzr_captured(self, argv, retcode=0, stdin=None):
> +        self.stdin = stdin
> +
> +    def test_stdin(self):
> +        # test that the stdin keyword to run_bzr is passed through to
> +        # run_bzr_captured as-is.
> +        self.run_bzr('foo', 'bar', stdin='gam')
> +        self.assertEqual('gam', self.stdin)
> +        self.run_bzr('foo', 'bar', stdin='zippy')
> +        self.assertEqual('zippy', self.stdin)

Could do with a comment about what you're doing here - overriding
run_bzr_captured to check that stdin is passed through correctly by
run_bzr?

> +
> +
> +class TestRunBzrCaptured(ExternalBase):
> +
> +    def apply_redirected(self, stdin=None, stdout=None, stderr=None,
> +                         a_callable=None, *args, **kwargs):
> +        self.stdin = stdin
> +        self.factory_stdin = getattr(bzrlib.ui.ui_factory, "stdin", None)
> +        return 0
> +
> +    def test_stdin(self):
> +        # test that the stdin keyword to run_bzr_captured is passed through to
> +        # apply_redirected as a StringIO
> +        self.run_bzr_captured(['foo', 'bar'], stdin='gam')
> +        self.assertEqual('gam', self.stdin.read())
> +        self.assertTrue(self.stdin is self.factory_stdin)
> +        self.run_bzr_captured(['foo', 'bar'], stdin='zippy')
> +        self.assertEqual('zippy', self.stdin.read())
> +        self.assertTrue(self.stdin is self.factory_stdin)

Similarly.

> +class TestBreakLock(TestCaseWithBzrDir):
> +
> +    def setUp(self):
> +        super(TestBreakLock, self).setUp()
> +        # we want a UI factory that accepts canned input for the tests:
> +        # while SilentUIFactory still accepts stdin, we need to customise
> +        # ours
> +        self.old_factory = bzrlib.ui.ui_factory
> +        bzrlib.ui.ui_factory = bzrlib.ui.SilentUIFactory()
> +
> +    def tearDown(self):
> +        bzrlib.ui.ui_factory = self.old_factory

Again

> === modified file 'a/bzrlib/tests/test_lockable_files.py'
> --- a/bzrlib/tests/test_lockable_files.py	
> +++ b/bzrlib/tests/test_lockable_files.py	

> @@ -106,9 +133,16 @@
>          super(TestLockableFiles_TransportLock, self).setUp()
>          transport = get_transport('.')
>          transport.mkdir('.bzr')
> -        sub_transport = transport.clone('.bzr')
> -        self.lockable = LockableFiles(sub_transport, 'my-lock', TransportLock)
> +        self.sub_transport = transport.clone('.bzr')
> +        self.lockable = self.get_lockable()
>          self.lockable.create_lock()
> +
> +    def tearDown(self):
> +        super(TestLockableFiles_TransportLock, self).tearDown()
> +        del self.sub_transport

Is there a reason why you can't just let the gc do this?  If so, add a
comment; otherwise this method doesn't seem worthwhile.

-- 
Martin