Python core.create_resource函数代码示例

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Python›Python编程经验

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

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

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

示例1: _test_create_security_group_rule

    def _test_create_security_group_rule(self, plugin, q_ctx, t_ctx, pod_id,
                                         top_sgs, bottom1_sgs):
        t_sg_id = uuidutils.generate_uuid()
        t_rule_id = uuidutils.generate_uuid()
        b_sg_id = uuidutils.generate_uuid()
        project_id = 'test_prject_id'
        t_sg = {'id': t_sg_id, 'name': 'test', 'description': '',
                'tenant_id': project_id,
                'security_group_rules': []}
        b_sg = {'id': b_sg_id, 'name': t_sg_id, 'description': '',
                'tenant_id': project_id,
                'security_group_rules': []}
        top_sgs.append(t_sg)
        bottom1_sgs.append(b_sg)
        route = {
            'top_id': t_sg_id,
            'pod_id': pod_id,
            'bottom_id': b_sg_id,
            'resource_type': constants.RT_SG}
        with t_ctx.session.begin():
            core.create_resource(t_ctx, models.ResourceRouting, route)

        rule = {
            'security_group_rule': self._build_test_rule(
                t_rule_id, t_sg_id, project_id, '10.0.0.0/24')}
        plugin.create_security_group_rule(q_ctx, rule)

        self.assertEqual(1, len(bottom1_sgs[0]['security_group_rules']))
        b_rule = bottom1_sgs[0]['security_group_rules'][0]
        self.assertEqual(b_sg_id, b_rule['security_group_id'])

开发者ID:LongXQ，项目名称:tricircle，代码行数:30，代码来源:test_security_groups.py

示例2: _test_handle_remote_group_invalid_input

    def _test_handle_remote_group_invalid_input(self, plugin, q_ctx, t_ctx,
                                                pod_id, top_sgs, top_rules,
                                                bottom1_sgs):
        t_sg1_id = uuidutils.generate_uuid()
        t_sg2_id = uuidutils.generate_uuid()
        t_rule1_id = uuidutils.generate_uuid()
        t_rule2_id = uuidutils.generate_uuid()
        b_sg_id = uuidutils.generate_uuid()
        project_id = 'test_prject_id'
        t_rule1 = self._build_test_rule(
            t_rule1_id, t_sg1_id, project_id, None, t_sg1_id)
        t_rule2 = self._build_test_rule(
            t_rule2_id, t_sg1_id, project_id, None, t_sg2_id)
        t_sg = {'id': t_sg1_id, 'name': 'test', 'description': '',
                'tenant_id': project_id,
                'security_group_rules': []}
        b_sg = {'id': b_sg_id, 'name': t_sg1_id, 'description': '',
                'tenant_id': project_id,
                'security_group_rules': []}
        top_sgs.append(t_sg)
        top_rules.append(t_rule1)
        bottom1_sgs.append(b_sg)
        route = {
            'top_id': t_sg1_id,
            'pod_id': pod_id,
            'bottom_id': b_sg_id,
            'resource_type': constants.RT_SG}
        with t_ctx.session.begin():
            core.create_resource(t_ctx, models.ResourceRouting, route)

        self.assertRaises(exceptions.RemoteGroupNotSupported,
                          plugin.create_security_group_rule, q_ctx,
                          {'security_group_rule': t_rule2})
        self.assertRaises(exceptions.RemoteGroupNotSupported,
                          plugin.delete_security_group_rule, q_ctx, t_rule1_id)

开发者ID:LongXQ，项目名称:tricircle，代码行数:35，代码来源:test_security_groups.py

示例3: test_get_bottom_mappings_by_top_id

 def test_get_bottom_mappings_by_top_id(self):
     for i in xrange(3):
         pod = {'pod_id': 'test_pod_uuid_%d' % i,
                'pod_name': 'test_pod_%d' % i,
                'az_name': 'test_az_uuid_%d' % i}
         api.create_pod(self.context, pod)
     route1 = {
         'top_id': 'top_uuid',
         'pod_id': 'test_pod_uuid_0',
         'resource_type': 'port'}
     route2 = {
         'top_id': 'top_uuid',
         'pod_id': 'test_pod_uuid_1',
         'bottom_id': 'bottom_uuid_1',
         'resource_type': 'port'}
     route3 = {
         'top_id': 'top_uuid',
         'pod_id': 'test_pod_uuid_2',
         'bottom_id': 'bottom_uuid_2',
         'resource_type': 'neutron'}
     routes = [route1, route2, route3]
     with self.context.session.begin():
         for route in routes:
             core.create_resource(
                 self.context, models.ResourceRouting, route)
     mappings = api.get_bottom_mappings_by_top_id(self.context,
                                                  'top_uuid', 'port')
     self.assertEqual('test_pod_uuid_1', mappings[0][0]['pod_id'])
     self.assertEqual('bottom_uuid_1', mappings[0][1])

开发者ID:Ronghui，项目名称:tricircle，代码行数:29，代码来源:test_api.py

示例4: get_pod_by_az_tenant

def get_pod_by_az_tenant(context, az_name, tenant_id):
    pod_bindings = core.query_resource(context,
                                       models.PodBinding,
                                       [{'key': 'tenant_id',
                                         'comparator': 'eq',
                                         'value': tenant_id}],
                                       [])
    for pod_b in pod_bindings:
        pod = core.get_resource(context,
                                models.Pod,
                                pod_b['pod_id'])
        if pod['az_name'] == az_name:
            return pod, pod['pod_az_name']

    # TODO(joehuang): schedule one dynamically in the future
    filters = [{'key': 'az_name', 'comparator': 'eq', 'value': az_name}]
    pods = db_api.list_pods(context, filters=filters)
    for pod in pods:
        if pod['pod_name'] != '':
            try:
                with context.session.begin():
                    core.create_resource(
                        context, models.PodBinding,
                        {'id': uuidutils.generate_uuid(),
                         'tenant_id': tenant_id,
                         'pod_id': pod['pod_id']})
                    return pod, pod['pod_az_name']
            except Exception as e:
                LOG.error(_LE('Fail to create pod binding: %(exception)s'),
                          {'exception': e})
                return None, None

    return None, None

开发者ID:Ronghui，项目名称:tricircle，代码行数:33，代码来源:az_ag.py

示例5: test_job_run_expire

    def test_job_run_expire(self):
        @xmanager._job_handle('fake_resource')
        def fake_handle(self, ctx, payload):
            pass

        fake_id = uuidutils.generate_uuid()
        payload = {'fake_resource': fake_id}
        expired_job = {
            'id': uuidutils.generate_uuid(),
            'type': 'fake_resource',
            'timestamp': datetime.datetime.now() - datetime.timedelta(0, 120),
            'status': constants.JS_Running,
            'resource_id': fake_id,
            'extra_id': constants.SP_EXTRA_ID
        }
        core.create_resource(self.context, models.Job, expired_job)
        fake_handle(None, self.context, payload=payload)

        jobs = core.query_resource(self.context, models.Job, [], [])
        expected_status = ['New', 'Fail', 'Success']
        job_status = [job['status'] for job in jobs]
        self.assertItemsEqual(expected_status, job_status)

        for i in xrange(3):
            self.assertEqual(fake_id, jobs[i]['resource_id'])
            self.assertEqual('fake_resource', jobs[i]['type'])

开发者ID:Ronghui，项目名称:tricircle，代码行数:26，代码来源:test_xmanager.py

示例6: _prepare_snat_test

 def _prepare_snat_test(self, top_router_id):
     ext_network = {'id': 'ext_network_id',
                    'router:external': True}
     ext_subnet = {
         'id': 'ext_subnet_id',
         'network_id': ext_network['id'],
         'cidr': '162.3.124.0/24',
         'gateway_ip': '162.3.124.1'
     }
     for router in TOP_ROUTER:
         if router['id'] == top_router_id:
             router['external_gateway_info'] = {
                 'network_id': ext_network['id']}
     router = {'id': 'ns_router_id'}
     for subnet in BOTTOM2_SUBNET:
         if 'bridge' in subnet['id']:
             bridge_subnet = subnet
     bridge_port = {
         'network_id': bridge_subnet['network_id'],
         'device_id': router['id'],
         'device_owner': 'network:router_interface',
         'fixed_ips': [{'subnet_id': bridge_subnet['id'],
                        'ip_address': bridge_subnet['gateway_ip']}]
     }
     BOTTOM2_NETWORK.append(ext_network)
     BOTTOM2_SUBNET.append(ext_subnet)
     BOTTOM2_PORT.append(bridge_port)
     BOTTOM2_ROUTER.append(router)
     route = {'top_id': top_router_id, 'bottom_id': router['id'],
              'pod_id': 'pod_id_2', 'resource_type': constants.RT_NS_ROUTER}
     with self.context.session.begin():
         core.create_resource(self.context, models.ResourceRouting, route)
     return bridge_subnet['gateway_ip'], router['id']

开发者ID:LongXQ，项目名称:tricircle，代码行数:33，代码来源:test_xmanager.py

示例7: test_job_run_expire

    def test_job_run_expire(self):
        job_type = 'fake_resource'

        @xmanager._job_handle(job_type)
        def fake_handle(self, ctx, payload):
            pass

        fake_id = uuidutils.generate_uuid()
        fake_project_id = uuidutils.generate_uuid()
        payload = {job_type: fake_id}
        db_api.new_job(self.context, fake_project_id, job_type, fake_id)
        expired_job = {
            'id': uuidutils.generate_uuid(),
            'type': job_type,
            'timestamp': datetime.datetime.now() - datetime.timedelta(0, 200),
            'status': constants.JS_Running,
            'resource_id': fake_id,
            'extra_id': constants.SP_EXTRA_ID
        }
        core.create_resource(self.context, models.AsyncJob, expired_job)
        fake_handle(None, self.context, payload=payload)

        logs = core.query_resource(self.context, models.AsyncJobLog, [], [])

        self.assertEqual(fake_id, logs[0]['resource_id'])
        self.assertEqual(job_type, logs[0]['type'])

开发者ID:LongXQ，项目名称:tricircle，代码行数:26，代码来源:test_xmanager.py

示例8: _prepare_port_test

    def _prepare_port_test(self, tenant_id, ctx, pod_name, net_id):
        t_port_id = uuidutils.generate_uuid()
        t_port = {
            'id': t_port_id,
            'network_id': net_id
        }
        TOP_PORTS.append(DotDict(t_port))
        b_port = {
            'id': t_port_id,
            'network_id': net_id
        }
        if pod_name == 'pod_1':
            BOTTOM1_PORTS.append(DotDict(b_port))
        else:
            BOTTOM2_PORTS.append(DotDict(b_port))

        pod_id = 'pod_id_1' if pod_name == 'pod_1' else 'pod_id_2'
        core.create_resource(ctx, models.ResourceRouting,
                             {'top_id': t_port_id,
                              'bottom_id': t_port_id,
                              'pod_id': pod_id,
                              'project_id': tenant_id,
                              'resource_type': constants.RT_PORT})

        return t_port_id

开发者ID:openstack，项目名称:tricircle，代码行数:25，代码来源:test_central_sfc_plugin.py

示例9: _test_delete_security_group_rule_exception

    def _test_delete_security_group_rule_exception(self, plugin, q_ctx, t_ctx,
                                                   pod_id, top_sgs, top_rules,
                                                   bottom1_sgs):
        t_sg_id = uuidutils.generate_uuid()
        t_rule_id = uuidutils.generate_uuid()
        b_sg_id = uuidutils.generate_uuid()
        project_id = 'test_prject_id'
        t_rule = self._build_test_rule(
            t_rule_id, t_sg_id, project_id, '10.0.1.0/24')
        b_rule = self._build_test_rule(
            t_rule_id, b_sg_id, project_id, '10.0.1.0/24')
        t_sg = {'id': t_sg_id, 'name': 'test', 'description': '',
                'tenant_id': project_id,
                'security_group_rules': [t_rule]}
        b_sg = {'id': b_sg_id, 'name': t_sg_id, 'description': '',
                'tenant_id': project_id,
                'security_group_rules': [b_rule]}
        top_sgs.append(t_sg)
        top_rules.append(t_rule)
        bottom1_sgs.append(b_sg)
        route = {
            'top_id': t_sg_id,
            'pod_id': pod_id,
            'bottom_id': b_sg_id,
            'resource_type': constants.RT_SG}
        with t_ctx.session.begin():
            core.create_resource(t_ctx, models.ResourceRouting, route)

        self.assertRaises(exceptions.BottomPodOperationFailure,
                          plugin.delete_security_group_rule, q_ctx, t_rule_id)

开发者ID:LongXQ，项目名称:tricircle，代码行数:30，代码来源:test_security_groups.py

示例10: _prepare_flow_classifier_test

    def _prepare_flow_classifier_test(self, project_id, t_ctx, pod_name,
                                      index, src_port_id, create_bottom):
        t_fc_id = uuidutils.generate_uuid()
        b_fc_id = uuidutils.generate_uuid()

        top_fc = {
            "source_port_range_min": None,
            "destination_ip_prefix": None,
            "protocol": None,
            "description": "",
            "l7_parameters": {},
            "source_port_range_max": None,
            "id": t_fc_id,
            "name": "t_fc_%s" % index,
            "ethertype": "IPv4",
            "tenant_id": project_id,
            "source_ip_prefix": "1.0.0.0/24",
            "logical_destination_port": None,
            "destination_port_range_min": None,
            "destination_port_range_max": None,
            "project_id": project_id,
            "logical_source_port": src_port_id}

        TOP_FLOWCLASSIFIERS.append(DotDict(top_fc))
        if create_bottom:
            btm_fc = {
                "source_port_range_min": None,
                "destination_ip_prefix": None,
                "protocol": None,
                "description": "",
                "l7_parameters": {},
                "source_port_range_max": None,
                "id": b_fc_id,
                "name": "b_fc_%s" % index,
                "ethertype": "IPv4",
                "tenant_id": project_id,
                "source_ip_prefix": "1.0.0.0/24",
                "logical_destination_port": None,
                "destination_port_range_min": None,
                "destination_port_range_max": None,
                "project_id": project_id,
                "logical_source_port": src_port_id}
            if pod_name == 'pod_1':
                BOTTOM1_FLOWCLASSIFIERS.append(DotDict(btm_fc))
            else:
                BOTTOM2_FLOWCLASSIFIERS.append(DotDict(btm_fc))

            pod_id = 'pod_id_1' if pod_name == 'pod_1' else 'pod_id_2'
            core.create_resource(t_ctx, models.ResourceRouting,
                                 {'top_id': t_fc_id,
                                  'bottom_id': b_fc_id,
                                  'pod_id': pod_id,
                                  'project_id': project_id,
                                  'resource_type':
                                      constants.RT_FLOW_CLASSIFIER})

        return t_fc_id, b_fc_id

开发者ID:openstack，项目名称:tricircle，代码行数:57，代码来源:test_central_sfc_plugin.py

示例11: _prepare_net_test

 def _prepare_net_test(self, project_id, ctx, pod_name):
     t_net_id = uuidutils.generate_uuid()
     pod_id = 'pod_id_1' if pod_name == 'pod_1' else 'pod_id_2'
     core.create_resource(ctx, models.ResourceRouting,
                          {'top_id': t_net_id,
                           'bottom_id': t_net_id,
                           'pod_id': pod_id,
                           'project_id': project_id,
                           'resource_type': constants.RT_NETWORK})
     return t_net_id

开发者ID:openstack，项目名称:tricircle，代码行数:10，代码来源:test_central_sfc_plugin.py

示例12: _prepare_server

 def _prepare_server(self, pod):
     t_server_id = uuidutils.generate_uuid()
     b_server_id = t_server_id
     with self.context.session.begin():
         core.create_resource(
             self.context, models.ResourceRouting,
             {'top_id': t_server_id, 'bottom_id': b_server_id,
              'pod_id': pod['pod_id'], 'project_id': self.project_id,
              'resource_type': constants.RT_SERVER})
     return t_server_id

开发者ID:dingboopt，项目名称:tricircle，代码行数:10，代码来源:test_action.py

示例13: _test_handle_network_dhcp_port

    def _test_handle_network_dhcp_port(self, dhcp_ip):
        t_pod, b_pod = self._prepare_pod()

        top_net_id = 'top_net_id'
        bottom_net_id = 'bottom_net_id'
        top_subnet_id = 'top_subnet_id'
        bottom_subnet_id = 'bottom_subnet_id'
        t_net = {'id': top_net_id}
        b_net = {'id': bottom_net_id}
        t_subnet = {'id': top_subnet_id,
                    'network_id': top_net_id,
                    'ip_version': 4,
                    'cidr': '10.0.0.0/24',
                    'gateway_ip': '10.0.0.1',
                    'allocation_pools': {'start': '10.0.0.2',
                                         'end': '10.0.0.254'},
                    'enable_dhcp': True}
        b_subnet = {'id': bottom_subnet_id,
                    'network_id': bottom_net_id,
                    'ip_version': 4,
                    'cidr': '10.0.0.0/24',
                    'gateway_ip': '10.0.0.1',
                    'allocation_pools': {'start': '10.0.0.2',
                                         'end': '10.0.0.254'},
                    'enable_dhcp': True}
        b_dhcp_port = {'id': 'bottom_dhcp_port_id',
                       'network_id': bottom_net_id,
                       'fixed_ips': [
                           {'subnet_id': bottom_subnet_id,
                            'ip_address': dhcp_ip}
                       ],
                       'mac_address': 'fa:16:3e:96:41:0a',
                       'binding:profile': {},
                       'device_id': 'reserved_dhcp_port',
                       'device_owner': 'network:dhcp'}
        TOP_NETS.append(t_net)
        TOP_SUBNETS.append(t_subnet)
        BOTTOM_NETS.append(b_net)
        BOTTOM_SUBNETS.append(b_subnet)
        BOTTOM_PORTS.append(b_dhcp_port)
        with self.context.session.begin():
            core.create_resource(
                self.context, models.ResourceRouting,
                {'top_id': top_net_id, 'bottom_id': bottom_net_id,
                 'pod_id': b_pod['pod_id'], 'project_id': self.project_id,
                 'resource_type': 'network'})
            core.create_resource(
                self.context, models.ResourceRouting,
                {'top_id': top_subnet_id, 'bottom_id': bottom_subnet_id,
                 'pod_id': b_pod['pod_id'], 'project_id': self.project_id,
                 'resource_type': 'subnet'})
        self.controller._handle_network(self.context,
                                        b_pod, t_net, [t_subnet])
        self._check_routes()

开发者ID:Ronghui，项目名称:tricircle，代码行数:54，代码来源:test_server.py

示例14: test_resource_routing_unique_key

 def test_resource_routing_unique_key(self):
     pod = {'pod_id': 'test_pod1_uuid',
            'pod_name': 'test_pod1',
            'az_name': 'test_az1_uuid'}
     api.create_pod(self.context, pod)
     routing = {'top_id': 'top_uuid',
                'pod_id': 'test_pod1_uuid',
                'resource_type': 'port'}
     with self.context.session.begin():
         core.create_resource(self.context, models.ResourceRouting, routing)
     self.assertRaises(oslo_db.exception.DBDuplicateEntry,
                       core.create_resource,
                       self.context, models.ResourceRouting, routing)

开发者ID:Ronghui，项目名称:tricircle，代码行数:13，代码来源:test_models.py

示例15: _prepare_port_pair_group_test

    def _prepare_port_pair_group_test(self, project_id, t_ctx, pod_name, index,
                                      t_pp_ids, create_bottom, b_pp_ids):
        t_ppg_id = uuidutils.generate_uuid()
        b_ppg_id = uuidutils.generate_uuid()

        t_client = FakeClient()
        b_client = FakeClient(pod_name)
        t_pps = [t_client.get_resource(
            'port_pair', t_ctx, e) for e in t_pp_ids]
        if create_bottom:
            b_pps = [b_client.get_resource(
                'port_pair', t_ctx, e) for e in b_pp_ids]

        top_ppg = {
            "group_id": 1,
            "description": "",
            "tenant_id": project_id,
            "port_pair_group_parameters": {"lb_fields": []},
            "port_pairs": t_pps,
            "project_id": project_id,
            "id": t_ppg_id,
            "name": 'top_ppg_%d' % index,
            "tap_enabled": False}
        TOP_PORTPAIRGROUPS.append(DotDict(top_ppg))
        if create_bottom:
            btm_ppg = {
                "group_id": 1,
                "description": "",
                "tenant_id": project_id,
                "port_pair_group_parameters": {"lb_fields": []},
                "port_pairs": b_pps,
                "project_id": project_id,
                "id": b_ppg_id,
                "name": 'btm_ppg_%d' % index,
                "tap_enabled": False}
            if pod_name == 'pod_1':
                BOTTOM1_PORTPAIRGROUPS.append(DotDict(btm_ppg))
            else:
                BOTTOM2_PORTPAIRGROUPS.append(DotDict(btm_ppg))

            pod_id = 'pod_id_1' if pod_name == 'pod_1' else 'pod_id_2'
            core.create_resource(t_ctx, models.ResourceRouting,
                                 {'top_id': t_ppg_id,
                                  'bottom_id': b_ppg_id,
                                  'pod_id': pod_id,
                                  'project_id': project_id,
                                  'resource_type':
                                      constants.RT_PORT_PAIR_GROUP})

        return t_ppg_id, b_ppg_id

开发者ID:openstack，项目名称:tricircle，代码行数:50，代码来源:test_central_sfc_plugin.py

示例16: create_ag_az

def create_ag_az(context, ag_name, az_name):
    aggregate = core.create_resource(context, models.Aggregate,
                                     {'name': ag_name})
    core.create_resource(
        context, models.AggregateMetadata,
        {'key': 'availability_zone',
         'value': az_name,
         'aggregate_id': aggregate['id']})
    extra_fields = {
        'availability_zone': az_name,
        'metadata': {'availability_zone': az_name}
    }
    aggregate.update(extra_fields)
    return aggregate

开发者ID:Ronghui，项目名称:tricircle，代码行数:14，代码来源:az_ag.py

