Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_formula_subparser():
"""
test to parse strings with a formula parser and retrieve the following results :
- "" : {}
- "--sub toto -b" : {a:True, sub: {'toto' : {b: True}}}
- "-b" : BadContextException(b, [toto])
Parser description :
- formula subparser toto binded to the argument sub with sub arguments : -b and --name
"""
parser = MainParser(help_arg=False)
subparser = ComponentSubParser('toto')
subparser.add_argument('b', flag=True, action=store_true)
parser.add_formula_subparser('sub', subparser)
check_parsing_result(parser, '', {})
check_parsing_result(parser, '--sub toto -b', {'sub': {'toto': {'b': True}}})
with pytest.raises(BadContextException):
check_parsing_result(parser, '-b', None)
def test_subparser_empty_token_list_default_value():
subparser = ComponentSubParser('toto')
subparser.add_argument('a', default=1)
acc, result = subparser.subparse([])
assert len(result) == 1
assert 'a' in result
assert result['a'] == 1
assert acc == []
def test_add_component_subparser_with_two_name():
"""
add a component subparser with one short name and one long name
parse a string and test if the value is only bind to the long name
"""
parser = MainParser(help_arg=False)
subparser = ComponentSubParser('titi')
subparser.add_argument('a', 'aaa', flag=True, action=store_true, default=False)
subparser.add_argument('n', 'name')
parser.add_component_subparser('sub', subparser)
check_parsing_result(parser, '--sub titi -a --name tutu', {'sub': {'titi': {'tutu': {'aaa': True, 'name': 'tutu'}}}})
def test_create_component_that_already_exist():
"""
Create two component with the same name with the following cli
--sub toto --name titi --sub toto --name titi
test if an ComponentAlreadyExistException is raised
"""
parser = MainParser(help_arg=False)
subparser = ComponentSubParser('toto')
subparser.add_argument('b', flag=True, action=store_true)
subparser.add_argument('n', 'name')
parser.add_component_subparser('sub', subparser)
with pytest.raises(ComponentAlreadyExistException):
check_parsing_result(parser, '--sub toto --name titi --sub toto --name titi', None)
def test_add_component_subparser_that_aldready_exists():
"""
Add a component_subparser with no argument 'name'
test if a SubParserWithoutNameArgumentException is raised
"""
parser = MainParser(help_arg=False)
subparser = ComponentSubParser('titi')
with pytest.raises(SubParserWithoutNameArgumentException):
parser.add_component_subparser('toto', subparser)
def test_create_two_component():
"""
Create two component of the same type with the following cli :
--sub toto --name titi --sub toto -b --name tutu
test if the result is :
{sub:{'toto' : {'titi': {'name': 'titi'}, 'tutu': {'name': 'tutu', 'b':False}}}}
"""
parser = MainParser(help_arg=False)
subparser = ComponentSubParser('toto')
subparser.add_argument('b', flag=True, action=store_true)
subparser.add_argument('n', 'name')
parser.add_component_subparser('sub', subparser)
check_parsing_result(parser, '--sub toto --name titi --sub toto -b --name tutu', {'sub': {'toto': {'titi': {'name': 'titi'}, 'tutu': {'name': 'tutu', 'b': True}}}})
self.add_argument('v', 'verbose', flag=True, action=enable_log, default=logging.NOTSET,
help='enable verbose mode')
self.add_argument('s', 'stream', flag=True, action=store_true, default=False, help='enable stream mode')
subparser_mongo_input = ComponentSubParser('mongodb')
subparser_mongo_input.add_argument('u', 'uri', help='sepcify MongoDB uri')
subparser_mongo_input.add_argument('d', 'db', help='specify MongoDB database name', )
subparser_mongo_input.add_argument('c', 'collection', help='specify MongoDB database collection')
subparser_mongo_input.add_argument('n', 'name', help='specify puller name', default='puller_mongodb')
subparser_mongo_input.add_argument('m', 'model', help='specify data type that will be storen in the database',
default='HWPCReport')
self.add_component_subparser('input', subparser_mongo_input,
help_str='specify a database input : --db_output database_name ARG1 ARG2 ... ')
subparser_csv_input = ComponentSubParser('csv')
subparser_csv_input.add_argument('f', 'files',
help='specify input csv files with this format : file1,file2,file3',
action=extract_file_names, default=[], check=check_csv_files,
check_msg='one or more csv files couldn\'t be read')
subparser_csv_input.add_argument('m', 'model', help='specify data type that will be storen in the database',
default='HWPCReport')
subparser_csv_input.add_argument('n', 'name', help='specify puller name', default='puller_csv')
self.add_component_subparser('input', subparser_csv_input,
help_str='specify a database input : --db_output database_name ARG1 ARG2 ... ')
subparser_mongo_output = ComponentSubParser('mongodb')
subparser_mongo_output.add_argument('u', 'uri', help='sepcify MongoDB uri')
subparser_mongo_output.add_argument('d', 'db', help='specify MongoDB database name')
subparser_mongo_output.add_argument('c', 'collection', help='specify MongoDB database collection')
subparser_mongo_output.add_argument('m', 'model', help='specify data type that will be storen in the database',
subparser_mongo_output.add_argument('m', 'model', help='specify data type that will be storen in the database',
default='PowerReport')
subparser_mongo_output.add_argument('n', 'name', help='specify puller name', default='pusher_mongodb')
self.add_component_subparser('output', subparser_mongo_output,
help_str='specify a database output : --db_output database_name ARG1 ARG2 ...')
subparser_csv_output = ComponentSubParser('csv')
subparser_csv_output.add_argument('d', 'directory',
help='specify directory where where output csv files will be writen')
subparser_csv_output.add_argument('m', 'model', help='specify data type that will be storen in the database',
default='PowerReport')
subparser_csv_output.add_argument('n', 'name', help='specify puller name', default='pusher_csv')
self.add_component_subparser('output', subparser_csv_output,
help_str='specify a database input : --db_output database_name ARG1 ARG2 ... ')
subparser_influx_output = ComponentSubParser('influxdb')
subparser_influx_output.add_argument('u', 'uri', help='sepcify InfluxDB uri')
subparser_influx_output.add_argument('d', 'db', help='specify InfluxDB database name')
subparser_influx_output.add_argument('p', 'port', help='specify InfluxDB connection port', type=int)
subparser_influx_output.add_argument('m', 'model', help='specify data type that will be storen in the database',
default='PowerReport')
subparser_influx_output.add_argument('n', 'name', help='specify puller name', default='pusher_influxdb')
self.add_component_subparser('output', subparser_influx_output,
help_str='specify a database input : --db_output database_name ARG1 ARG2 ... ')
subparser_opentsdb_output = ComponentSubParser('opentsdb')
subparser_opentsdb_output.add_argument('u', 'uri', help='sepcify openTSDB host')
subparser_opentsdb_output.add_argument('p', 'port', help='specify openTSDB connection port', type=int)
subparser_opentsdb_output.add_argument('metric_name', help='specify metric name')
subparser_opentsdb_output.add_argument('m', 'model', help='specify data type that will be storen in the database',
default='PowerReport')
default='PowerReport')
subparser_csv_output.add_argument('n', 'name', help='specify puller name', default='pusher_csv')
self.add_component_subparser('output', subparser_csv_output,
help_str='specify a database input : --db_output database_name ARG1 ARG2 ... ')
subparser_influx_output = ComponentSubParser('influxdb')
subparser_influx_output.add_argument('u', 'uri', help='sepcify InfluxDB uri')
subparser_influx_output.add_argument('d', 'db', help='specify InfluxDB database name')
subparser_influx_output.add_argument('p', 'port', help='specify InfluxDB connection port', type=int)
subparser_influx_output.add_argument('m', 'model', help='specify data type that will be storen in the database',
default='PowerReport')
subparser_influx_output.add_argument('n', 'name', help='specify puller name', default='pusher_influxdb')
self.add_component_subparser('output', subparser_influx_output,
help_str='specify a database input : --db_output database_name ARG1 ARG2 ... ')
subparser_opentsdb_output = ComponentSubParser('opentsdb')
subparser_opentsdb_output.add_argument('u', 'uri', help='sepcify openTSDB host')
subparser_opentsdb_output.add_argument('p', 'port', help='specify openTSDB connection port', type=int)
subparser_opentsdb_output.add_argument('metric_name', help='specify metric name')
subparser_opentsdb_output.add_argument('m', 'model', help='specify data type that will be storen in the database',
default='PowerReport')
subparser_opentsdb_output.add_argument('n', 'name', help='specify puller name', default='pusher_opentsdb')
self.add_component_subparser('output', subparser_opentsdb_output,
help_str='specify a database input : --db_output database_name ARG1 ARG2 ... ')
def __init__(self):
MainParser.__init__(self)
self.add_argument('v', 'verbose', flag=True, action=enable_log, default=logging.NOTSET,
help='enable verbose mode')
self.add_argument('s', 'stream', flag=True, action=store_true, default=False, help='enable stream mode')
subparser_mongo_input = ComponentSubParser('mongodb')
subparser_mongo_input.add_argument('u', 'uri', help='sepcify MongoDB uri')
subparser_mongo_input.add_argument('d', 'db', help='specify MongoDB database name', )
subparser_mongo_input.add_argument('c', 'collection', help='specify MongoDB database collection')
subparser_mongo_input.add_argument('n', 'name', help='specify puller name', default='puller_mongodb')
subparser_mongo_input.add_argument('m', 'model', help='specify data type that will be storen in the database',
default='HWPCReport')
self.add_component_subparser('input', subparser_mongo_input,
help_str='specify a database input : --db_output database_name ARG1 ARG2 ... ')
subparser_csv_input = ComponentSubParser('csv')
subparser_csv_input.add_argument('f', 'files',
help='specify input csv files with this format : file1,file2,file3',
action=extract_file_names, default=[], check=check_csv_files,
check_msg='one or more csv files couldn\'t be read')
subparser_csv_input.add_argument('m', 'model', help='specify data type that will be storen in the database',
default='HWPCReport')