def _num_of_matching_terms(basic_list, attack_list):
if not attack_list:
- return 1
+ return len(basic_list)
matching_terms = 0
for i, term in enumerate(basic_list):
_exec_cmd(executable, unparsed, parser.print_help)
-from command_shell import _fix_arguments, Shell
-
-
-def _start_shell():
- shell = Shell()
- shell.set_prompt(basename(argv[0]))
- from kamaki import __version__ as version
- shell.greet(version)
- shell.do_EOF = shell.do_exit
- return shell
-
-
-def run_shell(arguments):
- _fix_arguments()
- shell = _start_shell()
- _config = _arguments['config']
- from kamaki.cli.command_tree import CommandTree
- shell.cmd_tree = CommandTree(
- 'kamaki', 'A command line tool for poking clouds')
+def run_shell(exe_string, arguments):
+ from command_shell import _init_shell
+ shell = _init_shell(exe_string, arguments)
+ # Load all commands in shell CommandTree
+ _config = arguments['config']
for spec in [spec for spec in _config.get_groups()\
- if arguments['config'].get(spec, 'cli')]:
+ if _config.get(spec, 'cli')]:
try:
spec_module = _load_spec_module(spec, arguments, '_commands')
spec_commands = getattr(spec_module, '_commands')
parser.print_help()
_groups_help(_arguments)
else:
- run_shell(_arguments)
+ run_shell(exe, _arguments)
except CLIError as err:
if _debug:
raise err
from kamaki.cli.errors import CLIError
-def _fix_arguments():
- _arguments.pop('version', None)
- _arguments.pop('options', None)
- _arguments.pop('history', None)
+def _init_shell(exe_string, arguments):
+ arguments.pop('version', None)
+ arguments.pop('options', None)
+ arguments.pop('history', None)
+ shell = Shell()
+ shell.set_prompt(exe_string)
+ from kamaki import __version__ as version
+ shell.greet(version)
+ shell.do_EOF = shell.do_exit
+ from kamaki.cli.command_tree import CommandTree
+ shell.cmd_tree = CommandTree(
+ 'kamaki', 'A command line tool for poking clouds')
+ return shell
class Shell(Cmd):
def _print(self, server):
addr_dict = {}
if 'attachments' in server:
- for addr in server['attachments']['values']:
+ atts = server.pop('attachments')
+ for addr in atts['values']:
ips = addr.pop('values', [])
for ip in ips:
addr['IPv%s' % ip['version']] = ip['addr']
if 'firewallProfile' in addr:
addr['firewall'] = addr.pop('firewallProfile')
addr_dict[addr.pop('id')] = addr
- server['attachments'] = addr_dict if addr_dict is not {} else None
+ server['attachments'] = addr_dict if addr_dict else None
if 'metadata' in server:
server['metadata'] = server['metadata']['values']
print_dict(server, ident=2)
from kamaki.cli import get_cmd_terms, command
from kamaki.cli.commands import _command_init
from kamaki.cli.command_tree import CommandTree
-from kamaki.cli.argument import FlagArgument
+from kamaki.clients import tests
-sample_cmds = CommandTree(
- 'sample',
- 'Sample commands for developing your own')
-test_cmds = CommandTree(
- 'test',
- 'Test commands for testing clients')
-_commands = [sample_cmds, test_cmds]
+test_cmds = CommandTree('test', 'Unitest clients')
+_commands = [test_cmds]
-print('Command Terms: ', get_cmd_terms())
+#print('Command Terms: ', get_cmd_terms())
class _test_init(_command_init):
- def main(self, *args, **kwargs):
- print(self.__class__)
- for v in args:
- print('\t\targ: %s' % v)
- for k, v in kwargs.items():
- print('\t\tkwarg: %s: %s' % (k, v))
+ def main(self, client, method=None):
+ if method:
+ tests.main([client, method])
+ else:
+ tests.main([client])
-@command(sample_cmds)
-class sample_cmd0(_test_init):
- """ test cmd
- This is the zero command test and this is the long description of it
- """
-
- def main(self, mant):
- super(self.__class__, self).main(mant)
-
-
-@command(sample_cmds)
-class sample_cmd_all(_test_init):
- """test cmd all"""
+@command(test_cmds)
+class test_all(_test_init):
+ """test all clients"""
def main(self):
- super(self.__class__, self).main()
+ for client in ('pithos', 'cyclades', 'image', 'astakos'):
+ super(self.__class__, self).main(client)
-@command(sample_cmds)
-class sample_cmd_some(_test_init):
- """test_cmd_some"""
+@command(test_cmds)
+class test_pithos(_test_init):
+ """ test Pithos client"""
- def main(self, opt='lala'):
- super(self.__class__, self).main(opt=opt)
+ def main(self, method=None):
+ super(self.__class__, self).main('pithos', method)
@command(test_cmds)
-class test_cmd0(_test_init):
- """ test cmd"""
+class test_cyclades(_test_init):
+ """ test Cyclades client"""
- def main(self, mant):
- super(self.__class__, self).main(mant)
+ def main(self, method=None):
+ super(self.__class__, self).main('cyclades', method)
@command(test_cmds)
-class test_cmd_all(_test_init):
- """test cmd all"""
-
- def __init__(self, arguments={}):
- super(self.__class__, self).__init__(arguments)
- self.arguments['testarg'] = FlagArgument('a test arg', '--test')
+class test_image(_test_init):
+ """ test Image client"""
- def main(self):
- super(self.__class__, self).main()
+ def main(self, method=None):
+ super(self.__class__, self).main('image', method)
@command(test_cmds)
-class test_cmdion(_test_init):
- """test_cmd_some"""
+class test_astakos(_test_init):
+ """ test Astakos client"""
- def main(self, opt='lala'):
- super(self.__class__, self).main(opt=opt)
+ def main(self, method=None):
+ super(self.__class__, self).main('astakos', method)
@command(test_cmds)
-class test_cmd_cmdion_comedian(_test_init):
- """test_cmd_some"""
+class test_lala_lele(_test_init):
+ """test lala lele"""
- def main(self, opt='lala'):
- super(self.__class__, self).main(opt=opt)
+ def main(self, *args):
+ print('Do smth')
def test_get_server_console(self):
"""Test get_server_console"""
self.server2 = self._create_server(self.servname2,
- self.flavorid + 1,
+ self.flavorid + 2,
self.img)
+ self._wait_for_status(self.server2['id'], 'BUILD')
self._test_get_server_console()
def _test_get_server_console(self):
- self._wait_for_status(self.server2['id'], 'BUILD')
r = self.client.get_server_console(self.server2['id'])
self.assertTrue('host' in r)
self.assertTrue('password' in r)
r = self.client.list_server_nics(self.server1['id'])
len0 = len(r)
self.assertTrue(len0 > 0)
+ print(' ' + net)
self.assertTrue('1' in [net['network_id'] for net in r])
self.client.connect_server(self.server1['id'], self.network2['id'])
help="Show this help message and exit")
return parser
-if __name__ == '__main__':
- parser = init_parser()
- args, argv = parser.parse_known_args()
- if len(argv) > 2 or getattr(args, 'help') or len(argv) < 1:
- raise Exception('\tusage: tests.py <group> [command]')
- suiteFew = unittest.TestSuite()
+def main(argv):
+ suiteFew = unittest.TestSuite()
if len(argv) == 0 or argv[0] == 'pithos':
if len(argv) == 1:
suiteFew.addTest(unittest.makeSuite(testPithos))
suiteFew.addTest(testAstakos('test_' + argv[1]))
unittest.TextTestRunner(verbosity=2).run(suiteFew)
+
+if __name__ == '__main__':
+ parser = init_parser()
+ args, argv = parser.parse_known_args()
+ if len(argv) > 2 or getattr(args, 'help') or len(argv) < 1:
+ raise Exception('\tusage: tests.py <group> [command]')
+ main(argv)
projects / kamaki / commitdiff