Merge lp:~jimbaker/pyjuju/robust-zk-connect into lp:pyjuju

[Jim Baker]

This branch implements the change in the API, per this mailing list discussion:
https://lists.ubuntu.com/archives/juju/2011-November/000990.html

Some things not covered in this branch:

Documentation of the rationale, as requested by https://lists.ubuntu.com/archives/juju/2011-November/001010.html since However, there's not anything documenting this in general, so it's a bit larger in scope. Instead this should be covered in a separate doc branch.

Support for timeouts. However, no one discussed this on the mailing list, so this issue may be moot.

Running with --verbose is *very* verbose indeed with these retries, due to ZooKeeper client logging. This could be addressed by providing for more logging levels (maybe --verbose=DEBUG_ZK).

Bug 892254 was added to take in account that during retries, spurious messages are printed to stderr. This has no impact however on the robustness of the retry process, however.

Lastly, I removed what appears to be redundant test code in the orchestra and ec2 providers. From what I can tell, this was simply combining some aspects of testing the findzookeepers code, which is necessarily different for each provider, with the same connection tests in juju.providers.common.tests.test_connect, which is in fact common. So updating these tests to take in account that EnvironmentPending indicates that a retry should be attempted, or address a couple of gaps in testing, didn't make sense.

[William Reade (fwereade)]

Can't see anything to complain about at all. +1.

review: Approve

[Kapil Thangavelu (hazmat)]

It works :-) I'm a little concerned by the content and verbosity of
the output both when run normally or in verbose mode.

verbose mode -> http://paste.ubuntu.com/762451/
normal mode -> http://paste.ubuntu.com/762452/

[0]

The retry is a bit aggressive, i see 3-4 calls per second to try and
resolve the host ip address, over the course of avg ~30s. Some backoff
might be beneficial here, like 1s between retries on a machine without
an address.

[1]
2011-11-29 21:06:27,053:13891(0xb6500b70):ZOO_ERROR@handle_socket_error_msg@1579: Socket [127.0.0.1:35354] zk retcode=-4, errno=111(Connection refused): server refused to accept the client
Unhandled error in Deferred:
Unhandled error in Deferred:
Unhandled Error
Traceback (most recent call last):
Failure: txzookeeper.client.ConnectionTimeoutException: could not connect before timeout

it would be nice if this could be resolved as well in this branch, i
tracked it down to the called_aware_deferred_chain returning
failures if the parent deferred is called.

The more i look at this branch, i find the sshclient/forward code more
mystifying. Its not germane to this branch, but i've got a refactoring
branch that will followup this branch to make this code a bit cleaner.
The gist of the cleanup http://pastebin.ubuntu.com/761938/

[2] The zookeeper c library doesn't have a notion of timeouts, its an
internal thing the txzk library manages using a twisted
reactor.callLater. Effectively it means everytime we attempt a
connect, there will be a background attempt continously running in the
zk io thread.

An effective usage change would be to just increase the timeout on connect, the tunnel is already active, so its just a matter of waiting. I just commited r45 on txzookeeper to handle this issue there, namely tackle closing the conn should terminate the background connect activity, but i don't see why we wouldn't just go for the longer delay since its a continual reconnection attempt.

[3]

2011-12-05 17:54:46,490 ERROR Connection refused

Normal stdout log for the command has a few of these error
messages. Since this is normal activity it would be good to ellide
them unless in verbose mode.

review: Needs Fixing

[Kapil Thangavelu (hazmat)]

i've fixed the verbosity/output issues from [1-3] in a follow up branch (lp:~hazmat/juju/sshclient-refactor) that also cleanups the ssh client connect code.. so outside of [0] this branch should be fine.

[Jim Baker (jimbaker)]

For [0], I added a 1 second deferred sleep so as to implement the desired backoff. This also marks the addition of a sleep function finally to twistutils, after a number of thoughts of doing so.

[Kapil Thangavelu (hazmat)]

LGTM, let me know when your planning on merging, i'd like to get the followup branch in with it to avoid getting builds with this branch alone. Also this functionality could definitely use an email out to the list once landed.

[Kapil Thangavelu (hazmat)]

review: Approve

=== modified file 'juju/lib/tests/test_twistutils.py'
--- juju/lib/tests/test_twistutils.py	2011-09-15 18:50:23 +0000
+++ juju/lib/tests/test_twistutils.py	2011-12-08 20:20:29 +0000
@@ -1,4 +1,5 @@
 import os
