diff options
Diffstat (limited to 'common/src/leap/soledad')
| -rw-r--r-- | common/src/leap/soledad/common/tests/util.py | 169 |
1 files changed, 36 insertions, 133 deletions
diff --git a/common/src/leap/soledad/common/tests/util.py b/common/src/leap/soledad/common/tests/util.py index 2190eeaa..3187472f 100644 --- a/common/src/leap/soledad/common/tests/util.py +++ b/common/src/leap/soledad/common/tests/util.py @@ -27,10 +27,8 @@ import shutil import random import string import u1db -import subprocess -import time -import re import traceback +import couchdb from uuid import uuid4 from mock import Mock @@ -337,119 +335,6 @@ class BaseSoledadTest(BaseLeapTest, MockedSharedDBTest): self.assertEqual(exp_doc.content, doc.content) -# ----------------------------------------------------------------------------- -# A wrapper for running couchdb locally. -# ----------------------------------------------------------------------------- - -# from: https://github.com/smcq/paisley/blob/master/paisley/test/util.py -# TODO: include license of above project. -class CouchDBWrapper(object): - - """ - Wrapper for external CouchDB instance which is started and stopped for - testing. - """ - BOOT_TIMEOUT_SECONDS = 5 - RETRY_LIMIT = 3 - - def start(self): - tries = 0 - while tries < self.RETRY_LIMIT and not hasattr(self, 'port'): - try: - self._try_start() - return - except Exception, e: - print traceback.format_exc() - self.stop() - tries += 1 - raise Exception( - "Check your couchdb: Tried to start 3 times and failed badly") - - def _try_start(self): - """ - Start a CouchDB instance for a test. - """ - self.tempdir = tempfile.mkdtemp(suffix='.couch.test') - - path = os.path.join(os.path.dirname(__file__), - 'couchdb.ini.template') - handle = open(path) - conf = handle.read() % { - 'tempdir': self.tempdir, - } - handle.close() - - shutil.copy('/etc/couchdb/default.ini', self.tempdir) - defaultConfPath = os.path.join(self.tempdir, 'default.ini') - - confPath = os.path.join(self.tempdir, 'test.ini') - handle = open(confPath, 'w') - handle.write(conf) - handle.close() - - # create the dirs from the template - mkdir_p(os.path.join(self.tempdir, 'lib')) - mkdir_p(os.path.join(self.tempdir, 'log')) - args = ['/usr/bin/couchdb', '-n', - '-a', defaultConfPath, '-a', confPath] - null = open('/dev/null', 'w') - - self.process = subprocess.Popen( - args, env=None, stdout=null.fileno(), stderr=null.fileno(), - close_fds=True) - boot_time = time.time() - # find port - logPath = os.path.join(self.tempdir, 'log', 'couch.log') - while not os.path.exists(logPath): - if self.process.poll() is not None: - got_stdout, got_stderr = "", "" - if self.process.stdout is not None: - got_stdout = self.process.stdout.read() - - if self.process.stderr is not None: - got_stderr = self.process.stderr.read() - raise Exception(""" -couchdb exited with code %d. -stdout: -%s -stderr: -%s""" % ( - self.process.returncode, got_stdout, got_stderr)) - time.sleep(0.01) - if (time.time() - boot_time) > self.BOOT_TIMEOUT_SECONDS: - self.stop() - raise Exception("Timeout starting couch") - while os.stat(logPath).st_size == 0: - time.sleep(0.01) - if (time.time() - boot_time) > self.BOOT_TIMEOUT_SECONDS: - self.stop() - raise Exception("Timeout starting couch") - PORT_RE = re.compile( - 'Apache CouchDB has started on http://127.0.0.1:(?P<port>\d+)') - - handle = open(logPath) - line = handle.read() - handle.close() - m = PORT_RE.search(line) - if not m: - self.stop() - raise Exception("Cannot find port in line %s" % line) - self.port = int(m.group('port')) - - def stop(self): - """ - Terminate the CouchDB instance. - """ - try: - self.process.terminate() - self.process.communicate() - except: - # just to clean up - # if it can't, the process wasn't created anyway - pass - shutil.rmtree(self.tempdir) - - class CouchDBTestCase(unittest.TestCase, MockedSharedDBTest): """ @@ -460,15 +345,26 @@ class CouchDBTestCase(unittest.TestCase, MockedSharedDBTest): """ Make sure we have a CouchDB instance for a test. """ - self.wrapper = CouchDBWrapper() - self.wrapper.start() - # self.db = self.wrapper.db + server = self.couch_server = couchdb.Server() + self.previous_dbs = set([db for db in server]) + self.couch_port = 5984 + self.couch_url = 'http://localhost:%d' % self.couch_port def tearDown(self): """ Stop CouchDB instance for test. """ - self.wrapper.stop() + current_dbs = set([db for db in self.couch_server]) + remaining_dbs = current_dbs - self.previous_dbs + if remaining_dbs and False: + raise Exception("tests created %s and didn't clean up!", remaining_dbs) + + def delete_db(self, name): + try: + self.couch_server.delete(name) + except: + # ignore if already missing + pass class CouchServerStateForTests(CouchServerState): @@ -484,15 +380,25 @@ class CouchServerStateForTests(CouchServerState): which is less pleasant than allowing the db to be automatically created. """ - def _create_database(self, dbname): - return CouchDatabase.open_database( + def __init__(self, *args, **kwargs): + self.dbs = [] + super(CouchServerStateForTests, self).__init__(*args, **kwargs) + + def _create_database(self, replica_uid=None, dbname=None): + """ + Create db and append to a list, allowing test to close it later + """ + dbname = dbname or ('test-%s' % uuid4().hex) + db = CouchDatabase.open_database( urljoin(self.couch_url, dbname), True, - replica_uid=dbname, + replica_uid=replica_uid or 'test', ensure_ddocs=True) + self.dbs.append(db) + return db def ensure_database(self, dbname): - db = self._create_database(dbname) + db = self._create_database(dbname=dbname) return db, db.replica_uid @@ -506,23 +412,20 @@ class SoledadWithCouchServerMixin( main_test_class = getattr(self, 'main_test_class', None) if main_test_class is not None: main_test_class.setUp(self) - self.couch_url = 'http://localhost:%d' % self.wrapper.port def tearDown(self): main_test_class = getattr(self, 'main_test_class', None) if main_test_class is not None: main_test_class.tearDown(self) # delete the test database - try: - db = CouchDatabase(self.couch_url, 'test') - db.delete_database() - except DatabaseDoesNotExist: - pass BaseSoledadTest.tearDown(self) CouchDBTestCase.tearDown(self) def make_app(self): - couch_url = urljoin( - 'http://localhost:' + str(self.wrapper.port), 'tests') - self.request_state = CouchServerStateForTests(couch_url) + self.request_state = CouchServerStateForTests(self.couch_url) + self.addCleanup(self.delete_dbs) return self.make_app_with_state(self.request_state) + + def delete_dbs(self): + for db in self.request_state.dbs: + self.delete_db(db._dbname) |