本文整理汇总了Python中st2tests.fixturesloader.get_fixtures_packs_base_path函数的典型用法代码示例。如果您正苦于以下问题:Python get_fixtures_packs_base_path函数的具体用法?Python get_fixtures_packs_base_path怎么用?Python get_fixtures_packs_base_path使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了get_fixtures_packs_base_path函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_register_from_pack_action_metadata_fails_validation
def test_register_from_pack_action_metadata_fails_validation(self):
# No fail on failure flag, should succeed
pack_dir = os.path.join(get_fixtures_packs_base_path(), 'dummy_pack_4')
runner_dirs = os.path.join(get_fixtures_packs_base_path(), 'runners')
opts = [
'--register-pack=%s' % (pack_dir),
'--register-no-fail-on-failure',
'--register-runner-dir=%s' % (runner_dirs),
]
cmd = BASE_REGISTER_ACTIONS_CMD_ARGS + opts
exit_code, _, stderr = run_command(cmd=cmd)
self.assertTrue('Registered 0 actions.' in stderr)
self.assertEqual(exit_code, 0)
# Fail on failure, should fail
pack_dir = os.path.join(get_fixtures_packs_base_path(), 'dummy_pack_4')
opts = [
'--register-pack=%s' % (pack_dir),
'--register-fail-on-failure',
'--register-runner-dir=%s' % (runner_dirs),
]
cmd = BASE_REGISTER_ACTIONS_CMD_ARGS + opts
exit_code, _, stderr = run_command(cmd=cmd)
self.assertTrue('object has no attribute \'get\'' in stderr)
self.assertEqual(exit_code, 1)
开发者ID:StackStorm,项目名称:st2,代码行数:27,代码来源:test_register_content_script.py
示例2: test_get_pack_resource_file_abs_path
def test_get_pack_resource_file_abs_path(self):
# Mock the packs path to point to the fixtures directory
cfg.CONF.content.packs_base_paths = get_fixtures_packs_base_path()
# Invalid resource type
expected_msg = 'Invalid resource type: fooo'
self.assertRaisesRegexp(ValueError, expected_msg, get_pack_resource_file_abs_path,
pack_ref='dummy_pack_1',
resource_type='fooo',
file_path='test.py')
# Invalid paths (directory traversal and absolute paths)
file_paths = ['/tmp/foo.py', '../foo.py', '/etc/passwd', '../../foo.py']
for file_path in file_paths:
expected_msg = 'Invalid file path: .*%s' % (file_path)
self.assertRaisesRegexp(ValueError, expected_msg, get_pack_resource_file_abs_path,
pack_ref='dummy_pack_1',
resource_type='action',
file_path=file_path)
# Valid paths
file_paths = ['foo.py', 'a/foo.py', 'a/b/foo.py']
for file_path in file_paths:
expected = os.path.join(get_fixtures_packs_base_path(),
'dummy_pack_1/actions', file_path)
result = get_pack_resource_file_abs_path(pack_ref='dummy_pack_1',
resource_type='action',
file_path=file_path)
self.assertEqual(result, expected)
开发者ID:peak6,项目名称:st2,代码行数:29,代码来源:test_content_utils.py
示例3: test_register_from_pack_success
def test_register_from_pack_success(self):
pack_dir = os.path.join(get_fixtures_packs_base_path(), 'dummy_pack_1')
runner_dirs = os.path.join(get_fixtures_packs_base_path(), 'runners')
opts = [
'--register-pack=%s' % (pack_dir),
'--register-runner-dir=%s' % (runner_dirs),
]
cmd = BASE_REGISTER_ACTIONS_CMD_ARGS + opts
exit_code, _, stderr = run_command(cmd=cmd)
self.assertTrue('Registered 1 actions.' in stderr)
self.assertEqual(exit_code, 0)
开发者ID:StackStorm,项目名称:st2,代码行数:12,代码来源:test_register_content_script.py
示例4: test_get_pack_resource_file_abs_path
def test_get_pack_resource_file_abs_path(self):
# Mock the packs path to point to the fixtures directory
cfg.CONF.content.packs_base_paths = get_fixtures_packs_base_path()
# Invalid resource type
expected_msg = 'Invalid resource type: fooo'
self.assertRaisesRegexp(ValueError, expected_msg, get_pack_resource_file_abs_path,
pack_ref='dummy_pack_1',
resource_type='fooo',
file_path='test.py')
# Invalid paths (directory traversal and absolute paths)
file_paths = ['/tmp/foo.py', '../foo.py', '/etc/passwd', '../../foo.py',
'/opt/stackstorm/packs/invalid_pack/actions/my_action.py',
'../../foo.py']
for file_path in file_paths:
# action resource_type
expected_msg = (r'Invalid file path: ".*%s"\. File path needs to be relative to the '
r'pack actions directory (.*). For example "my_action.py"\.' %
(file_path))
self.assertRaisesRegexp(ValueError, expected_msg, get_pack_resource_file_abs_path,
pack_ref='dummy_pack_1',
resource_type='action',
file_path=file_path)
# sensor resource_type
expected_msg = (r'Invalid file path: ".*%s"\. File path needs to be relative to the '
r'pack sensors directory (.*). For example "my_sensor.py"\.' %
(file_path))
self.assertRaisesRegexp(ValueError, expected_msg, get_pack_resource_file_abs_path,
pack_ref='dummy_pack_1',
resource_type='sensor',
file_path=file_path)
# no resource type
expected_msg = (r'Invalid file path: ".*%s"\. File path needs to be relative to the '
r'pack directory (.*). For example "my_action.py"\.' %
(file_path))
self.assertRaisesRegexp(ValueError, expected_msg, get_pack_file_abs_path,
pack_ref='dummy_pack_1',
file_path=file_path)
# Valid paths
file_paths = ['foo.py', 'a/foo.py', 'a/b/foo.py']
for file_path in file_paths:
expected = os.path.join(get_fixtures_packs_base_path(),
'dummy_pack_1/actions', file_path)
result = get_pack_resource_file_abs_path(pack_ref='dummy_pack_1',
resource_type='action',
file_path=file_path)
self.assertEqual(result, expected)
开发者ID:nzlosh,项目名称:st2,代码行数:51,代码来源:test_content_utils.py
示例5: test_register_triggers_from_pack
def test_register_triggers_from_pack(self):
base_path = get_fixtures_packs_base_path()
pack_dir = os.path.join(base_path, 'dummy_pack_1')
trigger_type_dbs = TriggerType.get_all()
self.assertEqual(len(trigger_type_dbs), 0)
count = triggers_registrar.register_triggers(pack_dir=pack_dir)
self.assertEqual(count, 2)
# Verify TriggerTypeDB and corresponding TriggerDB objects have been created
trigger_type_dbs = TriggerType.get_all()
trigger_dbs = Trigger.get_all()
self.assertEqual(len(trigger_type_dbs), 2)
self.assertEqual(len(trigger_dbs), 2)
self.assertEqual(trigger_type_dbs[0].name, 'event_handler')
self.assertEqual(trigger_type_dbs[0].pack, 'dummy_pack_1')
self.assertEqual(trigger_dbs[0].name, 'event_handler')
self.assertEqual(trigger_dbs[0].pack, 'dummy_pack_1')
self.assertEqual(trigger_dbs[0].type, 'dummy_pack_1.event_handler')
self.assertEqual(trigger_type_dbs[1].name, 'head_sha_monitor')
self.assertEqual(trigger_type_dbs[1].pack, 'dummy_pack_1')
self.assertEqual(trigger_type_dbs[1].payload_schema['type'], 'object')
开发者ID:lyandut,项目名称:st2,代码行数:25,代码来源:test_triggers_registrar.py
示例6: test_register_all_policies
def test_register_all_policies(self):
policies_dbs = Policy.get_all()
self.assertEqual(len(policies_dbs), 0)
packs_base_path = get_fixtures_packs_base_path()
count = policies_registrar.register_policies(packs_base_paths=[packs_base_path])
# Verify PolicyDB objects have been created
policies_dbs = Policy.get_all()
policies = {
policies_db.name: {
'pack': policies_db.pack,
'type': policies_db.policy_type,
'parameters': policies_db.parameters
}
for policies_db in policies_dbs
}
expected_policies = {
'test_policy_1': {
'pack': 'dummy_pack_1',
'type': 'action.concurrency',
'parameters': {
'action': 'delay',
'threshold': 3
}
},
'test_policy_3': {
'pack': 'dummy_pack_1',
'type': 'action.retry',
'parameters': {
'retry_on': 'timeout',
'max_retry_count': 5
}
},
'cancel_on_concurrency': {
'pack': 'mistral_tests',
'type': 'action.concurrency',
'parameters': {
'action': 'cancel',
'threshold': 3
}
},
'cancel_on_concurrency_by_attr': {
'pack': 'mistral_tests',
'type': 'action.concurrency.attr',
'parameters': {
'action': 'cancel',
'threshold': 1,
'attributes': ['friend']
}
}
}
self.assertEqual(len(expected_policies), count)
self.assertEqual(len(expected_policies), len(policies_dbs))
self.assertDictEqual(expected_policies, policies)
开发者ID:lyandut,项目名称:st2,代码行数:58,代码来源:test_policies_registrar.py
示例7: _override_common_opts
def _override_common_opts():
packs_base_path = get_fixtures_packs_base_path()
CONF.set_override(name='base_path', override=packs_base_path, group='system')
CONF.set_override(name='system_packs_base_path', override=packs_base_path, group='content')
CONF.set_override(name='packs_base_paths', override=packs_base_path, group='content')
CONF.set_override(name='api_url', override='http://127.0.0.1', group='auth')
CONF.set_override(name='mask_secrets', override=True, group='log')
CONF.set_override(name='url', override='zake://', group='coordination')
CONF.set_override(name='lock_timeout', override=1, group='coordination')
开发者ID:Gagaro,项目名称:st2,代码行数:9,代码来源:config.py
示例8: test_register_policy_invalid_policy_type_references
def test_register_policy_invalid_policy_type_references(self):
# Policy references an invalid (inexistent) policy type
registrar = PolicyRegistrar()
policy_path = os.path.join(get_fixtures_packs_base_path(),
'dummy_pack_1/policies/policy_2.yaml')
expected_msg = 'Referenced policy_type "action.mock_policy_error" doesnt exist'
self.assertRaisesRegexp(ValueError, expected_msg, registrar._register_policy,
pack='dummy_pack_1', policy=policy_path)
开发者ID:LindsayHill,项目名称:st2,代码行数:9,代码来源:test_policies_registrar.py
示例9: test_make_sure_policy_parameters_are_validated_during_register
def test_make_sure_policy_parameters_are_validated_during_register(self):
# Policy where specified parameters fail schema validation
registrar = PolicyRegistrar()
policy_path = os.path.join(get_fixtures_packs_base_path(),
'dummy_pack_2/policies/policy_3.yaml')
expected_msg = '100 is greater than the maximum of 5'
self.assertRaisesRegexp(jsonschema.ValidationError, expected_msg,
registrar._register_policy, pack='dummy_pack_2',
policy=policy_path)
开发者ID:LindsayHill,项目名称:st2,代码行数:10,代码来源:test_policies_registrar.py
示例10: test_register_setup_virtualenvs
def test_register_setup_virtualenvs(self):
# Single pack
pack_dir = os.path.join(get_fixtures_packs_base_path(), 'dummy_pack_1')
cmd = BASE_CMD_ARGS + ['--register-pack=%s' % (pack_dir), '--register-setup-virtualenvs',
'--register-no-fail-on-failure']
exit_code, stdout, stderr = run_command(cmd=cmd)
self.assertTrue('Setting up virtualenv for pack "dummy_pack_1"' in stderr)
self.assertTrue('Setup virtualenv for 1 pack(s)' in stderr)
self.assertEqual(exit_code, 0)
开发者ID:StackStorm,项目名称:st2,代码行数:11,代码来源:test_register_content_script.py
示例11: _override_common_opts
def _override_common_opts():
packs_base_path = get_fixtures_packs_base_path()
CONF.set_override(name='base_path', override=packs_base_path, group='system')
CONF.set_override(name='validate_output_schema', override=True, group='system')
CONF.set_override(name='system_packs_base_path', override=packs_base_path, group='content')
CONF.set_override(name='packs_base_paths', override=packs_base_path, group='content')
CONF.set_override(name='api_url', override='http://127.0.0.1', group='auth')
CONF.set_override(name='mask_secrets', override=True, group='log')
CONF.set_override(name='jitter_interval', override=0, group='mistral')
CONF.set_override(name='query_interval', override=0.1, group='resultstracker')
CONF.set_override(name='stream_output', override=False, group='actionrunner')
开发者ID:StackStorm,项目名称:st2,代码行数:11,代码来源:config.py
示例12: test_register_all_triggers
def test_register_all_triggers(self):
trigger_type_dbs = TriggerType.get_all()
self.assertEqual(len(trigger_type_dbs), 0)
packs_base_path = get_fixtures_packs_base_path()
count = triggers_registrar.register_triggers(packs_base_paths=[packs_base_path])
self.assertEqual(count, 2)
# Verify TriggerTypeDB and corresponding TriggerDB objects have been created
trigger_type_dbs = TriggerType.get_all()
trigger_dbs = Trigger.get_all()
self.assertEqual(len(trigger_type_dbs), 2)
self.assertEqual(len(trigger_dbs), 2)
开发者ID:lyandut,项目名称:st2,代码行数:13,代码来源:test_triggers_registrar.py
示例13: test_register_policies
def test_register_policies(self):
# Note: Only one policy should be registered since second one fails validation
pack_dir = os.path.join(fixturesloader.get_fixtures_packs_base_path(), 'dummy_pack_1')
self.assertEqual(register_policies(pack_dir=pack_dir), 1)
p1 = Policy.get_by_ref('dummy_pack_1.test_policy_1')
self.assertEqual(p1.name, 'test_policy_1')
self.assertEqual(p1.pack, 'dummy_pack_1')
self.assertEqual(p1.resource_ref, 'dummy_pack_1.local')
self.assertEqual(p1.policy_type, 'action.concurrency')
p2 = Policy.get_by_ref('dummy_pack_1.test_policy_2')
self.assertEqual(p2, None)
开发者ID:Bala96,项目名称:st2,代码行数:13,代码来源:test_policies.py
示例14: test_register_policies_from_pack
def test_register_policies_from_pack(self):
pack_dir = os.path.join(get_fixtures_packs_base_path(), 'dummy_pack_1')
self.assertEqual(register_policies(pack_dir=pack_dir), 2)
p1 = Policy.get_by_ref('dummy_pack_1.test_policy_1')
self.assertEqual(p1.name, 'test_policy_1')
self.assertEqual(p1.pack, 'dummy_pack_1')
self.assertEqual(p1.resource_ref, 'dummy_pack_1.local')
self.assertEqual(p1.policy_type, 'action.concurrency')
# Verify that a default value for parameter "action" which isn't provided in the file is set
self.assertEqual(p1.parameters['action'], 'delay')
p2 = Policy.get_by_ref('dummy_pack_1.test_policy_2')
self.assertEqual(p2, None)
开发者ID:LindsayHill,项目名称:st2,代码行数:14,代码来源:test_policies_registrar.py
示例15: test_register_all_and_register_setup_virtualenvs
def test_register_all_and_register_setup_virtualenvs(self):
# Verify that --register-all works in combinations with --register-setup-virtualenvs
# Single pack
pack_dir = os.path.join(get_fixtures_packs_base_path(), 'dummy_pack_1')
cmd = BASE_CMD_ARGS + [
'--register-pack=%s' % (pack_dir),
'--register-all',
'--register-setup-virtualenvs',
'--register-no-fail-on-failure'
]
exit_code, stdout, stderr = run_command(cmd=cmd)
self.assertTrue('Registering actions' in stderr, 'Actual stderr: %s' % (stderr))
self.assertTrue('Registering rules' in stderr)
self.assertTrue('Setup virtualenv for %s pack(s)' % ('1') in stderr)
self.assertEqual(exit_code, 0)
开发者ID:StackStorm,项目名称:st2,代码行数:15,代码来源:test_register_content_script.py
示例16: test_get_pack_resource_file_abs_path
def test_get_pack_resource_file_abs_path(self):
# Mock the packs path to point to the fixtures directory
cfg.CONF.content.packs_base_paths = get_fixtures_packs_base_path()
# Invalid resource type
expected_msg = "Invalid resource type: fooo"
self.assertRaisesRegexp(
ValueError,
expected_msg,
get_pack_resource_file_abs_path,
pack_name="dummy_pack_1",
resource_type="fooo",
file_path="test.py",
)
# Invalid paths (directory traversal and absolute paths)
file_paths = ["/tmp/foo.py", "../foo.py", "/etc/passwd", "../../foo.py"]
for file_path in file_paths:
expected_msg = "Invalid file path: .*%s" % (file_path)
self.assertRaisesRegexp(
ValueError,
expected_msg,
get_pack_resource_file_abs_path,
pack_name="dummy_pack_1",
resource_type="action",
file_path=file_path,
)
# Valid paths
file_paths = ["foo.py", "a/foo.py", "a/b/foo.py"]
for file_path in file_paths:
expected = os.path.join(get_fixtures_packs_base_path(), "dummy_pack_1/actions", file_path)
result = get_pack_resource_file_abs_path(
pack_name="dummy_pack_1", resource_type="action", file_path=file_path
)
self.assertEqual(result, expected)
开发者ID:rlugojr,项目名称:st2,代码行数:36,代码来源:test_content_utils.py
示例17: test_register_all_policies
def test_register_all_policies(self):
policies_dbs = Policy.get_all()
self.assertEqual(len(policies_dbs), 0)
packs_base_path = get_fixtures_packs_base_path()
count = policies_registrar.register_policies(packs_base_paths=[packs_base_path])
self.assertEqual(count, 2)
# Verify PolicyDB objects have been created
policies_dbs = Policy.get_all()
self.assertEqual(len(policies_dbs), 2)
self.assertEqual(policies_dbs[0].name, 'test_policy_1')
self.assertEqual(policies_dbs[0].policy_type, 'action.concurrency')
# Verify that a default value for parameter "action" which isn't provided in the file is set
self.assertEqual(policies_dbs[0].parameters['action'], 'delay')
self.assertEqual(policies_dbs[0].parameters['threshold'], 3)
开发者ID:LindsayHill,项目名称:st2,代码行数:18,代码来源:test_policies_registrar.py
示例18: test_get_relative_path_to_pack_file
def test_get_relative_path_to_pack_file(self):
packs_base_paths = get_fixtures_packs_base_path()
pack_ref = 'dummy_pack_1'
# 1. Valid paths
file_path = os.path.join(packs_base_paths, 'dummy_pack_1/pack.yaml')
result = get_relative_path_to_pack_file(pack_ref=pack_ref, file_path=file_path)
self.assertEqual(result, 'pack.yaml')
file_path = os.path.join(packs_base_paths, 'dummy_pack_1/actions/action.meta.yaml')
result = get_relative_path_to_pack_file(pack_ref=pack_ref, file_path=file_path)
self.assertEqual(result, 'actions/action.meta.yaml')
file_path = os.path.join(packs_base_paths, 'dummy_pack_1/actions/lib/foo.py')
result = get_relative_path_to_pack_file(pack_ref=pack_ref, file_path=file_path)
self.assertEqual(result, 'actions/lib/foo.py')
# Already relative
file_path = 'actions/lib/foo2.py'
result = get_relative_path_to_pack_file(pack_ref=pack_ref, file_path=file_path)
self.assertEqual(result, 'actions/lib/foo2.py')
# 2. Invalid path - outside pack directory
expected_msg = r'file_path (.*?) is not located inside the pack directory (.*?)'
file_path = os.path.join(packs_base_paths, 'dummy_pack_2/actions/lib/foo.py')
self.assertRaisesRegexp(ValueError, expected_msg, get_relative_path_to_pack_file,
pack_ref=pack_ref, file_path=file_path)
file_path = '/tmp/foo/bar.py'
self.assertRaisesRegexp(ValueError, expected_msg, get_relative_path_to_pack_file,
pack_ref=pack_ref, file_path=file_path)
file_path = os.path.join(packs_base_paths, '../dummy_pack_1/pack.yaml')
self.assertRaisesRegexp(ValueError, expected_msg, get_relative_path_to_pack_file,
pack_ref=pack_ref, file_path=file_path)
file_path = os.path.join(packs_base_paths, '../../dummy_pack_1/pack.yaml')
self.assertRaisesRegexp(ValueError, expected_msg, get_relative_path_to_pack_file,
pack_ref=pack_ref, file_path=file_path)
开发者ID:nzlosh,项目名称:st2,代码行数:41,代码来源:test_content_utils.py
示例19: test_register_from_pack_fail_on_failure_pack_dir_doesnt_exist
def test_register_from_pack_fail_on_failure_pack_dir_doesnt_exist(self):
# No fail on failure flag, should succeed
pack_dir = 'doesntexistblah'
runner_dirs = os.path.join(get_fixtures_packs_base_path(), 'runners')
opts = [
'--register-pack=%s' % (pack_dir),
'--register-runner-dir=%s' % (runner_dirs),
'--register-no-fail-on-failure'
]
cmd = BASE_REGISTER_ACTIONS_CMD_ARGS + opts
exit_code, _, _ = run_command(cmd=cmd)
self.assertEqual(exit_code, 0)
# Fail on failure, should fail
opts = [
'--register-pack=%s' % (pack_dir),
'--register-runner-dir=%s' % (runner_dirs),
'--register-fail-on-failure'
]
cmd = BASE_REGISTER_ACTIONS_CMD_ARGS + opts
exit_code, _, stderr = run_command(cmd=cmd)
self.assertTrue('Directory "doesntexistblah" doesn\'t exist' in stderr)
self.assertEqual(exit_code, 1)
开发者ID:StackStorm,项目名称:st2,代码行数:24,代码来源:test_register_content_script.py
示例20: UnloadActionTestCase
from st2common.persistence.rule import Rule
from st2common.persistence.policy import Policy
from st2common.persistence.sensor import SensorType as Sensor
from st2common.persistence.trigger import TriggerType
from st2tests.base import BaseActionTestCase
from st2tests.base import CleanDbTestCase
from st2tests import fixturesloader
from pack_mgmt.unload import UnregisterPackAction
__all__ = [
'UnloadActionTestCase'
]
PACK_PATH_1 = os.path.join(fixturesloader.get_fixtures_packs_base_path(), 'dummy_pack_1')
class UnloadActionTestCase(BaseActionTestCase, CleanDbTestCase):
action_cls = UnregisterPackAction
def setUp(self):
super(UnloadActionTestCase, self).setUp()
# Register mock pack
# Verify DB is empty
pack_dbs = Pack.get_all()
config_schema_dbs = ConfigSchema.get_all()
config_dbs = Config.get_all()
self.assertEqual(len(pack_dbs), 0)
开发者ID:lyandut,项目名称:st2,代码行数:31,代码来源:test_action_unload.py
注:本文中的st2tests.fixturesloader.get_fixtures_packs_base_path函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论