本文整理汇总了Python中senlin.tests.unit.db.shared.create_cluster函数的典型用法代码示例。如果您正苦于以下问题:Python create_cluster函数的具体用法?Python create_cluster怎么用?Python create_cluster使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了create_cluster函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_cluster_get_by_name_diff_project
def test_cluster_get_by_name_diff_project(self):
cluster1 = shared.create_cluster(self.ctx, self.profile,
name='cluster_A',
project=UUID2)
shared.create_cluster(self.ctx, self.profile, name='cluster_B',
project=UUID2)
shared.create_cluster(self.ctx, self.profile, name='cluster_B',
project=UUID2)
res = db_api.cluster_get_by_name(self.ctx, 'cluster_A')
self.assertIsNone(res)
self.ctx.project = UUID3
self.assertIsNone(db_api.cluster_get_by_name(self.ctx,
'cluster_A'))
self.ctx.project = UUID2
res = db_api.cluster_get_by_name(self.ctx, 'cluster_A')
self.assertEqual(cluster1.id, res.id)
self.assertRaises(exception.MultipleChoices,
db_api.cluster_get_by_name,
self.ctx, 'cluster_B')
res = db_api.cluster_get_by_name(self.ctx, 'non-existent')
self.assertIsNone(res)
开发者ID:Alzon,项目名称:senlin,代码行数:27,代码来源:test_cluster_api.py
示例2: test_event_get_all_project_safe
def test_event_get_all_project_safe(self):
self.ctx.project = 'project_1'
cluster1 = shared.create_cluster(self.ctx, self.profile,
name='cluster1')
self.create_event(self.ctx, entity=cluster1)
self.ctx.project = 'project_2'
cluster2 = shared.create_cluster(self.ctx, self.profile,
name='cluster2')
self.create_event(self.ctx, entity=cluster2, action='CLUSTER_CREATE')
self.create_event(self.ctx, entity=cluster2, action='CLUSTER_DELETE')
# Default project_safe to true, only the last two events are visible
events = db_api.event_get_all(self.ctx)
self.assertEqual(2, len(events))
oids = [event.oid for event in events]
onames = [event.oname for event in events]
self.assertNotIn(cluster1.id, oids)
self.assertNotIn(cluster1.name, onames)
self.assertIn(cluster2.id, oids)
self.assertIn(cluster2.name, onames)
# Set project_safe to false, we should get all three events
events = db_api.event_get_all(self.ctx, project_safe=False)
self.assertEqual(3, len(events))
oids = [event.oid for event in events]
onames = [event.oname for event in events]
self.assertIn(cluster1.id, oids)
self.assertIn(cluster1.name, onames)
self.assertIn(cluster2.id, oids)
self.assertIn(cluster2.name, onames)
开发者ID:jonnary,项目名称:senlin,代码行数:32,代码来源:test_event_api.py
示例3: test_cluster_get_by_short_id
def test_cluster_get_by_short_id(self):
cid1 = 'same-part-unique-part'
cid2 = 'same-part-part-unique'
cluster1 = shared.create_cluster(self.ctx, self.profile,
id=cid1,
name='cluster-1')
cluster2 = shared.create_cluster(self.ctx, self.profile,
id=cid2,
name='cluster-2')
for x in range(len('same-part-')):
self.assertRaises(exception.MultipleChoices,
db_api.cluster_get_by_short_id,
self.ctx, cid1[:x])
res = db_api.cluster_get_by_short_id(self.ctx, cid1[:11])
self.assertEqual(cluster1.id, res.id)
res = db_api.cluster_get_by_short_id(self.ctx, cid2[:11])
self.assertEqual(cluster2.id, res.id)
res = db_api.cluster_get_by_short_id(self.ctx, 'non-existent')
self.assertIsNone(res)
ctx_new = utils.dummy_context(project='different_project_id')
res = db_api.cluster_get_by_short_id(ctx_new, cid1[:11])
self.assertIsNone(res)
开发者ID:onesafe,项目名称:senlin,代码行数:25,代码来源:test_cluster_api.py
示例4: test_cluster_get_all_returns_all_if_no_filters
def test_cluster_get_all_returns_all_if_no_filters(self):
shared.create_cluster(self.ctx, self.profile)
shared.create_cluster(self.ctx, self.profile)
filters = None
results = db_api.cluster_get_all(self.ctx, filters=filters)
self.assertEqual(2, len(results))
开发者ID:Alzon,项目名称:senlin,代码行数:8,代码来源:test_cluster_api.py
示例5: test_node_count_by_cluster
def test_node_count_by_cluster(self):
shared.create_cluster(self.ctx, self.profile)
shared.create_node(self.ctx, self.cluster, self.profile)
shared.create_node(self.ctx, self.cluster, self.profile)
res = db_api.node_count_by_cluster(self.ctx, self.cluster.id)
self.assertEqual(2, res)
开发者ID:onesafe,项目名称:senlin,代码行数:8,代码来源:test_node_api.py
示例6: test_event_get_all_filtered
def test_event_get_all_filtered(self):
cluster1 = shared.create_cluster(self.ctx, self.profile,
name='cluster1')
cluster2 = shared.create_cluster(self.ctx, self.profile,
name='cluster2')
self.create_event(self.ctx, entity=cluster1, action='CLUSTER_CREATE')
self.create_event(self.ctx, entity=cluster1, action='CLUSTER_DELETE')
self.create_event(self.ctx, entity=cluster2, action='CLUSTER_CREATE')
events = db_api.event_get_all_by_cluster(self.ctx, cluster1.id)
self.assertEqual(2, len(events))
# test filter by action
filters = {'action': 'CLUSTER_CREATE'}
events = db_api.event_get_all_by_cluster(self.ctx, cluster1.id,
filters=filters)
self.assertEqual(1, len(events))
self.assertEqual('CLUSTER_CREATE', events[0].action)
filters = {'action': 'CLUSTER_UPDATE'}
events = db_api.event_get_all_by_cluster(self.ctx, cluster1.id,
filters=filters)
self.assertEqual(0, len(events))
# test filter by oname
filters = {'oname': 'cluster1'}
events = db_api.event_get_all_by_cluster(self.ctx, cluster1.id,
filters=filters)
self.assertEqual(2, len(events))
self.assertEqual('cluster1', events[0].oname)
self.assertEqual('cluster1', events[1].oname)
filters = {'oname': 'cluster3'}
events = db_api.event_get_all_by_cluster(self.ctx, cluster1.id,
filters=filters)
self.assertEqual(0, len(events))
# test filter by otype
filters = {'otype': 'CLUSTER'}
events = db_api.event_get_all_by_cluster(self.ctx, cluster2.id,
filters=filters)
self.assertEqual(1, len(events))
self.assertEqual('CLUSTER', events[0].otype)
filters = {'otype': 'NODE'}
events = db_api.event_get_all_by_cluster(self.ctx, cluster2.id,
filters=filters)
self.assertEqual(0, len(events))
# test limit and marker
events_all = db_api.event_get_all_by_cluster(self.ctx, cluster1.id)
marker = events_all[0].id
expected = events_all[1].id
events = db_api.event_get_all_by_cluster(self.ctx, cluster1.id,
limit=1, marker=marker)
self.assertEqual(1, len(events))
self.assertEqual(expected, events[0].id)
开发者ID:jonnary,项目名称:senlin,代码行数:58,代码来源:test_event_api.py
示例7: test_cluster_count_all_show_nested
def test_cluster_count_all_show_nested(self):
cluster1 = shared.create_cluster(self.ctx, self.profile, name='c1')
shared.create_cluster(self.ctx, self.profile, name='c2',
parent=cluster1.id)
results = db_api.cluster_count_all(self.ctx)
self.assertEqual(1, results)
results = db_api.cluster_count_all(self.ctx, show_nested=True)
self.assertEqual(2, results)
开发者ID:Alzon,项目名称:senlin,代码行数:9,代码来源:test_cluster_api.py
示例8: test_node_count_by_cluster_admin_context
def test_node_count_by_cluster_admin_context(self):
shared.create_cluster(self.ctx, self.profile)
shared.create_node(self.ctx, self.cluster, self.profile)
shared.create_node(self.ctx, self.cluster, self.profile)
admin_ctx = utils.dummy_context(project='a_different_project',
is_admin=True)
res = db_api.node_count_by_cluster(admin_ctx, self.cluster.id,
project_safe=True)
self.assertEqual(2, res)
开发者ID:onesafe,项目名称:senlin,代码行数:10,代码来源:test_node_api.py
示例9: test_node_get_by_cluster_admin_context
def test_node_get_by_cluster_admin_context(self):
shared.create_cluster(self.ctx, self.profile)
node1 = shared.create_node(self.ctx, self.cluster, self.profile)
node2 = shared.create_node(self.ctx, self.cluster, self.profile)
admin_ctx = utils.dummy_context(project='a_different_project',
is_admin=True)
nodes = db_api.node_get_all_by_cluster(admin_ctx, self.cluster.id)
self.assertEqual(2, len(nodes))
self.assertEqual(set([node1.id, node2.id]),
set([nodes[0].id, nodes[1].id]))
开发者ID:onesafe,项目名称:senlin,代码行数:12,代码来源:test_node_api.py
示例10: test_event_get_all_by_cluster
def test_event_get_all_by_cluster(self):
cluster1 = shared.create_cluster(self.ctx, self.profile)
cluster2 = shared.create_cluster(self.ctx, self.profile)
node1_1 = shared.create_node(self.ctx, cluster1, self.profile)
node1_2 = shared.create_node(self.ctx, cluster1, self.profile)
node2_1 = shared.create_node(self.ctx, cluster2, self.profile)
node_orphan = shared.create_node(self.ctx, None, self.profile)
# 1 event for cluster 1
self.create_event(self.ctx, entity=cluster1)
events = db_api.event_get_all_by_cluster(self.ctx, cluster1.id)
self.assertEqual(1, len(events))
events = db_api.event_get_all_by_cluster(self.ctx, cluster2.id)
self.assertEqual(0, len(events))
# two more events for cluster 1, with one for an orphan node
self.create_event(self.ctx, entity=node1_1)
self.create_event(self.ctx, entity=node1_2)
self.create_event(self.ctx, entity=node_orphan)
events = db_api.event_get_all_by_cluster(self.ctx, cluster1.id)
self.assertEqual(3, len(events))
events = db_api.event_get_all_by_cluster(self.ctx, cluster2.id)
self.assertEqual(0, len(events))
# one more events for cluster 2, with one for an orphan node
self.create_event(self.ctx, entity=cluster2)
self.create_event(self.ctx, entity=node_orphan)
events = db_api.event_get_all_by_cluster(self.ctx, cluster1.id)
self.assertEqual(3, len(events))
events = db_api.event_get_all_by_cluster(self.ctx, cluster2.id)
self.assertEqual(1, len(events))
# two more events for cluster 2, with one for an orphan node
self.create_event(self.ctx, entity=node2_1)
self.create_event(self.ctx, entity=node2_1)
self.create_event(self.ctx, entity=node_orphan)
events = db_api.event_get_all_by_cluster(self.ctx, cluster1.id)
self.assertEqual(3, len(events))
events = db_api.event_get_all_by_cluster(self.ctx, cluster2.id)
self.assertEqual(3, len(events))
# two more events for cluster 1, with one for an orphan node
self.create_event(self.ctx, entity=cluster1)
self.create_event(self.ctx, entity=node1_1)
self.create_event(self.ctx, entity=node_orphan)
events = db_api.event_get_all_by_cluster(self.ctx, cluster1.id)
self.assertEqual(5, len(events))
events = db_api.event_get_all_by_cluster(self.ctx, cluster2.id)
self.assertEqual(3, len(events))
开发者ID:jonnary,项目名称:senlin,代码行数:53,代码来源:test_event_api.py
示例11: test_event_count_all_by_cluster_diff_project
def test_event_count_all_by_cluster_diff_project(self):
cluster1 = shared.create_cluster(self.ctx, self.profile)
cluster2 = shared.create_cluster(self.ctx, self.profile)
self.create_event(self.ctx, entity=cluster1)
self.create_event(self.ctx, entity=cluster2)
new_ctx = utils.dummy_context(project='a-different-project')
res = db_api.event_count_by_cluster(new_ctx, cluster1.id)
self.assertEqual(0, res)
res = db_api.event_count_by_cluster(new_ctx, cluster1.id,
project_safe=False)
self.assertEqual(1, res)
开发者ID:jonnary,项目名称:senlin,代码行数:12,代码来源:test_event_api.py
示例12: test_cluster_get_all_with_filters
def test_cluster_get_all_with_filters(self):
shared.create_cluster(self.ctx, self.profile, name='foo')
shared.create_cluster(self.ctx, self.profile, name='bar')
filters = {'name': ['bar', 'quux']}
results = db_api.cluster_get_all(self.ctx, filters=filters)
self.assertEqual(1, len(results))
self.assertEqual('bar', results[0]['name'])
filters = {'name': 'foo'}
results = db_api.cluster_get_all(self.ctx, filters=filters)
self.assertEqual(1, len(results))
self.assertEqual('foo', results[0]['name'])
开发者ID:Alzon,项目名称:senlin,代码行数:13,代码来源:test_cluster_api.py
示例13: test_node_count_by_cluster_diff_project
def test_node_count_by_cluster_diff_project(self):
ctx_new = utils.dummy_context(project='a_different_project')
shared.create_cluster(self.ctx, self.profile)
shared.create_node(self.ctx, self.cluster, self.profile)
shared.create_node(self.ctx, self.cluster, self.profile)
res = db_api.node_count_by_cluster(ctx_new, self.cluster.id)
self.assertEqual(0, res)
res = db_api.node_count_by_cluster(ctx_new, self.cluster.id,
project_safe=False)
self.assertEqual(2, res)
开发者ID:onesafe,项目名称:senlin,代码行数:13,代码来源:test_node_api.py
示例14: test_nested_cluster_get_by_name
def test_nested_cluster_get_by_name(self):
cluster1 = shared.create_cluster(self.ctx, self.profile,
name='cluster1')
cluster2 = shared.create_cluster(self.ctx, self.profile,
name='cluster2',
parent=cluster1.id)
result = db_api.cluster_get_by_name(self.ctx, 'cluster2')
self.assertEqual(cluster2.id, result.id)
db_api.cluster_delete(self.ctx, cluster2.id)
result = db_api.cluster_get_by_name(self.ctx, 'cluster2')
self.assertIsNone(result)
开发者ID:Alzon,项目名称:senlin,代码行数:13,代码来源:test_cluster_api.py
示例15: test_event_get_all_admin_context
def test_event_get_all_admin_context(self):
self.ctx.project = 'project_1'
cluster1 = shared.create_cluster(self.ctx, self.profile,
name='cluster1')
self.create_event(self.ctx, entity=cluster1)
self.ctx.project = 'project_2'
cluster2 = shared.create_cluster(self.ctx, self.profile,
name='cluster2')
self.create_event(self.ctx, entity=cluster2, action='CLUSTER_CREATE')
self.create_event(self.ctx, entity=cluster2, action='CLUSTER_DELETE')
admin_ctx = utils.dummy_context(project='another-project',
is_admin=True)
events = db_api.event_get_all(admin_ctx, project_safe=True)
self.assertEqual(3, len(events))
开发者ID:jonnary,项目名称:senlin,代码行数:15,代码来源:test_event_api.py
示例16: test_event_get_all_with_sort_keys_and_dir
def test_event_get_all_with_sort_keys_and_dir(self):
cluster1 = shared.create_cluster(self.ctx, self.profile)
event1 = self.create_event(self.ctx, entity=cluster1,
timestamp=tu.utcnow(),
action='action2')
event2 = self.create_event(self.ctx, entity=cluster1,
timestamp=tu.utcnow(),
action='action3')
event3 = self.create_event(self.ctx, entity=cluster1,
timestamp=tu.utcnow(),
action='action1')
events = db_api.event_get_all(self.ctx, sort_keys=['timestamp'])
self.assertEqual(event1.id, events[0].id)
self.assertEqual(event2.id, events[1].id)
self.assertEqual(event3.id, events[2].id)
events = db_api.event_get_all(self.ctx, sort_keys=['timestamp'],
sort_dir='desc')
self.assertEqual(event1.id, events[2].id)
self.assertEqual(event2.id, events[1].id)
self.assertEqual(event3.id, events[0].id)
events = db_api.event_get_all(self.ctx, sort_keys=['action'])
self.assertEqual(event1.id, events[1].id)
self.assertEqual(event2.id, events[2].id)
self.assertEqual(event3.id, events[0].id)
events = db_api.event_get_all(self.ctx, sort_keys=['action'],
sort_dir='desc')
self.assertEqual(event1.id, events[1].id)
self.assertEqual(event2.id, events[0].id)
self.assertEqual(event3.id, events[2].id)
开发者ID:KongJustin,项目名称:senlin,代码行数:34,代码来源:test_event_api.py
示例17: test_event_get_all_show_deleted
def test_event_get_all_show_deleted(self):
cluster1 = shared.create_cluster(self.ctx, self.profile)
cluster2 = shared.create_cluster(self.ctx, self.profile)
# Simulate deleted events by setting 'deleted_time' to not-None
now = tu.utcnow()
self.create_event(self.ctx, entity=cluster1, deleted_time=now)
self.create_event(self.ctx, entity=cluster1)
self.create_event(self.ctx, entity=cluster2, deleted_time=now)
# Default show_deleted is False
events = db_api.event_get_all(self.ctx)
self.assertEqual(1, len(events))
events = db_api.event_get_all(self.ctx, show_deleted=True)
self.assertEqual(3, len(events))
开发者ID:KongJustin,项目名称:senlin,代码行数:16,代码来源:test_event_api.py
示例18: test_event_get_all_with_sorting
def test_event_get_all_with_sorting(self):
cluster1 = shared.create_cluster(self.ctx, self.profile)
event1 = self.create_event(self.ctx, entity=cluster1,
timestamp=tu.utcnow(True),
action='action2')
event2 = self.create_event(self.ctx, entity=cluster1,
timestamp=tu.utcnow(True),
action='action3')
event3 = self.create_event(self.ctx, entity=cluster1,
timestamp=tu.utcnow(True),
action='action1')
events = db_api.event_get_all(self.ctx, sort='timestamp')
self.assertEqual(event1.id, events[0].id)
self.assertEqual(event2.id, events[1].id)
self.assertEqual(event3.id, events[2].id)
events = db_api.event_get_all(self.ctx, sort='timestamp:desc')
self.assertEqual(event1.id, events[2].id)
self.assertEqual(event2.id, events[1].id)
self.assertEqual(event3.id, events[0].id)
events = db_api.event_get_all(self.ctx, sort='action')
self.assertEqual(event1.id, events[1].id)
self.assertEqual(event2.id, events[2].id)
self.assertEqual(event3.id, events[0].id)
events = db_api.event_get_all(self.ctx, sort='action:desc')
self.assertEqual(event1.id, events[1].id)
self.assertEqual(event2.id, events[0].id)
self.assertEqual(event3.id, events[2].id)
开发者ID:jonnary,项目名称:senlin,代码行数:32,代码来源:test_event_api.py
示例19: test_event_get_all_with_limit_and_marker
def test_event_get_all_with_limit_and_marker(self):
cluster1 = shared.create_cluster(self.ctx, self.profile)
self.create_event(self.ctx, entity=cluster1)
self.create_event(self.ctx, entity=cluster1)
self.create_event(self.ctx, entity=cluster1)
events_all = db_api.event_get_all(self.ctx)
self.assertEqual(3, len(events_all))
marker = events_all[0].id
event1_id = events_all[1].id
event2_id = events_all[2].id
events = db_api.event_get_all(self.ctx, limit=1, marker=marker)
self.assertEqual(1, len(events))
self.assertEqual(event1_id, events[0].id)
events = db_api.event_get_all(self.ctx, limit=2, marker=marker)
self.assertEqual(2, len(events))
self.assertEqual(event1_id, events[0].id)
self.assertEqual(event2_id, events[1].id)
marker = event1_id
events = db_api.event_get_all(self.ctx, limit=1, marker=marker)
self.assertEqual(1, len(events))
self.assertEqual(event2_id, events[0].id)
开发者ID:jonnary,项目名称:senlin,代码行数:26,代码来源:test_event_api.py
示例20: test_cluster_get_all_show_nested
def test_cluster_get_all_show_nested(self):
cluster1 = shared.create_cluster(self.ctx, self.profile,
name='cluster1')
cluster2 = shared.create_cluster(self.ctx, self.profile,
name='cluster2',
parent=cluster1.id)
cl_db = db_api.cluster_get_all(self.ctx)
self.assertEqual(1, len(cl_db))
self.assertEqual(cluster1.id, cl_db[0].id)
cl_db = db_api.cluster_get_all(self.ctx, show_nested=True)
self.assertEqual(2, len(cl_db))
cl_ids = [s.id for s in cl_db]
self.assertIn(cluster1.id, cl_ids)
self.assertIn(cluster2.id, cl_ids)
开发者ID:Alzon,项目名称:senlin,代码行数:16,代码来源:test_cluster_api.py
注:本文中的senlin.tests.unit.db.shared.create_cluster函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论