+import time
 
 from twisted.internet.defer import (
     succeed, fail, Deferred, DeferredList, inlineCallbacks, returnValue)
@@ -8,7 +9,7 @@
 
 from juju.lib.testing import TestCase
 from juju.lib.twistutils import (
-    concurrent_execution_guard, gather_results, get_module_directory)
+    concurrent_execution_guard, gather_results, get_module_directory, sleep)
 
 
 class Bar(object):
@@ -144,3 +145,13 @@
         self.assertIn("juju", directory)
         self.assertNotIn("_trial_temp", directory)
         self.assertTrue(os.path.isdir(directory))
+
+
+class SleepTest(TestCase):
+
+    @inlineCallbacks
+    def test_sleep(self):
+        """Directly test deferred sleep."""
+        start = time.time()
+        yield sleep(0.1)
+        self.assertGreaterEqual(time.time() - start, 0.1)

=== modified file 'juju/lib/twistutils.py'
--- juju/lib/twistutils.py	2011-02-15 15:15:16 +0000
+++ juju/lib/twistutils.py	2011-12-08 20:20:29 +0000
@@ -1,7 +1,9 @@
 import inspect
 import os
 
-from twisted.internet.defer import maybeDeferred, succeed, DeferredList
+from twisted.internet import reactor
+from twisted.internet.defer import (
+    Deferred, maybeDeferred, succeed, DeferredList)
 from twisted.python.util import mergeFunctionMetadata
 
 
@@ -53,3 +55,15 @@
     """
     return os.path.abspath(os.path.dirname(inspect.getabsfile(module)).replace(
         "/_trial_temp", ""))
+
+
+def sleep(delay):
+    """Non-blocking sleep.
+
+    :param int delay: time in seconds to sleep.
+    :return: a Deferred that fires after the desired delay.
+    :rtype: :class:`twisted.internet.defer.Deferred`
+    """
+    deferred = Deferred()
+    reactor.callLater(delay, deferred.callback, None)
+    return deferred

=== modified file 'juju/providers/common/base.py'
--- juju/providers/common/base.py	2011-09-28 09:48:30 +0000
+++ juju/providers/common/base.py	2011-12-08 20:20:29 +0000
@@ -148,8 +148,6 @@
 
         :raises: :exc:`juju.errors.EnvironmentNotFound` when no zookeepers
             exist
-        :raises: :exc:`juju.errors.EnvironmentPending` when zookeepers
-            exist but connection attempt fails
         """
         return ZookeeperConnect(self).run(share=share)
 

=== modified file 'juju/providers/common/connect.py'
--- juju/providers/common/connect.py	2011-09-16 19:37:15 +0000
+++ juju/providers/common/connect.py	2011-12-08 20:20:29 +0000
@@ -1,8 +1,11 @@
+import random
+
 from twisted.internet.defer import inlineCallbacks, returnValue
 
 from txzookeeper.client import ConnectionTimeoutException
 
-from juju.errors import EnvironmentPending, NoConnection
+from juju.errors import EnvironmentNotFound, EnvironmentPending, NoConnection
+from juju.lib.twistutils import sleep
 from juju.state.sshclient import SSHClient
 
 from .utils import log
@@ -15,51 +18,69 @@
 
     @inlineCallbacks
     def run(self, share=False):
-        """Attempt to connect to a running zookeeper node.
+        """Attempt to connect to a running zookeeper node, retrying as needed.
 
         :param bool share: where feasible, attempt to share a connection with
-            other clients
+            other clients.
 
         :return: an open :class:`txzookeeper.client.ZookeeperClient`
         :rtype: :class:`twisted.internet.defer.Deferred`
 
         :raises: :exc:`juju.errors.EnvironmentNotFound` when no zookeepers
             exist
