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

Python provisioning_serializers.serialize函数代码示例

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

本文整理汇总了Python中nailgun.orchestrator.provisioning_serializers.serialize函数的典型用法代码示例。如果您正苦于以下问题:Python serialize函数的具体用法?Python serialize怎么用?Python serialize使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。



在下文中一共展示了serialize函数的14个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: message

    def message(cls, task, nodes_to_provisioning):
        logger.debug("ProvisionTask.message(task=%s)" % task.uuid)
        task = objects.Task.get_by_uid(
            task.id,
            fail_if_not_found=True,
            lock_for_update=True
        )
        objects.NodeCollection.lock_nodes(nodes_to_provisioning)
        serialized_cluster = provisioning_serializers.serialize(
            task.cluster, nodes_to_provisioning)

        for node in nodes_to_provisioning:
            if settings.FAKE_TASKS or settings.FAKE_TASKS_AMQP:
                continue

            admin_net_id = objects.Node.get_network_manager(
                node
            ).get_admin_network_group_id(node.id)

            TaskHelper.prepare_syslog_dir(node, admin_net_id)

        rpc_message = make_astute_message(
            task,
            'provision',
            'provision_resp',
            {
                'provisioning_info': serialized_cluster
            }
        )
        db().commit()
        return rpc_message
开发者ID:cxb811201,项目名称:fuel-web,代码行数:31,代码来源:task.py


示例2: test_node_serialization_w_bonded_admin_iface

 def test_node_serialization_w_bonded_admin_iface(self):
     self.cluster_db = self.env.clusters[0]
     # create additional node to test bonding
     admin_mac = self.env.generate_random_mac()
     meta = {
         'interfaces': [
             {'name': 'eth1', 'mac': self.env.generate_random_mac()},
             {'name': 'eth2', 'mac': self.env.generate_random_mac()},
             {'name': 'eth3', 'mac': self.env.generate_random_mac()},
             {'name': 'eth4', 'mac': self.env.generate_random_mac()}
         ]
     }
     node = self.env.create_node(
         pending_addition=True,
         cluster_id=self.cluster_db.id,
         meta=meta,
         mac=admin_mac
     )
     # get node from db
     node_db = objects.Node.get_by_uid(node['id'])
     # bond admin iface
     self.env.make_bond_via_api('lnx_bond',
                                '',
                                ['eth1', 'eth4'],
                                node['id'],
                                bond_properties={
                                    'mode': consts.BOND_MODES.balance_rr
                                })
     # check serialized data
     serialized_node = ps.serialize(self.cluster_db, [node_db])['nodes'][0]
     out_mac = serialized_node['kernel_options']['netcfg/choose_interface']
     self.assertEqual(out_mac, admin_mac)
开发者ID:blkart,项目名称:fuel-web,代码行数:32,代码来源:test_provisioning_serializer.py


示例3: message

    def message(cls, task, nodes_to_provisioning):
        logger.debug("ProvisionTask.message(task=%s)" % task.uuid)
        task = objects.Task.get_by_uid(
            task.id,
            fail_if_not_found=True,
            lock_for_update=True
        )
        objects.NodeCollection.lock_nodes(nodes_to_provisioning)
        serialized_cluster = provisioning_serializers.serialize(
            task.cluster, nodes_to_provisioning)

        for node in nodes_to_provisioning:
            if settings.FAKE_TASKS or settings.FAKE_TASKS_AMQP:
                continue
            logs_utils.prepare_syslog_dir(node)

        rpc_message = make_astute_message(
            task,
            cls._get_provision_method(task.cluster),
            'provision_resp',
            {
                'provisioning_info': serialized_cluster
            }
        )
        db().commit()
        return rpc_message
开发者ID:anbangr,项目名称:fuel-web,代码行数:26,代码来源:task.py


示例4: message

    def message(cls, task, nodes_to_provisioning):
        logger.debug("ProvisionTask.message(task=%s)" % task.uuid)

        serialized_cluster = task.cluster.replaced_provisioning_info or \
            provisioning_serializers.serialize(
                task.cluster, nodes_to_provisioning)

        for node in nodes_to_provisioning:
            if settings.FAKE_TASKS or settings.FAKE_TASKS_AMQP:
                continue

            admin_net_id = objects.Node.get_network_manager(
                node
            ).get_admin_network_group_id()

            TaskHelper.prepare_syslog_dir(node, admin_net_id)

        return make_astute_message(
            'provision',
            'provision_resp',
            {
                'task_uuid': task.uuid,
                'provisioning_info': serialized_cluster
            }
        )
开发者ID:iberezovskiy,项目名称:fuel-web,代码行数:25,代码来源:task.py


