moved tests to cement.test project

This commit is contained in:
BJ Dierkes 2010-10-23 18:05:56 -05:00
parent b89a9b201c
commit 8cc97bbbda
17 changed files with 86 additions and 331 deletions

View File

@ -11,7 +11,6 @@ from cement.core.exc import CementArgumentError
log = get_logger(__name__)
# FIXME: This method is so effing ugly.
def run_command(cmd_name=None):
"""
Run the command or namespace-subcommand as defined by the 'expose()'
@ -64,9 +63,10 @@ def run_command(cmd_name=None):
if namespaces[namespace].commands.has_key(actual_cmd):
cmd = namespaces[namespace].commands[actual_cmd]
log.debug("executing command '%s'" % actual_cmd)
run_controller_command(cmd['controller_namespace'], cmd['func'],
cli_opts, cli_args)
(res, out_txt) = run_controller_command(cmd['controller_namespace'],
cmd['func'],
cli_opts, cli_args)
return (res, out_txt)
else:
raise CementArgumentError, "Unknown command '%s', see --help?" % actual_cmd

View File

@ -23,7 +23,8 @@ class CementController(object):
def run_controller_command(namespace, func, cli_opts=None, cli_args=None,
*args, **kw):
"""
Cleanly run a command function from a controller.
Cleanly run a command function from a controller. Returns a tuple of
(result_dict, output_txt).
Arguments:
@ -56,8 +57,8 @@ def run_controller_command(namespace, func, cli_opts=None, cli_args=None,
set_config_opts_per_cli_opts(nam, cli_opts)
controller = namespaces[namespace].controller(cli_opts, cli_args)
res = getattr(controller, func)(*args, **kw)
return res
(res, out_txt) = getattr(controller, func)(*args, **kw)
return (res, out_txt)
class expose(object):
"""

View File

@ -25,14 +25,8 @@ class CementRuntimeError(CementError):
code = 1020
CementError.__init__(self, value, code)
class CementInternalServerError(CementError):
"""Unknown or private internal errors."""
def __init__(self, value):
code = 1030
CementError.__init__(self, value, code)
class CementArgumentError(CementError):
"""Argument errors."""
def __init__(self, value):
code = 1040
code = 1030
CementError.__init__(self, value, code)

View File

@ -2,6 +2,7 @@
from cement import handlers
from cement.core.log import get_logger
from cement.core.namespace import get_config
from cement.core.exc import CementRuntimeError
log = get_logger(__name__)
@ -10,8 +11,8 @@ def get_handler(handler_type, handler_name):
if handler_type in handlers:
if handler_name in handlers[handler_type]:
return handlers[handler_type][handler_name]
raise MFRuntimeError, "The handler handlers[%s][%s] does not exist!" \
(handler_type, handler_name)
raise CementRuntimeError, "The handler handlers[%s][%s] does not exist!" \
% (handler_type, handler_name)
def define_handler(type):
"""
@ -32,6 +33,7 @@ def define_handler(type):
define_handler('database')
"""
config = get_config()
log.debug("defining handler type '%s'", type)
if handlers.has_key(type):
raise CementRuntimeError, "Handler type '%s' already defined!" % type

View File

@ -6,8 +6,8 @@ from cement.core.log import get_logger
log = get_logger(__name__)
def clear_hooks():
hooks = {}
#def clear_hooks():
# hooks = {}
def define_hook(name):
"""
@ -70,7 +70,7 @@ class register_hook(object):
log.debug("registering hook func '%s' from %s into hooks['%s']" % \
(func.__name__, func.__module__, self.name))
if not hooks.has_key(self.name):
log.warn("Hook name '%s' is not define!" % self.name)
log.warn("Hook name '%s' is not defined!" % self.name)
return func
# Hooks are as follows: (weight, name, func)
hooks[self.name].append(
@ -78,7 +78,7 @@ class register_hook(object):
)
def run_hooks(*args, **kwargs):
def run_hooks(name, *args, **kwargs):
"""
Run all defined hooks in the namespace. Yields the result of each hook
function run.
@ -103,13 +103,12 @@ def run_hooks(*args, **kwargs):
# do something with result from each hook function
...
"""
name = args[0]
if not hooks.has_key(name):
CementRuntimeError, "Hook name '%s' is not defined!" % name
raise CementRuntimeError, "Hook name '%s' is not defined!" % name
hooks[name].sort() # Will order based on weight
for hook in hooks[name]:
log.debug("running hook '%s' from %s" % (name, hook[2].__module__))
res = hook[2](*args[1:], **kwargs)
res = hook[2](*args, **kwargs)
# Results are yielded, so you must fun a for loop on it, you can not
# simply call run_hooks().

View File

@ -95,7 +95,7 @@ def setup_logging(clear_loggers=True, level=None, to_console=True):
from logging.handlers import RotatingFileHandler
file_handler = RotatingFileHandler(
config['log_file'], maxBytes=int(config['log_max_bytes']),
backupCount=int(config['log_max_files'])
backupCount=int(config['log_max_bytes'])
)
else:
from logging import FileHandler

View File

@ -10,9 +10,6 @@ from cement.core.opt import init_parser
log = get_logger(__name__)
def clear_namespaces():
namespaces = {}
def get_namespace(namespace):
"""
Return the namespace object whose label is 'namespace'.
@ -26,10 +23,12 @@ def get_namespace(namespace):
if namespaces.has_key(namespace):
return namespaces[namespace]
else:
log.fatal("the namespace '%s' doesn't exist" % namespace)
raise CementRuntimeError, "the namespace '%s' doesn't exist" % \
namespace
def get_config(namespace='root'):
"""Get a namespace's config. Returns a ConfigObj object.
"""
Get a namespace's config. Returns a ConfigObj object.
Optional Arguments:
@ -40,7 +39,8 @@ def get_config(namespace='root'):
if namespaces.has_key(namespace):
return namespaces[namespace].config
else:
log.fatal("the namespace '%s' doesn't exist" % namespace)
raise CementRuntimeError, "the namespace '%s' doesn't exist" % \
namespace
class CementNamespace(object):
"""

View File

@ -14,10 +14,10 @@ from cement.core.configuration import ensure_api_compat
log = get_logger(__name__)
class CementPlugin(CementNamespace):
"""Wrapper for CementNamespace."""
def __init__(self, *args, **kwargs):
CementNamespace.__init__(self, *args, **kwargs)
#class CementPlugin(CementNamespace):
# """Wrapper for CementNamespace."""
# def __init__(self, *args, **kwargs):
# CementNamespace.__init__(self, *args, **kwargs)
def get_enabled_plugins():
"""

View File

@ -1,15 +1,26 @@
"""Helper functions for testing applications built on Cement."""
import os
import sys
from shutil import rmtree
from tempfile import mkdtemp
from cement import namespaces
from cement.core.namespace import get_config
from cement.core.exc import CementRuntimeError
from cement.core.controller import run_controller_command
from cement.core.opt import parse_options
def simulate():
def simulate(args=[]):
"""
Simulate running a command at command line. Requires sys.argv to have
Simulate running a command at command line. Requires args to have
the exact args set to it as would be passed at command line.
Required arguments:
args
The args to pass to sys.argv
Usage:
.. code-block:: python
@ -17,14 +28,35 @@ def simulate():
import sys
from cement.core.testing import simulate
sys.argv = ['helloworld', 'example', 'cmd1', '--test-option']
res = simulate()
args = ['helloworld', 'example', 'cmd1', '--test-option']
res = simulate(args)
"""
if not len(sys.argv) >= 1:
raise CementRuntimeError, "args must be set properly."
sys.argv = args
if sys.argv[1] in namespaces:
(opts, args) = parse_options(sys.argv[1], ignore_conflicts=True)
res = run_controller_command(sys.argv[1], sys.argv[2], opts, args)
if not len(sys.argv) >= 3:
raise CementRuntimeError, "A subcommand (additional arg) is required."
else:
namespace = sys.argv[1]
cmd = sys.argv[2]
else:
(opts, args) = parse_options('root', ignore_conflicts=True)
res = run_controller_command('root', sys.argv[1], opts, args)
return res
namespace = 'root'
cmd = sys.argv[1]
(opts, args) = parse_options(namespace, ignore_conflicts=True)
(res_dict, output_txt) = run_controller_command(namespace, cmd, opts, args)
return (res_dict, output_txt)
def setup_func():
"""A generic setup function for nose testing."""
config = get_config()
config['datadir'] = mkdtemp()
def teardown_func():
"""A generic teardown function for nose testing."""
config = get_config()
if os.path.exists(config['datadir']):
rmtree(config['datadir'])

View File

@ -41,7 +41,8 @@ def render_genshi_output(return_dict, template_content=None):
from genshi.template import NewTextTemplate
if template_content:
tmpl = NewTextTemplate(template_content)
return tmpl.generate(**return_dict).render()
res = tmpl.generate(**return_dict).render()
return res
else:
log.debug('template content is empty.')
return ''
@ -90,9 +91,17 @@ class render(object):
Keywork arguments:
output_handler
The name of the output handler to use for rendering
template
The module path to the template (default: None)
When called, a tuple is returned consisting of (dict, output), meaning
the first item is the result dictionary as returned by the original
function, and the second is the output as rendered by the output handler.
"""
def __init__(self, output_handler, template=None):
self.func = None
@ -158,9 +167,10 @@ class render(object):
elif out and self.config['log_to_console']:
out.write(out_txt)
# return res, because we want it to be readable when
# called directly from run_controller_command()
return res
# return res and out_txt, because we want it to be
# readable when called directly from
# run_controller_command()
return (res, out_txt)
else:
raise CementRuntimeError, \
"Handler name '%s' " % self.output_handler + \

View File

View File

View File

@ -1,60 +0,0 @@
from configobj import ConfigObj
from nose.tools import raises, with_setup
from cement import hooks, namespaces, handlers
from cement.core.exc import CementConfigError
from cement.core.app_setup import lay_cement, define_default_hooks
from cement.core.app_setup import define_default_handler_types
from cement.core.app_setup import register_default_handlers
def setup_func():
"set up test fixtures"
pass
def teardown_func():
"tear down test fixtures"
pass
@with_setup(setup_func, teardown_func)
def test_define_default_hooks():
global hooks
define_default_hooks()
expected_hooks = [
'options_hook', 'post_options_hook', 'validate_config_hook',
'pre_plugins_hook', 'post_plugins_hook', 'post_bootstrap_hook'
]
for hook_name in expected_hooks:
yield check_hook, hook_name
def check_hook(hook_name):
assert hooks.has_key(hook_name)
@with_setup(setup_func, teardown_func)
def test_define_default_handler_types():
global handlers
define_default_handler_types()
expected_handler_types = [
'output'
]
for handler_type in expected_handler_types:
yield check_handler_type, handler_type
def check_handler_type(handler_type):
assert handlers.has_key(handler_type)
@with_setup(setup_func, teardown_func)
def test_register_default_handlers():
global handlers
register_default_handlers()
expected_output = [
'genshi', 'json'
]
for handler in expected_output:
yield check_handler, 'output', handler
def check_handler(type, name):
assert handlers[type].has_key(name)
# FIXME: How do you test lay_cement()? Needs a full working (and installed)
# application.

View File

@ -1,90 +0,0 @@
import os
import shutil
from tempfile import mkdtemp
from configobj import ConfigObj
from nose.tools import with_setup, raises, ok_
from cement.core.exc import CementRuntimeError, CementConfigError
from cement.core.configuration import ensure_api_compat, CEMENT_API, t_f_pass
from cement.core.configuration import validate_config
tmpdir = None
def setup_func():
"set up test fixtures"
global tmpdir
tmpdir = mkdtemp()
def teardown_func():
"tear down test fixtures"
global tmpdir
shutil.rmtree(tmpdir)
@raises(CementRuntimeError)
@with_setup(setup_func, teardown_func)
def test_ensure_api_compat_bad():
ensure_api_compat(__name__, 'xxxxx')
@with_setup(setup_func, teardown_func)
def test_ensure_api_compat():
ok_(ensure_api_compat(__name__, CEMENT_API))
@with_setup(setup_func, teardown_func)
def test_validate_config():
global tmpdir
prefix = tmpdir
dcf = ConfigObj() # default config
dcf['config_source'] = ['defaults']
dcf['app_name'] = 'helloworld' # name for cli like /etc/<app_name>
dcf['app_egg_name'] = 'helloworld' # name from setup.py
dcf['app_module'] = 'helloworld' # name of the library dir
dcf['enabled_plugins'] = [] # no default plugins, add via the config file
dcf['debug'] = False
dcf['datadir'] = '%s/data' % prefix
dcf['tmpdir'] = '%s/tmp' % prefix
dcf['log_file'] = '%s/log/%s.log' % (prefix, dcf['app_name'])
dcf['plugin_config_dir'] = '%s/etc/plugins.d' % prefix
dcf['log_to_console'] = True
dcf['output_handler'] = 'genshi'
dcf['show_plugin_load'] = True
# By default look in /etc and ~/ for config files. You should probably
# symlink /etc/<your_app> => ./etc/<your_app> for easy development.
dcf['config_files'] = [
os.path.join(prefix, 'etc', '%s.conf' % dcf['app_name']),
]
validate_config(dcf)
@raises(CementConfigError)
@with_setup(setup_func, teardown_func)
def test_validate_config_bad():
global tmpdir
prefix = tmpdir
dcf = ConfigObj() # default config
validate_config(dcf)
@with_setup(setup_func, teardown_func)
def test_t_f_pass():
for val in ['true', 'True', True]:
yield check_true, val
for val in ['false', 'False', False]:
yield check_false, val
for val in ['a', 'Blah Hah', 100]:
yield check_pass, val
def check_true(val):
assert t_f_pass(val) == True
def check_false(val):
assert t_f_pass(val) == False
def check_pass(val):
assert t_f_pass(val) == val

View File

@ -1,31 +0,0 @@
import os
import shutil
from tempfile import mkdtemp
from configobj import ConfigObj
from nose.tools import with_setup, raises, ok_
from cement.core.exc import CementRuntimeError, CementConfigError
from cement.core.controller import CementController
tmpdir = None
def setup_func():
"set up test fixtures"
global tmpdir
tmpdir = mkdtemp()
def teardown_func():
"tear down test fixtures"
global tmpdir
shutil.rmtree(tmpdir)
@with_setup(setup_func, teardown_func)
def test_controller_class():
# FIXME: passing bogus cause cli_opts is actually an object, but we just
# want to test that self.cli_opts is getting assigned
c = CementController(cli_opts='bogus', cli_args=['a', 'b'])
assert c.cli_opts == 'bogus', "self.cli_opts is not getting set."
assert 'a' in c.cli_args, "self.cli_opts is not getting set."
# FIX ME: rest of stuff requires a running/installed cement app

View File

@ -1,75 +0,0 @@
import os
import shutil
from tempfile import mkdtemp
from configobj import ConfigObj
from nose.tools import with_setup, raises, eq_
from cement.core.exc import CementRuntimeError, CementConfigError
from cement.core.exc import CementArgumentError, CementError
from cement.core.exc import CementInternalServerError
from cement.core.controller import CementController
tmpdir = None
def setup_func():
"set up test fixtures"
global tmpdir
tmpdir = mkdtemp()
def teardown_func():
"tear down test fixtures"
global tmpdir
shutil.rmtree(tmpdir)
@raises(CementConfigError)
@with_setup(setup_func, teardown_func)
def test_exc_config_error():
raise CementConfigError, 'test'
@raises(CementRuntimeError)
@with_setup(setup_func, teardown_func)
def test_exc_runtime_error():
raise CementRuntimeError, 'test'
@raises(CementInternalServerError)
@with_setup(setup_func, teardown_func)
def test_exc_internal_server_error():
raise CementInternalServerError, 'test'
@raises(CementArgumentError)
@with_setup(setup_func, teardown_func)
def test_exc_argument_error():
raise CementArgumentError, 'test'
@with_setup(setup_func, teardown_func)
def test_exc_config_code():
try:
raise CementConfigError, 'test'
except CementConfigError, e:
eq_(e.code, 1010)
@with_setup(setup_func, teardown_func)
def test_exc_runtime_code():
try:
raise CementRuntimeError, 'test'
except CementRuntimeError, e:
eq_(e.code, 1020)
@with_setup(setup_func, teardown_func)
def test_exc_internal_server_code():
try:
raise CementInternalServerError, 'test'
except CementInternalServerError, e:
eq_(e.code, 1030)
@with_setup(setup_func, teardown_func)
def test_exc_argument_code():
try:
raise CementArgumentError, 'test'
except CementArgumentError, e:
eq_(e.code, 1040)

View File

@ -1,27 +0,0 @@
from nose.tools import raises, with_setup, eq_
from cement import hooks, namespaces, handlers
from cement.core.view import render_genshi_output, render_json_output
def setup_func():
"set up test fixtures"
pass
def teardown_func():
"tear down test fixtures"
pass
@with_setup(setup_func, teardown_func)
def test_render_genshi_output():
fake_dict = dict(foo='String', bar=100, list=[1,2,3,4,5])
tmpl_content = """$foo$bar{% for i in list %}${i}{% end %}"""
output = render_genshi_output(fake_dict, tmpl_content)
eq_(output, 'String10012345')
@with_setup(setup_func, teardown_func)
def test_render_json_output():
fake_dict = dict(foo='String', bar=100, list=[1,2,3,4,5])
tmpl_content = """$foo$bar{% for i in list %}${i}{% end %}"""
output = render_json_output(fake_dict, tmpl_content)
eq_(output, '{"bar": 100, "foo": "String", "list": [1, 2, 3, 4, 5], "stderr": "", "stdout": ""}')