-        :raises: :exc:`juju.errors.EnvironmentPending` when zookeepers
-            exist but connection attempt fails
+
+        Internally this method catches all
+        :exc:`juju.errors.EnvironmentPending`, since
+        this exception explicitly means that a retry is feasible.
+
+        TODO consider supporting a timeout for this method, instead of
+        any such timeouts being done externally.
         """
+        log.info("Connecting to environment...")
+        while True:
+            try:
+                client = yield self._internal_connect(share)
+                log.info("Connected to environment.")
+                returnValue(client)
+            except EnvironmentPending as e:
+                log.debug("Retrying connection: %s", e)
+            except EnvironmentNotFound:
+                # Expected if not bootstrapped, simply raise up
+                raise
+            except Exception as e:
+                # Otherwise this is unexpected, log with some details
+                log.exception("Cannot connect to environment: %s", e)
+                raise
+
+    @inlineCallbacks
+    def _internal_connect(self, share):
+        """Attempt connection to one of the ZK nodes."""
         candidates = yield self._provider.get_zookeeper_machines()
-        chosen = yield self._pick_machine(candidates)
-        client = yield self._connect_to_machine(chosen, share)
+        assigned = [machine for machine in candidates if machine.dns_name]
+        if not assigned:
+            yield sleep(1)  # Momentarily backoff
+            raise EnvironmentPending("No machines have assigned addresses")
+
+        chosen = random.choice(assigned)
+        log.debug("Connecting to environment using %s...", chosen.dns_name)
+        try:
+            client = yield SSHClient().connect(
+                chosen.dns_name + ":2181", timeout=30, share=share)
+        except (NoConnection, ConnectionTimeoutException) as e:
+            raise EnvironmentPending(
+                "Cannot connect to environment using %s "
+                "(perhaps still initializing): %s" % (
+                    chosen.dns_name, str(e)))
+
         yield self.wait_for_initialization(client)
         returnValue(client)
 
-    def _pick_machine(self, machines):
-        # TODO Should we pick a random entry from the nodes list?
-        for machine in machines:
-            if machine.dns_name:
-                return machine
-        raise EnvironmentPending("No machines have addresses assigned yet")
-
-    def _connect_to_machine(self, machine, share):
-        log.info("Connecting to environment.")
-        result = SSHClient().connect(
-            machine.dns_name + ":2181", timeout=30, share=share)
-        result.addErrback(self._cannot_connect, machine)
-        return result
-
-    def _cannot_connect(self, failure, machine):
-        failure.trap(NoConnection, ConnectionTimeoutException)
-        raise EnvironmentPending(
-            "Cannot connect to machine %s (perhaps still initializing): %s"
-            % (machine.instance_id, str(failure.value)))
-
     @inlineCallbacks
     def wait_for_initialization(self, client):
         exists_d, watch_d = client.exists_and_watch("/initialized")
         exists = yield exists_d
         if not exists:
-            log.info("Environment still initializing. Will wait.")
+            log.debug("Environment still initializing. Will wait.")
             yield watch_d
         else:
-            log.debug("Environment already initialized.")
+            log.debug("Environment is initialized.")

=== modified file 'juju/providers/common/tests/test_connect.py'
--- juju/providers/common/tests/test_connect.py	2011-09-15 18:50:23 +0000
+++ juju/providers/common/tests/test_connect.py	2011-12-08 20:20:29 +0000
@@ -1,3 +1,6 @@
+import logging
+import random
+
 from twisted.internet.defer import fail, inlineCallbacks, succeed
 
 import zookeeper
@@ -6,7 +9,7 @@
 from txzookeeper.client import ConnectionTimeoutException
 from txzookeeper.tests.utils import deleteTree
 
-from juju.errors import EnvironmentPending, NoConnection
+from juju.errors import EnvironmentNotFound, NoConnection
 from juju.lib.testing import TestCase
 from juju.machine import ProviderMachine
 from juju.providers.common.base import MachineProviderBase
@@ -16,16 +19,26 @@
 
 class DummyProvider(MachineProviderBase):
 
-    def __init__(self, second_zookeeeper):
-        self._second_zookeeper = second_zookeeeper
+    def __init__(self, *zookeepers):
+        self._zookeepers = zookeepers
 
     def get_zookeeper_machines(self):
         """
         Return a pair of possible zookeepers, the first of which is invalid
         """
-        return succeed([
-            ProviderMachine("i-havenodns"),
-            self._second_zookeeper])
+        machines = [ProviderMachine("i-havenodns")]
+        machines.extend(self._zookeepers)
+        return succeed(machines)
+
+
+class NotBootstrappedProvider(MachineProviderBase):
+    """Pretend to be an environment that has not been bootstrapped."""
+
+    def __init__(self):
+        pass
+
+    def get_zookeeper_machines(self):
+        return fail(EnvironmentNotFound("is the environment bootstrapped?"))
 
 
 class ConnectTest(TestCase):
@@ -35,33 +48,32 @@
         client.connect("foo.example.com:2181", timeout=30, share=share)
         self.mocker.result(result)
 
-    def assert_connect_error(self, error, expect_type, expect_message):
-        self.mock_connect(False, fail(error))
+    @inlineCallbacks
+    def test_none_have_dns_at_first(self):
+        """Verify retry of ZK machine that at first doesn't have a DNS name."""
+        log = self.capture_logging(level=logging.DEBUG)
+        mocked_machine = self.mocker.proxy(ProviderMachine(
+                "i-have-no-dns-at-first"))
+        mocked_machine.dns_name
+        self.mocker.result(None)
+        mocked_machine.dns_name
+        self.mocker.count(0, None)
+        self.mocker.result("foo.example.com")  # name assumed by mock_connect
+
+        # Provide for a valid mocked client once connected
+        client = self.mocker.mock(type=SSHClient)
+        self.mock_connect(True, succeed(client))
+        client.exists_and_watch("/initialized")
+        self.mocker.result((succeed(True), None))
+
         self.mocker.replay()
 
