X-Git-Url: https://gerrit.opnfv.org/gerrit/gitweb?a=blobdiff_plain;f=src%2Fceph%2Fsrc%2Ftest%2Fpybind%2Ftest_ceph_argparse.py;fp=src%2Fceph%2Fsrc%2Ftest%2Fpybind%2Ftest_ceph_argparse.py;h=bc9f2d929fc9f5e77232edc604b2b01f3f32934a;hb=812ff6ca9fcd3e629e49d4328905f33eee8ca3f5;hp=0000000000000000000000000000000000000000;hpb=15280273faafb77777eab341909a3f495cf248d9;p=stor4nfv.git diff --git a/src/ceph/src/test/pybind/test_ceph_argparse.py b/src/ceph/src/test/pybind/test_ceph_argparse.py new file mode 100755 index 0000000..bc9f2d9 --- /dev/null +++ b/src/ceph/src/test/pybind/test_ceph_argparse.py @@ -0,0 +1,1184 @@ +#!/usr/bin/env nosetests +# -*- mode:python; tab-width:4; indent-tabs-mode:t; coding:utf-8 -*- +# vim: ts=4 sw=4 smarttab expandtab fileencoding=utf-8 +# +# Ceph - scalable distributed file system +# +# Copyright (C) 2013,2014 Cloudwatt +# Copyright (C) 2014 Red Hat +# +# Author: Loic Dachary +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# + +from nose.tools import eq_ as eq +from nose.tools import * + +from ceph_argparse import validate_command, parse_json_funcsigs + +import os +import re +import json + +def get_command_descriptions(what): + CEPH_BIN = os.environ['CEPH_BIN'] + if CEPH_BIN == "": + CEPH_BIN = "." + return os.popen(CEPH_BIN + "/get_command_descriptions " + "--" + what).read() + +def test_parse_json_funcsigs(): + commands = get_command_descriptions("all") + cmd_json = parse_json_funcsigs(commands, 'cli') + + # syntax error https://github.com/ceph/ceph/pull/585 + commands = get_command_descriptions("pull585") + assert_raises(TypeError, parse_json_funcsigs, commands, 'cli') + +sigdict = parse_json_funcsigs(get_command_descriptions("all"), 'cli') + + +class TestArgparse: + + def assert_valid_command(self, args): + result = validate_command(sigdict, args) + assert_not_equal(result,None) + assert_not_equal(result,{}) + + def check_1_natural_arg(self, prefix, command): + self.assert_valid_command([prefix, command, '1']) + assert_equal({}, validate_command(sigdict, [prefix, command])) + assert_equal({}, validate_command(sigdict, [prefix, command, '-1'])) + assert_equal({}, validate_command(sigdict, [prefix, command, '1', + '1'])) + + def check_0_or_1_natural_arg(self, prefix, command): + self.assert_valid_command([prefix, command, '1']) + self.assert_valid_command([prefix, command]) + assert_equal({}, validate_command(sigdict, [prefix, command, '-1'])) + assert_equal({}, validate_command(sigdict, [prefix, command, '1', + '1'])) + + def check_1_string_arg(self, prefix, command): + assert_equal({}, validate_command(sigdict, [prefix, command])) + self.assert_valid_command([prefix, command, 'string']) + assert_equal({}, validate_command(sigdict, [prefix, + command, + 'string', + 'toomany'])) + + def check_1_or_more_string_args(self, prefix, command): + assert_equal({}, validate_command(sigdict, [prefix, + command])) + self.assert_valid_command([prefix, + command, + 'string']) + self.assert_valid_command([prefix, + command, + 'string', + 'more string']) + + def check_no_arg(self, prefix, command): + self.assert_valid_command([prefix, + command]) + assert_equal({}, validate_command(sigdict, [prefix, + command, + 'toomany'])) + + +class TestBasic: + + def test_non_ascii_in_non_options(self): + # ArgumentPrefix("no match for {0}".format(s)) is not able to convert + # unicode str parameter into str. and validate_command() should not + # choke on it. + assert_is_none(validate_command(sigdict, [u'章鱼和鱿鱼'])) + assert_is_none(validate_command(sigdict, [u'–w'])) + # actually we always pass unicode strings to validate_command() in "ceph" + # CLI, but we also use bytestrings in our tests, so make sure it does not + # break. + assert_is_none(validate_command(sigdict, ['章鱼和鱿鱼'])) + assert_is_none(validate_command(sigdict, ['–w'])) + + +class TestPG(TestArgparse): + + def test_stat(self): + self.assert_valid_command(['pg', 'stat']) + + def test_getmap(self): + self.assert_valid_command(['pg', 'getmap']) + + def test_dump(self): + self.assert_valid_command(['pg', 'dump']) + self.assert_valid_command(['pg', 'dump', + 'all', + 'summary', + 'sum', + 'delta', + 'pools', + 'osds', + 'pgs', + 'pgs_brief']) + assert_equal({}, validate_command(sigdict, ['pg', 'dump', 'invalid'])) + + def test_dump_json(self): + self.assert_valid_command(['pg', 'dump_json']) + self.assert_valid_command(['pg', 'dump_json', + 'all', + 'summary', + 'sum', + 'pools', + 'osds', + 'pgs']) + assert_equal({}, validate_command(sigdict, ['pg', 'dump_json', + 'invalid'])) + + def test_dump_pools_json(self): + self.assert_valid_command(['pg', 'dump_pools_json']) + + def test_dump_pools_stuck(self): + self.assert_valid_command(['pg', 'dump_stuck']) + self.assert_valid_command(['pg', 'dump_stuck', + 'inactive', + 'unclean', + 'stale']) + assert_equal({}, validate_command(sigdict, ['pg', 'dump_stuck', + 'invalid'])) + self.assert_valid_command(['pg', 'dump_stuck', + 'inactive', + '1234']) + + def one_pgid(self, command): + self.assert_valid_command(['pg', command, '1.1']) + assert_equal({}, validate_command(sigdict, ['pg', command])) + assert_equal({}, validate_command(sigdict, ['pg', command, '1'])) + + def test_map(self): + self.one_pgid('map') + + def test_scrub(self): + self.one_pgid('scrub') + + def test_deep_scrub(self): + self.one_pgid('deep-scrub') + + def test_repair(self): + self.one_pgid('repair') + + def test_debug(self): + self.assert_valid_command(['pg', + 'debug', + 'unfound_objects_exist']) + self.assert_valid_command(['pg', + 'debug', + 'degraded_pgs_exist']) + assert_equal({}, validate_command(sigdict, ['pg', 'debug'])) + assert_equal({}, validate_command(sigdict, ['pg', 'debug', + 'invalid'])) + + +class TestAuth(TestArgparse): + + def test_export(self): + self.assert_valid_command(['auth', 'export']) + self.assert_valid_command(['auth', + 'export', + 'string']) + assert_equal({}, validate_command(sigdict, ['auth', + 'export', + 'string', + 'toomany'])) + + def test_get(self): + self.check_1_string_arg('auth', 'get') + + def test_get_key(self): + self.check_1_string_arg('auth', 'get-key') + + def test_print_key(self): + self.check_1_string_arg('auth', 'print-key') + self.check_1_string_arg('auth', 'print_key') + + def test_list(self): + self.check_no_arg('auth', 'list') + + def test_import(self): + self.check_no_arg('auth', 'import') + + def test_add(self): + self.check_1_or_more_string_args('auth', 'add') + + def test_get_or_create_key(self): + self.check_1_or_more_string_args('auth', 'get-or-create-key') + + def test_get_or_create(self): + self.check_1_or_more_string_args('auth', 'get-or-create') + + def test_caps(self): + assert_equal({}, validate_command(sigdict, ['auth', + 'caps'])) + assert_equal({}, validate_command(sigdict, ['auth', + 'caps', + 'string'])) + self.assert_valid_command(['auth', + 'caps', + 'string', + 'more string']) + + def test_del(self): + self.check_1_string_arg('auth', 'del') + + +class TestMonitor(TestArgparse): + + def test_compact(self): + self.assert_valid_command(['compact']) + + def test_scrub(self): + self.assert_valid_command(['scrub']) + + def test_fsid(self): + self.assert_valid_command(['fsid']) + + def test_log(self): + assert_equal({}, validate_command(sigdict, ['log'])) + self.assert_valid_command(['log', 'a logtext']) + self.assert_valid_command(['log', 'a logtext', 'and another']) + + def test_injectargs(self): + assert_equal({}, validate_command(sigdict, ['injectargs'])) + self.assert_valid_command(['injectargs', 'one']) + self.assert_valid_command(['injectargs', 'one', 'two']) + + def test_status(self): + self.assert_valid_command(['status']) + + def test_health(self): + self.assert_valid_command(['health']) + self.assert_valid_command(['health', 'detail']) + assert_equal({}, validate_command(sigdict, ['health', 'invalid'])) + assert_equal({}, validate_command(sigdict, ['health', 'detail', + 'toomany'])) + + def test_df(self): + self.assert_valid_command(['df']) + self.assert_valid_command(['df', 'detail']) + assert_equal({}, validate_command(sigdict, ['df', 'invalid'])) + assert_equal({}, validate_command(sigdict, ['df', 'detail', + 'toomany'])) + + def test_report(self): + self.assert_valid_command(['report']) + self.assert_valid_command(['report', 'tag1']) + self.assert_valid_command(['report', 'tag1', 'tag2']) + + def test_quorum_status(self): + self.assert_valid_command(['quorum_status']) + + def test_mon_status(self): + self.assert_valid_command(['mon_status']) + + def test_sync_force(self): + self.assert_valid_command(['sync', + 'force', + '--yes-i-really-mean-it', + '--i-know-what-i-am-doing']) + self.assert_valid_command(['sync', + 'force', + '--yes-i-really-mean-it']) + self.assert_valid_command(['sync', + 'force']) + assert_equal({}, validate_command(sigdict, ['sync'])) + assert_equal({}, validate_command(sigdict, ['sync', + 'force', + '--yes-i-really-mean-it', + '--i-know-what-i-am-doing', + 'toomany'])) + + def test_heap(self): + assert_equal({}, validate_command(sigdict, ['heap'])) + assert_equal({}, validate_command(sigdict, ['heap', 'invalid'])) + self.assert_valid_command(['heap', 'dump']) + self.assert_valid_command(['heap', 'start_profiler']) + self.assert_valid_command(['heap', 'stop_profiler']) + self.assert_valid_command(['heap', 'release']) + self.assert_valid_command(['heap', 'stats']) + + def test_quorum(self): + assert_equal({}, validate_command(sigdict, ['quorum'])) + assert_equal({}, validate_command(sigdict, ['quorum', 'invalid'])) + self.assert_valid_command(['quorum', 'enter']) + self.assert_valid_command(['quorum', 'exit']) + assert_equal({}, validate_command(sigdict, ['quorum', + 'enter', + 'toomany'])) + + def test_tell(self): + assert_equal({}, validate_command(sigdict, ['tell'])) + assert_equal({}, validate_command(sigdict, ['tell', 'invalid'])) + for name in ('osd', 'mon', 'client', 'mds'): + assert_equal({}, validate_command(sigdict, ['tell', name])) + assert_equal({}, validate_command(sigdict, ['tell', + name + ".42"])) + self.assert_valid_command(['tell', name + ".42", 'something']) + self.assert_valid_command(['tell', name + ".42", + 'something', + 'something else']) + + +class TestMDS(TestArgparse): + + def test_stat(self): + self.check_no_arg('mds', 'stat') + + def test_dump(self): + self.check_0_or_1_natural_arg('mds', 'dump') + + def test_tell(self): + self.assert_valid_command(['mds', 'tell', + 'someone', + 'something']) + self.assert_valid_command(['mds', 'tell', + 'someone', + 'something', + 'something else']) + assert_equal({}, validate_command(sigdict, ['mds', 'tell'])) + assert_equal({}, validate_command(sigdict, ['mds', 'tell', + 'someone'])) + + def test_compat_show(self): + self.assert_valid_command(['mds', 'compat', 'show']) + assert_equal({}, validate_command(sigdict, ['mds', 'compat'])) + assert_equal({}, validate_command(sigdict, ['mds', 'compat', + 'show', 'toomany'])) + + def test_stop(self): + self.assert_valid_command(['mds', 'stop', 'someone']) + assert_equal({}, validate_command(sigdict, ['mds', 'stop'])) + assert_equal({}, validate_command(sigdict, ['mds', 'stop', + 'someone', 'toomany'])) + + def test_deactivate(self): + self.assert_valid_command(['mds', 'deactivate', 'someone']) + assert_equal({}, validate_command(sigdict, ['mds', 'deactivate'])) + assert_equal({}, validate_command(sigdict, ['mds', 'deactivate', + 'someone', 'toomany'])) + + def test_set_max_mds(self): + self.check_1_natural_arg('mds', 'set_max_mds') + + def test_set_state(self): + self.assert_valid_command(['mds', 'set_state', '1', '2']) + assert_equal({}, validate_command(sigdict, ['mds', 'set_state'])) + assert_equal({}, validate_command(sigdict, ['mds', 'set_state', '-1'])) + assert_equal({}, validate_command(sigdict, ['mds', 'set_state', + '1', '-1'])) + assert_equal({}, validate_command(sigdict, ['mds', 'set_state', + '1', '21'])) + + def test_fail(self): + self.check_1_string_arg('mds', 'fail') + + def test_rm(self): + # Valid: single GID argument present + self.assert_valid_command(['mds', 'rm', '1']) + + # Missing GID arg: invalid + assert_equal({}, validate_command(sigdict, ['mds', 'rm'])) + # Extra arg: invalid + assert_equal({}, validate_command(sigdict, ['mds', 'rm', '1', 'mds.42'])) + + def test_rmfailed(self): + self.assert_valid_command(['mds', 'rmfailed', '0']) + self.assert_valid_command(['mds', 'rmfailed', '0', '--yes-i-really-mean-it']) + assert_equal({}, validate_command(sigdict, ['mds', 'rmfailed', '0', + '--yes-i-really-mean-it', + 'toomany'])) + + def test_cluster_down(self): + self.check_no_arg('mds', 'cluster_down') + + def test_cluster_up(self): + self.check_no_arg('mds', 'cluster_up') + + def test_compat_rm_compat(self): + self.assert_valid_command(['mds', 'compat', 'rm_compat', '1']) + assert_equal({}, validate_command(sigdict, ['mds', + 'compat', + 'rm_compat'])) + assert_equal({}, validate_command(sigdict, ['mds', + 'compat', + 'rm_compat', '-1'])) + assert_equal({}, validate_command(sigdict, ['mds', + 'compat', + 'rm_compat', '1', '1'])) + + def test_incompat_rm_incompat(self): + self.assert_valid_command(['mds', 'compat', 'rm_incompat', '1']) + assert_equal({}, validate_command(sigdict, ['mds', + 'compat', + 'rm_incompat'])) + assert_equal({}, validate_command(sigdict, ['mds', + 'compat', + 'rm_incompat', '-1'])) + assert_equal({}, validate_command(sigdict, ['mds', + 'compat', + 'rm_incompat', '1', '1'])) + + def test_mds_set(self): + self.assert_valid_command(['mds', 'set', 'max_mds', '2']) + self.assert_valid_command(['mds', 'set', 'max_file_size', '2']) + self.assert_valid_command(['mds', 'set', 'allow_new_snaps', 'no']) + assert_equal({}, validate_command(sigdict, ['mds', + 'set', + 'invalid'])) + + def test_add_data_pool(self): + self.assert_valid_command(['mds', 'add_data_pool', '1']) + self.assert_valid_command(['mds', 'add_data_pool', 'foo']) + + def test_remove_data_pool(self): + self.assert_valid_command(['mds', 'remove_data_pool', '1']) + self.assert_valid_command(['mds', 'remove_data_pool', 'foo']) + + def test_newfs(self): + self.assert_valid_command(['mds', 'newfs', '1', '2', + '--yes-i-really-mean-it']) + self.assert_valid_command(['mds', 'newfs', '1', '2']) + assert_equal({}, validate_command(sigdict, ['mds', 'newfs'])) + assert_equal({}, validate_command(sigdict, ['mds', 'newfs', '1'])) + assert_equal({}, validate_command(sigdict, ['mds', + 'newfs', + '1', + '2', + '--yes-i-really-mean-it', + 'toomany'])) + assert_equal({}, validate_command(sigdict, ['mds', + 'newfs', + '-1', + '2', + '--yes-i-really-mean-it'])) + assert_equal({}, validate_command(sigdict, ['mds', + 'newfs', + '1', + '-1', + '--yes-i-really-mean-it'])) + + +class TestFS(TestArgparse): + + def test_dump(self): + self.check_0_or_1_natural_arg('fs', 'dump') + + def test_fs_new(self): + self.assert_valid_command(['fs', 'new', 'default', 'metadata', 'data']) + + def test_fs_rm(self): + self.assert_valid_command(['fs', 'rm', 'default']) + self.assert_valid_command(['fs', 'rm', 'default', '--yes-i-really-mean-it']) + assert_equal({}, validate_command(sigdict, ['fs', 'rm', 'default', '--yes-i-really-mean-it', 'toomany'])) + + def test_fs_ls(self): + self.assert_valid_command(['fs', 'ls']) + assert_equal({}, validate_command(sigdict, ['fs', 'ls', 'toomany'])) + + def test_fs_set_default(self): + self.assert_valid_command(['fs', 'set-default', 'cephfs']) + assert_equal({}, validate_command(sigdict, ['fs', 'set-default'])) + assert_equal({}, validate_command(sigdict, ['fs', 'set-default', 'cephfs', 'toomany'])) + +class TestMon(TestArgparse): + + def test_dump(self): + self.check_0_or_1_natural_arg('mon', 'dump') + + def test_stat(self): + self.check_no_arg('mon', 'stat') + + def test_getmap(self): + self.check_0_or_1_natural_arg('mon', 'getmap') + + def test_add(self): + self.assert_valid_command(['mon', 'add', 'name', '1.2.3.4:1234']) + assert_equal({}, validate_command(sigdict, ['mon', 'add'])) + assert_equal({}, validate_command(sigdict, ['mon', 'add', 'name'])) + assert_equal({}, validate_command(sigdict, ['mon', 'add', + 'name', + '400.500.600.700'])) + assert_equal({}, validate_command(sigdict, ['mon', 'add', 'name', + '1.2.3.4:1234', + 'toomany'])) + + def test_remove(self): + self.assert_valid_command(['mon', 'remove', 'name']) + assert_equal({}, validate_command(sigdict, ['mon', 'remove'])) + assert_equal({}, validate_command(sigdict, ['mon', 'remove', + 'name', 'toomany'])) + + +class TestOSD(TestArgparse): + + def test_stat(self): + self.check_no_arg('osd', 'stat') + + def test_dump(self): + self.check_0_or_1_natural_arg('osd', 'dump') + + def test_osd_tree(self): + self.check_0_or_1_natural_arg('osd', 'tree') + + def test_osd_ls(self): + self.check_0_or_1_natural_arg('osd', 'ls') + + def test_osd_getmap(self): + self.check_0_or_1_natural_arg('osd', 'getmap') + + def test_osd_getcrushmap(self): + self.check_0_or_1_natural_arg('osd', 'getcrushmap') + + def test_perf(self): + self.check_no_arg('osd', 'perf') + + def test_getmaxosd(self): + self.check_no_arg('osd', 'getmaxosd') + + def test_find(self): + self.check_1_natural_arg('osd', 'find') + + def test_map(self): + self.assert_valid_command(['osd', 'map', 'poolname', 'objectname']) + self.assert_valid_command(['osd', 'map', 'poolname', 'objectname', 'nspace']) + assert_equal({}, validate_command(sigdict, ['osd', 'map'])) + assert_equal({}, validate_command(sigdict, ['osd', 'map', 'poolname'])) + assert_equal({}, validate_command(sigdict, ['osd', 'map', + 'poolname', 'objectname', 'nspace', + 'toomany'])) + + def test_metadata(self): + self.check_0_or_1_natural_arg('osd', 'metadata') + + def test_scrub(self): + self.check_1_string_arg('osd', 'scrub') + + def test_deep_scrub(self): + self.check_1_string_arg('osd', 'deep-scrub') + + def test_repair(self): + self.check_1_string_arg('osd', 'repair') + + def test_lspools(self): + self.assert_valid_command(['osd', 'lspools']) + self.assert_valid_command(['osd', 'lspools', '1']) + self.assert_valid_command(['osd', 'lspools', '-1']) + assert_equal({}, validate_command(sigdict, ['osd', 'lspools', + '1', 'toomany'])) + + def test_blacklist_ls(self): + self.assert_valid_command(['osd', 'blacklist', 'ls']) + assert_equal({}, validate_command(sigdict, ['osd', 'blacklist'])) + assert_equal({}, validate_command(sigdict, ['osd', 'blacklist', + 'ls', 'toomany'])) + + def test_crush_rule(self): + assert_equal({}, validate_command(sigdict, ['osd', 'crush'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', 'rule'])) + for subcommand in ('list', 'ls'): + self.assert_valid_command(['osd', 'crush', 'rule', subcommand]) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'rule', subcommand, + 'toomany'])) + + def test_crush_rule_dump(self): + self.assert_valid_command(['osd', 'crush', 'rule', 'dump']) + self.assert_valid_command(['osd', 'crush', 'rule', 'dump', 'RULE']) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'rule', 'dump', + 'RULE', + 'toomany'])) + + def test_crush_dump(self): + self.assert_valid_command(['osd', 'crush', 'dump']) + assert_equal({}, validate_command(sigdict, ['osd', 'crush'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'dump', + 'toomany'])) + + def test_setcrushmap(self): + self.check_no_arg('osd', 'setcrushmap') + + def test_crush_add_bucket(self): + self.assert_valid_command(['osd', 'crush', 'add-bucket', + 'name', 'type']) + assert_equal({}, validate_command(sigdict, ['osd', 'crush'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'add-bucket'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'add-bucket', 'name', + 'type', + 'toomany'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'add-bucket', '^^^', + 'type'])) + + def test_crush_rename_bucket(self): + self.assert_valid_command(['osd', 'crush', 'rename-bucket', + 'srcname', 'dstname']) + assert_equal({}, validate_command(sigdict, ['osd', 'crush'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'rename-bucket'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'rename-bucket', + 'srcname'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'rename-bucket', 'srcname', + 'dstname', + 'toomany'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'rename-bucket', '^^^', + 'dstname'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'rename-bucket', 'srcname', + '^^^^'])) + + def check_crush_setter(self, setter): + self.assert_valid_command(['osd', 'crush', setter, + '*', '2.3', 'AZaz09-_.=']) + self.assert_valid_command(['osd', 'crush', setter, + 'osd.0', '2.3', 'AZaz09-_.=']) + self.assert_valid_command(['osd', 'crush', setter, + '0', '2.3', 'AZaz09-_.=']) + self.assert_valid_command(['osd', 'crush', setter, + '0', '2.3', 'AZaz09-_.=', 'AZaz09-_.=']) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + setter, + 'osd.0'])) + ret = validate_command(sigdict, ['osd', 'crush', + setter, + 'osd.0', + '-1.0']) + assert ret in [None, {}] + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + setter, + 'osd.0', + '1.0', + '^^^'])) + + def test_crush_set(self): + assert_equal({}, validate_command(sigdict, ['osd', 'crush'])) + self.check_crush_setter('set') + + def test_crush_add(self): + assert_equal({}, validate_command(sigdict, ['osd', 'crush'])) + self.check_crush_setter('add') + + def test_crush_create_or_move(self): + assert_equal({}, validate_command(sigdict, ['osd', 'crush'])) + self.check_crush_setter('create-or-move') + + def test_crush_move(self): + self.assert_valid_command(['osd', 'crush', 'move', + 'AZaz09-_.', 'AZaz09-_.=']) + self.assert_valid_command(['osd', 'crush', 'move', + '0', 'AZaz09-_.=', 'AZaz09-_.=']) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'move'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'move', 'AZaz09-_.'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'move', '^^^', + 'AZaz09-_.='])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'move', 'AZaz09-_.', + '^^^'])) + + def test_crush_link(self): + self.assert_valid_command(['osd', 'crush', 'link', + 'name', 'AZaz09-_.=']) + self.assert_valid_command(['osd', 'crush', 'link', + 'name', 'AZaz09-_.=', 'AZaz09-_.=']) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'link'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'link', + 'name'])) + + def test_crush_rm(self): + for alias in ('rm', 'remove', 'unlink'): + self.assert_valid_command(['osd', 'crush', alias, 'AZaz09-_.']) + self.assert_valid_command(['osd', 'crush', alias, + 'AZaz09-_.', 'AZaz09-_.']) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + alias])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + alias, + 'AZaz09-_.', + 'AZaz09-_.', + 'toomany'])) + + def test_crush_reweight(self): + self.assert_valid_command(['osd', 'crush', 'reweight', + 'AZaz09-_.', '2.3']) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'reweight'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'reweight', + 'AZaz09-_.'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'reweight', + 'AZaz09-_.', + '-1.0'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'reweight', + '^^^', + '2.3'])) + + def test_crush_tunables(self): + for tunable in ('legacy', 'argonaut', 'bobtail', 'firefly', + 'optimal', 'default'): + self.assert_valid_command(['osd', 'crush', 'tunables', + tunable]) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'tunables'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'tunables', + 'default', 'toomany'])) + + def test_crush_rule_create_simple(self): + self.assert_valid_command(['osd', 'crush', 'rule', 'create-simple', + 'AZaz09-_.', 'AZaz09-_.', 'AZaz09-_.']) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', 'rule', + 'create-simple'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', 'rule', + 'create-simple', + 'AZaz09-_.'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', 'rule', + 'create-simple', + 'AZaz09-_.', + 'AZaz09-_.'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', 'rule', + 'create-simple', + '^^^', + 'AZaz09-_.', + 'AZaz09-_.'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', 'rule', + 'create-simple', + 'AZaz09-_.', + '|||', + 'AZaz09-_.'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', 'rule', + 'create-simple', + 'AZaz09-_.', + 'AZaz09-_.', + '+++'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', 'rule', + 'create-simple', + 'AZaz09-_.', + 'AZaz09-_.', + 'AZaz09-_.', + 'toomany'])) + + def test_crush_rule_create_erasure(self): + self.assert_valid_command(['osd', 'crush', 'rule', 'create-erasure', + 'AZaz09-_.']) + self.assert_valid_command(['osd', 'crush', 'rule', 'create-erasure', + 'AZaz09-_.', 'whatever']) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', 'rule', + 'create-erasure'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', 'rule', + 'create-erasure', + '^^^'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', 'rule', + 'create-erasure', + 'name', '^^^'])) + + def test_crush_rule_rm(self): + self.assert_valid_command(['osd', 'crush', 'rule', 'rm', 'AZaz09-_.']) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'rule', 'rm'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'rule', 'rm', + '^^^^'])) + assert_equal({}, validate_command(sigdict, ['osd', 'crush', + 'rule', 'rm', + 'AZaz09-_.', + 'toomany'])) + + def test_setmaxosd(self): + self.check_1_natural_arg('osd', 'setmaxosd') + + def test_pause(self): + self.check_no_arg('osd', 'pause') + + def test_unpause(self): + self.check_no_arg('osd', 'unpause') + + def test_erasure_code_profile_set(self): + self.assert_valid_command(['osd', 'erasure-code-profile', 'set', + 'name']) + self.assert_valid_command(['osd', 'erasure-code-profile', 'set', + 'name', 'A=B']) + self.assert_valid_command(['osd', 'erasure-code-profile', 'set', + 'name', 'A=B', 'C=D']) + assert_equal({}, validate_command(sigdict, ['osd', + 'erasure-code-profile', + 'set'])) + assert_equal({}, validate_command(sigdict, ['osd', + 'erasure-code-profile', + 'set', + '^^^^'])) + + def test_erasure_code_profile_get(self): + self.assert_valid_command(['osd', 'erasure-code-profile', 'get', + 'name']) + assert_equal({}, validate_command(sigdict, ['osd', + 'erasure-code-profile', + 'get'])) + assert_equal({}, validate_command(sigdict, ['osd', + 'erasure-code-profile', + 'get', + '^^^^'])) + + def test_erasure_code_profile_rm(self): + self.assert_valid_command(['osd', 'erasure-code-profile', 'rm', + 'name']) + assert_equal({}, validate_command(sigdict, ['osd', + 'erasure-code-profile', + 'rm'])) + assert_equal({}, validate_command(sigdict, ['osd', + 'erasure-code-profile', + 'rm', + '^^^^'])) + + def test_erasure_code_profile_ls(self): + self.assert_valid_command(['osd', 'erasure-code-profile', 'ls']) + assert_equal({}, validate_command(sigdict, ['osd', + 'erasure-code-profile', + 'ls', + 'toomany'])) + + def test_set_unset(self): + for action in ('set', 'unset'): + for flag in ('pause', 'noup', 'nodown', 'noout', 'noin', + 'nobackfill', 'norecover', 'noscrub', 'nodeep-scrub'): + self.assert_valid_command(['osd', action, flag]) + assert_equal({}, validate_command(sigdict, ['osd', action])) + assert_equal({}, validate_command(sigdict, ['osd', action, + 'invalid'])) + assert_equal({}, validate_command(sigdict, ['osd', action, + 'pause', 'toomany'])) + + def test_cluster_snap(self): + assert_equal(None, validate_command(sigdict, ['osd', 'cluster_snap'])) + + def test_down(self): + self.check_1_or_more_string_args('osd', 'down') + + def test_out(self): + self.check_1_or_more_string_args('osd', 'out') + + def test_in(self): + self.check_1_or_more_string_args('osd', 'in') + + def test_rm(self): + self.check_1_or_more_string_args('osd', 'rm') + + def test_reweight(self): + self.assert_valid_command(['osd', 'reweight', '1', '0.1']) + assert_equal({}, validate_command(sigdict, ['osd', 'reweight'])) + assert_equal({}, validate_command(sigdict, ['osd', 'reweight', + '1'])) + assert_equal({}, validate_command(sigdict, ['osd', 'reweight', + '1', '2.0'])) + assert_equal({}, validate_command(sigdict, ['osd', 'reweight', + '-1', '0.1'])) + assert_equal({}, validate_command(sigdict, ['osd', 'reweight', + '1', '0.1', + 'toomany'])) + + def test_lost(self): + self.assert_valid_command(['osd', 'lost', '1', + '--yes-i-really-mean-it']) + self.assert_valid_command(['osd', 'lost', '1']) + assert_equal({}, validate_command(sigdict, ['osd', 'lost'])) + assert_equal({}, validate_command(sigdict, ['osd', 'lost', + '1', + 'what?'])) + assert_equal({}, validate_command(sigdict, ['osd', 'lost', + '-1', + '--yes-i-really-mean-it'])) + assert_equal({}, validate_command(sigdict, ['osd', 'lost', + '1', + '--yes-i-really-mean-it', + 'toomany'])) + + def test_create(self): + uuid = '12345678123456781234567812345678' + self.assert_valid_command(['osd', 'create']) + self.assert_valid_command(['osd', 'create', + uuid]) + assert_equal({}, validate_command(sigdict, ['osd', 'create', + 'invalid'])) + assert_equal({}, validate_command(sigdict, ['osd', 'create', + uuid, + 'toomany'])) + + def test_blacklist(self): + for action in ('add', 'rm'): + self.assert_valid_command(['osd', 'blacklist', action, + '1.2.3.4/567']) + self.assert_valid_command(['osd', 'blacklist', action, + '1.2.3.4']) + self.assert_valid_command(['osd', 'blacklist', action, + '1.2.3.4/567', '600.40']) + self.assert_valid_command(['osd', 'blacklist', action, + '1.2.3.4', '600.40']) + assert_equal({}, validate_command(sigdict, ['osd', 'blacklist', + action, + 'invalid', + '600.40'])) + assert_equal({}, validate_command(sigdict, ['osd', 'blacklist', + action, + '1.2.3.4/567', + '-1.0'])) + assert_equal({}, validate_command(sigdict, ['osd', 'blacklist', + action, + '1.2.3.4/567', + '600.40', + 'toomany'])) + + def test_pool_mksnap(self): + self.assert_valid_command(['osd', 'pool', 'mksnap', + 'poolname', 'snapname']) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', 'mksnap'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', 'mksnap', + 'poolname'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', 'mksnap', + 'poolname', 'snapname', + 'toomany'])) + + def test_pool_rmsnap(self): + self.assert_valid_command(['osd', 'pool', 'rmsnap', + 'poolname', 'snapname']) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', 'rmsnap'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', 'rmsnap', + 'poolname'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', 'rmsnap', + 'poolname', 'snapname', + 'toomany'])) + + def test_pool_create(self): + self.assert_valid_command(['osd', 'pool', 'create', + 'poolname', '128']) + self.assert_valid_command(['osd', 'pool', 'create', + 'poolname', '128', '128']) + self.assert_valid_command(['osd', 'pool', 'create', + 'poolname', '128', '128', + 'replicated']) + self.assert_valid_command(['osd', 'pool', 'create', + 'poolname', '128', '128', + 'erasure', 'A-Za-z0-9-_.', 'ruleset^^']) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', 'create'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', 'create', + 'poolname'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', 'create', + 'poolname', '-1'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', 'create', + 'poolname', + '128', '128', + 'erasure', '^^^', + 'ruleset'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', 'create', + 'poolname', + '128', '128', + 'erasure', 'profile', + 'ruleset', + 'toomany'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', 'create', + 'poolname', + '128', '128', + 'INVALID', 'profile', + 'ruleset'])) + + def test_pool_delete(self): + self.assert_valid_command(['osd', 'pool', 'delete', + 'poolname', 'poolname', + '--yes-i-really-really-mean-it']) + self.assert_valid_command(['osd', 'pool', 'delete', + 'poolname', 'poolname']) + self.assert_valid_command(['osd', 'pool', 'delete', + 'poolname']) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', 'delete'])) + assert_equal({}, validate_command(sigdict, + ['osd', 'pool', 'delete', + 'poolname', 'poolname', + '--yes-i-really-really-mean-it', + 'toomany'])) + + def test_pool_rename(self): + self.assert_valid_command(['osd', 'pool', 'rename', + 'poolname', 'othername']) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', 'rename'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', 'rename', + 'poolname'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', 'rename', + 'poolname', 'othername', + 'toomany'])) + + def test_pool_get(self): + for var in ('size', 'min_size', 'crash_replay_interval', + 'pg_num', 'pgp_num', 'crush_rule', 'auid', 'fast_read', + 'scrub_min_interval', 'scrub_max_interval', + 'deep_scrub_interval', 'recovery_priority', + 'recovery_op_priority'): + self.assert_valid_command(['osd', 'pool', 'get', 'poolname', var]) + assert_equal({}, validate_command(sigdict, ['osd', 'pool'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', + 'get'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', + 'get', 'poolname'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', + 'get', 'poolname', + 'size', 'toomany'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', + 'get', 'poolname', + 'invalid'])) + + def test_pool_set(self): + for var in ('size', 'min_size', 'crash_replay_interval', + 'pg_num', 'pgp_num', 'crush_rule', + 'hashpspool', 'auid', 'fast_read', + 'scrub_min_interval', 'scrub_max_interval', + 'deep_scrub_interval', 'recovery_priority', + 'recovery_op_priority'): + self.assert_valid_command(['osd', 'pool', + 'set', 'poolname', var, 'value']) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', + 'set'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', + 'set', 'poolname'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', + 'set', 'poolname', + 'size', 'value', + 'toomany'])) + + def test_pool_set_quota(self): + for field in ('max_objects', 'max_bytes'): + self.assert_valid_command(['osd', 'pool', 'set-quota', + 'poolname', field, '10K']) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', + 'set-quota'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', + 'set-quota', + 'poolname'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', + 'set-quota', + 'poolname', + 'max_objects'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', + 'set-quota', + 'poolname', + 'invalid', + '10K'])) + assert_equal({}, validate_command(sigdict, ['osd', 'pool', + 'set-quota', + 'poolname', + 'max_objects', + '10K', + 'toomany'])) + + def test_reweight_by_utilization(self): + self.assert_valid_command(['osd', 'reweight-by-utilization']) + self.assert_valid_command(['osd', 'reweight-by-utilization', '100']) + self.assert_valid_command(['osd', 'reweight-by-utilization', '100', '.1']) + self.assert_valid_command(['osd', 'reweight-by-utilization', '--no-increasing']) + assert_equal({}, validate_command(sigdict, ['osd', + 'reweight-by-utilization', + '100', + 'toomany'])) + + def test_tier_op(self): + for op in ('add', 'remove', 'set-overlay'): + self.assert_valid_command(['osd', 'tier', op, + 'poolname', 'othername']) + assert_equal({}, validate_command(sigdict, ['osd', 'tier', op])) + assert_equal({}, validate_command(sigdict, ['osd', 'tier', op, + 'poolname'])) + assert_equal({}, validate_command(sigdict, ['osd', 'tier', op, + 'poolname', + 'othername', + 'toomany'])) + + def test_tier_cache_mode(self): + for mode in ('none', 'writeback', 'forward', 'readonly', 'readforward', 'readproxy'): + self.assert_valid_command(['osd', 'tier', 'cache-mode', + 'poolname', mode]) + assert_equal({}, validate_command(sigdict, ['osd', 'tier', + 'cache-mode'])) + assert_equal({}, validate_command(sigdict, ['osd', 'tier', + 'cache-mode', + 'invalid'])) + + def test_tier_remove_overlay(self): + self.assert_valid_command(['osd', 'tier', 'remove-overlay', + 'poolname']) + assert_equal({}, validate_command(sigdict, ['osd', 'tier', + 'remove-overlay'])) + assert_equal({}, validate_command(sigdict, ['osd', 'tier', + 'remove-overlay', + 'poolname', + 'toomany'])) + + def set_ratio(self, command): + self.assert_valid_command(['osd', + command, + '0.0']) + assert_equal({}, validate_command(sigdict, ['osd', command])) + assert_equal({}, validate_command(sigdict, ['osd', + command, + '2.0'])) + + def test_set_full_ratio(self): + self.set_ratio('set-full-ratio') + + def test_set_backfillfull_ratio(self): + self.set_ratio('set-backfillfull-ratio') + + def test_set_nearfull_ratio(self): + self.set_ratio('set-nearfull-ratio') + + +class TestConfigKey(TestArgparse): + + def test_get(self): + self.check_1_string_arg('config-key', 'get') + + def test_put(self): + self.assert_valid_command(['config-key', 'put', + 'key']) + self.assert_valid_command(['config-key', 'put', + 'key', 'value']) + assert_equal({}, validate_command(sigdict, ['config-key', 'put'])) + assert_equal({}, validate_command(sigdict, ['config-key', 'put', + 'key', 'value', + 'toomany'])) + + def test_del(self): + self.check_1_string_arg('config-key', 'del') + + def test_exists(self): + self.check_1_string_arg('config-key', 'exists') + + def test_dump(self): + self.check_no_arg('config-key', 'dump') + + def test_list(self): + self.check_no_arg('config-key', 'list') +# Local Variables: +# compile-command: "cd ../.. ; make -j4 && +# PYTHONPATH=pybind nosetests --stop \ +# test/pybind/test_ceph_argparse.py # test_ceph_argparse.py:TestOSD.test_rm" +# End: