SLIDER-628 Cleanup a few unwanted test related kazoo files
diff --git a/slider-agent/src/main/python/kazoo/testing/__init__.py b/slider-agent/src/main/python/kazoo/testing/__init__.py
deleted file mode 100644
index 660546b..0000000
--- a/slider-agent/src/main/python/kazoo/testing/__init__.py
+++ /dev/null
@@ -1,6 +0,0 @@
-"""license: Apache License 2.0, see LICENSE for more details."""
-from kazoo.testing.harness import KazooTestCase
-from kazoo.testing.harness import KazooTestHarness
-
-
-__all__ = ('KazooTestHarness', 'KazooTestCase', )
diff --git a/slider-agent/src/main/python/kazoo/testing/common.py b/slider-agent/src/main/python/kazoo/testing/common.py
deleted file mode 100644
index b497a8e..0000000
--- a/slider-agent/src/main/python/kazoo/testing/common.py
+++ /dev/null
@@ -1,284 +0,0 @@
-"""license: Apache License 2.0, see LICENSE for more details."""
-#
-# Copyright (C) 2010-2011, 2011 Canonical Ltd. All Rights Reserved
-#
-# This file was originally taken from txzookeeper and modified later.
-#
-# Authors:
-# Kapil Thangavelu and the Kazoo team
-#
-# txzookeeper is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Lesser General Public License as published by
-# the Free Software Foundation, either version 3 of the License, or
-# (at your option) any later version.
-#
-# txzookeeper is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU Lesser General Public License for more details.
-#
-# You should have received a copy of the GNU Lesser General Public License
-# along with txzookeeper. If not, see <http://www.gnu.org/licenses/>.
-
-
-import code
-import os
-import os.path
-import shutil
-import signal
-import subprocess
-import tempfile
-import traceback
-
-from itertools import chain
-from collections import namedtuple
-from glob import glob
-
-
-def debug(sig, frame):
- """Interrupt running process, and provide a python prompt for
- interactive debugging."""
- d = {'_frame': frame} # Allow access to frame object.
- d.update(frame.f_globals) # Unless shadowed by global
- d.update(frame.f_locals)
-
- i = code.InteractiveConsole(d)
- message = "Signal recieved : entering python shell.\nTraceback:\n"
- message += ''.join(traceback.format_stack(frame))
- i.interact(message)
-
-
-def listen():
- if os.name != 'nt': # SIGUSR1 is not supported on Windows
- signal.signal(signal.SIGUSR1, debug) # Register handler
-listen()
-
-
-def to_java_compatible_path(path):
- if os.name == 'nt':
- path = path.replace('\\', '/')
- return path
-
-ServerInfo = namedtuple(
- "ServerInfo", "server_id client_port election_port leader_port")
-
-
-class ManagedZooKeeper(object):
- """Class to manage the running of a ZooKeeper instance for testing.
-
- Note: no attempt is made to probe the ZooKeeper instance is
- actually available, or that the selected port is free. In the
- future, we may want to do that, especially when run in a
- Hudson/Buildbot context, to ensure more test robustness."""
-
- def __init__(self, software_path, server_info, peers=(), classpath=None):
- """Define the ZooKeeper test instance.
-
- @param install_path: The path to the install for ZK
- @param port: The port to run the managed ZK instance
- """
- self.install_path = software_path
- self._classpath = classpath
- self.server_info = server_info
- self.host = "127.0.0.1"
- self.peers = peers
- self.working_path = tempfile.mkdtemp()
- self._running = False
-
- def run(self):
- """Run the ZooKeeper instance under a temporary directory.
-
- Writes ZK log messages to zookeeper.log in the current directory.
- """
- if self.running:
- return
- config_path = os.path.join(self.working_path, "zoo.cfg")
- log_path = os.path.join(self.working_path, "log")
- log4j_path = os.path.join(self.working_path, "log4j.properties")
- data_path = os.path.join(self.working_path, "data")
-
- # various setup steps
- if not os.path.exists(self.working_path):
- os.mkdir(self.working_path)
- if not os.path.exists(log_path):
- os.mkdir(log_path)
- if not os.path.exists(data_path):
- os.mkdir(data_path)
-
- with open(config_path, "w") as config:
- config.write("""
-tickTime=2000
-dataDir=%s
-clientPort=%s
-maxClientCnxns=0
-""" % (to_java_compatible_path(data_path), self.server_info.client_port))
-
- # setup a replicated setup if peers are specified
- if self.peers:
- servers_cfg = []
- for p in chain((self.server_info,), self.peers):
- servers_cfg.append("server.%s=localhost:%s:%s" % (
- p.server_id, p.leader_port, p.election_port))
-
- with open(config_path, "a") as config:
- config.write("""
-initLimit=4
-syncLimit=2
-%s
-""" % ("\n".join(servers_cfg)))
-
- # Write server ids into datadir
- with open(os.path.join(data_path, "myid"), "w") as myid_file:
- myid_file.write(str(self.server_info.server_id))
-
- with open(log4j_path, "w") as log4j:
- log4j.write("""
-# DEFAULT: console appender only
-log4j.rootLogger=INFO, ROLLINGFILE
-log4j.appender.ROLLINGFILE.layout=org.apache.log4j.PatternLayout
-log4j.appender.ROLLINGFILE.layout.ConversionPattern=%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L] - %m%n
-log4j.appender.ROLLINGFILE=org.apache.log4j.RollingFileAppender
-log4j.appender.ROLLINGFILE.Threshold=DEBUG
-log4j.appender.ROLLINGFILE.File=""" + to_java_compatible_path(
- self.working_path + os.sep + "zookeeper.log\n"))
-
- self.process = subprocess.Popen(
- args=["java",
- "-cp", self.classpath,
- "-Dreadonlymode.enabled=true",
- "-Dzookeeper.log.dir=%s" % log_path,
- "-Dzookeeper.root.logger=INFO,CONSOLE",
- "-Dlog4j.configuration=file:%s" % log4j_path,
- # "-Dlog4j.debug",
- "org.apache.zookeeper.server.quorum.QuorumPeerMain",
- config_path])
- self._running = True
-
- @property
- def classpath(self):
- """Get the classpath necessary to run ZooKeeper."""
-
- if self._classpath:
- return self._classpath
-
- # Two possibilities, as seen in zkEnv.sh:
- # Check for a release - top-level zookeeper-*.jar?
- jars = glob((os.path.join(
- self.install_path, 'zookeeper-*.jar')))
- if jars:
- # Release build (`ant package`)
- jars.extend(glob(os.path.join(
- self.install_path,
- "lib/*.jar")))
- # support for different file locations on Debian/Ubuntu
- jars.extend(glob(os.path.join(
- self.install_path,
- "log4j-*.jar")))
- jars.extend(glob(os.path.join(
- self.install_path,
- "slf4j-api-*.jar")))
- jars.extend(glob(os.path.join(
- self.install_path,
- "slf4j-log4j-*.jar")))
- else:
- # Development build (plain `ant`)
- jars = glob((os.path.join(
- self.install_path, 'build/zookeeper-*.jar')))
- jars.extend(glob(os.path.join(
- self.install_path,
- "build/lib/*.jar")))
-
- return os.pathsep.join(jars)
-
- @property
- def address(self):
- """Get the address of the ZooKeeper instance."""
- return "%s:%s" % (self.host, self.client_port)
-
- @property
- def running(self):
- return self._running
-
- @property
- def client_port(self):
- return self.server_info.client_port
-
- def reset(self):
- """Stop the zookeeper instance, cleaning out its on disk-data."""
- self.stop()
- shutil.rmtree(os.path.join(self.working_path, "data"))
- os.mkdir(os.path.join(self.working_path, "data"))
- with open(os.path.join(self.working_path, "data", "myid"), "w") as fh:
- fh.write(str(self.server_info.server_id))
-
- def stop(self):
- """Stop the Zookeeper instance, retaining on disk state."""
- if not self.running:
- return
- self.process.terminate()
- self.process.wait()
- self._running = False
-
- def destroy(self):
- """Stop the ZooKeeper instance and destroy its on disk-state"""
- # called by at exit handler, reimport to avoid cleanup race.
- import shutil
- self.stop()
-
- shutil.rmtree(self.working_path)
-
-
-class ZookeeperCluster(object):
-
- def __init__(self, install_path=None, classpath=None, size=3, port_offset=20000):
- self._install_path = install_path
- self._classpath = classpath
- self._servers = []
-
- # Calculate ports and peer group
- port = port_offset
- peers = []
-
- for i in range(size):
- info = ServerInfo(i + 1, port, port + 1, port + 2)
- peers.append(info)
- port += 10
-
- # Instantiate Managed ZK Servers
- for i in range(size):
- server_peers = list(peers)
- server_info = server_peers.pop(i)
- self._servers.append(
- ManagedZooKeeper(
- self._install_path, server_info, server_peers, classpath=self._classpath))
-
- def __getitem__(self, k):
- return self._servers[k]
-
- def __iter__(self):
- return iter(self._servers)
-
- def start(self):
- # Zookeeper client expresses a preference for either lower ports or
- # lexicographical ordering of hosts, to ensure that all servers have a
- # chance to startup, start them in reverse order.
- for server in reversed(list(self)):
- server.run()
- # Giving the servers a moment to start, decreases the overall time
- # required for a client to successfully connect (2s vs. 4s without
- # the sleep).
- import time
- time.sleep(2)
-
- def stop(self):
- for server in self:
- server.stop()
- self._servers = []
-
- def terminate(self):
- for server in self:
- server.destroy()
-
- def reset(self):
- for server in self:
- server.reset()
diff --git a/slider-agent/src/main/python/kazoo/testing/harness.py b/slider-agent/src/main/python/kazoo/testing/harness.py
deleted file mode 100644
index 93cc744..0000000
--- a/slider-agent/src/main/python/kazoo/testing/harness.py
+++ /dev/null
@@ -1,181 +0,0 @@
-"""license: Apache License 2.0, see LICENSE for more details."""
-"""Kazoo testing harnesses"""
-import atexit
-import logging
-import os
-import uuid
-import threading
-import unittest
-
-from kazoo.client import KazooClient
-from kazoo.exceptions import NotEmptyError
-from kazoo.protocol.states import (
- KazooState
-)
-from kazoo.testing.common import ZookeeperCluster
-from kazoo.protocol.connection import _SESSION_EXPIRED
-
-log = logging.getLogger(__name__)
-
-CLUSTER = None
-
-
-def get_global_cluster():
- global CLUSTER
- if CLUSTER is None:
- ZK_HOME = os.environ.get("ZOOKEEPER_PATH")
- ZK_CLASSPATH = os.environ.get("ZOOKEEPER_CLASSPATH")
- ZK_PORT_OFFSET = int(os.environ.get("ZOOKEEPER_PORT_OFFSET", 20000))
-
- assert ZK_HOME or ZK_CLASSPATH, (
- "either ZOOKEEPER_PATH or ZOOKEEPER_CLASSPATH environment variable "
- "must be defined.\n"
- "For deb package installations this is /usr/share/java")
-
- CLUSTER = ZookeeperCluster(
- install_path=ZK_HOME,
- classpath=ZK_CLASSPATH,
- port_offset=ZK_PORT_OFFSET,
- )
- atexit.register(lambda cluster: cluster.terminate(), CLUSTER)
- return CLUSTER
-
-
-class KazooTestHarness(unittest.TestCase):
- """Harness for testing code that uses Kazoo
-
- This object can be used directly or as a mixin. It supports starting
- and stopping a complete ZooKeeper cluster locally and provides an
- API for simulating errors and expiring sessions.
-
- Example::
-
- class MyTestCase(KazooTestHarness):
- def setUp(self):
- self.setup_zookeeper()
-
- # additional test setup
-
- def tearDown(self):
- self.teardown_zookeeper()
-
- def test_something(self):
- something_that_needs_a_kazoo_client(self.client)
-
- def test_something_else(self):
- something_that_needs_zk_servers(self.servers)
-
- """
-
- def __init__(self, *args, **kw):
- super(KazooTestHarness, self).__init__(*args, **kw)
- self.client = None
- self._clients = []
-
- @property
- def cluster(self):
- return get_global_cluster()
-
- @property
- def servers(self):
- return ",".join([s.address for s in self.cluster])
-
- def _get_nonchroot_client(self):
- return KazooClient(self.servers)
-
- def _get_client(self, **kwargs):
- c = KazooClient(self.hosts, **kwargs)
- try:
- self._clients.append(c)
- except AttributeError:
- self._client = [c]
- return c
-
- def expire_session(self, client_id=None):
- """Force ZK to expire a client session
-
- :param client_id: id of client to expire. If unspecified, the id of
- self.client will be used.
-
- """
- client_id = client_id or self.client.client_id
-
- lost = threading.Event()
- safe = threading.Event()
-
- def watch_loss(state):
- if state == KazooState.LOST:
- lost.set()
- if lost.is_set() and state == KazooState.CONNECTED:
- safe.set()
- return True
-
- self.client.add_listener(watch_loss)
-
- self.client._call(_SESSION_EXPIRED, None)
-
- lost.wait(5)
- if not lost.isSet():
- raise Exception("Failed to get notified of session loss")
-
- # Wait for the reconnect now
- safe.wait(15)
- if not safe.isSet():
- raise Exception("Failed to see client reconnect")
- self.client.retry(self.client.get_async, '/')
-
- def setup_zookeeper(self, **client_options):
- """Create a ZK cluster and chrooted :class:`KazooClient`
-
- The cluster will only be created on the first invocation and won't be
- fully torn down until exit.
- """
- if not self.cluster[0].running:
- self.cluster.start()
- namespace = "/kazootests" + uuid.uuid4().hex
- self.hosts = self.servers + namespace
-
- if 'timeout' not in client_options:
- client_options['timeout'] = 0.8
- self.client = self._get_client(**client_options)
- self.client.start()
- self.client.ensure_path("/")
-
- def teardown_zookeeper(self):
- """Clean up any ZNodes created during the test
- """
- if not self.cluster[0].running:
- self.cluster.start()
-
- tries = 0
- if self.client and self.client.connected:
- while tries < 3:
- try:
- self.client.retry(self.client.delete, '/', recursive=True)
- break
- except NotEmptyError:
- pass
- tries += 1
- self.client.stop()
- self.client.close()
- del self.client
- else:
- client = self._get_client()
- client.start()
- client.retry(client.delete, '/', recursive=True)
- client.stop()
- client.close()
- del client
-
- for client in self._clients:
- client.stop()
- del client
- self._clients = None
-
-
-class KazooTestCase(KazooTestHarness):
- def setUp(self):
- self.setup_zookeeper()
-
- def tearDown(self):
- self.teardown_zookeeper()