-        provider = DummyProvider(ProviderMachine("i-exist", "foo.example.com"))
-        d = provider.connect()
-
-        def check_error(error):
-            self.assertEqual(str(error), expect_message)
-        self.assertFailure(d, expect_type)
-        d.addCallback(check_error)
-        return d
-
-    def test_none_have_dns(self):
-        """
-        `EnvironmentPending` should be raised if no zookeeper nodes have dns
-        names
-        """
-        provider = DummyProvider(ProviderMachine("i-havenodnseither"))
-        d = provider.connect()
-        self.assertFailure(d, EnvironmentPending)
-
-        def check(error):
-            self.assertEquals(
-                str(error), "No machines have addresses assigned yet")
-        d.addCallback(check)
-        return d
+        provider = DummyProvider(mocked_machine)
+        client = yield provider.connect(share=True)
+
+        self.assertIn(
+            "Retrying connection: No machines have assigned addresses",
+            log.getvalue())
 
     def test_share_kwarg(self):
         """The `share` kwarg should be passed through to `SSHClient.connect`"""
@@ -79,26 +91,80 @@
         d.addCallback(verify)
         return d
 
+    @inlineCallbacks
     def test_no_connection(self):
-        """`NoConnection` errors should become `EnvironmentPending`s"""
-        return self.assert_connect_error(
-            NoConnection("KABOOM!"), EnvironmentPending,
-            "Cannot connect to machine i-exist (perhaps still initializing): "
-            "KABOOM!")
-
+        """Verify retry of `NoConnection` errors."""
+        log = self.capture_logging(level=logging.DEBUG)
+
+        # First connection attempts fails
+        mock_client = self.mocker.patch(SSHClient)
+        mock_client.connect("foo.example.com:2181", timeout=30, share=False)
+        self.mocker.result(fail(NoConnection("KABOOM!")))
+
+        # Subsequent connection attempt then succeeds with a valid
+        # (mocked) client that can be waited on.
+        mock_client.connect("foo.example.com:2181", timeout=30, share=False)
+        self.mocker.result(succeed(mock_client))
+        mock_client.exists_and_watch("/initialized")
+        self.mocker.result((succeed(True), None))
+
+        self.mocker.replay()
+
+        provider = DummyProvider(ProviderMachine("i-exist", "foo.example.com"))
+        yield provider.connect()
+        self.assertIn(
+            "Retrying connection: Cannot connect to environment "
+            "using foo.example.com (perhaps still initializing): KABOOM!",
+            log.getvalue())
+
+    @inlineCallbacks
+    def test_not_bootstrapped(self):
+        """Verify failure when the provider has not been bootstrapped."""
+        provider = NotBootstrappedProvider()
+        e = yield self.assertFailure(provider.connect(), EnvironmentNotFound)
+        self.assertEqual(
+            str(e),
+            "juju environment not found: is the environment bootstrapped?")
+
+    @inlineCallbacks
     def test_txzookeeper_error(self):
