• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号

Python shared.create_cluster函数代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了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;未经允许,请勿转载。


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Python shared.create_node函数代码示例发布时间:2022-05-27
下一篇:
Python utils.dummy_context函数代码示例发布时间:2022-05-27
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap