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_compile_error(self):
with self.assertRaises(ReKeywordsChangedError):
class _G1(Grammar):
k_test = Keyword('test')
RE_KEYWORDS = '[a-z]+'
with self.assertRaises(KeywordError):
class _G2(Grammar):
k_test = Keyword('test-nomatch')
with self.assertRaises(KeywordError):
# test 'deep' keyword checks.
class _G3(Grammar):
START = Optional(Keyword('test-nomatch'))
with self.assertRaises(NameAssignedError):
class _G4(Grammar):
k_test = Keyword('test')
k_test2 = k_test
with self.assertRaises(NameAssignedError):
class _G5(Grammar):
k_test = Keyword('test')
k_Test = Keyword('duplicate')
with self.assertRaises(MissingStartError):
class _G6(Grammar):
k_test = Keyword('test')
with self.assertRaises(UnusedElementError):
def _walk(cls, path, structure):
name = '_'.join(path)
try:
if structure:
for child_path, child in structure.items():
_walk(cls, child_path, child)
opt = Optional(Choice(*[
getattr(cls, '_'.join(p)) for p in structure.keys()]))
value = Sequence(*[getattr(
cls,
'k_' + path[-1],
Keyword(path[-1])), opt])
else:
value = Keyword(path[-1])
except AttributeError:
logging.critical('Cannot parse help file: {!r}'.format(
'{}.md'.format('_'.join(path))))
import sys
sys.exit(1)
else:
if name != 'help':
_set_attribute(cls, name, value)
list_groups = Sequence(
k_groups, Optional(group_columns), Optional(where_group))
list_pools = Sequence(
k_pools, Optional(pool_columns), Optional(where_pool))
list_series = Sequence(
k_series,
Optional(series_columns),
Optional(series_match),
Optional(where_series))
list_servers = Sequence(
k_servers, Optional(server_columns), Optional(where_server))
list_shards = Sequence(
k_shards, Optional(shard_columns), Optional(where_shard))
list_users = Sequence(
k_users, Optional(user_columns), Optional(where_user))
revoke_user = Sequence(k_user, string)
alter_stmt = Sequence(k_alter, Choice(
alter_user,
alter_group,
alter_server,
alter_servers,
alter_database,
most_greedy=False))
calc_stmt = Repeat(time_expr, 1, 1)
count_stmt = Sequence(k_count, Choice(
count_groups,
count_pools,
drop_user = Sequence(k_user, string)
grant_user = Sequence(
k_user, string, Optional(set_password))
list_groups = Sequence(
k_groups, Optional(group_columns), Optional(where_group))
list_pools = Sequence(
k_pools, Optional(pool_columns), Optional(where_pool))
list_series = Sequence(
k_series,
Optional(series_columns),
Optional(series_match),
Optional(where_series))
list_servers = Sequence(
k_servers, Optional(server_columns), Optional(where_server))
list_shards = Sequence(
k_shards, Optional(shard_columns), Optional(where_shard))
list_users = Sequence(
k_users, Optional(user_columns), Optional(where_user))
revoke_user = Sequence(k_user, string)
alter_stmt = Sequence(k_alter, Choice(
alter_user,
alter_group,
alter_server,
alter_servers,
alter_database,
most_greedy=False))
calc_stmt = Repeat(time_expr, 1, 1)
drop_group = Sequence(k_group, group_name)
# Drop statement needs at least a series_math or where STMT or both
drop_series = Sequence(
k_series,
Optional(series_match),
Optional(where_series),
Optional(set_ignore_threshold))
drop_shards = Sequence(
k_shards,
Optional(where_shard),
Optional(set_ignore_threshold))
drop_server = Sequence(k_server, uuid)
drop_user = Sequence(k_user, string)
grant_user = Sequence(
k_user, string, Optional(set_password))
list_groups = Sequence(
k_groups, Optional(group_columns), Optional(where_group))
list_pools = Sequence(
k_pools, Optional(pool_columns), Optional(where_pool))
list_series = Sequence(
k_series,
Optional(series_columns),
Optional(series_match),
Optional(where_series))
list_servers = Sequence(
k_servers, Optional(server_columns), Optional(where_server))
list_shards = Sequence(
k_shards, Optional(shard_columns), Optional(where_shard))
list_users = Sequence(
k_users, Optional(user_columns), Optional(where_user))
drop_user,
most_greedy=False))
grant_stmt = Sequence(k_grant, access_expr, k_to, Choice(
grant_user,
most_greedy=False))
list_stmt = Sequence(k_list, Choice(
list_series,
list_users,
list_shards,
list_groups,
list_servers,
list_pools,
most_greedy=False
), Optional(limit_expr))
revoke_stmt = Sequence(k_revoke, access_expr, k_from, Choice(
revoke_user,
most_greedy=False))
select_stmt = Sequence(
k_select,
select_aggregates,
k_from,
series_match,
Optional(where_series),
Optional(Choice(
after_expr,
between_expr,
before_expr,
most_greedy=False)),
alter_servers = Sequence(k_servers, Optional(where_server), Choice(
set_log_level,
set_tee_pipe_name,
most_greedy=False))
alter_user = Sequence(k_user, string, Choice(
set_password,
set_name,
most_greedy=False))
count_groups = Sequence(
k_groups, Optional(where_group))
count_pools = Sequence(
k_pools, Optional(where_pool))
count_series = Sequence(
k_series, Optional(series_match), Optional(where_series))
count_servers = Sequence(
k_servers, Optional(where_server))
count_servers_received = Sequence(
k_servers,
k_received_points,
Optional(where_server))
count_servers_selected = Sequence(
k_servers,
k_selected_points,
Optional(where_server))
count_shards = Sequence(
k_shards, Optional(where_shard))
count_shards_size = Sequence(
k_shards, k_size, Optional(where_shard))
count_users = Sequence(
k_users, Optional(where_user))
f_count,
f_variance,
f_pvariance,
f_stddev,
f_first,
f_last,
f_difference,
f_derivative,
f_filter,
f_points,
most_greedy=False), '=>', 1)
select_aggregate = Sequence(
aggregate_functions,
Optional(prefix_expr),
Optional(suffix_expr))
select_aggregates = List(select_aggregate, ',', 1)
merge_as = Sequence(
k_merge,
k_as,
string,
Optional(Sequence(k_using, aggregate_functions)))
set_address = Sequence(k_set, k_address, string)
set_tee_pipe_name = Sequence(k_set, k_tee_pipe_name, Choice(
k_false,
string,
most_greedy=False))
set_backup_mode = Sequence(k_set, k_backup_mode, _boolean)
set_drop_threshold = Sequence(k_set, k_drop_threshold, r_float)
count_series = Sequence(
k_series, Optional(series_match), Optional(where_series))
count_servers = Sequence(
k_servers, Optional(where_server))
count_servers_received = Sequence(
k_servers,
k_received_points,
Optional(where_server))
count_servers_selected = Sequence(
k_servers,
k_selected_points,
Optional(where_server))
count_shards = Sequence(
k_shards, Optional(where_shard))
count_shards_size = Sequence(
k_shards, k_size, Optional(where_shard))
count_users = Sequence(
k_users, Optional(where_user))
count_series_length = Sequence(
k_series,
k_length,
Optional(series_match),
Optional(where_series))
create_group = Sequence(
k_group, group_name, k_for, r_regex)
create_user = Sequence(
k_user, string, set_password)
drop_group = Sequence(k_group, group_name)
# Drop statement needs at least a series_math or where STMT or both
drop_series = Sequence(