/usr/share/pyshared/txzookeeper/tests/__init__.py is in python-txzookeeper 0.9.5-0ubuntu1.
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 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 | #
# Copyright (C) 2010-2011 Canonical Ltd. All Rights Reserved
#
# This file is part of txzookeeper.
#
# Authors:
# Kapil Thangavelu
#
# 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 logging
import StringIO
import sys
import zookeeper
from twisted.internet.defer import Deferred
from twisted.trial.unittest import TestCase
from mocker import MockerTestCase
class ZookeeperTestCase(TestCase, MockerTestCase):
def setUp(self):
super(ZookeeperTestCase, self).setUp()
zookeeper.set_debug_level(0)
def tearDown(self):
super(ZookeeperTestCase, self).tearDown()
def get_log(self):
return open(self.log_file_path).read()
def sleep(self, delay):
"""Non-blocking sleep."""
from twisted.internet import reactor
deferred = Deferred()
reactor.callLater(delay, deferred.callback, None)
return deferred
_missing_attr = object()
def patch(self, object, attr, value):
"""Replace an object's attribute, and restore original value later.
Returns the original value of the attribute if any or None.
"""
original_value = getattr(object, attr, self._missing_attr)
@self.addCleanup
def restore_original():
if original_value is self._missing_attr:
try:
delattr(object, attr)
except AttributeError:
pass
else:
setattr(object, attr, original_value)
setattr(object, attr, value)
if original_value is self._missing_attr:
return None
return original_value
def capture_log(
self, name="", level=logging.INFO, log_file=None, formatter=None):
"""Capture log channel to StringIO"""
if log_file is None:
log_file = StringIO.StringIO()
log_handler = logging.StreamHandler(log_file)
if formatter:
log_handler.setFormatter(formatter)
logger = logging.getLogger(name)
logger.addHandler(log_handler)
old_logger_level = logger.level
logger.setLevel(level)
@self.addCleanup
def reset_logging():
logger.removeHandler(log_handler)
logger.setLevel(old_logger_level)
return log_file
def egg_test_runner():
"""
Test collector and runner for setup.py test
"""
from twisted.scripts.trial import run
original_args = list(sys.argv)
sys.argv = ["", "txzookeeper"]
try:
return run()
finally:
sys.argv = original_args
|