示例5: message

    def message(cls, task):
        logger.debug("ProvisionTask.message(task=%s)" % task.uuid)
        nodes = TaskHelper.nodes_to_provision(task.cluster)
        USE_FAKE = settings.FAKE_TASKS or settings.FAKE_TASKS_AMQP

        # We need to assign admin ips
        # and only after that prepare syslog
        # directories
        task.cluster.prepare_for_provisioning()

        for node in nodes:
            if USE_FAKE:
                continue

            if node.offline:
                raise errors.NodeOffline(
                    u'Node "%s" is offline.' " Remove it from environment and try again." % node.full_name
                )

            TaskHelper.prepare_syslog_dir(node)

        serialized_cluster = task.cluster.replaced_provisioning_info or provisioning_serializers.serialize(task.cluster)

        message = {
            "method": "provision",
            "respond_to": "provision_resp",
            "args": {"task_uuid": task.uuid, "provisioning_info": serialized_cluster},
        }

        return message
开发者ID:rsokolkov,项目名称:fuel-web,代码行数:30,代码来源:task.py


示例6: test_provisioning_serialization_ignore_customized_false

    def test_provisioning_serialization_ignore_customized_false(self):
        cluster = self._create_cluster_with_extensions(
            nodes_kwargs=[
                {'roles': ['controller'], 'pending_addition': True},
                {'roles': ['controller'], 'pending_addition': True},
                {'roles': ['controller'], 'pending_addition': True},
                {'roles': ['controller'], 'pending_addition': True},
            ]
        )

        data = {"nodes": [{"uid": n.uid} for n in cluster.nodes]}
        expected_data = {"nodes": copy.deepcopy(data["nodes"][1:])}

        mserializer = mock.MagicMock()
        mserializer.serialize.return_value = data

        with mock.patch(
                'nailgun.orchestrator.provisioning_serializers.'
                'get_serializer_for_cluster',
                return_value=mserializer):
            with mock.patch('nailgun.orchestrator.provisioning_serializers.'
                            'fire_callback_on_provisioning_data_serialization'
                            ) as mfire_callback:

                replaced_data = {"it's": "something"}
                with mock.patch.object(
                        cluster.nodes[0], 'replaced_provisioning_info',
                        new_callable=mock.Mock(return_value=replaced_data)):

                    provisioning_serializers.serialize(
                        cluster, cluster.nodes, ignore_customized=False)

        self.assertEqual(mfire_callback.call_args[0][0], expected_data)
        self.assertIs(mfire_callback.call_args[0][1], cluster)
        self.assertItemsEqual(
            mfire_callback.call_args[0][2], cluster.nodes[1:])
开发者ID:huyupeng,项目名称:fuel-web,代码行数:36,代码来源:test_extensions.py


示例7: test_provisioning_serialization_ignore_customized

    def test_provisioning_serialization_ignore_customized(self):
        cluster = self._create_cluster_with_extensions()

        data = {"nodes": cluster.nodes}
        mserializer = mock.MagicMock()
        mserializer.serialize.return_value = data

        with mock.patch(
                'nailgun.orchestrator.provisioning_serializers.'
                'get_serializer_for_cluster',
                return_value=mserializer):
            with mock.patch('nailgun.orchestrator.provisioning_serializers.'
                            'fire_callback_on_provisioning_data_serialization'
                            ) as mfire_callback:

                replaced_data = {"it's": "something"}
                with mock.patch.object(
                        cluster.nodes[0], 'replaced_provisioning_info',
                        new_callable=mock.Mock(return_value=replaced_data)):

                    provisioning_serializers.serialize(
                        cluster, cluster.nodes, ignore_customized=True)

        mfire_callback.assert_called_once_with(data, cluster, cluster.nodes)
开发者ID:huyupeng,项目名称:fuel-web,代码行数:24,代码来源:test_extensions.py


示例8: test_pipeline_change_data

    def test_pipeline_change_data(self):
        cluster = self.env.create(
            cluster_kwargs={'api': False},
            nodes_kwargs=[{'roles': ['controller'], 'pending_addition': True}]
        )
        cluster.extensions = [self.extension.name]
        self.db.flush()

        class PipelinePlus1(BasePipeline):

            @classmethod
            def process_provisioning(cls, data, cluster, nodes, **kwargs):
                data['key'] += 1
                return data

        class PipelinePlus2(BasePipeline):

            @classmethod
            def process_provisioning(cls, data, cluster, nodes, **kwargs):
                data['key'] += 2
                return data

        class Extension(BaseExtension):
            name = 'ext_name'
            version = '1.0.0'
            description = 'ext description'
            data_pipelines = (PipelinePlus1, PipelinePlus2)

        extension = Extension()

        cluster.extensions = [extension.name]
        self.db.flush()

        data = {'key': 0, 'nodes': []}

        mserializer = mock.MagicMock()
        mserializer.serialize.return_value = data

        with mock.patch('nailgun.extensions.manager.get_all_extensions',
                        return_value=[extension]):
            with mock.patch('nailgun.orchestrator.provisioning_serializers.'
                            'get_serializer_for_cluster',
                            return_value=mserializer):
                new_data = provisioning_serializers.serialize(
                    cluster, cluster.nodes)

        self.assertEqual(new_data['key'], 3)
开发者ID:huyupeng,项目名称:fuel-web,代码行数:47,代码来源:test_extensions.py


示例9: test_ubuntu_serializer

    def test_ubuntu_serializer(self):
        release = self.env.create_release(api=False, operating_system="Ubuntu")

        self.env.create(
            cluster_kwargs={"release_id": release.id},
            nodes_kwargs=[
                {"roles": ["controller"], "pending_addition": True},
                {"roles": ["compute"], "pending_addition": True},
            ],
        )

        cluster_db = self.env.clusters[0]
        serialized_cluster = serialize(cluster_db, cluster_db.nodes)

        for node in serialized_cluster["nodes"]:
            node_db = self.db.query(Node).filter_by(fqdn=node["hostname"]).first()
            self.assertEqual(node["kernel_options"]["netcfg/choose_interface"], node_db.admin_interface.mac)
开发者ID:koder-ua,项目名称:nailgun-fcert,代码行数:17,代码来源:test_provisioning_serializer.py


示例10: test_ubuntu_serializer

    def test_ubuntu_serializer(self):
        release = self.env.create_release(
            api=False, operating_system='Ubuntu')

        cluster = self.env.create(
            cluster_kwargs={
                'mode': 'multinode',
                'release': release.id},
            nodes_kwargs=[
                {'roles': ['controller'], 'pending_addition': True},
                {'roles': ['compute'], 'pending_addition': True}])

        cluster_db = self.db.query(Cluster).get(cluster['id'])
        serialized_cluster = serialize(cluster_db)

        for node in serialized_cluster['nodes']:
            node_db = db().query(Node).filter_by(fqdn=node['hostname']).first()
            self.assertEquals(
                node['kernel_options']['netcfg/choose_interface'],
                node_db.admin_interface.name)
开发者ID:AlexeyKasatkin,项目名称:fuel-web,代码行数:20,代码来源:test_provisioning_serializer.py


示例11: message

    def message(cls, task, nodes_to_provisioning):
        logger.debug("ProvisionTask.message(task=%s)" % task.uuid)

        serialized_cluster = task.cluster.replaced_provisioning_info or \
            provisioning_serializers.serialize(
                task.cluster, nodes_to_provisioning)

        for node in nodes_to_provisioning:
            if settings.FAKE_TASKS or settings.FAKE_TASKS_AMQP:
                continue

            TaskHelper.prepare_syslog_dir(node)

        message = {
            'method': 'provision',
            'respond_to': 'provision_resp',
            'args': {
                'task_uuid': task.uuid,
                'provisioning_info': serialized_cluster}}

        return message
开发者ID:stamak,项目名称:fuel-web,代码行数:21,代码来源:task.py


示例12: setUp

 def setUp(self):
     super(TestProvisioningSerializer, self).setUp()
     self.cluster_db = self.env.create()
     self.env.create_nodes_w_interfaces_count(
         1, 1,
         **{
             'roles': ['controller'],
             'pending_addition': True,
             'cluster_id': self.cluster_db.id
         }
     )
     self.env.create_nodes_w_interfaces_count(
         1, 1,
         **{
             'roles': ['compute'],
             'pending_addition': True,
             'cluster_id': self.cluster_db.id
         }
     )
     self.attributes = self.cluster_db.attributes.editable
     self.serialized_cluster = ps.serialize(
         self.cluster_db, self.cluster_db.nodes)
开发者ID:mmalchuk,项目名称:openstack-fuel-web,代码行数:22,代码来源:test_provisioning_serializer.py


示例13: message

    def message(cls, task):
        logger.debug("ProvisionTask.message(task=%s)" % task.uuid)
        nodes = TaskHelper.nodes_to_provision(task.cluster)
        USE_FAKE = settings.FAKE_TASKS or settings.FAKE_TASKS_AMQP

        # We need to assign admin ips
        # and only after that prepare syslog
        # directories
        task.cluster.prepare_for_provisioning()

        for node in nodes:
            if USE_FAKE:
                continue

            if node.offline:
                raise errors.NodeOffline(
                    u'Node "%s" is offline.'
                    ' Remove it from environment and try again.' %
                    node.full_name)

            TaskHelper.prepare_syslog_dir(node)

            node.status = 'provisioning'
            db().commit()

        serialized_cluster = task.cluster.replaced_provisioning_info or \
            provisioning_serializers.serialize(task.cluster)

        message = {
            'method': 'provision',
            'respond_to': 'provision_resp',
            'args': {
                'task_uuid': task.uuid,
                'provisioning_info': serialized_cluster}}

        return message
开发者ID:mrasskazov,项目名称:fuelweb,代码行数:36,代码来源:task.py


示例14: _serialize

 def _serialize(self, cluster, nodes):
     return provisioning_serializers.serialize(
         cluster, nodes, ignore_customized=True)
开发者ID:TorstenS73,项目名称:fuel-web,代码行数:3,代码来源:orchestrator.py



注:本文中的nailgun.orchestrator.provisioning_serializers.serialize函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。


鲜花

握手

雷人

路过

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

请发表评论

全部评论

专题导读
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

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

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

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