示例17: _test_get_security_group

    def _test_get_security_group(self, plugin, q_ctx, t_ctx,
                                 pod_id, top_sgs, bottom1_sgs):
        t_sg_id = uuidutils.generate_uuid()
        t_rule1_id = uuidutils.generate_uuid()
        t_rule2_id = uuidutils.generate_uuid()
        b_sg_id = uuidutils.generate_uuid()
        project_id = 'test_prject_id'
        t_rule1 = self._build_test_rule(
            t_rule1_id, t_sg_id, project_id, '10.0.0.0/24')
        t_rule2 = self._build_test_rule(
            t_rule2_id, t_sg_id, project_id, '192.168.56.0/24')
        t_sg = {'id': t_sg_id, 'name': 'top_sg', 'description': '',
                'tenant_id': project_id,
                'security_group_rules': [t_rule1, t_rule2]}
        b_sg = {'id': b_sg_id, 'name': 'bottom_sg', 'description': '',
                'tenant_id': project_id,
                'security_group_rules': [t_rule1, t_rule2]}
        top_sgs.append(t_sg)
        bottom1_sgs.append(b_sg)

        route1 = {
            'top_id': t_sg_id,
            'pod_id': pod_id,
            'bottom_id': b_sg_id,
            'resource_type': constants.RT_SG}
        with t_ctx.session.begin():
            core.create_resource(t_ctx, models.ResourceRouting, route1)

        # test get_sg for normal situation
        res = plugin.get_security_group(q_ctx, t_sg_id)
        self.assertTrue(res['id'] == t_sg_id and res['name'] == 'top_sg')

        # test get_sg when the top_sg is under deleting
        dict_para = {'resource_id': t_sg_id,
                     'resource_type': t_constants.RT_SG}
        with t_ctx.session.begin():
            core.create_resource(t_ctx, models.DeletingResources,
                                 dict_para)

        q_ctx.USER_AGENT = t_constants.LOCAL
        self.assertRaises(t_exceptions.ResourceNotFound,
                          plugin.get_security_group,
                          q_ctx, t_sg_id)

        # test get_sg when the request is from user_agent
        q_ctx.USER_AGENT = t_constants.USER_AGENT
        self.assertRaises(t_exceptions.ResourceIsInDeleting,
                          plugin.get_security_group,
                          q_ctx, t_sg_id)

开发者ID:daespinel，项目名称:tricircle，代码行数:49，代码来源:test_security_groups.py

示例18: test_get_failed_or_new_jobs

    def test_get_failed_or_new_jobs(self, mock_now):
        mock_now.return_value = datetime.datetime(2000, 1, 2, 12, 0, 0)
        job_dict_list = [
            {'timestamp': datetime.datetime(2000, 1, 1, 12, 0, 0),
             'resource_id': 'uuid1', 'type': 'res1', 'project_id': "uuid1",
             'status': constants.JS_Fail},  # job_uuid1
            {'timestamp': datetime.datetime(2000, 1, 1, 12, 5, 0),
             'resource_id': 'uuid1', 'type': 'res1', 'project_id': "uuid1",
             'status': constants.JS_Fail},  # job_uuid3
            {'timestamp': datetime.datetime(2000, 1, 1, 12, 20, 0),
             'resource_id': 'uuid2', 'type': 'res2', 'project_id': "uuid1",
             'status': constants.JS_Fail},  # job_uuid5
            {'timestamp': datetime.datetime(2000, 1, 1, 12, 15, 0),
             'resource_id': 'uuid2', 'type': 'res2', 'project_id': "uuid1",
             'status': constants.JS_Fail},  # job_uuid7
            {'timestamp': datetime.datetime(2000, 1, 1, 12, 25, 0),
             'resource_id': 'uuid3', 'type': 'res3', 'project_id': "uuid1",
             'status': constants.JS_Success},  # job_uuid9
            {'timestamp': datetime.datetime(2000, 1, 1, 12, 30, 0),
             'resource_id': 'uuid4', 'type': 'res4', 'project_id': "uuid1",
             'status': constants.JS_New},  # job_uuid11
            {'timestamp': datetime.datetime(1999, 12, 31, 12, 0, 0),
             'resource_id': 'uuid5', 'type': 'res5', 'project_id': "uuid1",
             'status': constants.JS_Fail},  # job_uuid13
            {'timestamp': datetime.datetime(1999, 12, 31, 11, 59, 59),
             'resource_id': 'uuid6', 'type': 'res6', 'project_id': "uuid1",
             'status': constants.JS_Fail}]  # job_uuid15
        for i, job_dict in enumerate(job_dict_list, 1):
            job_dict['id'] = 'job_uuid%d' % (2 * i - 1)
            job_dict['extra_id'] = 'extra_uuid%d' % (2 * i - 1)
            core.create_resource(self.context, models.AsyncJob, job_dict)
            job_dict['id'] = 'job_uuid%d' % (2 * i)
            job_dict['extra_id'] = 'extra_uuid%d' % (2 * i)
            job_dict['status'] = constants.JS_New
            core.create_resource(self.context, models.AsyncJob, job_dict)

        # for res3 + uuid3, the latest job's status is "Success", not returned
        # for res6 + uuid6, the latest job is out of the redo time span
        expected_failed_jobs = [
            {'resource_id': 'uuid1', 'type': 'res1', 'project_id': "uuid1"},
            {'resource_id': 'uuid2', 'type': 'res2', 'project_id': "uuid1"},
            {'resource_id': 'uuid5', 'type': 'res5', 'project_id': "uuid1"}]
        expected_new_jobs = [{'resource_id': 'uuid4', 'type': 'res4',
                              'project_id': "uuid1"}]
        (failed_jobs,
         new_jobs) = db_api.get_latest_failed_or_new_jobs(self.context)
        six.assertCountEqual(self, expected_failed_jobs, failed_jobs)
        six.assertCountEqual(self, expected_new_jobs, new_jobs)