-        """
-        `ConnectionTimeoutException` errors should become `EnvironmentPending`s
-        """
-        return self.assert_connect_error(
-            ConnectionTimeoutException("SPLAT!"), EnvironmentPending,
-            "Cannot connect to machine i-exist (perhaps still initializing): "
-            "SPLAT!")
-
+        """Verify retry of `ConnectionTimeoutException` errors."""
+        log = self.capture_logging(level=logging.DEBUG)
+
+        # First connection attempts fails
+        mock_client = self.mocker.patch(SSHClient)
+        mock_client.connect("foo.example.com:2181", timeout=30, share=False)
+        self.mocker.result(fail(ConnectionTimeoutException("SPLAT!")))
+
+        # Subsequent connection attempt then succeeds with a valid
+        # (mocked) client that can be waited on.
+        mock_client.connect("foo.example.com:2181", timeout=30, share=False)
+        self.mocker.result(succeed(mock_client))
+        mock_client.exists_and_watch("/initialized")
+        self.mocker.result((succeed(True), None))
+
+        self.mocker.replay()
+
+        provider = DummyProvider(ProviderMachine("i-exist", "foo.example.com"))
+        yield provider.connect()
+        self.assertIn(
+            "Retrying connection: Cannot connect to environment "
+            "using foo.example.com (perhaps still initializing): SPLAT!",
+            log.getvalue())
+
+    @inlineCallbacks
     def test_other_error(self):
         """Other errors should propagate"""
-        return self.assert_connect_error(
-            TypeError("THUD!"), TypeError, "THUD!")
+        log = self.capture_logging()
+        self.mock_connect(False, fail(TypeError("THUD!")))
+        self.mocker.replay()
+
+        provider = DummyProvider(ProviderMachine("i-exist", "foo.example.com"))
+        ex = yield self.assertFailure(provider.connect(), TypeError)
+        self.assertEqual(str(ex), "THUD!")
+        self.assertIn(
+            "Cannot connect to environment: THUD!",
+            log.getvalue())
 
     @inlineCallbacks
     def test_wait_for_initialize(self):
