113 lines
3.2 KiB
Python
113 lines
3.2 KiB
Python
"""
|
|
Utility and helper functions for the packaged tasks
|
|
|
|
:copyright: Copyright 2013-2015, Dorian Pula <dorian.pula@amber-penguin-software.ca>
|
|
:license: AGPL v3+
|
|
"""
|
|
|
|
import contextlib
|
|
import copy
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import traceback
|
|
|
|
import arrow
|
|
import pathlib
|
|
import pycouchdb
|
|
|
|
|
|
def delete_in_path_list(path_list, dry_run=False):
|
|
"""
|
|
Deletes all the elements in the given path list.
|
|
|
|
:param path_list: A list of paths to delete.
|
|
:param dry_run: Mode for switching between a dry run and actual deletion run.
|
|
"""
|
|
if dry_run:
|
|
print('Running in: Dry run mode.')
|
|
|
|
for folder in path_list:
|
|
path = pathlib.Path(folder)
|
|
if not path.exists():
|
|
continue
|
|
print('Deleting... {}'.format(path))
|
|
|
|
if not dry_run and path.is_dir():
|
|
shutil.rmtree(path.as_posix())
|
|
continue
|
|
|
|
if not dry_run:
|
|
path.unlink()
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def cd(path):
|
|
"""
|
|
Context manager to change a current working directory.
|
|
|
|
:param path: The path to change the current working directory.
|
|
"""
|
|
previous_path = os.getcwd()
|
|
os.chdir(path)
|
|
yield
|
|
os.chdir(previous_path)
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def environment(environ_modifier):
|
|
"""
|
|
Context manager to store all the environment variables, modify them and then restore them.
|
|
|
|
:param environ_modifier: A dictionary of environment variables and the values to change.
|
|
"""
|
|
current_env = copy.deepcopy(os.environ)
|
|
if not isinstance(environ_modifier, dict):
|
|
raise ValueError('Provided environment "{}" is a not a dict!'.format(environ_modifier))
|
|
|
|
non_string_keys = [key for key in environ_modifier.keys() if not isinstance(key, basestring)]
|
|
if non_string_keys:
|
|
raise ValueError('Provided environment contains invalid keys: "{}"!'.format(non_string_keys))
|
|
|
|
for environ_key, value in environ_modifier.items():
|
|
os.environ[environ_key] = value
|
|
yield
|
|
os.environ = current_env
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def test_database(couchdb_connection='http://admin:password@localhost:5984/', keep_test_db=False,
|
|
test_db_name='rookeries_{}'.format(arrow.now().format('YYYY_MM_DD_HHmmss'))):
|
|
"""
|
|
Context for creating and managing a test database.
|
|
|
|
:param couchdb_connection: Connection to the CouchDB database to go against.
|
|
:param keep_test_db: Keep the test data and database around.
|
|
:param test_db_name: The name of the test database.
|
|
"""
|
|
|
|
couchdb_server = None
|
|
try:
|
|
couchdb_server = pycouchdb.Server(couchdb_connection)
|
|
test_db = couchdb_server.create(test_db_name)
|
|
|
|
from tests.fixtures import sample_site
|
|
test_db.save_bulk(sample_site.bulk_docs_export())
|
|
modified_environ = {
|
|
'ROOKERIES_COUCHDB_DATABASE_URI': couchdb_connection,
|
|
'ROOKERIES_COUCHDB_DATABASE': test_db_name,
|
|
}
|
|
|
|
with environment(modified_environ):
|
|
pass
|
|
yield
|
|
|
|
except Exception as err:
|
|
print('Unable to run server test because of "{message}" \n Stack: \n'.format(message=err.message))
|
|
traceback.print_tb(sys.exc_traceback)
|
|
raise err
|
|
|
|
finally:
|
|
if not keep_test_db and couchdb_server:
|
|
couchdb_server.delete(test_db_name)
|