本文整理汇总了Python中quantum.openstack.common.jsonutils.loads函数的典型用法代码示例。如果您正苦于以下问题:Python loads函数的具体用法?Python loads怎么用?Python loads使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了loads函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_plugin_prefix_with_parent_resource
def test_plugin_prefix_with_parent_resource(self):
controller = self.DummySvcPlugin()
parent = dict(member_name="tenant",
collection_name="tenants")
member = {'custom_member_action': "GET"}
collections = {'collection_action': "GET"}
res_ext = extensions.ResourceExtension('tweedles', controller, parent,
path_prefix="/dummy_svc",
member_actions=member,
collection_actions=collections)
test_app = _setup_extensions_test_app(SimpleExtensionManager(res_ext))
index_response = test_app.get("/dummy_svc/tenants/1/tweedles")
self.assertEqual(200, index_response.status_int)
response = test_app.get("/dummy_svc/tenants/1/"
"tweedles/1/custom_member_action")
self.assertEqual(200, response.status_int)
self.assertEqual(jsonutils.loads(response.body)['member_action'],
"value")
response = test_app.get("/dummy_svc/tenants/2/"
"tweedles/collection_action")
self.assertEqual(200, response.status_int)
self.assertEqual(jsonutils.loads(response.body)['collection'],
"value")
开发者ID:Apsu,项目名称:quantum,代码行数:26,代码来源:test_extensions.py
示例2: test_plug_l2_gw_port_attachment
def test_plug_l2_gw_port_attachment(self):
tenant_id = 'pippo'
node_uuid = _uuid()
lswitch = nvplib.create_lswitch(self.fake_cluster, tenant_id,
'fake-switch')
gw_id = self._create_gw_service(node_uuid, 'fake-gw')['uuid']
lport = nvplib.create_lport(self.fake_cluster,
lswitch['uuid'],
tenant_id,
_uuid(),
'fake-gw-port',
gw_id,
True)
json.loads(nvplib.plug_l2_gw_service(self.fake_cluster,
lswitch['uuid'],
lport['uuid'],
gw_id))
uri = nvplib._build_uri_path(nvplib.LSWITCHPORT_RESOURCE,
lport['uuid'],
lswitch['uuid'],
is_attachment=True)
resp_obj = json.loads(
nvplib.do_single_request("GET", uri,
cluster=self.fake_cluster))
self.assertIn('LogicalPortAttachment', resp_obj)
self.assertEqual(resp_obj['LogicalPortAttachment']['type'],
'L2GatewayAttachment')
开发者ID:StackOps,项目名称:quantum,代码行数:27,代码来源:test_nvplib.py
示例3: test_resource_extension_for_get_custom_collection_action
def test_resource_extension_for_get_custom_collection_action(self):
controller = self.ResourceExtensionController()
collections = {"custom_collection_action": "GET"}
res_ext = extensions.ResourceExtension("tweedles", controller, collection_actions=collections)
test_app = _setup_extensions_test_app(SimpleExtensionManager(res_ext))
response = test_app.get("/tweedles/custom_collection_action")
self.assertEqual(200, response.status_int)
LOG.debug(jsonutils.loads(response.body))
self.assertEqual(jsonutils.loads(response.body)["collection"], "value")
开发者ID:cuiwow,项目名称:quantum,代码行数:10,代码来源:test_extensions.py
示例4: test_resource_ext_with_custom_member_action_gets_plugin_prefix
def test_resource_ext_with_custom_member_action_gets_plugin_prefix(self):
controller = self.DummySvcPlugin()
member = {"custom_member_action": "GET"}
collections = {"collection_action": "GET"}
res_ext = extensions.ResourceExtension(
"tweedles", controller, path_prefix="/dummy_svc", member_actions=member, collection_actions=collections
)
test_app = _setup_extensions_test_app(SimpleExtensionManager(res_ext))
response = test_app.get("/dummy_svc/tweedles/1/custom_member_action")
self.assertEqual(200, response.status_int)
self.assertEqual(jsonutils.loads(response.body)["member_action"], "value")
response = test_app.get("/dummy_svc/tweedles/collection_action")
self.assertEqual(200, response.status_int)
self.assertEqual(jsonutils.loads(response.body)["collection"], "value")
开发者ID:cuiwow,项目名称:quantum,代码行数:16,代码来源:test_extensions.py
示例5: _goose_handler
def _goose_handler(req, res):
#NOTE: This only handles JSON responses.
# You can use content type header to test for XML.
data = jsonutils.loads(res.body)
data['FOXNSOX:googoose'] = req.GET.get('chewing')
res.body = jsonutils.dumps(data)
return res
开发者ID:FreescaleSemiconductor,项目名称:quantum,代码行数:7,代码来源:foxinsocks.py
示例6: _bands_handler
def _bands_handler(req, res):
#NOTE: This only handles JSON responses.
# You can use content type header to test for XML.
data = jsonutils.loads(res.body)
data['FOXNSOX:big_bands'] = 'Pig Bands!'
res.body = jsonutils.dumps(data)
return res
开发者ID:FreescaleSemiconductor,项目名称:quantum,代码行数:7,代码来源:foxinsocks.py
示例7: _handler
def _handler(self, client_sock, client_addr):
"""Handle incoming lease relay stream connection.
This method will only read the first 1024 bytes and then close the
connection. The limit exists to limit the impact of misbehaving
clients.
"""
try:
msg = client_sock.recv(1024)
data = jsonutils.loads(msg)
client_sock.close()
network_id = data['network_id']
if not uuidutils.is_uuid_like(network_id):
raise ValueError(_("Network ID %s is not a valid UUID") %
network_id)
ip_address = str(netaddr.IPAddress(data['ip_address']))
lease_remaining = int(data['lease_remaining'])
self.callback(network_id, ip_address, lease_remaining)
except ValueError as e:
LOG.warn(_('Unable to parse lease relay msg to dict.'))
LOG.warn(_('Exception value: %s'), e)
LOG.warn(_('Message representation: %s'), repr(msg))
except Exception as e:
LOG.exception(_('Unable update lease. Exception'))
开发者ID:alanmeadows,项目名称:quantum,代码行数:25,代码来源:dhcp_agent.py
示例8: get_port_status
def get_port_status(controller, lswitch_id, port_id):
"""Retrieve the operational status of the port"""
# Make sure the network exists first
try:
do_single_request("GET", "/ws.v1/lswitch/%s" % (lswitch_id),
controller=controller)
except NvpApiClient.ResourceNotFound as e:
LOG.error("Network not found, Error: %s" % str(e))
raise exception.NetworkNotFound(net_id=lswitch_id)
except NvpApiClient.NvpApiException as e:
raise exception.QuantumException()
try:
r = do_single_request(
"GET",
"/ws.v1/lswitch/%s/lport/%s/status" % (lswitch_id, port_id),
controller=controller)
r = jsonutils.loads(r)
except NvpApiClient.ResourceNotFound as e:
LOG.error("Port not found, Error: %s" % str(e))
raise exception.PortNotFound(port_id=port_id, net_id=lswitch_id)
except NvpApiClient.NvpApiException as e:
raise exception.QuantumException()
if r['link_status_up'] is True:
return "UP"
else:
return "DOWN"
开发者ID:LuizOz,项目名称:quantum,代码行数:26,代码来源:nvplib.py
示例9: create_port
def create_port(tenant, network, port_init_state, **params):
# Check initial state -- this throws an exception if the port state is
# invalid
check_port_state(port_init_state)
controller = params["controller"]
ls_uuid = network
admin_status = True
if port_init_state == "DOWN":
admin_status = False
lport_obj = {"admin_status_enabled": admin_status}
path = "/ws.v1/lswitch/" + ls_uuid + "/lport"
try:
resp_obj = do_single_request("POST",
path,
jsonutils.dumps(lport_obj),
controller=controller)
except NvpApiClient.ResourceNotFound as e:
LOG.error("Network not found, Error: %s" % str(e))
raise exception.NetworkNotFound(net_id=network)
except NvpApiClient.NvpApiException as e:
raise exception.QuantumException()
result = jsonutils.loads(resp_obj)
result['port-op-status'] = get_port_status(controller, ls_uuid,
result['uuid'])
return result
开发者ID:LuizOz,项目名称:quantum,代码行数:30,代码来源:nvplib.py
示例10: update_port
def update_port(network, port_id, **params):
controller = params["controller"]
lport_obj = {}
if "state" in params:
state = params["state"]
check_port_state(state)
admin_status = True
if state == "DOWN":
admin_status = False
lport_obj["admin_status_enabled"] = admin_status
uri = "/ws.v1/lswitch/" + network + "/lport/" + port_id
try:
resp_obj = do_single_request("PUT",
uri,
jsonutils.dumps(lport_obj),
controller=controller)
except NvpApiClient.ResourceNotFound as e:
LOG.error("Port or Network not found, Error: %s" % str(e))
raise exception.PortNotFound(port_id=port_id, net_id=network)
except NvpApiClient.NvpApiException as e:
raise exception.QuantumException()
obj = jsonutils.loads(resp_obj)
obj["port-op-status"] = get_port_status(controller, network, obj["uuid"])
return obj
开发者ID:LuizOz,项目名称:quantum,代码行数:27,代码来源:nvplib.py
示例11: _check_is_create_user
def _check_is_create_user(self,request,re,info):
'''Withdraw tenant_id from re ,read body from write in database'''
if 'resource' in info and info['resource']=='users' \
and 'method' in info and info['method']=='POST':
print '[auth] create user'
re_json=json.loads(re[0])
print '[auth] re_json = ',re_json
userid=re_json['user']['id']
body=json.loads(request.body)
username=body['user']['name']
userpasswd=body['user']['passwd']
usercompany=body['user']['company']
session=Session()
session.add(User(name=username,passwd=userpasswd,id=userid,company=usercompany))
session.commit()
return True
return False
开发者ID:wangchang,项目名称:OpenStack-MPLS-FOR-WAN,代码行数:17,代码来源:huawei_auth.py
示例12: test_get_resources
def test_get_resources(self):
app = _setup_extensions_test_app()
response = app.get("/dummy_resources/1?chewing=newblue")
response_data = jsonutils.loads(response.body)
self.assertEqual('newblue', response_data['FOXNSOX:googoose'])
self.assertEqual("Pig Bands!", response_data['FOXNSOX:big_bands'])
开发者ID:Apsu,项目名称:quantum,代码行数:8,代码来源:test_extensions.py
示例13: test_resource_extension_with_custom_member_action
def test_resource_extension_with_custom_member_action(self):
controller = self.ResourceExtensionController()
member = {"custom_member_action": "GET"}
res_ext = extensions.ResourceExtension("tweedles", controller, member_actions=member)
test_app = _setup_extensions_test_app(SimpleExtensionManager(res_ext))
response = test_app.get("/tweedles/some_id/custom_member_action")
self.assertEqual(200, response.status_int)
self.assertEqual(jsonutils.loads(response.body)["member_action"], "value")
开发者ID:cuiwow,项目名称:quantum,代码行数:9,代码来源:test_extensions.py
示例14: delete_all_ports
def delete_all_ports(controller, ls_uuid):
res = do_single_request("GET", "/ws.v1/lswitch/%s/lport?fields=uuid" %
ls_uuid, controller=controller)
res = jsonutils.loads(res)
for r in res["results"]:
do_single_request(
"DELETE",
"/ws.v1/lswitch/%s/lport/%s" % (ls_uuid, r["uuid"]),
controller=controller)
开发者ID:LuizOz,项目名称:quantum,代码行数:9,代码来源:nvplib.py
示例15: load_json
def load_json(cls, data, default_rule=None):
"""
Allow loading of JSON rule data.
"""
# Suck in the JSON data and parse the rules
rules = dict((k, parse_rule(v)) for k, v in jsonutils.loads(data).items())
return cls(rules, default_rule)
开发者ID:habuka036,项目名称:quantum,代码行数:9,代码来源:policy.py
示例16: get_configuration_dict
def get_configuration_dict(self, agent_db):
try:
conf = jsonutils.loads(agent_db.configurations)
except Exception:
msg = _('Configuration for agent %(agent_type)s on host %(host)s'
' is invalid.')
LOG.warn(msg, {'agent_type': agent_db.agent_type,
'host': agent_db.host})
conf = {}
return conf
开发者ID:Frostman,项目名称:quantum,代码行数:10,代码来源:agents_db.py
示例17: test_resource_ext_for_formatted_req_on_custom_collection_action
def test_resource_ext_for_formatted_req_on_custom_collection_action(self):
controller = self.ResourceExtensionController()
collections = {'custom_collection_action': "GET"}
res_ext = extensions.ResourceExtension('tweedles', controller,
collection_actions=collections)
test_app = _setup_extensions_test_app(SimpleExtensionManager(res_ext))
response = test_app.get("/tweedles/custom_collection_action.json")
self.assertEqual(200, response.status_int)
self.assertEqual(jsonutils.loads(response.body)['collection'], "value")
开发者ID:Apsu,项目名称:quantum,代码行数:11,代码来源:test_extensions.py
示例18: test_resource_extension_for_delete_custom_collection_action
def test_resource_extension_for_delete_custom_collection_action(self):
controller = self.ResourceExtensionController()
collections = {'custom_collection_action': "DELETE"}
res_ext = extensions.ResourceExtension('tweedles', controller,
collection_actions=collections)
test_app = _setup_extensions_test_app(SimpleExtensionManager(res_ext))
response = test_app.delete("/tweedles/custom_collection_action")
self.assertEqual(200, response.status_int)
self.assertEqual(jsonutils.loads(response.body)['collection'], 'value')
开发者ID:Apsu,项目名称:quantum,代码行数:11,代码来源:test_extensions.py
示例19: get_network
def get_network(controller, net_id):
path = "/ws.v1/lswitch/%s" % net_id
try:
resp_obj = do_single_request("GET", path, controller=controller)
network = jsonutils.loads(resp_obj)
except NvpApiClient.ResourceNotFound as e:
raise exception.NetworkNotFound(net_id=net_id)
except NvpApiClient.NvpApiException as e:
raise exception.QuantumException()
LOG.debug("Got network \"%s\": %s" % (net_id, network))
return network
开发者ID:LuizOz,项目名称:quantum,代码行数:11,代码来源:nvplib.py
示例20: _create_tz
def _create_tz(self, name):
post_uri = "/ws.v1/transport-zone"
body = {"display_name": name, "tags": [{"tag": "plugin-test"}]}
try:
resp_obj = self.quantum.api_client.request("POST",
post_uri,
jsonutils.dumps(body))
except NvpApiClient.NvpApiException as e:
LOG.error("Unknown API Error: %s" % str(e))
raise exception.QuantumException()
return jsonutils.loads(resp_obj)["uuid"]
开发者ID:jkoelker,项目名称:quantum,代码行数:11,代码来源:test_port.py
注:本文中的quantum.openstack.common.jsonutils.loads函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论