Removing py.test-specifics.
This commit is contained in:
Ryan Petrello
@@ -1,226 +0,0 @@
|
||||
"""
|
||||
Plugin for py.test that sets up the app.
|
||||
|
||||
App configuration inspired by the Pylons nose equivalent:
|
||||
https://github.com/Pylons/pylons/blob/master/pylons/test.py
|
||||
|
||||
Handling of multiprocessing inspired by pytest-cov.
|
||||
"""
|
||||
from configuration import set_config, _runtime_conf as conf
|
||||
from tempfile import mkdtemp
|
||||
|
||||
import py
|
||||
import py.test
|
||||
import os
|
||||
import shutil
|
||||
import socket
|
||||
import sys
|
||||
|
||||
|
||||
def pytest_addoption(parser):
|
||||
"""
|
||||
Adds the custom "with-config" option to take in the config file.
|
||||
"""
|
||||
group = parser.getgroup('pecan')
|
||||
group._addoption('--with-config',
|
||||
dest='config_file',
|
||||
metavar='path',
|
||||
default='./test.py',
|
||||
action='store',
|
||||
type='string',
|
||||
help='configuration file for pecan tests')
|
||||
|
||||
|
||||
def pytest_configure(config):
|
||||
"""
|
||||
Loads the Pecan plugin if using a configuration file.
|
||||
"""
|
||||
if config.getvalue('config_file'):
|
||||
config.pluginmanager.register(PecanPlugin(config), '_pecan')
|
||||
|
||||
|
||||
class PecanPlugin(object):
|
||||
"""
|
||||
Plugin for a Pecan application. Sets up and tears down the
|
||||
WSGI application based on the configuration and session type.
|
||||
"""
|
||||
|
||||
def __init__(self, config):
|
||||
self.config = config
|
||||
self.impl = None
|
||||
|
||||
def pytest_namespace(self):
|
||||
"""
|
||||
Add the session variables to the namespace.
|
||||
"""
|
||||
return {
|
||||
'temp_dir': None,
|
||||
'wsgi_app': None
|
||||
}
|
||||
|
||||
def pytest_sessionstart(self, session):
|
||||
"""
|
||||
Set up the testing session.
|
||||
"""
|
||||
self.impl = PecanPluginImpl.create_from_session(session)
|
||||
self.impl.sessionstart(session)
|
||||
|
||||
def pytest_configure_node(self, node):
|
||||
"""
|
||||
Configures a new slave node.
|
||||
"""
|
||||
if self.impl:
|
||||
self.impl.configure_node(node)
|
||||
|
||||
def pytest_runtest_setup(self, item):
|
||||
if self.impl:
|
||||
self.impl.ensure_config_loaded()
|
||||
|
||||
def pytest_testnodedown(self, node, error):
|
||||
"""
|
||||
Tears down an exiting node.
|
||||
"""
|
||||
if self.impl:
|
||||
self.impl.testnodedown(node, error)
|
||||
|
||||
def pytest_sessionfinish(self, session, exitstatus):
|
||||
"""
|
||||
Cleans up the testing session.
|
||||
"""
|
||||
if self.impl:
|
||||
self.impl.sessionfinish(session, exitstatus)
|
||||
|
||||
|
||||
class PecanPluginImpl(object):
|
||||
"""
|
||||
Actual implementation of the Pecan plugin. This ensures the proper
|
||||
environment is configured for each session type.
|
||||
"""
|
||||
|
||||
def __init__(self, config):
|
||||
self.config = config
|
||||
self.log = py.log.Producer('pecan-%s' % self.name)
|
||||
if not config.option.debug:
|
||||
py.log.setconsumer(self.log._keywords, None)
|
||||
self.log('Created %s instance' % self.__class__.__name__)
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
return 'main'
|
||||
|
||||
def _setup_app(self):
|
||||
self.log('Invoking setup_app')
|
||||
path = os.getcwd()
|
||||
if path not in sys.path:
|
||||
sys.path.insert(0, path)
|
||||
set_config(self.config.getvalue('config_file'))
|
||||
py.test.wsgi_app = self._load_app(conf)
|
||||
|
||||
def _get_package_names(self, config):
|
||||
if not hasattr(config.app, 'modules'):
|
||||
return []
|
||||
return [module.__name__ for module in config.app.modules if hasattr(module, '__name__')]
|
||||
|
||||
def _can_import(self, name):
|
||||
try:
|
||||
__import__(name)
|
||||
return True
|
||||
except ImportError:
|
||||
return False
|
||||
|
||||
def _load_app(self, config):
|
||||
for package_name in self._get_package_names(config):
|
||||
module_name = '%s.app' % package_name
|
||||
if self._can_import(module_name):
|
||||
module = sys.modules[module_name]
|
||||
if hasattr(module, 'setup_app'):
|
||||
return module.setup_app(config)
|
||||
raise RuntimeError('No app.setup_app found in any of the configured app.modules')
|
||||
|
||||
def _create_temp_directory(self):
|
||||
temp_dir = mkdtemp()
|
||||
self.log('Created temporary directory %s' % temp_dir)
|
||||
py.test.temp_dir = temp_dir
|
||||
|
||||
def _delete_temp_directory(self):
|
||||
if py.test.temp_dir and os.path.exists(py.test.temp_dir):
|
||||
self.log('Removing temporary directory %s' % py.test.temp_dir)
|
||||
shutil.rmtree(py.test.temp_dir)
|
||||
|
||||
def sessionstart(self, session):
|
||||
self.log('Starting session')
|
||||
self._create_temp_directory()
|
||||
|
||||
def ensure_config_loaded(self):
|
||||
if not hasattr(py.test, 'wsgi_app') or py.test.wsgi_app is None:
|
||||
self._setup_app()
|
||||
|
||||
def configure_node(self, node):
|
||||
pass
|
||||
|
||||
def testnodedown(self, node, error):
|
||||
pass
|
||||
|
||||
def sessionfinish(self, session, exitstatus):
|
||||
self.log('Stopping session')
|
||||
self._delete_temp_directory()
|
||||
|
||||
@staticmethod
|
||||
def create_from_session(session):
|
||||
if session.config.option.dist != 'no':
|
||||
impl_cls = MasterPecanPluginImpl
|
||||
elif getattr(session.config, 'slaveinput', {}).get('slaveid'):
|
||||
impl_cls = SlavePecanPluginImpl
|
||||
else:
|
||||
impl_cls = PecanPluginImpl
|
||||
return impl_cls(session.config)
|
||||
|
||||
|
||||
class MasterPecanPluginImpl(PecanPluginImpl):
|
||||
"""
|
||||
Plugin implementation for distributed master.
|
||||
"""
|
||||
|
||||
def sessionstart(self, session):
|
||||
self.log('Starting master session')
|
||||
self._create_temp_directory()
|
||||
self._setup_app()
|
||||
|
||||
def configure_node(self, node):
|
||||
self.log('Configuring slave node %s' % node.gateway.id)
|
||||
node.slaveinput['pecan_master_host'] = socket.gethostname()
|
||||
node.slaveinput['pecan_temp_dir'] = py.test.temp_dir
|
||||
|
||||
def sessionfinish(self, session, exitstatus):
|
||||
self.log('Stopping master session')
|
||||
self._delete_temp_directory()
|
||||
|
||||
|
||||
class SlavePecanPluginImpl(PecanPluginImpl):
|
||||
"""
|
||||
Plugin implementation for distributed slaves.
|
||||
"""
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
return self.config.slaveinput['slaveid']
|
||||
|
||||
def _is_collocated(self, session):
|
||||
return (socket.gethostname() == session.config.slaveinput['pecan_master_host'])
|
||||
|
||||
def _set_temp_directory(self, session):
|
||||
self.log('Setting temporary directory to %s' % session.config.slaveinput['pecan_temp_dir'])
|
||||
py.test.temp_dir = session.config.slaveinput['pecan_temp_dir']
|
||||
|
||||
def sessionstart(self, session):
|
||||
self.log('Starting slave session')
|
||||
if self._is_collocated(session):
|
||||
self._set_temp_directory(session)
|
||||
else:
|
||||
self._create_temp_directory()
|
||||
self._setup_app()
|
||||
|
||||
def sessionfinish(self, session, exitstatus):
|
||||
self.log('Stopping slave session')
|
||||
if not self._is_collocated(session):
|
||||
self._delete_temp_directory()
|
||||
Reference in New Issue
Block a user