global _debug
global kloger
descriptions = {}
- acceptable_groups = arguments['config'].get_groups()
- for cmd_group, spec in arguments['config'].get_cli_specs():
+ acceptable_groups = arguments['config'].groups
+ for cmd_group, spec in arguments['config'].cli_specs:
pkg = _load_spec_module(spec, arguments, '_commands')
if pkg:
cmds = getattr(pkg, '_commands')
def _load_all_commands(cmd_tree, arguments):
_cnf = arguments['config']
- for cmd_group, spec in _cnf.get_cli_specs():
+ for cmd_group, spec in _cnf.cli_specs:
try:
spec_module = _load_spec_module(spec, arguments, '_commands')
spec_commands = getattr(spec_module, '_commands')
def get_command_group(unparsed, arguments):
- groups = arguments['config'].get_groups()
+ groups = arguments['config'].groups
for term in unparsed:
if term.startswith('-'):
continue
class ConfigArgument(Argument):
"""Manage a kamaki configuration (file)"""
- _config_file = None
+ def __init__(self, help, parsed_name=('-c', '--config')):
+ super(ConfigArgument, self).__init__(1, help, parsed_name, None)
+ self.file_path = None
@property
def value(self):
- """A Config object"""
- super(self.__class__, self).value
- return super(self.__class__, self).value
+ return super(ConfigArgument, self).value
@value.setter
def value(self, config_file):
if config_file:
self._value = Config(config_file)
- self._config_file = config_file
- elif self._config_file:
- self._value = Config(self._config_file)
+ self.file_path = config_file
+ elif self.file_path:
+ self._value = Config(self.file_path)
else:
self._value = Config()
"""Get a configuration setting from the Config object"""
return self.value.get(group, term)
- def get_groups(self):
+ @property
+ def groups(self):
suffix = '_cli'
slen = len(suffix)
return [term[:-slen] for term in self.value.keys('global') if (
term.endswith(suffix))]
- def get_cli_specs(self):
+ @property
+ def cli_specs(self):
suffix = '_cli'
slen = len(suffix)
return [(k[:-slen], v) for k, v in self.value.items('global') if (
return self.value.get_cloud(cloud, option)
-_config_arg = ConfigArgument(
- 1, 'Path to configuration file', ('-c', '--config'))
+_config_arg = ConfigArgument('Path to config file')
class CmdLineConfigArgument(Argument):
#from mock import patch, call
from unittest import TestCase
from StringIO import StringIO
+from sys import stdin, stdout
#from itertools import product
from kamaki.cli import argument
+from kamaki.cli.config import Config
class Argument(TestCase):
del arp
+class ConfigArgument(TestCase):
+
+ # A cloud name in config with a URL but no TOKEN
+ SEMI_CLOUD = 'production'
+
+ # A cloud name that is not configured in config
+ INVALID_CLOUD = 'QWERTY_123456'
+
+ def setUp(self):
+ argument._config_arg = argument.ConfigArgument('Recovered Path')
+
+ def test_value(self):
+ c = argument._config_arg
+ self.assertEqual(c.value, None)
+ exp = '/some/random/path'
+ c.value = exp
+ self.assertTrue(isinstance(c.value, Config))
+ self.assertEqual(c.file_path, exp)
+ self.assertEqual(c.value.path, exp)
+
+ def test_get(self):
+ c = argument._config_arg
+ c.value = None
+ self.assertEqual(c.value.get('global', 'config_cli'), 'config')
+
+ def test_groups(self):
+ c = argument._config_arg
+ c.value = None
+ self.assertTrue(set(c.groups).issuperset([
+ 'image', 'config', 'history']))
+
+ def test_cli_specs(self):
+ c = argument._config_arg
+ c.value = None
+ self.assertTrue(set(c.cli_specs).issuperset([
+ ('image', 'image'), ('config', 'config'), ('history', 'history')]))
+
+ def test_get_global(self):
+ c = argument._config_arg
+ c.value = None
+ for k, v in (
+ ('config_cli', 'config'),
+ ('image_cli', 'image'),
+ ('history_cli', 'history')):
+ self.assertEqual(c.get_global(k), v)
+
+ def test_get_cloud(self):
+ """test_get_cloud (!! hard-set SEMI/INVALID_CLOUD to run this !!)"""
+ c = argument._config_arg
+ c.value = None
+ if not self.SEMI_CLOUD:
+ stdout.write(
+ '\n\tA cloud name set in config file with URL but no TOKEN: ')
+ self.SEMI_CLOUD = stdin.readline()[:-1]
+ self.assertTrue(len(c.get_cloud(self.SEMI_CLOUD, 'url')) > 0)
+ self.assertRaises(KeyError, c.get_cloud, self.SEMI_CLOUD, 'token')
+
+ if not self.INVALID_CLOUD:
+ stdout.write('\tok\n\tA cloud name NOT in your config file: ')
+ self.INVALID_CLOUD = stdin.readline()[:-1]
+ self.assertRaises(KeyError, c.get_cloud, self.INVALID_CLOUD, 'url')
+
+
if __name__ == '__main__':
from sys import argv
from kamaki.cli.test import runTestCase
runTestCase(Argument, 'Argument', argv[1:])
+ runTestCase(ConfigArgument, 'ConfigArgument', argv[1:])
else:
intro = self.cmd_tree.name
- acceptable = parser.arguments['config'].get_groups()
+ acceptable = parser.arguments['config'].groups
total = self.cmd_tree.groups.keys()
self.cmd_tree.exclude(set(total).difference(acceptable))
except NoSectionError:
pass
- def remote_from_cloud(self, cloud, option):
+ def remove_from_cloud(self, cloud, option):
d = self.get(CLOUD_PREFIX, cloud)
if isinstance(d, dict):
d.pop(option)
from inspect import getmembers, isclass
from kamaki.cli.command_tree.test import Command, CommandTree
-from kamaki.cli.argument.test import Argument
+from kamaki.cli.argument.test import Argument, ConfigArgument
# TestCase auxiliary methods