mirror of
https://github.com/datafolklabs/cement.git
synced 2026-02-06 13:56:49 +00:00
Resolves Issue #363
This commit is contained in:
parent
fbd58da54d
commit
a71d9e78d1
@ -28,6 +28,8 @@ development will happen as 2.9.x under the git master branch.
|
||||
|
||||
Bugs:
|
||||
|
||||
* :issue:`363` - CementTestCase does not delete temporary
|
||||
files/directories
|
||||
* :issue:`346` - AttributeError: 'module' object has no attribute 'SIGHUP'
|
||||
on Windows
|
||||
* :issue:`352` - ``CementApp.extend()`` Breaks on ``CementApp.reload()``
|
||||
|
||||
@ -1,6 +1,8 @@
|
||||
"""Cement testing utilities."""
|
||||
|
||||
import os
|
||||
import unittest
|
||||
import shutil
|
||||
from tempfile import mkstemp, mkdtemp
|
||||
from ..core import backend, foundation
|
||||
from ..utils.misc import rando
|
||||
@ -40,6 +42,8 @@ class CementTestCase(unittest.TestCase):
|
||||
|
||||
def __init__(self, *args, **kw):
|
||||
super(CementTestCase, self).__init__(*args, **kw)
|
||||
self.tmp_file = None
|
||||
self.tmp_dir = None
|
||||
|
||||
def setUp(self):
|
||||
"""
|
||||
@ -49,8 +53,21 @@ class CementTestCase(unittest.TestCase):
|
||||
|
||||
"""
|
||||
self.app = self.make_app()
|
||||
_, self.tmp_file = mkstemp()
|
||||
self.tmp_dir = mkdtemp()
|
||||
|
||||
# recreate temp file and dir for each test
|
||||
_prefix = "cement.tests.%s.tmp" % self.__class__.__name__
|
||||
_, self.tmp_file = mkstemp(prefix=_prefix)
|
||||
self.tmp_dir = mkdtemp(prefix=_prefix)
|
||||
|
||||
def tearDown(self):
|
||||
"""
|
||||
Tears down the test environment (if necessary), removes any temporary
|
||||
files/directories, etc.
|
||||
"""
|
||||
if os.path.exists(self.tmp_file):
|
||||
os.remove(self.tmp_file)
|
||||
if os.path.exists(self.tmp_dir):
|
||||
shutil.rmtree(self.tmp_dir)
|
||||
|
||||
def make_app(self, *args, **kw):
|
||||
"""
|
||||
|
||||
@ -1,7 +1,6 @@
|
||||
"""Tests for cement.core.config."""
|
||||
|
||||
import os
|
||||
from tempfile import mkstemp
|
||||
from cement.core import exc, config, backend
|
||||
from cement.utils import test
|
||||
|
||||
@ -25,10 +24,9 @@ class ConfigTestCase(test.CementCoreTestCase):
|
||||
|
||||
@test.raises(NotImplementedError)
|
||||
def test_parse_file_not_implemented(self):
|
||||
_, tmppath = mkstemp()
|
||||
c = config.CementConfigHandler()
|
||||
c._setup(self.app)
|
||||
c._parse_file(tmppath)
|
||||
c._parse_file(self.tmp_file)
|
||||
|
||||
def test_has_key(self):
|
||||
self.app.setup()
|
||||
@ -80,10 +78,9 @@ class ConfigTestCase(test.CementCoreTestCase):
|
||||
self.app.setup()
|
||||
|
||||
def test_parse_file(self):
|
||||
_, tmppath = mkstemp()
|
||||
f = open(tmppath, 'w+')
|
||||
f = open(self.tmp_file, 'w+')
|
||||
f.write(CONFIG)
|
||||
f.close()
|
||||
self.app._meta.config_files = [tmppath]
|
||||
self.app._meta.config_files = [self.tmp_file]
|
||||
self.app.setup()
|
||||
self.eq(self.app.config.get('my_section', 'my_param'), 'my_value')
|
||||
|
||||
@ -70,6 +70,7 @@ class TestHandler(meta.MetaMixin):
|
||||
class HandlerTestCase(test.CementCoreTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(HandlerTestCase, self).setUp()
|
||||
self.app = self.make_app()
|
||||
|
||||
@test.raises(exc.FrameworkError)
|
||||
@ -187,6 +188,7 @@ class HandlerTestCase(test.CementCoreTestCase):
|
||||
class DeprecatedHandlerTestCase(test.CementCoreTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(DeprecatedHandlerTestCase, self).setUp()
|
||||
self.app = self.make_app()
|
||||
|
||||
@test.raises(exc.FrameworkError)
|
||||
|
||||
@ -37,6 +37,7 @@ def cement_hook_six(*args, **kw):
|
||||
class HookTestCase(test.CementCoreTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(HookTestCase, self).setUp()
|
||||
self.app = self.make_app()
|
||||
self.app.hook.define('nosetests_hook')
|
||||
|
||||
@ -113,6 +114,7 @@ class HookTestCase(test.CementCoreTestCase):
|
||||
class DeprecatedHookTestCase(test.CementCoreTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(DeprecatedHookTestCase, self).setUp()
|
||||
self.app = self.make_app()
|
||||
hook.define('nosetests_hook')
|
||||
|
||||
|
||||
@ -32,6 +32,7 @@ class TestHandler3():
|
||||
class InterfaceTestCase(test.CementCoreTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(InterfaceTestCase, self).setUp()
|
||||
self.app = self.make_app()
|
||||
|
||||
@test.raises(exc.InterfaceError)
|
||||
|
||||
@ -16,6 +16,7 @@ class BogusHandler1(log.CementLogHandler):
|
||||
class LogTestCase(test.CementCoreTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(LogTestCase, self).setUp()
|
||||
self.app = self.make_app()
|
||||
|
||||
@test.raises(exc.InterfaceError)
|
||||
|
||||
@ -1,7 +1,6 @@
|
||||
"""Tests for cement.core.output."""
|
||||
|
||||
import os
|
||||
from tempfile import mkdtemp
|
||||
from cement.core import exc, backend, output
|
||||
from cement.utils import test
|
||||
from cement.utils.misc import init_defaults, rando
|
||||
@ -24,11 +23,11 @@ TEST_TEMPLATE = "%(foo)s"
|
||||
class OutputTestCase(test.CementCoreTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(OutputTestCase, self).setUp()
|
||||
self.app = self.make_app()
|
||||
|
||||
def test_load_template_from_file(self):
|
||||
tmpdir = mkdtemp()
|
||||
template = os.path.join(tmpdir, 'mytemplate.txt')
|
||||
template = os.path.join(self.tmp_dir, 'mytemplate.txt')
|
||||
|
||||
f = open(template, 'w')
|
||||
f.write(TEST_TEMPLATE)
|
||||
@ -36,7 +35,7 @@ class OutputTestCase(test.CementCoreTestCase):
|
||||
|
||||
app = self.make_app(APP,
|
||||
config_files=[],
|
||||
template_dir=tmpdir,
|
||||
template_dir=self.tmp_dir,
|
||||
output_handler=TestOutputHandler,
|
||||
)
|
||||
app.setup()
|
||||
@ -45,12 +44,11 @@ class OutputTestCase(test.CementCoreTestCase):
|
||||
|
||||
@test.raises(exc.FrameworkError)
|
||||
def test_load_template_from_bad_file(self):
|
||||
tmpdir = mkdtemp()
|
||||
template = os.path.join(tmpdir, 'my-bogus-template.txt')
|
||||
template = os.path.join(self.tmp_dir, 'my-bogus-template.txt')
|
||||
|
||||
app = self.make_app(APP,
|
||||
config_files=[],
|
||||
template_dir=tmpdir,
|
||||
template_dir=self.tmp_dir,
|
||||
output_handler=TestOutputHandler,
|
||||
)
|
||||
app.setup()
|
||||
|
||||
@ -3,7 +3,6 @@
|
||||
import os
|
||||
import sys
|
||||
import shutil
|
||||
from tempfile import mkdtemp
|
||||
from cement.core import exc, backend, plugin
|
||||
from cement.utils import test
|
||||
from cement.utils.misc import init_defaults, rando
|
||||
@ -60,52 +59,47 @@ def load(app):
|
||||
class PluginTestCase(test.CementCoreTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(PluginTestCase, self).setUp()
|
||||
self.app = self.make_app()
|
||||
|
||||
def test_load_plugins_from_files(self):
|
||||
tmpdir = mkdtemp()
|
||||
f = open(os.path.join(tmpdir, 'myplugin.conf'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'myplugin.conf'), 'w')
|
||||
f.write(CONF)
|
||||
f.close()
|
||||
|
||||
f = open(os.path.join(tmpdir, 'myplugin.py'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'myplugin.py'), 'w')
|
||||
f.write(PLUGIN)
|
||||
f.close()
|
||||
|
||||
app = self.make_app(APP,
|
||||
config_files=[],
|
||||
plugin_config_dir=tmpdir,
|
||||
plugin_dir=tmpdir,
|
||||
plugin_config_dir=self.tmp_dir,
|
||||
plugin_dir=self.tmp_dir,
|
||||
plugin_bootstrap=None,
|
||||
)
|
||||
app.setup()
|
||||
|
||||
try:
|
||||
han = app.handler.get('output', 'test_output_handler')()
|
||||
self.eq(han._meta.label, 'test_output_handler')
|
||||
finally:
|
||||
shutil.rmtree(tmpdir)
|
||||
han = app.handler.get('output', 'test_output_handler')()
|
||||
self.eq(han._meta.label, 'test_output_handler')
|
||||
|
||||
def test_load_order_presedence_one(self):
|
||||
# App config defines it as enabled, even though the plugin config has
|
||||
# it disabled... app trumps
|
||||
defaults = init_defaults(APP, 'myplugin')
|
||||
defaults['myplugin']['enable_plugin'] = True
|
||||
tmpdir = mkdtemp()
|
||||
|
||||
f = open(os.path.join(tmpdir, 'myplugin.conf'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'myplugin.conf'), 'w')
|
||||
f.write(CONF2)
|
||||
f.close()
|
||||
|
||||
f = open(os.path.join(tmpdir, 'myplugin.py'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'myplugin.py'), 'w')
|
||||
f.write(PLUGIN)
|
||||
f.close()
|
||||
|
||||
app = self.make_app(APP,
|
||||
config_defaults=defaults,
|
||||
config_files=[],
|
||||
plugin_config_dir=tmpdir,
|
||||
plugin_dir=tmpdir,
|
||||
plugin_config_dir=self.tmp_dir,
|
||||
plugin_dir=self.tmp_dir,
|
||||
plugin_bootstrap=None,
|
||||
)
|
||||
app.setup()
|
||||
@ -118,167 +112,142 @@ class PluginTestCase(test.CementCoreTestCase):
|
||||
self.ok(res)
|
||||
|
||||
finally:
|
||||
shutil.rmtree(tmpdir)
|
||||
shutil.rmtree(self.tmp_dir)
|
||||
|
||||
def test_load_order_presedence_two(self):
|
||||
# App config defines it as false, even though the plugin config has
|
||||
# it enabled... app trumps
|
||||
defaults = init_defaults(APP, 'myplugin')
|
||||
defaults['myplugin']['enable_plugin'] = False
|
||||
tmpdir = mkdtemp()
|
||||
|
||||
f = open(os.path.join(tmpdir, 'myplugin.conf'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'myplugin.conf'), 'w')
|
||||
f.write(CONF)
|
||||
f.close()
|
||||
|
||||
f = open(os.path.join(tmpdir, 'myplugin.py'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'myplugin.py'), 'w')
|
||||
f.write(PLUGIN)
|
||||
f.close()
|
||||
|
||||
app = self.make_app(APP,
|
||||
config_defaults=defaults,
|
||||
config_files=[],
|
||||
plugin_config_dir=tmpdir,
|
||||
plugin_dir=tmpdir,
|
||||
plugin_config_dir=self.tmp_dir,
|
||||
plugin_dir=self.tmp_dir,
|
||||
plugin_bootstrap=None,
|
||||
)
|
||||
app.setup()
|
||||
res = 'myplugin' in app.plugin._disabled_plugins
|
||||
self.ok(res)
|
||||
|
||||
try:
|
||||
res = 'myplugin' in app.plugin._disabled_plugins
|
||||
self.ok(res)
|
||||
|
||||
res = 'myplugin' not in app.plugin._enabled_plugins
|
||||
self.ok(res)
|
||||
|
||||
finally:
|
||||
shutil.rmtree(tmpdir)
|
||||
res = 'myplugin' not in app.plugin._enabled_plugins
|
||||
self.ok(res)
|
||||
|
||||
def test_load_order_presedence_three(self):
|
||||
# Multiple plugin configs, first plugin conf defines it as disabled,
|
||||
# but last read should make it enabled.
|
||||
defaults = init_defaults(APP, 'myplugin')
|
||||
tmpdir = mkdtemp()
|
||||
|
||||
f = open(os.path.join(tmpdir, 'a.conf'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'a.conf'), 'w')
|
||||
f.write(CONF2) # disabled config
|
||||
f.close()
|
||||
|
||||
f = open(os.path.join(tmpdir, 'b.conf'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'b.conf'), 'w')
|
||||
f.write(CONF) # enabled config
|
||||
f.close()
|
||||
|
||||
f = open(os.path.join(tmpdir, 'myplugin.py'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'myplugin.py'), 'w')
|
||||
f.write(PLUGIN)
|
||||
f.close()
|
||||
|
||||
app = self.make_app(APP,
|
||||
config_defaults=defaults,
|
||||
config_files=[],
|
||||
plugin_config_dir=tmpdir,
|
||||
plugin_dir=tmpdir,
|
||||
plugin_config_dir=self.tmp_dir,
|
||||
plugin_dir=self.tmp_dir,
|
||||
plugin_bootstrap=None,
|
||||
)
|
||||
app.setup()
|
||||
res = 'myplugin' in app.plugin._enabled_plugins
|
||||
self.ok(res)
|
||||
|
||||
try:
|
||||
res = 'myplugin' in app.plugin._enabled_plugins
|
||||
self.ok(res)
|
||||
|
||||
res = 'myplugin' not in app.plugin._disabled_plugins
|
||||
self.ok(res)
|
||||
|
||||
finally:
|
||||
shutil.rmtree(tmpdir)
|
||||
res = 'myplugin' not in app.plugin._disabled_plugins
|
||||
self.ok(res)
|
||||
|
||||
def test_load_order_presedence_four(self):
|
||||
# Multiple plugin configs, first plugin conf defines it as enabled,
|
||||
# but last read should make it disabled.
|
||||
defaults = init_defaults(APP, 'myplugin')
|
||||
tmpdir = mkdtemp()
|
||||
|
||||
f = open(os.path.join(tmpdir, 'a.conf'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'a.conf'), 'w')
|
||||
f.write(CONF) # enabled config
|
||||
f.close()
|
||||
|
||||
f = open(os.path.join(tmpdir, 'b.conf'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'b.conf'), 'w')
|
||||
f.write(CONF2) # disabled config
|
||||
f.close()
|
||||
|
||||
f = open(os.path.join(tmpdir, 'myplugin.py'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'myplugin.py'), 'w')
|
||||
f.write(PLUGIN)
|
||||
f.close()
|
||||
|
||||
app = self.make_app(APP,
|
||||
config_defaults=defaults,
|
||||
config_files=[],
|
||||
plugin_config_dir=tmpdir,
|
||||
plugin_dir=tmpdir,
|
||||
plugin_config_dir=self.tmp_dir,
|
||||
plugin_dir=self.tmp_dir,
|
||||
plugin_bootstrap=None,
|
||||
)
|
||||
app.setup()
|
||||
res = 'myplugin' in app.plugin._disabled_plugins
|
||||
self.ok(res)
|
||||
|
||||
try:
|
||||
res = 'myplugin' in app.plugin._disabled_plugins
|
||||
self.ok(res)
|
||||
|
||||
res = 'myplugin' not in app.plugin._enabled_plugins
|
||||
self.ok(res)
|
||||
|
||||
finally:
|
||||
shutil.rmtree(tmpdir)
|
||||
res = 'myplugin' not in app.plugin._enabled_plugins
|
||||
self.ok(res)
|
||||
|
||||
def test_load_order_presedence_five(self):
|
||||
# Multiple plugin configs, enable -> disabled -> enable
|
||||
defaults = init_defaults(APP, 'myplugin')
|
||||
tmpdir = mkdtemp()
|
||||
|
||||
f = open(os.path.join(tmpdir, 'a.conf'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'a.conf'), 'w')
|
||||
f.write(CONF) # enabled config
|
||||
f.close()
|
||||
|
||||
f = open(os.path.join(tmpdir, 'b.conf'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'b.conf'), 'w')
|
||||
f.write(CONF2) # disabled config
|
||||
f.close()
|
||||
|
||||
f = open(os.path.join(tmpdir, 'c.conf'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'c.conf'), 'w')
|
||||
f.write(CONF) # enabled config
|
||||
f.close()
|
||||
|
||||
f = open(os.path.join(tmpdir, 'e.conf'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'e.conf'), 'w')
|
||||
f.write(CONF2) # disabled config
|
||||
f.close()
|
||||
|
||||
f = open(os.path.join(tmpdir, 'f.conf'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'f.conf'), 'w')
|
||||
f.write(CONF) # enabled config
|
||||
f.close()
|
||||
|
||||
f = open(os.path.join(tmpdir, 'myplugin.py'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'myplugin.py'), 'w')
|
||||
f.write(PLUGIN)
|
||||
f.close()
|
||||
|
||||
app = self.make_app(APP,
|
||||
config_defaults=defaults,
|
||||
config_files=[],
|
||||
plugin_config_dir=tmpdir,
|
||||
plugin_dir=tmpdir,
|
||||
plugin_config_dir=self.tmp_dir,
|
||||
plugin_dir=self.tmp_dir,
|
||||
plugin_bootstrap=None,
|
||||
)
|
||||
app.setup()
|
||||
res = 'myplugin' in app.plugin._enabled_plugins
|
||||
self.ok(res)
|
||||
|
||||
try:
|
||||
res = 'myplugin' in app.plugin._enabled_plugins
|
||||
self.ok(res)
|
||||
|
||||
res = 'myplugin' not in app.plugin._disabled_plugins
|
||||
self.ok(res)
|
||||
|
||||
finally:
|
||||
shutil.rmtree(tmpdir)
|
||||
res = 'myplugin' not in app.plugin._disabled_plugins
|
||||
self.ok(res)
|
||||
|
||||
def test_load_plugins_from_config(self):
|
||||
tmpdir = mkdtemp()
|
||||
f = open(os.path.join(tmpdir, 'myplugin.py'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'myplugin.py'), 'w')
|
||||
f.write(PLUGIN)
|
||||
f.close()
|
||||
|
||||
@ -289,17 +258,13 @@ class PluginTestCase(test.CementCoreTestCase):
|
||||
defaults['myplugin2']['enable_plugin'] = False
|
||||
app = self.make_app(APP, config_defaults=defaults,
|
||||
config_files=[],
|
||||
plugin_config_dir=tmpdir,
|
||||
plugin_dir=tmpdir,
|
||||
plugin_config_dir=self.tmp_dir,
|
||||
plugin_dir=self.tmp_dir,
|
||||
plugin_bootstrap=None,
|
||||
)
|
||||
app.setup()
|
||||
|
||||
try:
|
||||
han = app.handler.get('output', 'test_output_handler')()
|
||||
self.eq(han._meta.label, 'test_output_handler')
|
||||
finally:
|
||||
shutil.rmtree(tmpdir)
|
||||
han = app.handler.get('output', 'test_output_handler')()
|
||||
self.eq(han._meta.label, 'test_output_handler')
|
||||
|
||||
# some more checks
|
||||
res = 'myplugin' in app.plugin.get_enabled_plugins()
|
||||
@ -318,23 +283,22 @@ class PluginTestCase(test.CementCoreTestCase):
|
||||
self.ok(res)
|
||||
|
||||
def test_disabled_plugins_from_files(self):
|
||||
tmpdir = mkdtemp()
|
||||
f = open(os.path.join(tmpdir, 'myplugin.conf'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'myplugin.conf'), 'w')
|
||||
f.write(CONF2)
|
||||
f.close()
|
||||
|
||||
f = open(os.path.join(tmpdir, 'myplugin.py'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'myplugin.py'), 'w')
|
||||
f.write(PLUGIN)
|
||||
f.close()
|
||||
|
||||
app = self.make_app(APP,
|
||||
config_files=[],
|
||||
plugin_config_dir=tmpdir,
|
||||
plugin_dir=tmpdir,
|
||||
plugin_config_dir=self.tmp_dir,
|
||||
plugin_dir=self.tmp_dir,
|
||||
plugin_bootstrap=None,
|
||||
)
|
||||
app.setup()
|
||||
shutil.rmtree(tmpdir)
|
||||
shutil.rmtree(self.tmp_dir)
|
||||
|
||||
res = 'test_output_handler' not in app.handler.__handlers__['output']
|
||||
self.ok(res)
|
||||
@ -343,38 +307,36 @@ class PluginTestCase(test.CementCoreTestCase):
|
||||
self.ok(res)
|
||||
|
||||
def test_bogus_plugin_from_files(self):
|
||||
tmpdir = mkdtemp()
|
||||
f = open(os.path.join(tmpdir, 'myplugin.conf'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'myplugin.conf'), 'w')
|
||||
f.write(CONF3)
|
||||
f.close()
|
||||
|
||||
# do this for coverage... empty config file
|
||||
f = open(os.path.join(tmpdir, 'bogus.conf'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'bogus.conf'), 'w')
|
||||
f.write(CONF5)
|
||||
f.close()
|
||||
|
||||
app = self.make_app(APP,
|
||||
config_files=[],
|
||||
plugin_config_dir=tmpdir,
|
||||
plugin_dir=tmpdir,
|
||||
plugin_config_dir=self.tmp_dir,
|
||||
plugin_dir=self.tmp_dir,
|
||||
plugin_bootstrap=None,
|
||||
)
|
||||
app.setup()
|
||||
shutil.rmtree(tmpdir)
|
||||
shutil.rmtree(self.tmp_dir)
|
||||
|
||||
res = 'bogus_plugin' not in app.plugin.get_enabled_plugins()
|
||||
self.ok(res)
|
||||
|
||||
@test.raises(exc.FrameworkError)
|
||||
def test_bad_plugin_dir(self):
|
||||
tmpdir = mkdtemp()
|
||||
f = open(os.path.join(tmpdir, 'myplugin.conf'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'myplugin.conf'), 'w')
|
||||
f.write(CONF)
|
||||
f.close()
|
||||
|
||||
app = self.make_app(APP,
|
||||
config_files=[],
|
||||
plugin_config_dir=tmpdir,
|
||||
plugin_config_dir=self.tmp_dir,
|
||||
plugin_dir='./some/bogus/path',
|
||||
plugin_bootstrap=None,
|
||||
)
|
||||
@ -384,27 +346,22 @@ class PluginTestCase(test.CementCoreTestCase):
|
||||
raise
|
||||
except exc.FrameworkError as e:
|
||||
raise
|
||||
finally:
|
||||
shutil.rmtree(tmpdir)
|
||||
|
||||
def test_load_plugin_from_module(self):
|
||||
# We mock this out by loading a cement ext, but it is essentially the
|
||||
# same type of code.
|
||||
tmpdir = mkdtemp()
|
||||
del sys.modules['cement.ext.ext_json']
|
||||
f = open(os.path.join(tmpdir, 'ext_json.conf'), 'w')
|
||||
f = open(os.path.join(self.tmp_dir, 'ext_json.conf'), 'w')
|
||||
f.write(CONF4)
|
||||
f.close()
|
||||
|
||||
app = self.make_app(APP,
|
||||
config_files=[],
|
||||
plugin_config_dir=tmpdir,
|
||||
plugin_dir=tmpdir,
|
||||
plugin_config_dir=self.tmp_dir,
|
||||
plugin_dir=self.tmp_dir,
|
||||
plugin_bootstrap='cement.ext',
|
||||
)
|
||||
app.setup()
|
||||
|
||||
res = 'ext_json' in app.plugin.get_enabled_plugins()
|
||||
self.ok(res)
|
||||
|
||||
shutil.rmtree(tmpdir)
|
||||
|
||||
@ -10,6 +10,7 @@ from cement.utils import test
|
||||
class AlarmExtTestCase(test.CementExtTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(AlarmExtTestCase, self).setUp()
|
||||
self.app = self.make_app('tests',
|
||||
extensions=['alarm'],
|
||||
argv=[]
|
||||
|
||||
@ -285,37 +285,43 @@ class ArgparseExtTestCase(test.CementExtTestCase):
|
||||
app._meta.argv = ['cmd2']
|
||||
res = app.run()
|
||||
self.eq(res, "Inside Second.cmd2")
|
||||
|
||||
self.tearDown()
|
||||
|
||||
self.setUp()
|
||||
with self.app as app:
|
||||
app._meta.argv = ['third', 'cmd3']
|
||||
res = app.run()
|
||||
self.eq(res, "Inside Third.cmd3")
|
||||
self.tearDown()
|
||||
|
||||
self.setUp()
|
||||
with self.app as app:
|
||||
app._meta.argv = ['third', 'cmd4']
|
||||
res = app.run()
|
||||
self.eq(res, "Inside Fourth.cmd4")
|
||||
self.tearDown()
|
||||
|
||||
self.setUp()
|
||||
with self.app as app:
|
||||
app._meta.argv = ['third', 'fifth', 'cmd5']
|
||||
res = app.run()
|
||||
self.eq(res, "Inside Fifth.cmd5")
|
||||
self.tearDown()
|
||||
|
||||
self.setUp()
|
||||
with self.app as app:
|
||||
app._meta.argv = ['third', 'fifth', 'sixth', 'cmd6']
|
||||
res = app.run()
|
||||
self.eq(res, "Inside Sixth.cmd6")
|
||||
self.tearDown()
|
||||
|
||||
self.setUp()
|
||||
with self.app as app:
|
||||
app._meta.argv = ['third', 'cmd7']
|
||||
res = app.run()
|
||||
self.eq(res, "Inside Seventh.cmd7")
|
||||
|
||||
self.tearDown()
|
||||
|
||||
def test_base_cmd1_parsing(self):
|
||||
with self.app as app:
|
||||
app._meta.argv = ['--foo=bar', 'cmd1']
|
||||
|
||||
@ -2,7 +2,6 @@
|
||||
|
||||
import os
|
||||
import sys
|
||||
from tempfile import mkstemp
|
||||
from cement.core import handler, backend, log
|
||||
from cement.utils import test
|
||||
from cement.utils.misc import rando
|
||||
@ -25,21 +24,17 @@ my_param = my_value
|
||||
class ConfigObjExtTestCase(test.CementTestCase):
|
||||
|
||||
def setUp(self):
|
||||
_, self.tmppath = mkstemp()
|
||||
f = open(self.tmppath, 'w+')
|
||||
super(ConfigObjExtTestCase, self).setUp()
|
||||
f = open(self.tmp_file, 'w+')
|
||||
f.write(CONFIG)
|
||||
f.close()
|
||||
self.app = self.make_app(APP,
|
||||
extensions=['configobj'],
|
||||
config_handler='configobj',
|
||||
config_files=[self.tmppath],
|
||||
config_files=[self.tmp_file],
|
||||
argv=[]
|
||||
)
|
||||
|
||||
def tearDown(self):
|
||||
if os.path.exists(self.tmppath):
|
||||
os.remove(self.tmppath)
|
||||
|
||||
def test_configobj(self):
|
||||
self.app.setup()
|
||||
|
||||
|
||||
@ -5,7 +5,7 @@
|
||||
# sub-process is forked.
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
#import tempfile
|
||||
from random import random
|
||||
from cement.core import handler, backend, log, hook, exc
|
||||
from cement.utils import shell
|
||||
@ -19,6 +19,7 @@ APP = rando()[:12]
|
||||
class DaemonExtTestCase(test.CementExtTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(DaemonExtTestCase, self).setUp()
|
||||
self.app = self.make_app()
|
||||
|
||||
def test_switch(self):
|
||||
@ -26,32 +27,24 @@ class DaemonExtTestCase(test.CementExtTestCase):
|
||||
env.switch()
|
||||
|
||||
def test_switch_with_pid(self):
|
||||
(_, tmpfile) = tempfile.mkstemp()
|
||||
os.remove(tmpfile)
|
||||
env = ext_daemon.Environment(pid_file=tmpfile)
|
||||
os.remove(self.tmp_file)
|
||||
env = ext_daemon.Environment(pid_file=self.tmp_file)
|
||||
env.switch()
|
||||
|
||||
try:
|
||||
self.ok(os.path.exists(tmpfile))
|
||||
finally:
|
||||
os.remove(tmpfile)
|
||||
self.ok(os.path.exists(self.tmp_file))
|
||||
|
||||
@test.raises(exc.FrameworkError)
|
||||
def test_pid_exists(self):
|
||||
(_, tmpfile) = tempfile.mkstemp()
|
||||
|
||||
env = ext_daemon.Environment(pid_file=tmpfile)
|
||||
env = ext_daemon.Environment(pid_file=self.tmp_file)
|
||||
env.switch()
|
||||
|
||||
try:
|
||||
self.ok(os.path.exists(tmpfile))
|
||||
self.ok(os.path.exists(self.tmp_file))
|
||||
except exc.FrameworkError as e:
|
||||
self.ok(e.msg.startswith('Process already running'))
|
||||
raise
|
||||
finally:
|
||||
env = ext_daemon.Environment()
|
||||
env.switch()
|
||||
os.remove(tmpfile)
|
||||
|
||||
@test.raises(exc.FrameworkError)
|
||||
def test_bogus_user(self):
|
||||
@ -80,8 +73,7 @@ class DaemonExtTestCase(test.CementExtTestCase):
|
||||
env.switch()
|
||||
|
||||
def test_daemon(self):
|
||||
(_, tmpfile) = tempfile.mkstemp()
|
||||
os.remove(tmpfile)
|
||||
os.remove(self.tmp_file)
|
||||
from cement.utils import shell
|
||||
|
||||
# Test in a sub-process to avoid Nose hangup
|
||||
@ -90,7 +82,7 @@ class DaemonExtTestCase(test.CementExtTestCase):
|
||||
extensions=['daemon'])
|
||||
|
||||
app.setup()
|
||||
app.config.set('daemon', 'pid_file', tmpfile)
|
||||
app.config.set('daemon', 'pid_file', self.tmp_file)
|
||||
|
||||
try:
|
||||
# FIX ME: Can't daemonize, because nose loses sight of it
|
||||
|
||||
@ -15,6 +15,7 @@ else:
|
||||
class GenshiExtTestCase(test.CementExtTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(GenshiExtTestCase, self).setUp()
|
||||
self.app = self.make_app('tests',
|
||||
extensions=['genshi'],
|
||||
output_handler='genshi',
|
||||
|
||||
@ -11,6 +11,7 @@ from cement.utils import test
|
||||
class Jinja2ExtTestCase(test.CementExtTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(Jinja2ExtTestCase, self).setUp()
|
||||
self.app = self.make_app('tests',
|
||||
extensions=['jinja2'],
|
||||
output_handler='jinja2',
|
||||
|
||||
@ -2,7 +2,6 @@
|
||||
|
||||
import json
|
||||
import sys
|
||||
from tempfile import mkstemp
|
||||
from cement.core import handler, backend, hook
|
||||
from cement.utils import test
|
||||
|
||||
@ -37,14 +36,14 @@ class JsonConfigObjExtTestCase(test.CementExtTestCase):
|
||||
)
|
||||
|
||||
def setUp(self):
|
||||
_, self.tmppath = mkstemp()
|
||||
f = open(self.tmppath, 'w+')
|
||||
super(JsonConfigObjExtTestCase, self).setUp()
|
||||
f = open(self.tmp_file, 'w+')
|
||||
f.write(self.CONFIG)
|
||||
f.close()
|
||||
self.app = self.make_app('tests',
|
||||
extensions=['json_configobj'],
|
||||
config_handler='json_configobj',
|
||||
config_files=[self.tmppath],
|
||||
config_files=[self.tmp_file],
|
||||
)
|
||||
|
||||
def test_has_section(self):
|
||||
|
||||
@ -2,7 +2,6 @@
|
||||
|
||||
import json
|
||||
import sys
|
||||
from tempfile import mkstemp
|
||||
from cement.core import handler, backend, hook
|
||||
from cement.utils import test
|
||||
from cement.utils.misc import rando
|
||||
@ -34,15 +33,15 @@ class JsonExtTestCase(test.CementExtTestCase):
|
||||
)
|
||||
|
||||
def setUp(self):
|
||||
_, self.tmppath = mkstemp()
|
||||
f = open(self.tmppath, 'w+')
|
||||
super(JsonExtTestCase, self).setUp()
|
||||
f = open(self.tmp_file, 'w+')
|
||||
f.write(self.CONFIG)
|
||||
f.close()
|
||||
self.app = self.make_app('tests',
|
||||
extensions=['json'],
|
||||
output_handler='json',
|
||||
config_handler='json',
|
||||
config_files=[self.tmppath],
|
||||
config_files=[self.tmp_file],
|
||||
argv=['-o', 'json']
|
||||
)
|
||||
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
|
||||
import os
|
||||
import logging
|
||||
from tempfile import mkstemp
|
||||
import shutil
|
||||
from cement.core import handler, backend, log
|
||||
from cement.ext import ext_logging
|
||||
from cement.utils import test
|
||||
@ -105,13 +105,12 @@ class LoggingExtTestCase(test.CementExtTestCase):
|
||||
self.eq(os.path.exists("%s.3" % log_file), False)
|
||||
|
||||
def test_missing_log_dir(self):
|
||||
_, tmp_path = mkstemp()
|
||||
if os.path.exists(tmp_path):
|
||||
os.remove(tmp_path)
|
||||
if os.path.exists(self.tmp_dir):
|
||||
shutil.rmtree(self.tmp_dir)
|
||||
|
||||
defaults = init_defaults()
|
||||
defaults['log.logging'] = dict(
|
||||
file=os.path.join(tmp_path, '%s.log' % APP),
|
||||
file=os.path.join(self.tmp_dir, '%s.log' % APP),
|
||||
)
|
||||
app = self.make_app(config_defaults=defaults)
|
||||
app.setup()
|
||||
|
||||
@ -12,6 +12,7 @@ from cement.utils.misc import init_defaults
|
||||
class MemcachedExtTestCase(test.CementTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(MemcachedExtTestCase, self).setUp()
|
||||
self.key = "cement-tests-random-key-%s" % random()
|
||||
defaults = init_defaults('tests', 'cache.memcached')
|
||||
defaults['cache.memcached']['hosts'] = '127.0.0.1, localhost'
|
||||
@ -23,6 +24,7 @@ class MemcachedExtTestCase(test.CementTestCase):
|
||||
self.app.setup()
|
||||
|
||||
def tearDown(self):
|
||||
super(MemcachedExtTestCase, self).tearDown()
|
||||
self.app.cache.delete(self.key)
|
||||
|
||||
def test_memcache_list_type_config(self):
|
||||
|
||||
@ -10,6 +10,7 @@ from cement.utils import test
|
||||
class MustacheExtTestCase(test.CementExtTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(MustacheExtTestCase, self).setUp()
|
||||
self.app = self.make_app('tests',
|
||||
extensions=['mustache'],
|
||||
output_handler='mustache',
|
||||
|
||||
@ -12,6 +12,7 @@ from cement.utils.misc import init_defaults
|
||||
class RedisExtTestCase(test.CementTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(RedisExtTestCase, self).setUp()
|
||||
self.key = "cement-tests-random-key-%s" % random()
|
||||
defaults = init_defaults('tests', 'cache.redis')
|
||||
defaults['cache.redis']['host'] = '127.0.0.1'
|
||||
@ -25,6 +26,7 @@ class RedisExtTestCase(test.CementTestCase):
|
||||
self.app.setup()
|
||||
|
||||
def tearDown(self):
|
||||
super(RedisExtTestCase, self).tearDown()
|
||||
self.app.cache.delete(self.key)
|
||||
|
||||
def test_redis_set(self):
|
||||
|
||||
@ -8,6 +8,7 @@ from cement.utils import test
|
||||
class TabulateExtTestCase(test.CementExtTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(TabulateExtTestCase, self).setUp()
|
||||
self.app = self.make_app('tests',
|
||||
extensions=['tabulate'],
|
||||
output_handler='tabulate',
|
||||
|
||||
@ -3,7 +3,6 @@
|
||||
import os
|
||||
import sys
|
||||
import yaml
|
||||
from tempfile import mkstemp
|
||||
from cement.core import handler, hook
|
||||
from cement.utils import test
|
||||
|
||||
@ -38,20 +37,16 @@ class YamlConfigObjExtTestCase(test.CementTestCase):
|
||||
)
|
||||
|
||||
def setUp(self):
|
||||
_, self.tmppath = mkstemp()
|
||||
f = open(self.tmppath, 'w+')
|
||||
super(YamlConfigObjExtTestCase, self).setUp()
|
||||
f = open(self.tmp_file, 'w+')
|
||||
f.write(self.CONFIG)
|
||||
f.close()
|
||||
self.app = self.make_app('tests',
|
||||
extensions=['yaml_configobj'],
|
||||
config_handler='yaml_configobj',
|
||||
config_files=[self.tmppath],
|
||||
config_files=[self.tmp_file],
|
||||
)
|
||||
|
||||
def tearDown(self):
|
||||
if os.path.exists(self.tmppath):
|
||||
os.remove(self.tmppath)
|
||||
|
||||
def test_has_section(self):
|
||||
self.app.setup()
|
||||
self.ok(self.app.config.has_section('section'))
|
||||
|
||||
@ -3,7 +3,6 @@
|
||||
import os
|
||||
import sys
|
||||
import yaml
|
||||
from tempfile import mkstemp
|
||||
from cement.core import handler, hook
|
||||
from cement.utils import test
|
||||
from cement.utils.misc import rando
|
||||
@ -35,22 +34,18 @@ class YamlExtTestCase(test.CementTestCase):
|
||||
)
|
||||
|
||||
def setUp(self):
|
||||
_, self.tmppath = mkstemp()
|
||||
f = open(self.tmppath, 'w+')
|
||||
super(YamlExtTestCase, self).setUp()
|
||||
f = open(self.tmp_file, 'w+')
|
||||
f.write(self.CONFIG)
|
||||
f.close()
|
||||
self.app = self.make_app('tests',
|
||||
extensions=['yaml'],
|
||||
config_handler='yaml',
|
||||
output_handler='yaml',
|
||||
config_files=[self.tmppath],
|
||||
config_files=[self.tmp_file],
|
||||
argv=['-o', 'yaml']
|
||||
)
|
||||
|
||||
def tearDown(self):
|
||||
if os.path.exists(self.tmppath):
|
||||
os.remove(self.tmppath)
|
||||
|
||||
def test_yaml(self):
|
||||
self.app.setup()
|
||||
self.app.run()
|
||||
|
||||
@ -1,7 +1,7 @@
|
||||
"""Tests for cement.utils.fs"""
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
import shutil
|
||||
from cement.utils import fs, test
|
||||
|
||||
|
||||
@ -12,19 +12,24 @@ class FsUtilsTestCase(test.CementCoreTestCase):
|
||||
self.ok(path.startswith('/'))
|
||||
|
||||
def test_backup(self):
|
||||
_, tmpfile = tempfile.mkstemp()
|
||||
bkfile = fs.backup(tmpfile)
|
||||
self.eq("%s.bak" % os.path.basename(tmpfile), os.path.basename(bkfile))
|
||||
bkfile = fs.backup(tmpfile)
|
||||
self.eq("%s.bak.0" %
|
||||
os.path.basename(tmpfile), os.path.basename(bkfile))
|
||||
bkfile = fs.backup(tmpfile)
|
||||
self.eq("%s.bak.1" %
|
||||
os.path.basename(tmpfile), os.path.basename(bkfile))
|
||||
tmp_file = os.path.join(self.tmp_dir, 'test.file')
|
||||
tmp_dir = os.path.join(self.tmp_dir, 'test.dir')
|
||||
open(tmp_file, 'w').close()
|
||||
os.makedirs(tmp_dir)
|
||||
|
||||
tmpdir = tempfile.mkdtemp()
|
||||
bkdir = fs.backup(tmpdir)
|
||||
self.eq("%s.bak" % os.path.basename(tmpdir), os.path.basename(bkdir))
|
||||
bkfile = fs.backup(tmp_file)
|
||||
self.eq("%s.bak" % os.path.basename(tmp_file),
|
||||
os.path.basename(bkfile))
|
||||
bkfile = fs.backup(tmp_file)
|
||||
self.eq("%s.bak.0" %
|
||||
os.path.basename(tmp_file), os.path.basename(bkfile))
|
||||
bkfile = fs.backup(tmp_file)
|
||||
self.eq("%s.bak.1" %
|
||||
os.path.basename(tmp_file), os.path.basename(bkfile))
|
||||
|
||||
bkdir = fs.backup(tmp_dir)
|
||||
self.eq("%s.bak" % os.path.basename(tmp_dir),
|
||||
os.path.basename(bkdir))
|
||||
|
||||
res = fs.backup('someboguspath')
|
||||
self.eq(res, None)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user