本文整理汇总了Python中st2common.models.db.action.ActionDB类的典型用法代码示例。如果您正苦于以下问题：Python ActionDB类的具体用法？Python ActionDB怎么用？Python ActionDB使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了ActionDB类的19个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: setup_action_models
def setup_action_models(cls):
action_db = ActionDB()
action_db.name = 'action-1'
action_db.description = 'awesomeness'
action_db.enabled = True
action_db.pack = 'wolfpack'
action_db.ref = ResourceReference(name=action_db.name, pack=action_db.pack).ref
action_db.entry_point = ''
action_db.runner_type = {'name': 'test-runner'}
action_db.parameters = {
'actionstr': {'type': 'string', 'position': 1, 'required': True},
'actionint': {'type': 'number', 'default': 10, 'position': 0},
'runnerdummy': {'type': 'string', 'default': 'actiondummy'}
}
ActionDBUtilsTestCase.action_db = Action.add_or_update(action_db)
liveaction_db = LiveActionDB()
liveaction_db.status = 'initializing'
liveaction_db.start_timestamp = get_datetime_utc_now()
liveaction_db.action = ActionDBUtilsTestCase.action_db.ref
params = {
'actionstr': 'foo',
'some_key_that_aint_exist_in_action_or_runner': 'bar',
'runnerint': 555
}
liveaction_db.parameters = params
ActionDBUtilsTestCase.liveaction_db = LiveAction.add_or_update(liveaction_db)
开发者ID:ipv1337，项目名称:st2，代码行数:27，代码来源:test_action_db_utils.py
示例2: test_get_uid
def test_get_uid(self):
pack_1_db = PackDB(ref='test_pack')
pack_2_db = PackDB(ref='examples')
self.assertEqual(pack_1_db.get_uid(), 'pack:test_pack')
self.assertEqual(pack_2_db.get_uid(), 'pack:examples')
action_1_db = ActionDB(pack='examples', name='my_action', ref='examples.my_action')
action_2_db = ActionDB(pack='core', name='local', ref='core.local')
self.assertEqual(action_1_db.get_uid(), 'action:examples:my_action')
self.assertEqual(action_2_db.get_uid(), 'action:core:local')
开发者ID:StackStorm，项目名称:st2，代码行数:11，代码来源:test_db_uid_mixin.py
示例3: _register_action
def _register_action(self, pack, action):
with open(action, 'r') as fd:
try:
content = json.load(fd)
except ValueError:
LOG.exception('Failed loading action json from %s.', action)
raise
try:
model = Action.get_by_name(str(content['name']))
except ValueError:
model = ActionDB()
model.name = content['name']
model.description = content['description']
model.enabled = content['enabled']
model.pack = pack
model.entry_point = content['entry_point']
model.parameters = content.get('parameters', {})
runner_type = str(content['runner_type'])
valid_runner_type, runner_type_db = self._has_valid_runner_type(runner_type)
if valid_runner_type:
model.runner_type = {'name': runner_type_db.name}
else:
LOG.exception('Runner type %s doesn\'t exist.')
raise
try:
model = Action.add_or_update(model)
LOG.audit('Action created. Action %s from %s.', model, action)
except Exception:
LOG.exception('Failed to write action to db %s.', model.name)
raise
开发者ID:ojacques，项目名称:st2，代码行数:32，代码来源:actionsregistrar.py
示例4: test_uid_is_populated_on_save
def test_uid_is_populated_on_save(self):
pack_1_db = PackDB(ref='test_pack', name='test', description='foo', version='1.0.0',
author='dev', email='[email protected]')
pack_1_db = Pack.add_or_update(pack_1_db)
pack_1_db.reload()
self.assertEqual(pack_1_db.uid, 'pack:test_pack')
action_1_db = ActionDB(name='local', pack='core', ref='core.local', entry_point='',
runner_type={'name': 'local-shell-cmd'})
action_1_db = Action.add_or_update(action_1_db)
action_1_db.reload()
self.assertEqual(action_1_db.uid, 'action:core:local')
开发者ID:StackStorm，项目名称:st2，代码行数:14，代码来源:test_db_uid_mixin.py
示例5: _register_action
def _register_action(self, pack, action):
content = self._meta_loader.load(action)
action_ref = ResourceReference(pack=pack, name=str(content['name']))
model = action_utils.get_action_by_ref(action_ref)
if not model:
model = ActionDB()
model.name = content['name']
model.description = content['description']
model.enabled = content['enabled']
model.pack = pack
model.entry_point = content['entry_point']
model.parameters = content.get('parameters', {})
runner_type = str(content['runner_type'])
valid_runner_type, runner_type_db = self._has_valid_runner_type(runner_type)
if valid_runner_type:
model.runner_type = {'name': runner_type_db.name}
else:
LOG.exception('Runner type %s doesn\'t exist.', runner_type)
raise
try:
model = Action.add_or_update(model)
LOG.audit('Action created. Action %s from %s.', model, action)
except Exception:
LOG.exception('Failed to write action to db %s.', model.name)
raise
开发者ID:nagyist，项目名称:StackStorm-st2，代码行数:26，代码来源:actionsregistrar.py
示例6: _create_save_action
def _create_save_action(runnertype, metadata=False):
name = 'action-1'
pack = 'wolfpack'
ref = ResourceReference(pack=pack, name=name).ref
created = ActionDB(name=name, description='awesomeness', enabled=True,
entry_point='/tmp/action.py', pack=pack,
ref=ref,
runner_type={'name': runnertype.name})
if not metadata:
created.parameters = {'p1': None, 'p2': None, 'p3': None}
else:
created.parameters = {
'p1': {'type': 'string', 'required': True},
'p2': {'type': 'number', 'default': 2868},
'p3': {'type': 'boolean', 'default': False}
}
return Action.add_or_update(created)
开发者ID:azamsheriff，项目名称:st2，代码行数:18，代码来源:test_db.py
示例7: test_get_uid
def test_get_uid(self):
pack_db = PackDB(ref='ma_pack')
self.assertEqual(pack_db.get_uid(), 'pack:ma_pack')
sensor_type_db = SensorTypeDB(name='sname', pack='spack')
self.assertEqual(sensor_type_db.get_uid(), 'sensor_type:spack:sname')
action_db = ActionDB(name='aname', pack='apack', runner_type={})
self.assertEqual(action_db.get_uid(), 'action:apack:aname')
rule_db = RuleDB(name='rname', pack='rpack')
self.assertEqual(rule_db.get_uid(), 'rule:rpack:rname')
trigger_type_db = TriggerTypeDB(name='ttname', pack='ttpack')
self.assertEqual(trigger_type_db.get_uid(), 'trigger_type:ttpack:ttname')
trigger_db = TriggerDB(name='tname', pack='tpack')
self.assertTrue(trigger_db.get_uid().startswith('trigger:tpack:tname:'))
开发者ID:Bala96，项目名称:st2，代码行数:18，代码来源:test_db_model_uids.py
示例8: test_get_uid
def test_get_uid(self):
pack_db = PackDB(ref='ma_pack')
self.assertEqual(pack_db.get_uid(), 'pack:ma_pack')
sensor_type_db = SensorTypeDB(name='sname', pack='spack')
self.assertEqual(sensor_type_db.get_uid(), 'sensor_type:spack:sname')
action_db = ActionDB(name='aname', pack='apack', runner_type={})
self.assertEqual(action_db.get_uid(), 'action:apack:aname')
rule_db = RuleDB(name='rname', pack='rpack')
self.assertEqual(rule_db.get_uid(), 'rule:rpack:rname')
trigger_type_db = TriggerTypeDB(name='ttname', pack='ttpack')
self.assertEqual(trigger_type_db.get_uid(), 'trigger_type:ttpack:ttname')
trigger_db = TriggerDB(name='tname', pack='tpack')
self.assertTrue(trigger_db.get_uid().startswith('trigger:tpack:tname:'))
# Verify that same set of parameters always results in the same hash
parameters = {'a': 1, 'b': 2, 'c': [1, 2, 3], 'd': {'g': 1, 'h': 2}, 'b': u'unicode'}
paramers_hash = json.dumps(parameters, sort_keys=True)
paramers_hash = hashlib.md5(paramers_hash).hexdigest()
parameters = {'a': 1, 'b': 2, 'c': [1, 2, 3], 'b': u'unicode', 'd': {'g': 1, 'h': 2}}
trigger_db = TriggerDB(name='tname', pack='tpack', parameters=parameters)
self.assertEqual(trigger_db.get_uid(), 'trigger:tpack:tname:%s' % (paramers_hash))
parameters = {'c': [1, 2, 3], 'b': u'unicode', 'd': {'h': 2, 'g': 1}, 'a': 1}
trigger_db = TriggerDB(name='tname', pack='tpack', parameters=parameters)
self.assertEqual(trigger_db.get_uid(), 'trigger:tpack:tname:%s' % (paramers_hash))
parameters = {'b': u'unicode', 'c': [1, 2, 3], 'd': {'h': 2, 'g': 1}, 'a': 1}
trigger_db = TriggerDB(name='tname', pack='tpack', parameters=parameters)
self.assertEqual(trigger_db.get_uid(), 'trigger:tpack:tname:%s' % (paramers_hash))
parameters = OrderedDict({'c': [1, 2, 3], 'b': u'unicode', 'd': {'h': 2, 'g': 1}, 'a': 1})
trigger_db = TriggerDB(name='tname', pack='tpack', parameters=parameters)
self.assertEqual(trigger_db.get_uid(), 'trigger:tpack:tname:%s' % (paramers_hash))
开发者ID:LindsayHill，项目名称:st2，代码行数:39，代码来源:test_db_model_uids.py
示例9: test_format_secret_action_parameters_are_masked
def test_format_secret_action_parameters_are_masked(self):
formatter = ConsoleLogFormatter()
mock_message = 'test message 1'
mock_action_db = ActionDB()
mock_action_db.name = 'test.action'
mock_action_db.pack = 'testpack'
mock_action_db.parameters = {
'parameter1': {
'type': 'string',
'required': False
},
'parameter2': {
'type': 'string',
'required': False,
'secret': True
}
}
mock_action_execution_db = ActionExecutionDB()
mock_action_execution_db.action = mock_action_db.to_serializable_dict()
mock_action_execution_db.parameters = {
'parameter1': 'value1',
'parameter2': 'value2'
}
record = MockRecord()
record.msg = mock_message
# Add "extra" attributes
record._action_execution_db = mock_action_execution_db
expected_msg_part = "'parameters': {'parameter1': 'value1', 'parameter2': '********'}"
message = formatter.format(record=record)
self.assertTrue('test message 1' in message)
self.assertTrue(expected_msg_part in message)
开发者ID:ipv1337，项目名称:st2，代码行数:38，代码来源:test_logger.py
示例10: _create_save_action
def _create_save_action(runnertype, metadata=False):
name = "action-1"
pack = "wolfpack"
ref = ResourceReference(pack=pack, name=name).ref
created = ActionDB(
name=name,
description="awesomeness",
enabled=True,
entry_point="/tmp/action.py",
pack=pack,
ref=ref,
runner_type={"name": runnertype.name},
)
if not metadata:
created.parameters = {"p1": None, "p2": None, "p3": None}
else:
created.parameters = {
"p1": {"type": "string", "required": True},
"p2": {"type": "number", "default": 2868},
"p3": {"type": "boolean", "default": False},
}
return Action.add_or_update(created)
开发者ID:rlugojr，项目名称:st2，代码行数:23，代码来源:test_db.py
示例11: _setup_action_models
def _setup_action_models(cls):
action_db = ActionDB()
action_db.name = 'action-1'
action_db.description = 'awesomeness'
action_db.enabled = True
action_db.pack = 'wolfpack'
action_db.entry_point = ''
action_db.runner_type = {'name': 'test-runner'}
action_db.parameters = {
'actionstr': {'type': 'string', 'required': True},
'actionint': {'type': 'number', 'default': 10},
'runnerdummy': {'type': 'string', 'default': 'actiondummy', 'immutable': True},
'runnerimmutable': {'type': 'string', 'default': 'failed_override'},
'actionimmutable': {'type': 'string', 'default': 'actionimmutable', 'immutable': True}
}
ParamsUtilsTest.action_db = action_db
开发者ID:bjoernbessert，项目名称:st2，代码行数:16，代码来源:test_param_utils.py
示例12: test_format_secret_action_parameters_are_masked
def test_format_secret_action_parameters_are_masked(self):
formatter = ConsoleLogFormatter()
mock_message = 'test message 1'
parameters = {
'parameter1': {
'type': 'string',
'required': False
},
'parameter2': {
'type': 'string',
'required': False,
'secret': True
}
}
mock_action_db = ActionDB(pack='testpack', name='test.action', parameters=parameters)
action = mock_action_db.to_serializable_dict()
parameters = {
'parameter1': 'value1',
'parameter2': 'value2'
}
mock_action_execution_db = ActionExecutionDB(action=action, parameters=parameters)
record = MockRecord()
record.msg = mock_message
# Add "extra" attributes
record._action_execution_db = mock_action_execution_db
expected_msg_part = (r"'parameters': {u?'parameter1': u?'value1', "
r"u?'parameter2': u?'\*\*\*\*\*\*\*\*'}")
message = formatter.format(record=record)
self.assertTrue('test message 1' in message)
self.assertRegexpMatches(message, expected_msg_part)
开发者ID:nzlosh，项目名称:st2，代码行数:37，代码来源:test_logger.py
示例13: test_get_uid
def test_get_uid(self):
pack_db = PackDB(ref='ma_pack')
self.assertEqual(pack_db.get_uid(), 'pack:ma_pack')
self.assertTrue(pack_db.has_valid_uid())
sensor_type_db = SensorTypeDB(name='sname', pack='spack')
self.assertEqual(sensor_type_db.get_uid(), 'sensor_type:spack:sname')
self.assertTrue(sensor_type_db.has_valid_uid())
action_db = ActionDB(name='aname', pack='apack', runner_type={})
self.assertEqual(action_db.get_uid(), 'action:apack:aname')
self.assertTrue(action_db.has_valid_uid())
rule_db = RuleDB(name='rname', pack='rpack')
self.assertEqual(rule_db.get_uid(), 'rule:rpack:rname')
self.assertTrue(rule_db.has_valid_uid())
trigger_type_db = TriggerTypeDB(name='ttname', pack='ttpack')
self.assertEqual(trigger_type_db.get_uid(), 'trigger_type:ttpack:ttname')
self.assertTrue(trigger_type_db.has_valid_uid())
trigger_db = TriggerDB(name='tname', pack='tpack')
self.assertTrue(trigger_db.get_uid().startswith('trigger:tpack:tname:'))
# Verify that same set of parameters always results in the same hash
parameters = {'a': 1, 'b': 'unicode', 'c': [1, 2, 3], 'd': {'g': 1, 'h': 2}}
paramers_hash = json.dumps(parameters, sort_keys=True)
paramers_hash = hashlib.md5(paramers_hash.encode()).hexdigest()
parameters = {'a': 1, 'b': 'unicode', 'c': [1, 2, 3], 'd': {'g': 1, 'h': 2}}
trigger_db = TriggerDB(name='tname', pack='tpack', parameters=parameters)
self.assertEqual(trigger_db.get_uid(), 'trigger:tpack:tname:%s' % (paramers_hash))
self.assertTrue(trigger_db.has_valid_uid())
parameters = {'c': [1, 2, 3], 'b': u'unicode', 'd': {'h': 2, 'g': 1}, 'a': 1}
trigger_db = TriggerDB(name='tname', pack='tpack', parameters=parameters)
self.assertEqual(trigger_db.get_uid(), 'trigger:tpack:tname:%s' % (paramers_hash))
self.assertTrue(trigger_db.has_valid_uid())
parameters = {'b': u'unicode', 'c': [1, 2, 3], 'd': {'h': 2, 'g': 1}, 'a': 1}
trigger_db = TriggerDB(name='tname', pack='tpack', parameters=parameters)
self.assertEqual(trigger_db.get_uid(), 'trigger:tpack:tname:%s' % (paramers_hash))
self.assertTrue(trigger_db.has_valid_uid())
parameters = OrderedDict({'c': [1, 2, 3], 'b': u'unicode', 'd': {'h': 2, 'g': 1}, 'a': 1})
trigger_db = TriggerDB(name='tname', pack='tpack', parameters=parameters)
self.assertEqual(trigger_db.get_uid(), 'trigger:tpack:tname:%s' % (paramers_hash))
self.assertTrue(trigger_db.has_valid_uid())
policy_type_db = PolicyTypeDB(resource_type='action', name='concurrency')
self.assertEqual(policy_type_db.get_uid(), 'policy_type:action:concurrency')
self.assertTrue(policy_type_db.has_valid_uid())
policy_db = PolicyDB(pack='dummy', name='policy1')
self.assertEqual(policy_db.get_uid(), 'policy:dummy:policy1')
api_key_db = ApiKeyDB(key_hash='valid')
self.assertEqual(api_key_db.get_uid(), 'api_key:valid')
self.assertTrue(api_key_db.has_valid_uid())
api_key_db = ApiKeyDB()
self.assertEqual(api_key_db.get_uid(), 'api_key:')
self.assertFalse(api_key_db.has_valid_uid())
开发者ID:StackStorm，项目名称:st2，代码行数:63，代码来源:test_db_model_uids.py
示例14: test_command_construction_correct_default_parameter_values_are_used
def test_command_construction_correct_default_parameter_values_are_used(self):
runner_parameters = {}
action_db_parameters = {
'project': {
'type': 'string',
'default': 'st2',
'position': 0,
},
'version': {
'type': 'string',
'position': 1,
'required': True
},
'fork': {
'type': 'string',
'position': 2,
'default': 'StackStorm',
},
'branch': {
'type': 'string',
'position': 3,
'default': 'master',
},
'update_mistral': {
'type': 'boolean',
'position': 4,
'default': False
},
'update_changelog': {
'type': 'boolean',
'position': 5,
'default': False
},
'local_repo': {
'type': 'string',
'position': 6,
}
}
context = {}
action_db = ActionDB(pack='dummy', name='action')
runner = LocalShellScriptRunner('id')
runner.runner_parameters = {}
runner.action = action_db
# 1. All default values used
live_action_db_parameters = {
'project': 'st2flow',
'version': '3.0.0',
'fork': 'StackStorm',
'local_repo': '/tmp/repo'
}
runner_params, action_params = param_utils.render_final_params(runner_parameters,
action_db_parameters,
live_action_db_parameters,
context)
self.assertDictEqual(action_params, {
'project': 'st2flow',
'version': '3.0.0',
'fork': 'StackStorm',
'branch': 'master', # default value used
'update_mistral': False, # default value used
'update_changelog': False, # default value used
'local_repo': '/tmp/repo'
})
action_db.parameters = action_db_parameters
positional_args, named_args = runner._get_script_args(action_params)
named_args = runner._transform_named_args(named_args)
shell_script_action = ShellScriptAction(name='dummy', action_exec_id='dummy',
script_local_path_abs='/tmp/local.sh',
named_args=named_args,
positional_args=positional_args)
command_string = shell_script_action.get_full_command_string()
expected = '/tmp/local.sh st2flow 3.0.0 StackStorm master 0 0 /tmp/repo'
self.assertEqual(command_string, expected)
# 2. Some default values used
live_action_db_parameters = {
'project': 'st2web',
'version': '3.1.0',
'fork': 'StackStorm1',
'update_changelog': True,
'local_repo': '/tmp/repob'
}
runner_params, action_params = param_utils.render_final_params(runner_parameters,
action_db_parameters,
live_action_db_parameters,
context)
self.assertDictEqual(action_params, {
'project': 'st2web',
'version': '3.1.0',
'fork': 'StackStorm1',
#.........这里部分代码省略.........
开发者ID:StackStorm，项目名称:st2，代码行数:101，代码来源:test_shell_action_system_model.py
示例15: setUp
def setUp(self):
super(InquiryPermissionsResolverTestCase, self).setUp()
# Create some mock users
user_1_db = UserDB(name='custom_role_inquiry_list_grant')
user_1_db = User.add_or_update(user_1_db)
self.users['custom_role_inquiry_list_grant'] = user_1_db
user_2_db = UserDB(name='custom_role_inquiry_view_grant')
user_2_db = User.add_or_update(user_2_db)
self.users['custom_role_inquiry_view_grant'] = user_2_db
user_3_db = UserDB(name='custom_role_inquiry_respond_grant')
user_3_db = User.add_or_update(user_3_db)
self.users['custom_role_inquiry_respond_grant'] = user_3_db
user_4_db = UserDB(name='custom_role_inquiry_all_grant')
user_4_db = User.add_or_update(user_4_db)
self.users['custom_role_inquiry_all_grant'] = user_4_db
user_5_db = UserDB(name='custom_role_inquiry_inherit')
user_5_db = User.add_or_update(user_5_db)
self.users['custom_role_inquiry_inherit'] = user_5_db
# Create a workflow for testing inheritance of action_execute permission
# to inquiry_respond permission
wf_db = ActionDB(pack='examples', name='mistral-ask-basic', entry_point='',
runner_type={'name': 'mistral-v2'})
wf_db = Action.add_or_update(wf_db)
self.resources['wf'] = wf_db
runner = {'name': 'mistral-v2'}
liveaction = {'action': 'examples.mistral-ask-basic'}
status = action_constants.LIVEACTION_STATUS_PAUSED
# Spawn workflow
action = {'uid': wf_db.get_uid(), 'pack': 'examples'}
wf_exc_db = ActionExecutionDB(action=action, runner=runner, liveaction=liveaction,
status=status)
wf_exc_db = ActionExecution.add_or_update(wf_exc_db)
# Create an Inquiry on which permissions can be granted
action_1_db = ActionDB(pack='core', name='ask', entry_point='',
runner_type={'name': 'inquirer'})
action_1_db = Action.add_or_update(action_1_db)
self.resources['action_1'] = action_1_db
runner = {'name': 'inquirer'}
liveaction = {'action': 'core.ask'}
status = action_constants.LIVEACTION_STATUS_PENDING
# For now, Inquiries are "borrowing" the ActionExecutionDB model,
# so we have to test with that model
action = {'uid': action_1_db.get_uid(), 'pack': 'core'}
inquiry_1_db = ActionExecutionDB(action=action, runner=runner, liveaction=liveaction,
status=status)
# A separate inquiry that has a parent (so we can test workflow permission inheritance)
inquiry_2_db = ActionExecutionDB(action=action, runner=runner, liveaction=liveaction,
status=status, parent=str(wf_exc_db.id))
# A bit gross, but it's what we have to do since Inquiries
# don't yet have their own data model
def get_uid():
return "inquiry"
inquiry_1_db.get_uid = get_uid
inquiry_2_db.get_uid = get_uid
inquiry_1_db = ActionExecution.add_or_update(inquiry_1_db)
inquiry_2_db = ActionExecution.add_or_update(inquiry_2_db)
self.resources['inquiry_1'] = inquiry_1_db
self.resources['inquiry_2'] = inquiry_2_db
############################################################
# Create some mock roles with associated permission grants #
############################################################
# Custom role - "inquiry_list" grant
grant_db = PermissionGrantDB(resource_uid=self.resources['inquiry_1'].get_uid(),
resource_type=ResourceType.INQUIRY,
permission_types=[PermissionType.INQUIRY_LIST])
grant_db = PermissionGrant.add_or_update(grant_db)
permission_grants = [str(grant_db.id)]
role_db = RoleDB(name='custom_role_inquiry_list_grant',
permission_grants=permission_grants)
role_db = Role.add_or_update(role_db)
self.roles['custom_role_inquiry_list_grant'] = role_db
# Custom role - "inquiry_view" grant
grant_db = PermissionGrantDB(resource_uid=self.resources['inquiry_1'].get_uid(),
resource_type=ResourceType.INQUIRY,
permission_types=[PermissionType.INQUIRY_VIEW])
grant_db = PermissionGrant.add_or_update(grant_db)
permission_grants = [str(grant_db.id)]
role_db = RoleDB(name='custom_role_inquiry_view_grant',
permission_grants=permission_grants)
role_db = Role.add_or_update(role_db)
self.roles['custom_role_inquiry_view_grant'] = role_db
# Custom role - "inquiry_respond" grant
grant_db = PermissionGrantDB(resource_uid=self.resources['inquiry_1'].get_uid(),
#.........这里部分代码省略.........
开发者ID:lyandut，项目名称:st2，代码行数:101，代码来源:test_rbac_resolvers_inquiry.py
示例16: setUp
def setUp(self):
super(ExecutionPermissionsResolverTestCase, self).setUp()
# Create some mock users
user_1_db = UserDB(name='custom_role_unrelated_pack_action_grant')
user_1_db = User.add_or_update(user_1_db)
self.users['custom_role_unrelated_pack_action_grant'] = user_1_db
user_2_db = UserDB(name='custom_role_pack_action_grant_unrelated_permission')
user_2_db = User.add_or_update(user_2_db)
self.users['custom_role_pack_action_grant_unrelated_permission'] = user_2_db
user_3_db = UserDB(name='custom_role_pack_action_view_grant')
user_3_db = User.add_or_update(user_3_db)
self.users['custom_role_pack_action_view_grant'] = user_3_db
user_4_db = UserDB(name='custom_role_action_view_grant')
user_4_db = User.add_or_update(user_4_db)
self.users['custom_role_action_view_grant'] = user_4_db
user_5_db = UserDB(name='custom_role_pack_action_execute_grant')
user_5_db = User.add_or_update(user_5_db)
self.users['custom_role_pack_action_execute_grant'] = user_5_db
user_6_db = UserDB(name='custom_role_action_execute_grant')
user_6_db = User.add_or_update(user_6_db)
self.users['custom_role_action_execute_grant'] = user_6_db
user_7_db = UserDB(name='custom_role_pack_action_all_grant')
user_7_db = User.add_or_update(user_7_db)
self.users['custom_role_pack_action_all_grant'] = user_7_db
user_8_db = UserDB(name='custom_role_action_all_grant')
user_8_db = User.add_or_update(user_8_db)
self.users['custom_role_action_all_grant'] = user_8_db
# Create some mock resources on which permissions can be granted
action_1_db = ActionDB(pack='test_pack_2', name='action1', entry_point='',
runner_type={'name': 'run-local'})
action_1_db = Action.add_or_update(action_1_db)
self.resources['action_1'] = action_1_db
runner = {'name': 'run-python'}
liveaction = {'action': 'test_pack_2.action1'}
status = action_constants.LIVEACTION_STATUS_REQUESTED
action = {'uid': action_1_db.get_uid(), 'pack': 'test_pack_2'}
exec_1_db = ActionExecutionDB(action=action, runner=runner, liveaction=liveaction,
status=status)
exec_1_db = ActionExecution.add_or_update(exec_1_db)
self.resources['exec_1'] = exec_1_db
# Create some mock roles with associated permission grants
# Custom role - one grant to an unrelated pack
grant_db = PermissionGrantDB(resource_uid=self.resources['pack_1'].get_uid(),
resource_type=ResourceType.PACK,
permission_types=[PermissionType.ACTION_VIEW])
grant_db = PermissionGrant.add_or_update(grant_db)
permission_grants = [str(grant_db.id)]
role_db = RoleDB(name='custom_role_unrelated_pack_action_grant',
permission_grants=permission_grants)
role_db = Role.add_or_update(role_db)
self.roles['custom_role_unrelated_pack_action_grant'] = role_db
# Custom role - one grant of unrelated permission type to parent action pack
grant_db = PermissionGrantDB(resource_uid=self.resources['pack_2'].get_uid(),
resource_type=ResourceType.PACK,
permission_types=[PermissionType.RULE_VIEW])
grant_db = PermissionGrant.add_or_update(grant_db)
permission_grants = [str(grant_db.id)]
role_db = RoleDB(name='custom_role_pack_action_grant_unrelated_permission',
permission_grants=permission_grants)
role_db = Role.add_or_update(role_db)
self.roles['custom_role_pack_action_grant_unrelated_permission'] = role_db
# Custom role - one grant of "action_view" to the parent pack of the action the execution
# belongs to
grant_db = PermissionGrantDB(resource_uid=self.resources['pack_2'].get_uid(),
resource_type=ResourceType.PACK,
permission_types=[PermissionType.ACTION_VIEW])
grant_db = PermissionGrant.add_or_update(grant_db)
permission_grants = [str(grant_db.id)]
role_db = RoleDB(name='custom_role_pack_action_view_grant',
permission_grants=permission_grants)
role_db = Role.add_or_update(role_db)
self.roles['custom_role_pack_action_view_grant'] = role_db
# Custom role - one grant of "action_view" to the action the execution belongs to
grant_db = PermissionGrantDB(resource_uid=self.resources['action_1'].get_uid(),
resource_type=ResourceType.ACTION,
permission_types=[PermissionType.ACTION_VIEW])
grant_db = PermissionGrant.add_or_update(grant_db)
permission_grants = [str(grant_db.id)]
role_db = RoleDB(name='custom_role_action_view_grant',
permission_grants=permission_grants)
role_db = Role.add_or_update(role_db)
self.roles['custom_role_action_view_grant'] = role_db
# Custom role - one grant of "action_execute" to the parent pack of the action the
# execution belongs to
#.........这里部分代码省略.........
开发者ID:agilee，项目名称:st2，代码行数:101，代码来源:test_rbac_resolvers_execution.py
示例17: setup_action_models
def setup_action_models(cls):
action_db = ActionDB()
action_db.name = 'action-1'
action_db.description = 'awesomeness'
action_db.enabled = True
action_db.pack = 'wolfpack'
action_db.entry_point = ''
action_db.runner_type = {'name': 'test-runner'}
action_db.parameters = {
'actionstr': {'type': 'string', 'required': True},
'actionint': {'type': 'number', 'default': 10},
'runnerdummy': {'type': 'string', 'default': 'actiondummy'},
'runnerimmutable': {'type': 'string', 'default': 'failed_override'},
'actionimmutable': {'type': 'string', 'default': 'actionimmutable', 'immutable': True}
}
RunnerContainerTest.action_db = Action.add_or_update(action_db)
action_db = ActionDB()
action_db.name = 'action-2'
action_db.description = 'awesomeness'
action_db.enabled = True
action_db.pack = 'wolfpack'
action_db.entry_point = ''
action_db.runner_type = {'name': 'test-failingrunner'}
action_db.parameters = {}
RunnerContainerTest.failingaction_db = Action.add_or_update(action_db)
开发者ID:ojacques，项目名称:st2，代码行数:26，代码来源:test_runner_container.py
示例18: TriggerDB
from st2common.util import reference
from st2reactor.rules.enforcer import RuleEnforcer
import st2tests.config as tests_config
MOCK_TRIGGER = TriggerDB()
MOCK_TRIGGER.id = 'trigger-test.id'
MOCK_TRIGGER.name = 'trigger-test.name'
MOCK_TRIGGER.pack = 'dummypack1'
MOCK_TRIGGER_INSTANCE = TriggerInstanceDB()
MOCK_TRIGGER_INSTANCE.id = 'triggerinstance-test'
MOCK_TRIGGER_INSTANCE.trigger = reference.get_ref_from_model(MOCK_TRIGGER)
MOCK_TRIGGER_INSTANCE.payload = {}
MOCK_TRIGGER_INSTANCE.occurrence_time = datetime.datetime.utcnow()
MOCK_ACTION = ActionDB()
MOCK_ACTION.id = 'action-test-1.id'
MOCK_ACTION.name = 'action-test-1.name'
MOCK_ACTION_EXECUTION = ActionExecutionDB()
MOCK_ACTION_EXECUTION.id = 'actionexec-test-1.id'
MOCK_ACTION_EXECUTION.name = 'actionexec-test-1.name'
MOCK_ACTION_EXECUTION.status = 'scheduled'
MOCK_RULE_1 = RuleDB()
MOCK_RULE_1.id = 'rule-test-1'
MOCK_RULE_1.trigger = reference.get_str_resource_ref_from_model(MOCK_TRIGGER)
MOCK_RULE_1.criteria = {}
MOCK_RULE_1.action = ActionExecutionSpecDB()
MOCK_RULE_1.action.ref = reference.get_ref_from_model(MOCK_ACTION)
MOCK_RULE_1.enabled = True
开发者ID:bjoernbessert，项目名称:st2，代码行数:31，代码来源:test_enforce.py
示例19: TriggerDB
from st2tests import DbTestCase
MOCK_TRIGGER = TriggerDB()
MOCK_TRIGGER.id = bson.ObjectId()
MOCK_TRIGGER.name = 'trigger-test.name'
MOCK_TRIGGER.pack = 'dummy_pack_1'
MOCK_TRIGGER.type = 'system.test'
MOCK_TRIGGER_INSTANCE = TriggerInstanceDB()
MOCK_TRIGGER_INSTANCE.id = bson.ObjectId()
MOCK_TRIGGER_INSTANCE.trigger = MOCK_TRIGGER.get_reference().ref
MOCK_TRIGGER_INSTANCE.payload = {'p1': 'v1'}
MOCK_TRIGGER_INSTANCE.occurrence_time = datetime.datetime.utcnow()
MOCK_ACTION = ActionDB()
MOCK_ACTION.id = bson.ObjectId()
MOCK_ACTION.name = 'action-test-1.name'
MOCK_RULE_1 = RuleDB()
MOCK_RULE_1.id = bson.ObjectId()
MOCK_RULE_1.name = "some1"
MOCK_RULE_1.trigger = reference.get_str_resource_ref_from_model(MOCK_TRIGGER)
MOCK_RULE_1.criteria = {}
MOCK_RULE_1.action = ActionExecutionSpecDB(ref="somepack.someaction")
MOCK_RULE_2 = RuleDB()
MOCK_RULE_2.id = bson.ObjectId()
MOCK_RULE_1.name = "some2"
MOCK_RULE_2.trigger = reference.get_str_resource_ref_from_model(MOCK_TRIGGER)
MOCK_RULE_2.criteria = {}
开发者ID:bjoernbessert，项目名称:st2，代码行数:31，代码来源:test_filter.py
注：本文中的st2common.models.db.action.ActionDB类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论