@@ -107,7 +173,7 @@
         is not ready, should wait until that state is ready.
         """
         client = ZookeeperClient()
-        self.client = client # for poke_zk
+        self.client = client  # for poke_zk
         self.mock_connect(False, succeed(client))
         self.mocker.replay()
 
@@ -131,3 +197,43 @@
         finally:
             deleteTree("/", client.handle)
             client.close()
+
+    @inlineCallbacks
+    def test_fast_connection(self):
+        """Verify connection when requirements are available at time of conn.
+
+        This includes /initialized is already set. In addition, also
+        verifies that if multiple ZKs are available, one is selected
+        via random.choice.
+        """
+        log = self.capture_logging(level=logging.DEBUG)
+        client = ZookeeperClient()
+        self.mock_connect(False, succeed(client))
+
+        def get_choice(lst):
+            for item in lst:
+                if item.dns_name == "foo.example.com":
+                    return item
+            raise AssertionError("expected choice not seen")
+
+        self.patch(random, "choice", get_choice)
+        self.mocker.replay()
+
+        provider = DummyProvider(
+            ProviderMachine("i-am-picked", "foo.example.com"),
+            ProviderMachine("i-was-not", "bar.example.com"))
+
+        zookeeper.set_debug_level(0)
+        yield client.connect(get_test_zookeeper_address())
+        try:
+            yield client.create("/initialized")
+            yield provider.connect()
+            self.assertEqual(
+                "Connecting to environment...\n"
+                "Connecting to environment using foo.example.com...\n"
+                "Environment is initialized.\n"
+                "Connected to environment.\n",
+                log.getvalue())
+        finally:
+            deleteTree("/", client.handle)
+            client.close()

=== removed file 'juju/providers/ec2/tests/test_connect.py'
--- juju/providers/ec2/tests/test_connect.py	2011-09-15 18:50:23 +0000
+++ juju/providers/ec2/tests/test_connect.py	1970-01-01 00:00:00 +0000
@@ -1,143 +0,0 @@
-from yaml import dump
-
-from twisted.internet.defer import inlineCallbacks, succeed, fail
-
-import zookeeper
-
-from txzookeeper import ZookeeperClient
-from txzookeeper.client import ConnectionTimeoutException
-from txzookeeper.tests.utils import deleteTree
-
-from juju.lib.testing import TestCase
-
-from juju.errors import EnvironmentPending, NoConnection
-from juju.state.sshclient import SSHClient
-from juju.providers.ec2.tests.common import EC2TestMixin
-from juju.tests.common import get_test_zookeeper_address
-
-
-class EC2ConnectTest(EC2TestMixin, TestCase):
-
-    def mock_find_zookeepers(self, instance):
-        self.s3.get_object(self.env_name, "provider-state")
-        self.mocker.result(succeed(dump(
-            {"zookeeper-instances": ["i-foobar"]})))
-        self.ec2.describe_instances("i-foobar")
-        self.mocker.result(succeed([instance]))
-
-    def mock_connect(self, share, instance, result):
-        self.mock_find_zookeepers(instance)
-        client = self.mocker.patch(SSHClient)
-        client.connect("foo.example.com:2181", timeout=30, share=share)
-        self.mocker.result(result)
-
-    def assert_connect_error(self, error, expect_type, expect_message):
-        instance = self.get_instance("i-foobar", dns_name="foo.example.com")
-        self.mock_connect(False, instance, fail(error))
-        self.mocker.replay()
-
-        provider = self.get_provider()
-        d = provider.connect()
-
-        def check_error(error):
-            self.assertEqual(str(error), expect_message)
-        self.assertFailure(d, expect_type)
-        d.addCallback(check_error)
-        return d
-
-    def test_no_dns_name(self):
-        """
-        `EnvironmentPending` should be raised if no zookeeper nodes have dns
-        names
-        """
-        instance = self.get_instance("i-foobar")
-        self.mock_find_zookeepers(instance)
-        self.mocker.replay()
-
-        provider = self.get_provider()
-        d = provider.connect()
-
-        def check_error(error):
-            self.assertEqual(
-                str(error), "No machines have addresses assigned yet")
-
-        self.assertFailure(d, NoConnection)
-        d.addCallback(check_error)
-        return d
-
-    def test_provider_connect_forwards_share_option(self):
-        """The `share` kwarg should be passed through to `SSHClient.connect`"""
-        instance = self.get_instance("i-foobar", dns_name="foo.example.com")
-        connected_client = self.mocker.mock(type=SSHClient)
-        self.mock_connect(True, instance, succeed(connected_client))
-        # We'll test the wait on initialization separately.
-        connected_client.exists_and_watch("/initialized")
-        self.mocker.result((succeed(True), None))
-        self.mocker.replay()
-
-        provider = self.get_provider()
-        d = provider.connect(share=True)
-
-        def verify_result(result):
-            self.assertIdentical(connected_client, result)
-        d.addCallback(verify_result)
-        return d
-
-    def test_no_connection(self):
-        """`NoConnection` errors should become `EnvironmentPending`s"""
-        return self.assert_connect_error(
-            NoConnection("KABOOM!"), EnvironmentPending,
-            "Cannot connect to machine i-foobar (perhaps still initializing): "
-            "KABOOM!")
-
-    def test_txzookeeper_error(self):
-        """
-        `ConnectionTimeoutException` errors should become `EnvironmentPending`s
-        """
-        return self.assert_connect_error(
-            ConnectionTimeoutException("SPLAT!"), EnvironmentPending,
-            "Cannot connect to machine i-foobar (perhaps still initializing): "
-            "SPLAT!")
-
-    def test_other_error(self):
-        """Other errors should propagate"""
-        return self.assert_connect_error(
-            TypeError("THUD!"), TypeError, "THUD!")
-
-    @inlineCallbacks
-    def test_provider_connect_waits_on_initialization(self):
-        """
-        A connection to a zookeeper that is running, but whose juju state
-        is not ready, should wait until that state is ready.
-        """
-        # Hand back a real connected client to test the wait on initialization.
-        instance = self.get_instance("i-foobar", dns_name="foo.example.com")
-        connected_client = ZookeeperClient()
-        self.client = connected_client  # for poke_zk
-        self.mock_connect(False, instance, succeed(connected_client))
-
-        self.mocker.replay()
-
-        zookeeper.set_debug_level(0)
-        yield connected_client.connect(get_test_zookeeper_address())
-
-        client_result = []
-
-        provider = self.get_provider()
-        client_deferred = provider.connect()
-        client_deferred.addCallback(client_result.append)
-
-        # Give it a chance to do it incorrectly.
-        yield self.poke_zk()
-
-        try:
-            self.assertEquals(client_result, [])
-
-            yield connected_client.create("/initialized")
-
-            yield client_deferred
-            self.assertTrue(client_result, client_result)
-            self.assertIdentical(client_result[0], connected_client)
-        finally:
-            deleteTree("/", connected_client.handle)
-            connected_client.close()

=== removed file 'juju/providers/orchestra/tests/test_connect.py'
--- juju/providers/orchestra/tests/test_connect.py	2011-10-17 13:49:47 +0000
+++ juju/providers/orchestra/tests/test_connect.py	1970-01-01 00:00:00 +0000
@@ -1,111 +0,0 @@
-from yaml import dump
-
-from twisted.internet.defer import fail, inlineCallbacks, succeed
-
-import zookeeper
-
-from txzookeeper import ZookeeperClient
-from txzookeeper.client import ConnectionTimeoutException
-from txzookeeper.tests.utils import deleteTree
-
-from juju.errors import EnvironmentPending, NoConnection
-from juju.lib.testing import TestCase
-from juju.state.sshclient import SSHClient
-from juju.tests.common import get_test_zookeeper_address
-
-from .common import OrchestraTestMixin
-
-
-class ConnectTest(TestCase, OrchestraTestMixin):
-
-    def mock_connect(self, share, result):
-        self.setup_mocks()
-        self.mock_fs_get(
-            "http://somewhe.re/webdav/provider-state", 200, dump(
-                {"zookeeper-instances": ["foo"]}))
-        self.proxy_m.callRemote("get_systems")
-        self.mocker.result(succeed([{
-            "uid": "foo", "name": "foo.example.com",
-            "mgmt_classes": ["acquired"], "netboot_enabled": True}]))
-        client = self.mocker.patch(SSHClient)
-        client.connect("foo.example.com:2181", timeout=30, share=share)
-        self.mocker.result(result)
-
-    def assert_connect_error(self, error, expect_type, expect_message):
-        self.mock_connect(False, fail(error))
-        self.mocker.replay()
-
-        d = self.get_provider().connect()
-
-        def check_error(error):
-            self.assertEqual(str(error), expect_message)
-        self.assertFailure(d, expect_type)
-        d.addCallback(check_error)
-        return d
-
-    def test_share_kwarg(self):
-        """The `share` kwarg should be passed through to `SSHClient.connect`"""
-        client = self.mocker.mock(type=SSHClient)
-        self.mock_connect(True, succeed(client))
-        client.exists_and_watch("/initialized")
-        self.mocker.result((succeed(True), None))
-        self.mocker.replay()
-
-        d = self.get_provider().connect(share=True)
-
-        def verify(result):
-            self.assertIdentical(result, client)
-        d.addCallback(verify)
-        return d
-
-    def test_no_connection(self):
-        """`NoConnection` errors should become `EnvironmentPending`s"""
-        return self.assert_connect_error(
-            NoConnection("KABOOM!"), EnvironmentPending,
-            "Cannot connect to machine foo (perhaps still initializing): "
-            "KABOOM!")
-
-    def test_txzookeeper_error(self):
-        """
-        `ConnectionTimeoutException` errors should become `EnvironmentPending`s
-        """
-        return self.assert_connect_error(
-            ConnectionTimeoutException("SPLAT!"), EnvironmentPending,
-            "Cannot connect to machine foo (perhaps still initializing): "
-            "SPLAT!")
-
-    def test_other_error(self):
-        """Other errors should propagate"""
-        return self.assert_connect_error(
-            TypeError("THUD!"), TypeError, "THUD!")
-
-    @inlineCallbacks
-    def test_wait_for_initialize(self):
-        """
-        A connection to a zookeeper that is running, but whose juju state
-        is not ready, should wait until that state is ready.
-        """
-        client = ZookeeperClient()
-        self.client = client # for poke_zk
-        self.mock_connect(False, succeed(client))
-        self.mocker.replay()
-
-        zookeeper.set_debug_level(0)
-        yield client.connect(get_test_zookeeper_address())
-
-        d = self.get_provider().connect()
-        client_result = []
-        d.addCallback(client_result.append)
-
-        # Give it an opportunity to do it incorrectly.
-        yield self.poke_zk()
-
-        try:
-            self.assertEquals(client_result, [])
-            yield client.create("/initialized")
-            yield d
-            self.assertTrue(client_result, client_result)
-            self.assertIdentical(client_result[0], client)
-        finally:
-            deleteTree("/", client.handle)
-            client.close()