本文整理汇总了Python中st2common.persistence.keyvalue.KeyValuePair类的典型用法代码示例。如果您正苦于以下问题:Python KeyValuePair类的具体用法?Python KeyValuePair怎么用?Python KeyValuePair使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了KeyValuePair类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_lookup_cast
def test_lookup_cast(self):
KeyValuePair.add_or_update(KeyValuePairDB(name='count', value='5.5'))
lookup = KeyValueLookup(scope=FULL_SYSTEM_SCOPE)
self.assertEqual(str(lookup.count), '5.5')
self.assertEqual(float(lookup.count), 5.5)
self.assertEqual(int(lookup.count), 5)
开发者ID:StackStorm,项目名称:st2,代码行数:7,代码来源:test_keyvalue_lookup.py
示例2: test_user_scope_lookups_user_sep_in_name
def test_user_scope_lookups_user_sep_in_name(self):
KeyValuePair.add_or_update(KeyValuePairDB(name='stanley:r:i:p', value='v4',
scope=FULL_USER_SCOPE))
lookup = UserKeyValueLookup(scope=FULL_USER_SCOPE, user='stanley')
# This is the only way to lookup because USER_SEPARATOR (':') cannot be a part of
# variable name in Python.
self.assertEquals(str(lookup['r:i:p']), 'v4')
开发者ID:StackStorm,项目名称:st2,代码行数:7,代码来源:test_keyvalue_lookup.py
示例3: test_get_config_dynamic_config_item_list
def test_get_config_dynamic_config_item_list(self):
pack_name = 'dummy_pack_schema_with_nested_object_7'
loader = ContentPackConfigLoader(pack_name=pack_name)
KeyValuePair.add_or_update(KeyValuePairDB(name='k0', value='v0'))
KeyValuePair.add_or_update(KeyValuePairDB(name='k1', value='v1'))
####################
# values in list
values = {
'level0_key': [
'a',
'{{st2kv.system.k0}}',
'b',
'{{st2kv.system.k1}}',
]
}
config_db = ConfigDB(pack=pack_name, values=values)
config_db = Config.add_or_update(config_db)
config_rendered = loader.get_config()
self.assertEquals(config_rendered,
{
'level0_key': [
'a',
'v0',
'b',
'v1'
]
})
config_db.delete()
开发者ID:StackStorm,项目名称:st2,代码行数:33,代码来源:test_config_loader.py
示例4: test_get_finalized_params_system_values
def test_get_finalized_params_system_values(self):
KeyValuePair.add_or_update(KeyValuePairDB(name='actionstr', value='foo'))
KeyValuePair.add_or_update(KeyValuePairDB(name='actionnumber', value='1.0'))
params = {
'runnerint': 555
}
liveaction_db = self._get_liveaction_model(params)
runner_params, action_params = param_utils.get_finalized_params(
ParamsUtilsTest.runnertype_db.runner_parameters,
ParamsUtilsTest.action_system_default_db.parameters,
liveaction_db.parameters,
liveaction_db.context)
# Asserts for runner params.
# Assert that default values for runner params are resolved.
self.assertEqual(runner_params.get('runnerstr'), 'defaultfoo')
# Assert that a runner param from action exec is picked up.
self.assertEqual(runner_params.get('runnerint'), 555)
# Assert that an immutable param cannot be overridden by action param or execution param.
self.assertEqual(runner_params.get('runnerimmutable'), 'runnerimmutable')
# Asserts for action params.
self.assertEqual(action_params.get('actionstr'), 'foo')
self.assertEqual(action_params.get('actionnumber'), 1.0)
开发者ID:ruslantum,项目名称:st2,代码行数:25,代码来源:test_param_utils.py
示例5: delete
def delete(self, name):
"""
Delete the key value pair.
Handles requests:
DELETE /keys/1
"""
lock_name = self._get_lock_name_for_key(name=name)
# Note: We use lock to avoid a race
with self._coordinator.get_lock(lock_name):
kvp_db = self._get_by_name(resource_name=name)
if not kvp_db:
abort(http_client.NOT_FOUND)
return
LOG.debug('DELETE /keys/ lookup with name=%s found object: %s', name, kvp_db)
try:
KeyValuePair.delete(kvp_db)
except Exception as e:
LOG.exception('Database delete encountered exception during '
'delete of name="%s". ', name)
abort(http_client.INTERNAL_SERVER_ERROR, str(e))
return
extra = {'kvp_db': kvp_db}
LOG.audit('KeyValuePair deleted. KeyValuePair.id=%s' % (kvp_db.id), extra=extra)
开发者ID:MohammadHabbab,项目名称:st2,代码行数:29,代码来源:keyvalue.py
示例6: test_non_hierarchical_lookup
def test_non_hierarchical_lookup(self):
k1 = KeyValuePair.add_or_update(KeyValuePairDB(name='k1', value='v1'))
k2 = KeyValuePair.add_or_update(KeyValuePairDB(name='k2', value='v2'))
k3 = KeyValuePair.add_or_update(KeyValuePairDB(name='k3', value='v3'))
lookup = KeyValueLookup()
self.assertEquals(str(lookup.k1), k1.value)
self.assertEquals(str(lookup.k2), k2.value)
self.assertEquals(str(lookup.k3), k3.value)
开发者ID:MohammadHabbab,项目名称:st2,代码行数:9,代码来源:test_keyvalue_lookup.py
示例7: test_hierarchical_lookup_dict
def test_hierarchical_lookup_dict(self):
k1 = KeyValuePair.add_or_update(KeyValuePairDB(name='a.b', value='v1'))
k2 = KeyValuePair.add_or_update(KeyValuePairDB(name='a.b.c', value='v2'))
k3 = KeyValuePair.add_or_update(KeyValuePairDB(name='b.c', value='v3'))
lookup = KeyValueLookup()
self.assertEquals(str(lookup['a']['b']), k1.value)
self.assertEquals(str(lookup['a']['b']['c']), k2.value)
self.assertEquals(str(lookup['b']['c']), k3.value)
self.assertEquals(str(lookup['a']), '')
开发者ID:MohammadHabbab,项目名称:st2,代码行数:10,代码来源:test_keyvalue_lookup.py
示例8: test_lookups_older_scope_names_backward_compatibility
def test_lookups_older_scope_names_backward_compatibility(self):
k1 = KeyValuePair.add_or_update(KeyValuePairDB(name='a.b', value='v1',
scope=FULL_SYSTEM_SCOPE))
lookup = KeyValueLookup(scope=SYSTEM_SCOPE)
self.assertEquals(str(lookup['a']['b']), k1.value)
k2 = KeyValuePair.add_or_update(KeyValuePairDB(name='stanley:r.i.p', value='v4',
scope=FULL_USER_SCOPE))
user_lookup = UserKeyValueLookup(scope=USER_SCOPE, user='stanley')
self.assertEquals(str(user_lookup['r']['i']['p']), k2.value)
开发者ID:StackStorm,项目名称:st2,代码行数:10,代码来源:test_keyvalue_lookup.py
示例9: setUp
def setUp(self):
super(KeyValuesControllerRBACTestCase, self).setUp()
self.kvps = {}
# Insert mock users
user_1_db = UserDB(name='user1')
user_1_db = User.add_or_update(user_1_db)
self.users['user_1'] = user_1_db
user_2_db = UserDB(name='user2')
user_2_db = User.add_or_update(user_2_db)
self.users['user_2'] = user_2_db
# Insert mock kvp objects
kvp_api = KeyValuePairSetAPI(name='test_system_scope', value='value1',
scope=FULL_SYSTEM_SCOPE)
kvp_db = KeyValuePairSetAPI.to_model(kvp_api)
kvp_db = KeyValuePair.add_or_update(kvp_db)
kvp_db = KeyValuePairAPI.from_model(kvp_db)
self.kvps['kvp_1'] = kvp_db
kvp_api = KeyValuePairSetAPI(name='test_system_scope_secret', value='value_secret',
scope=FULL_SYSTEM_SCOPE, secret=True)
kvp_db = KeyValuePairSetAPI.to_model(kvp_api)
kvp_db = KeyValuePair.add_or_update(kvp_db)
kvp_db = KeyValuePairAPI.from_model(kvp_db)
self.kvps['kvp_2'] = kvp_db
name = get_key_reference(scope=FULL_USER_SCOPE, name='test_user_scope_1', user='user1')
kvp_db = KeyValuePairDB(name=name, value='valueu12', scope=FULL_USER_SCOPE)
kvp_db = KeyValuePair.add_or_update(kvp_db)
kvp_db = KeyValuePairAPI.from_model(kvp_db)
self.kvps['kvp_3'] = kvp_db
name = get_key_reference(scope=FULL_USER_SCOPE, name='test_user_scope_2', user='user1')
kvp_api = KeyValuePairSetAPI(name=name, value='user_secret', scope=FULL_USER_SCOPE,
secret=True)
kvp_db = KeyValuePairSetAPI.to_model(kvp_api)
kvp_db = KeyValuePair.add_or_update(kvp_db)
kvp_db = KeyValuePairAPI.from_model(kvp_db)
self.kvps['kvp_4'] = kvp_db
name = get_key_reference(scope=FULL_USER_SCOPE, name='test_user_scope_3', user='user2')
kvp_db = KeyValuePairDB(name=name, value='valueu21', scope=FULL_USER_SCOPE)
kvp_db = KeyValuePair.add_or_update(kvp_db)
kvp_db = KeyValuePairAPI.from_model(kvp_db)
self.kvps['kvp_5'] = kvp_db
self.system_scoped_items_count = 2
self.user_scoped_items_count = 3
self.user_scoped_items_per_user_count = {
'user1': 2,
'user2': 1
}
开发者ID:lyandut,项目名称:st2,代码行数:55,代码来源:test_kvps_rbac.py
示例10: test_lookup_with_key_prefix
def test_lookup_with_key_prefix(self):
KeyValuePair.add_or_update(KeyValuePairDB(name='some:prefix:stanley:k5', value='v5',
scope=FULL_USER_SCOPE))
# No prefix provided, should return None
lookup = UserKeyValueLookup(user='stanley', scope=FULL_USER_SCOPE)
self.assertEqual(str(lookup.k5), '')
# Prefix provided
lookup = UserKeyValueLookup(prefix='some:prefix', user='stanley', scope=FULL_USER_SCOPE)
self.assertEqual(str(lookup.k5), 'v5')
开发者ID:StackStorm,项目名称:st2,代码行数:11,代码来源:test_keyvalue_lookup.py
示例11: delete
def delete(self, name, requester_user, scope=FULL_SYSTEM_SCOPE, user=None):
"""
Delete the key value pair.
Handles requests:
DELETE /keys/1
"""
if not scope:
scope = FULL_SYSTEM_SCOPE
if not requester_user:
requester_user = UserDB(cfg.CONF.system_user.user)
scope = get_datastore_full_scope(scope)
self._validate_scope(scope=scope)
user = user or requester_user.name
# Validate that the authenticated user is admin if user query param is provided
rbac_utils = get_rbac_backend().get_utils_class()
rbac_utils.assert_user_is_admin_if_user_query_param_is_provided(user_db=requester_user,
user=user,
require_rbac=True)
key_ref = get_key_reference(scope=scope, name=name, user=user)
lock_name = self._get_lock_name_for_key(name=key_ref, scope=scope)
# Note: We use lock to avoid a race
with self._coordinator.get_lock(lock_name):
from_model_kwargs = {'mask_secrets': True}
kvp_api = self._get_one_by_scope_and_name(
name=key_ref,
scope=scope,
from_model_kwargs=from_model_kwargs
)
kvp_db = KeyValuePairAPI.to_model(kvp_api)
LOG.debug('DELETE /keys/ lookup with scope=%s name=%s found object: %s',
scope, name, kvp_db)
try:
KeyValuePair.delete(kvp_db)
except Exception as e:
LOG.exception('Database delete encountered exception during '
'delete of name="%s". ', name)
abort(http_client.INTERNAL_SERVER_ERROR, six.text_type(e))
return
extra = {'kvp_db': kvp_db}
LOG.audit('KeyValuePair deleted. KeyValuePair.id=%s' % (kvp_db.id), extra=extra)
return Response(status=http_client.NO_CONTENT)
开发者ID:StackStorm,项目名称:st2,代码行数:53,代码来源:keyvalue.py
示例12: test_secret_lookup
def test_secret_lookup(self):
secret_value = '0055A2D9A09E1071931925933744965EEA7E23DCF59A8D1D7A3' + \
'64338294916D37E83C4796283C584751750E39844E2FD97A3727DB5D553F638'
k1 = KeyValuePair.add_or_update(KeyValuePairDB(
name='k1', value=secret_value,
secret=True, encrypted=True)
)
k2 = KeyValuePair.add_or_update(KeyValuePairDB(name='k2', value='v2'))
lookup = KeyValueLookup()
self.assertEquals(str(lookup.k1), k1.value)
self.assertEquals(str(lookup.k2), k2.value)
开发者ID:MohammadHabbab,项目名称:st2,代码行数:12,代码来源:test_keyvalue_lookup.py
示例13: setUp
def setUp(self):
super(KeyValuesControllerRBACTestCase, self).setUp()
self.kvps = {}
# Insert mock users
user_1_db = UserDB(name="user1")
user_1_db = User.add_or_update(user_1_db)
self.users["user_1"] = user_1_db
user_2_db = UserDB(name="user2")
user_2_db = User.add_or_update(user_2_db)
self.users["user_2"] = user_2_db
# Insert mock kvp objects
kvp_api = KeyValuePairSetAPI(name="test_system_scope", value="value1", scope=FULL_SYSTEM_SCOPE)
kvp_db = KeyValuePairSetAPI.to_model(kvp_api)
kvp_db = KeyValuePair.add_or_update(kvp_db)
kvp_db = KeyValuePairAPI.from_model(kvp_db)
self.kvps["kvp_1"] = kvp_db
kvp_api = KeyValuePairSetAPI(
name="test_system_scope_secret", value="value_secret", scope=FULL_SYSTEM_SCOPE, secret=True
)
kvp_db = KeyValuePairSetAPI.to_model(kvp_api)
kvp_db = KeyValuePair.add_or_update(kvp_db)
kvp_db = KeyValuePairAPI.from_model(kvp_db)
self.kvps["kvp_2"] = kvp_db
name = get_key_reference(scope=FULL_USER_SCOPE, name="test_user_scope_1", user="user1")
kvp_db = KeyValuePairDB(name=name, value="valueu12", scope=FULL_USER_SCOPE)
kvp_db = KeyValuePair.add_or_update(kvp_db)
kvp_db = KeyValuePairAPI.from_model(kvp_db)
self.kvps["kvp_3"] = kvp_db
name = get_key_reference(scope=FULL_USER_SCOPE, name="test_user_scope_2", user="user1")
kvp_api = KeyValuePairSetAPI(name=name, value="user_secret", scope=FULL_USER_SCOPE, secret=True)
kvp_db = KeyValuePairSetAPI.to_model(kvp_api)
kvp_db = KeyValuePair.add_or_update(kvp_db)
kvp_db = KeyValuePairAPI.from_model(kvp_db)
self.kvps["kvp_4"] = kvp_db
name = get_key_reference(scope=FULL_USER_SCOPE, name="test_user_scope_3", user="user2")
kvp_db = KeyValuePairDB(name=name, value="valueu21", scope=FULL_USER_SCOPE)
kvp_db = KeyValuePair.add_or_update(kvp_db)
kvp_db = KeyValuePairAPI.from_model(kvp_db)
self.kvps["kvp_5"] = kvp_db
self.system_scoped_items_count = 2
self.user_scoped_items_count = 3
self.user_scoped_items_per_user_count = {"user1": 2, "user2": 1}
开发者ID:Pulsant,项目名称:st2,代码行数:51,代码来源:test_kvps_rbac.py
示例14: setUp
def setUp(self):
super(TemplatingUtilsTestCase, self).setUp()
# Insert mock DB objects
kvp_1_db = KeyValuePairDB(name='key1', value='valuea')
kvp_1_db = KeyValuePair.add_or_update(kvp_1_db)
kvp_2_db = KeyValuePairDB(name='key2', value='valueb')
kvp_2_db = KeyValuePair.add_or_update(kvp_2_db)
kvp_3_db = KeyValuePairDB(name='stanley:key1', value='valuestanley1', scope=USER_SCOPE)
kvp_3_db = KeyValuePair.add_or_update(kvp_3_db)
kvp_4_db = KeyValuePairDB(name='joe:key1', value='valuejoe1', scope=USER_SCOPE)
kvp_4_db = KeyValuePair.add_or_update(kvp_4_db)
开发者ID:E-LLP,项目名称:st2,代码行数:15,代码来源:test_util_templating.py
示例15: put
def put(self, name, kvp):
"""
Create a new entry or update an existing one.
"""
# TODO: There is a race, add custom add_or_update which updates by non
# id field
existing_kvp = self.__get_by_name(name=name)
kvp.name = name
try:
kvp_db = KeyValuePairAPI.to_model(kvp)
if existing_kvp:
kvp_db.id = existing_kvp.id
kvp_db = KeyValuePair.add_or_update(kvp_db)
except (ValidationError, ValueError) as e:
LOG.exception('Validation failed for key value data=%s', kvp)
abort(http_client.BAD_REQUEST, str(e))
return
extra = {'kvp_db': kvp_db}
LOG.audit('KeyValuePair updated. KeyValuePair.id=%s' % (kvp_db.id), extra=extra)
kvp_api = KeyValuePairAPI.from_model(kvp_db)
return kvp_api
开发者ID:Kailashkatheth1,项目名称:st2,代码行数:27,代码来源:keyvalue.py
示例16: test_filter_decrypt_kv_with_user_scope_value
def test_filter_decrypt_kv_with_user_scope_value(self):
KeyValuePair.add_or_update(KeyValuePairDB(name='stanley:k8', value=self.secret_value,
scope=FULL_USER_SCOPE,
secret=True))
context = {}
context.update({USER_SCOPE: UserKeyValueLookup(user='stanley', scope=USER_SCOPE)})
context.update({
DATASTORE_PARENT_SCOPE: {
USER_SCOPE: UserKeyValueLookup(user='stanley', scope=FULL_USER_SCOPE)
}
})
template = '{{st2kv.user.k8 | decrypt_kv}}'
actual = self.env.from_string(template).render(context)
self.assertEqual(actual, self.secret)
开发者ID:StackStorm,项目名称:st2,代码行数:16,代码来源:test_jinja_render_crypto_filters.py
示例17: test_filter_decrypt_kv
def test_filter_decrypt_kv(self):
KeyValuePair.add_or_update(KeyValuePairDB(name='k8', value=self.secret_value,
scope=FULL_SYSTEM_SCOPE,
secret=True))
context = {}
context.update({SYSTEM_SCOPE: KeyValueLookup(scope=SYSTEM_SCOPE)})
context.update({
DATASTORE_PARENT_SCOPE: {
SYSTEM_SCOPE: KeyValueLookup(scope=FULL_SYSTEM_SCOPE)
}
})
template = '{{st2kv.system.k8 | decrypt_kv}}'
actual = self.env.from_string(template).render(context)
self.assertEqual(actual, self.secret)
开发者ID:StackStorm,项目名称:st2,代码行数:16,代码来源:test_jinja_render_crypto_filters.py
示例18: _get_kv
def _get_kv(self, key):
scope = self._scope
LOG.debug('Lookup system kv: scope: %s and key: %s', scope, key)
kvp = KeyValuePair.get_by_scope_and_name(scope=scope, name=key)
if kvp:
LOG.debug('Got value %s from datastore.', kvp.value)
return kvp.value if kvp else ''
开发者ID:mahak,项目名称:st2,代码行数:7,代码来源:keyvalues.py
示例19: get_kvp_for_name
def get_kvp_for_name(name):
try:
kvp_db = KeyValuePair.get_by_name(name)
except ValueError:
kvp_db = None
return kvp_db
开发者ID:mahak,项目名称:st2,代码行数:7,代码来源:keyvalues.py
示例20: put
def put(self, kvp, name, scope=FULL_SYSTEM_SCOPE):
"""
Create a new entry or update an existing one.
"""
if not scope:
scope = FULL_SYSTEM_SCOPE
requester_user = get_requester()
scope = getattr(kvp, 'scope', scope)
scope = get_datastore_full_scope(scope)
self._validate_scope(scope=scope)
user = getattr(kvp, 'user', requester_user) or requester_user
# Validate that the authenticated user is admin if user query param is provided
assert_request_user_is_admin_if_user_query_param_is_provider(request=pecan.request,
user=user)
key_ref = get_key_reference(scope=scope, name=name, user=user)
lock_name = self._get_lock_name_for_key(name=key_ref, scope=scope)
LOG.debug('PUT scope: %s, name: %s', scope, name)
# TODO: Custom permission check since the key doesn't need to exist here
# Note: We use lock to avoid a race
with self._coordinator.get_lock(lock_name):
try:
existing_kvp_api = self._get_one_by_scope_and_name(
scope=scope,
name=key_ref
)
except StackStormDBObjectNotFoundError:
existing_kvp_api = None
kvp.name = key_ref
kvp.scope = scope
try:
kvp_db = KeyValuePairAPI.to_model(kvp)
if existing_kvp_api:
kvp_db.id = existing_kvp_api.id
kvp_db = KeyValuePair.add_or_update(kvp_db)
except (ValidationError, ValueError) as e:
LOG.exception('Validation failed for key value data=%s', kvp)
abort(http_client.BAD_REQUEST, str(e))
return
except CryptoKeyNotSetupException as e:
LOG.exception(str(e))
abort(http_client.BAD_REQUEST, str(e))
return
except InvalidScopeException as e:
LOG.exception(str(e))
abort(http_client.BAD_REQUEST, str(e))
return
extra = {'kvp_db': kvp_db}
LOG.audit('KeyValuePair updated. KeyValuePair.id=%s' % (kvp_db.id), extra=extra)
kvp_api = KeyValuePairAPI.from_model(kvp_db)
return kvp_api
开发者ID:LindsayHill,项目名称:st2,代码行数:60,代码来源:keyvalue.py
注:本文中的st2common.persistence.keyvalue.KeyValuePair类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论