本文整理汇总了Python中solar.dblayer.solar_models.Resource类的典型用法代码示例。如果您正苦于以下问题：Python Resource类的具体用法？Python Resource怎么用？Python Resource使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了Resource类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_nested_simple_listdict
def test_nested_simple_listdict(rk):
k1 = next(rk)
k2 = next(rk)
k3 = next(rk)
k4 = next(rk)
k5 = next(rk)
r1 = create_resource(k1, {'name': 'first',
'inputs': {'config': [{"backends": [{}],
'listen_port': 1}]}})
r2 = create_resource(k2, {'name': 'second', 'inputs': {'backend': {}}})
r3 = create_resource(k3, {'name': 'third', 'inputs': {'backend': {}}})
r5 = create_resource(k5, {'name': 'fifth',
'inputs': {"port": 5,
"host": "fifth_host"}})
r4 = create_resource(k4, {'name': 'fourth',
'inputs': {"port": 4,
"host": "fourth_host"}})
r4.connect(r2, {'port': "backend:port", 'host': 'backend:host'})
r5.connect(r3, {'port': "backend:port", 'host': 'backend:host'})
assert r2.inputs['backend'] == {'host': 'fourth_host', 'port': 4}
assert r3.inputs['backend'] == {'host': 'fifth_host', 'port': 5}
r2.connect(r1, {'backend': 'config:backends'})
r3.connect(r1, {'backend': 'config:backends'})
Resource.save_all_lazy()
backends = next(x['backends'] for x in r1.inputs['config']
if 'backends' in x)
assert len(backends) == 2
开发者ID:aglarendil，项目名称:solar，代码行数:33，代码来源:test_real.py
示例2: test_discard_all_pending_changes_resources_created
def test_discard_all_pending_changes_resources_created():
res1 = DBResource.from_dict(
"test1",
{
"name": "test1",
"base_path": "x",
"state": RESOURCE_STATE.created.name,
"meta_inputs": {"a": {"value": None, "schema": "str"}},
},
)
res1.inputs["a"] = "9"
res1.save_lazy()
res2 = DBResource.from_dict(
"test2",
{
"name": "test2",
"base_path": "x",
"state": RESOURCE_STATE.created.name,
"meta_inputs": {"a": {"value": None, "schema": "str"}},
},
)
res2.inputs["a"] = "0"
res2.save_lazy()
ModelMeta.save_all_lazy()
staged_log = change.stage_changes()
assert len(staged_log) == 2
change.discard_all()
staged_log = change.stage_changes()
assert len(staged_log) == 0
assert resource.load_all() == []
开发者ID:aglarendil，项目名称:solar，代码行数:33，代码来源:test_system_log_api.py
示例3: test_discard_connection
def test_discard_connection():
res1 = DBResource.from_dict('test1',
{'name': 'test1',
'base_path': 'x',
'state': RESOURCE_STATE.created.name,
'meta_inputs': {'a': {'value': None,
'schema': 'str'}}})
res1.inputs['a'] = '9'
res1.save_lazy()
res2 = DBResource.from_dict('test2',
{'name': 'test2',
'base_path': 'x',
'state': RESOURCE_STATE.created.name,
'meta_inputs': {'a': {'value': None,
'schema': 'str'}}})
res2.inputs['a'] = '0'
res2.save_lazy()
ModelMeta.save_all_lazy()
staged_log = change.stage_changes()
for item in staged_log:
operations.move_to_commited(item.log_action)
res1 = resource.load('test1')
res2 = resource.load('test2')
res1.connect(res2, {'a': 'a'})
staged_log = change.stage_changes()
assert len(staged_log) == 1
assert res2.args == {'a': '9'}
change.discard_all()
assert res2.args == {'a': '0'}
assert len(change.stage_changes()) == 0
开发者ID:pigmej，项目名称:solar，代码行数:33，代码来源:test_system_log_api.py
示例4: test_discard_all_pending_changes_resources_created
def test_discard_all_pending_changes_resources_created():
res1 = DBResource.from_dict('test1',
{'name': 'test1',
'base_path': 'x',
'state': RESOURCE_STATE.created.name,
'meta_inputs': {'a': {'value': None,
'schema': 'str'}}})
res1.inputs['a'] = '9'
res1.save_lazy()
res2 = DBResource.from_dict('test2',
{'name': 'test2',
'base_path': 'x',
'state': RESOURCE_STATE.created.name,
'meta_inputs': {'a': {'value': None,
'schema': 'str'}}})
res2.inputs['a'] = '0'
res2.save_lazy()
ModelMeta.save_all_lazy()
staged_log = change.stage_changes()
assert len(staged_log) == 2
change.discard_all()
staged_log = change.stage_changes()
assert len(staged_log) == 0
assert resource.load_all() == []
开发者ID:pigmej，项目名称:solar，代码行数:27，代码来源:test_system_log_api.py
示例5: test_nested_two_listdict
def test_nested_two_listdict(rk):
k1 = next(rk)
k2 = next(rk)
k3 = next(rk)
r1 = create_resource(k1, {'name': 'first',
'inputs': {'config': [{"backends": [{}],
'something': 0}]}})
r2 = create_resource(k2, {'name': 'second',
'inputs': {"backends": [{"host": "second_host",
"port": 2}],
'something': 1}})
r3 = create_resource(k3, {'name': 'third',
'inputs': {"backends": [{"host": "third_host",
"port": 3}],
'something': 2}})
r2.connect(r1, {'backends': 'config:backends',
'something': 'config:something'})
r3.connect(r1, {'backends': 'config:backends',
'something': 'config:something'})
Resource.save_all_lazy()
for sc in r1.inputs['config']:
assert 'something' in sc
assert 'backends' in sc
assert isinstance(sc['backends'], list)
assert isinstance(sc['something'], int)
开发者ID:aglarendil，项目名称:solar，代码行数:29，代码来源:test_real.py
示例6: load_updated
def load_updated(since=None, with_childs=True):
if since is None:
startkey = StrInt.p_min()
else:
startkey = since
candids = DBResource.updated.filter(startkey, StrInt.p_max())
if with_childs:
candids = DBResource.childs(candids)
return [Resource(r) for r in DBResource.multi_get(candids)]
开发者ID:Mirantis，项目名称:solar，代码行数:9，代码来源:resource.py
示例7: test_connect_other_list
def test_connect_other_list(rk):
k1 = next(rk)
k2 = next(rk)
r1 = create_resource(k1, {'name': 'first',
'inputs': {'config': {"trackers": []}}})
r2 = create_resource(k2, {'name': 'second',
'inputs': {"trackers": ["t1", "t2"]}})
r2.connect(r1, {'trackers': 'config:trackers'})
Resource.save_all_lazy()
assert r1.inputs['config']['trackers'] == ["t1", "t2"]
开发者ID:Mirantis，项目名称:solar，代码行数:12，代码来源:test_real.py
示例8: tagged_resources
def tagged_resources():
tags = ['n1', 'n2', 'n3']
t1 = Resource.from_dict('t1',
{'name': 't1', 'tags': tags, 'base_path': 'x'})
t1.save_lazy()
t2 = Resource.from_dict('t2',
{'name': 't2', 'tags': tags, 'base_path': 'x'})
t2.save_lazy()
t3 = Resource.from_dict('t3',
{'name': 't3', 'tags': tags, 'base_path': 'x'})
t3.save_lazy()
ModelMeta.save_all_lazy()
return [t1, t2, t3]
开发者ID:aglarendil，项目名称:solar，代码行数:13，代码来源:test_operations_with_tags.py
示例9: test_revert_update
def test_revert_update():
commit = {'a': '10'}
previous = {'a': '9'}
res = DBResource.from_dict('test1',
{'name': 'test1',
'base_path': 'x',
'meta_inputs': {'a': {'value': None,
'schema': 'str'}}})
res.save()
action = 'update'
res.inputs['a'] = '9'
resource_obj = resource.load(res.name)
assert resource_obj.args == previous
log = data.SL()
logitem = change.create_logitem(res.name,
action,
change.create_diff(commit, previous),
[],
base_path=res.base_path)
log.append(logitem)
resource_obj.update(commit)
operations.move_to_commited(logitem.log_action)
assert logitem.diff == [['change', 'a', ['9', '10']]]
assert resource_obj.args == commit
change.revert(logitem.uid)
assert resource_obj.args == previous
开发者ID:pigmej，项目名称:solar，代码行数:30，代码来源:test_system_log_api.py
示例10: test_revert_create
def test_revert_create():
res = DBResource.from_dict('test1',
{'name': 'test1',
'base_path': 'x',
'state': RESOURCE_STATE.created.name,
'meta_inputs': {'a': {'value': None,
'schema': 'str'}}})
res.inputs['a'] = '9'
res.save_lazy()
ModelMeta.save_all_lazy()
staged_log = change.stage_changes()
assert len(staged_log) == 1
logitem = staged_log[0]
operations.move_to_commited(logitem.log_action)
assert logitem.diff == [['add', '', [['a', '9']]]]
commited = CommitedResource.get('test1')
assert commited.inputs == {'a': '9'}
change.revert(logitem.uid)
staged_log = change.stage_changes()
assert len(staged_log) == 1
for item in staged_log:
operations.move_to_commited(item.log_action)
assert resource.load_all() == []
开发者ID:pigmej，项目名称:solar，代码行数:28，代码来源:test_system_log_api.py
示例11: test_remove_events
def test_remove_events(events_example):
r = Resource.from_dict(dict(key='e1', name='e1', base_path='x'))
r.save()
to_be_removed = events_example[2]
evapi.add_events('e1', events_example)
evapi.remove_event(to_be_removed)
assert to_be_removed not in evapi.all_events('e1')
开发者ID:loles，项目名称:solar-1，代码行数:7，代码来源:test_events.py
示例12: test_discard_removed
def test_discard_removed():
res1 = DBResource.from_dict(
"test1",
{
"name": "test1",
"base_path": "x",
"state": RESOURCE_STATE.created.name,
"meta_inputs": {"a": {"value": None, "schema": "str"}},
},
)
res1.inputs["a"] = "9"
res1.save_lazy()
ModelMeta.save_all_lazy()
staged_log = change.stage_changes()
for item in staged_log:
operations.move_to_commited(item.log_action)
res1 = resource.load("test1")
res1.remove()
assert len(change.stage_changes()) == 1
assert res1.to_be_removed()
change.discard_all()
assert len(change.stage_changes()) == 0
assert not resource.load("test1").to_be_removed()
开发者ID:aglarendil，项目名称:solar，代码行数:25，代码来源:test_system_log_api.py
示例13: load_by_tags
def load_by_tags(tags):
tags = set(tags)
candids_all = set()
for tag in tags:
candids = DBResource.tags.filter(tag)
candids_all.update(set(candids))
return [Resource(r) for r in DBResource.multi_get(candids_all)]
开发者ID:Mirantis，项目名称:solar，代码行数:7，代码来源:resource.py
示例14: test_revert_create
def test_revert_create():
res = DBResource.from_dict(
"test1",
{
"name": "test1",
"base_path": "x",
"state": RESOURCE_STATE.created.name,
"meta_inputs": {"a": {"value": None, "schema": "str"}},
},
)
res.inputs["a"] = "9"
res.save_lazy()
ModelMeta.save_all_lazy()
staged_log = change.stage_changes()
assert len(staged_log) == 1
logitem = staged_log[0]
operations.move_to_commited(logitem.log_action)
assert logitem.diff == [["add", "", [["a", "9"]]]]
commited = CommitedResource.get("test1")
assert commited.inputs == {"a": "9"}
change.revert(logitem.uid)
staged_log = change.stage_changes()
assert len(staged_log) == 1
for item in staged_log:
operations.move_to_commited(item.log_action)
assert resource.load_all() == []
开发者ID:aglarendil，项目名称:solar，代码行数:30，代码来源:test_system_log_api.py
示例15: load
def load(name):
r = DBResource.get(name)
if not r:
raise Exception('Resource {} does not exist in DB'.format(name))
return Resource(r)
开发者ID:Mirantis，项目名称:solar，代码行数:7，代码来源:resource.py
示例16: test_riak
def test_riak():
events = {
'riak_service1': [
evapi.React('riak_service1', 'run', 'success', 'riak_service2',
'run'),
evapi.React('riak_service1', 'run', 'success',
'riak_service3', 'run')
],
'riak_service3': [
evapi.React('riak_service3', 'join', 'success', 'riak_service1',
'commit'),
evapi.React('riak_service3', 'run', 'success', 'riak_service3',
'join')
],
'riak_service2': [
evapi.React('riak_service2', 'run', 'success', 'riak_service2',
'join'),
evapi.React('riak_service2', 'join', 'success', 'riak_service1',
'commit')
],
}
for name in events:
res = Resource.from_dict({'key': name, 'name': name})
res.save()
res.inputs.add_new('location_id', '1')
evapi.add_events(name, events[name])
changes_graph = nx.MultiDiGraph()
changes_graph.add_node('riak_service1.run')
evapi.build_edges(changes_graph, events)
assert set(changes_graph.predecessors('riak_service1.commit')) == {
'riak_service2.join', 'riak_service3.join'
}
开发者ID:loles，项目名称:solar-1，代码行数:34，代码来源:test_events.py
示例17: load_all
def load_all(startswith=None):
if startswith:
start = startswith
end = startswith + '~'
candids = DBResource.bucket.get_index("$key", start, end).results
else:
candids = DBResource.updated.filter(StrInt.p_min(), StrInt.p_max())
return [Resource(r) for r in DBResource.multi_get(candids)]
开发者ID:loles，项目名称:solar-1，代码行数:8，代码来源:resource.py
示例18: add_events
def add_events(resource, lst):
resource = Resource.get(resource)
events = resource.events
# TODO: currently we don't track mutable objects
events.extend([ev.to_dict() for ev in lst])
resource.events = events
# import pdb; pdb.settrace()
resource.save_lazy()
开发者ID:aglarendil，项目名称:solar，代码行数:8，代码来源:api.py
示例19: test_events
def test_events(rk):
k = next(rk)
r1 = Resource.from_dict(k, {'events': ['event1', 'event2']})
r1.save()
assert r1.events == ['event1', 'event2']
r1.events.pop()
assert r1.events == ['event1']
开发者ID:aglarendil，项目名称:solar，代码行数:8，代码来源:test_real.py
示例20: remove_event
def remove_event(ev):
to_remove = ev.to_dict()
resource = ev.parent
resource = Resource.get(resource)
# TODO: currently we don't track mutable objects
events = resource.events
events.remove(to_remove)
resource.events = events
resource.save_lazy()
开发者ID:aglarendil，项目名称:solar，代码行数:9，代码来源:api.py
注：本文中的solar.dblayer.solar_models.Resource类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论