/usr/share/pyshared/juju/tests/common.py is in juju-0.7 0.7-0ubuntu2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 | import logging
import os
import tempfile
from juju.lib.zk import Zookeeper
from contextlib import contextmanager
__all__ = ("ManagedZooKeeper",
"zookeeper_test_context",
"get_zookeeper_test_address")
log = logging.getLogger("juju.tests.common")
"""Global to manage the ZK test address - only for testing of course!"""
_zookeeper_address = "127.0.0.1:2181"
def get_test_zookeeper_address():
"""Get the current test ZK address, such as '127.0.0.1:2181'"""
return _zookeeper_address
@contextmanager
def zookeeper_test_context(install_path, port=28181, fsync=False):
"""Manage context to run/stop a ZooKeeper for testing and related vars.
@param install_path: The path to the install for ZK. Bare word "system"
causes special behavior to use system conf for ZK
@param port: The port to run the managed ZK instance
"""
global _zookeeper_address
saved_zookeeper_address = _zookeeper_address
saved_env = os.environ.get("ZOOKEEPER_ADDRESS")
test_zookeeper = Zookeeper(
tempfile.mkdtemp(), port,
zk_location=install_path, use_deferred=False,
fsync=fsync)
test_zookeeper.start()
os.environ["ZOOKEEPER_ADDRESS"] = test_zookeeper.address
_zookeeper_address = test_zookeeper.address
try:
yield test_zookeeper
finally:
test_zookeeper.stop()
_zookeeper_address = saved_zookeeper_address
if saved_env:
os.environ["ZOOKEEPER_ADDRESS"] = saved_env
else:
del os.environ["ZOOKEEPER_ADDRESS"]
|