开发者ID:LongXQ，项目名称:tricircle，代码行数:48，代码来源:test_xmanager.py

示例19: test_resources

 def test_resources(self):
     """Create all the resources to test model definition"""
     try:
         model_list = []
         for _, model_class in inspect.getmembers(models):
             if inspect.isclass(model_class) and (
                     issubclass(model_class, core.ModelBase)):
                 model_list.append(model_class)
         for model_class in _sort_model_by_foreign_key(model_list):
             create_dict = _construct_resource_dict(model_class)
             with self.context.session.begin():
                 core.create_resource(
                     self.context, model_class, create_dict)
     except Exception as e:
         msg = str(e)
         self.fail('test_resources raised Exception unexpectedly %s' % msg)

开发者ID:Ronghui，项目名称:tricircle，代码行数:16，代码来源:test_models.py

示例20: _prepare_port_pair_test

    def _prepare_port_pair_test(self, project_id, t_ctx, pod_name,
                                index, ingress, egress, create_bottom,
                                portpairgroup_id=None):
        t_pp_id = uuidutils.generate_uuid()
        b_pp_id = uuidutils.generate_uuid()
        top_pp = {
            'id': t_pp_id,
            'project_id': project_id,
            'tenant_id': project_id,
            'ingress': ingress,
            'egress': egress,
            'name': 'top_pp_%d' % index,
            'service_function_parameters': {
                "weight": 1,
                "correlation": DotDict({'value': 'null'})},
            'description': "description",
            'portpairgroup_id': portpairgroup_id
        }
        TOP_PORTPAIRS.append(DotDict(top_pp))
        if create_bottom:
            btm_pp = {
                'id': b_pp_id,
                'project_id': project_id,
                'tenant_id': project_id,
                'ingress': ingress,
                'egress': egress,
                'name': 'btm_pp_%d' % index,
                'service_function_parameters': {
                    "weight": 1,
                    "correlation": DotDict({'value': 'null'})},
                'description': "description",
                'portpairgroup_id': portpairgroup_id
            }
            if pod_name == 'pod_1':
                BOTTOM1_PORTPAIRS.append(DotDict(btm_pp))
            else:
                BOTTOM2_PORTPAIRS.append(DotDict(btm_pp))

            pod_id = 'pod_id_1' if pod_name == 'pod_1' else 'pod_id_2'
            core.create_resource(t_ctx, models.ResourceRouting,
                                 {'top_id': t_pp_id,
                                  'bottom_id': b_pp_id,
                                  'pod_id': pod_id,
                                  'project_id': project_id,
                                  'resource_type': constants.RT_PORT_PAIR})

        return t_pp_id, b_pp_id

开发者ID:openstack，项目名称:tricircle，代码行数:47，代码来源:test_central_sfc_plugin.py

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

鲜花

握手

雷人

路过

鸡蛋

该文章已有0人参与评论

请发表评论

全部评论

专题导读

More+

10-27 六六分期app的软件客服如何联系？(六六分期

11-06 可心卡盟:win10系统火狐flash插件崩溃怎么

11-06 亲亲特价:怎么删除回收站图标

11-06 济南大学虚拟社区:鲁大师节能降温的具体办

11-06 xlueops.exe:无线网络安装向导

11-06 女斗合众国:win7系统cf与主机连接不稳定怎

11-06 0xc000022-[cf烟雾头]cf怎么调烟雾头

11-06 qizideyouhuo:应用程序无法正常启动0xc0000

11-06 ipz-185:win7系统vcf文件怎么打开

11-06 傻哥蹦迪:win10系统s4怎么打开usb调试

11-06 八神浩树gtaste:回收站清空了怎么恢复

11-06 妖尾之黑色守护:win10系统电脑没有1440x900

11-06 校园至尊魔王小说:win7系统浏览网页时字体

11-06 女斗合众国:win10系统访问共享文件夹提示请

11-06 tokyo hot n0654:恢复win7系统默认字体一招

11-06 雨酷仙境:设置win7系统转移临时文件夹腾出

11-06 阿穆纳伊之杖:win7系统开始菜单在右边还原

11-06 tunespotting:win10系统火狐flash插件总是

11-06 甘尔葛分析师：计谋网站seo关键词暴涨有什

11-06 蔡贵霖: 计谋网站seo关键词暴涨有什么秘密

11-06 博益网首页:ao3网页版进入不了解决方法

11-06 漏斗子专栏: 网站数据分析小白易懂精华篇

11-06 见证双虹怎么做:win7系统开启telnet命令的

11-06 颾狐蝶蜋:系统资源不足无法完成请求的服务

11-06 国光中学校歌:提交网站到alexa查询详细步骤

11-06 西安有情天:静态网页和动态网页的区别

11-06 红木雅尚斋:外部链接构造对网站的好处

11-06 前官礼遇：防止域名劫持–增强域安全性的10

11-06 密传二转答案: 中文分词算法有哪些

11-06 金泉家园邮编:百度快照劫持的表现及应对方

Python core.get_engine函数代码示例发布时间：2022-05-27

Python api.get_bottom_mappings_by_top_id函数代码示例发布时间：2022-05-27

Python util.grid_equal函数代码示例

1 Python 入门教程

Python入门教程 Python 是一种解释型、面向对象、动态数据类型的高级程序设计语言。 P

阅读：13925|2022-01-22

2 Python wikiutil.getFrontPage函数代码示例

Python wikiutil.getFrontPage函数代码示例

阅读：10291|2022-05-24

3 Python 简介

Python 简介 Python 是一个高层次的结合了解释性、编译性、互动性和面向对象的脚本

阅读：4165|2022-01-22

4 Python tests.group函数代码示例

Python tests.group函数代码示例

阅读：4064|2022-05-27

5 Python util.check_if_user_has_permission

Python util.check_if_user_has_permission函数代码示例

阅读：3889|2022-05-27

6 Python 操练实例98

Python 练习实例98 Python 100例题目：从键盘输入一个字符串，将小写字母全部转换成大

阅读：3538|2022-01-22

7 Python 环境搭建

Python 环境搭建本章节我们将向大家介绍如何在本地搭建 Python 开发环境。 Py

阅读：3066|2022-01-22

8 Python 基础语法

Python 基础语法 Python 语言与 Perl，C 和 Java 等语言有许多相似之处。但是，也

阅读：2723|2022-01-22

9 Python output.darkgreen函数代码示例

Python output.darkgreen函数代码示例

阅读：2682|2022-05-25

10 Python 中文编码

Python 中文编码前面章节中我们已经学会了如何用 Python 输出 Hello, World!，英文没

阅读：2344|2022-01-22

客服电话

电子邮件

Python core.create_resource函数代码示例

示例1: _test_create_security_group_rule

示例2: _test_handle_remote_group_invalid_input

示例3: test_get_bottom_mappings_by_top_id

示例4: get_pod_by_az_tenant

示例5: test_job_run_expire

示例6: _prepare_snat_test

示例7: test_job_run_expire

示例8: _prepare_port_test

示例9: _test_delete_security_group_rule_exception

示例10: _prepare_flow_classifier_test

示例11: _prepare_net_test

示例12: _prepare_server

示例13: _test_handle_network_dhcp_port

示例14: test_resource_routing_unique_key

示例15: _prepare_port_pair_group_test

示例16: create_ag_az

示例17: _test_get_security_group

示例18: test_get_failed_or_new_jobs

示例19: test_resources

示例20: _prepare_port_pair_test

请发表评论

全部评论

上一篇：

下一篇：

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.get_stdout函数代码示例

关于我们

产品与服务

解决方案

139-2527-9053