def on_delete(self, req, resp, account_id):
"""
Handles DELETE requests. Deletes the account from the database
"""
raise falcon.HTTPMethodNotAllowed
servers = Actions.get_servers(account_id)
for server_id in servers:
server.Actions.remove_account(server_id, account_id)
result = server.Actions.delete_account(account_id)
resp.status = falcon.HTTP_200
if result == 1:
result = 'success'
else:
result = 'failed'
jsonresp = {'deleted': result}
resp.body = json.dumps(jsonresp)
python类HTTP_200的实例源码
def on_get(self, req, resp):
"""
Echo back user provided content query string in TwiML:
Example:
Query strings:
content: OK
TwiML:
<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Say language="en-US" voice="alice">OK</Say>
</Response>
"""
content = req.get_param('content')
loop = req.get_param('loop')
r = twiml.Response()
r.say(content, voice='alice', loop=loop, language="en-US")
resp.status = falcon.HTTP_200
resp.body = str(r)
resp.content_type = 'application/xml'
def on_get(self, req, resp):
"""Handles GET requests"""
resp.body = """
<html>
<head>
<title>Quote API Server</title>
</head>
<body>
<p>This is a toy JSON API server example.</p>
<p>Make a GET request to <a href="%s/quote">%s/quote</a></p>
</body>
</html>
""" % (self.prefix, self.prefix)
resp.content_type = "text/html"
resp.status = falcon.HTTP_200
# A Falcon middleware to implement validation of the Host header in requests
def test_status_retrieve(self):
"""
Verify retrieving Status.
"""
with mock.patch('cherrypy.engine.publish') as _publish:
manager = mock.MagicMock(StoreHandlerManager)
_publish.return_value = [manager]
child = MagicMock(value='')
self.return_value._children = [child]
self.return_value.leaves = self.return_value._children
manager.get.return_value = self.return_value
body = self.simulate_request('/api/v0/status')
self.assertEqual(self.srmock.status, falcon.HTTP_200)
self.assertEqual(
json.loads(self.astatus),
json.loads(body[0]))
def test_clusters_listing(self):
"""
Verify listing Clusters.
"""
with mock.patch('cherrypy.engine.publish') as _publish:
manager = mock.MagicMock(StoreHandlerManager)
_publish.return_value = [manager]
return_value = clusters.Clusters(
clusters=[clusters.Cluster.new(
name=self.cluster_name, status='', hostset=[])])
manager.list.return_value = return_value
body = self.simulate_request('/api/v0/clusters')
self.assertEqual(falcon.HTTP_200, self.srmock.status)
self.assertEqual(
[self.cluster_name],
json.loads(body[0]))
def test_cluster_delete(self):
"""
Verify deleting a cluster.
"""
with mock.patch('cherrypy.engine.publish') as _publish:
manager = mock.MagicMock(StoreHandlerManager)
_publish.return_value = [manager]
# Verify with proper deletion
manager.get.return_value = MagicMock()
body = self.simulate_request(
'/api/v0/cluster/development', method='DELETE')
# Get is called to verify cluster exists
self.assertEquals(falcon.HTTP_200, self.srmock.status)
self.assertEquals('{}', body[0])
# Verify when key doesn't exist
manager.delete.side_effect = etcd.EtcdKeyNotFound
body = self.simulate_request(
'/api/v0/cluster/development', method='DELETE')
self.assertEquals(falcon.HTTP_404, self.srmock.status)
self.assertEquals('{}', body[0])
def test_cluster_hosts_retrieve(self):
"""
Verify retrieving a cluster host list.
"""
with mock.patch('cherrypy.engine.publish') as _publish:
manager = mock.MagicMock(StoreHandlerManager)
_publish.return_value = [manager]
# Verify if the cluster exists the host list is returned
manager.get.return_value = make_new(CLUSTER_WITH_FLAT_HOST)
body = self.simulate_request('/api/v0/cluster/cluster/hosts')
self.assertEqual(falcon.HTTP_200, self.srmock.status)
self.assertEqual(
['10.2.0.2'],
json.loads(body[0]))
# Verify bad cluster name returns the proper result
manager.get.side_effect = Exception
body = self.simulate_request('/api/v0/cluster/bogus/hosts')
self.assertEqual(falcon.HTTP_404, self.srmock.status)
self.assertEqual({}, json.loads(body[0]))
def test_cluster_host_insert(self):
"""
Verify insertion of host in a cluster.
"""
with mock.patch('cherrypy.engine.publish') as _publish:
manager = mock.MagicMock(StoreHandlerManager)
_publish.return_value = [manager]
# Verify inserting host returns the proper result
manager.get.return_value = make_new(CLUSTER_WITH_FLAT_HOST)
body = self.simulate_request(
'/api/v0/cluster/developent/hosts/10.2.0.3', method='PUT')
self.assertEqual(falcon.HTTP_200, self.srmock.status)
self.assertEqual({}, json.loads(body[0]))
# Verify bad cluster name returns the proper result
manager.get.side_effect = Exception
body = self.simulate_request(
'/api/v0/cluster/bogus/hosts/10.2.0.3', method='PUT')
self.assertEqual(falcon.HTTP_404, self.srmock.status)
self.assertEqual({}, json.loads(body[0]))
def test_cluster_host_delete(self):
"""
Verify deletion of host in a cluster.
"""
with mock.patch('cherrypy.engine.publish') as _publish:
manager = mock.MagicMock(StoreHandlerManager)
_publish.return_value = [manager]
# Verify deleting host returns the proper result
manager.get.return_value = make_new(CLUSTER_WITH_FLAT_HOST)
body = self.simulate_request(
'/api/v0/cluster/development/hosts/10.2.0.2', method='DELETE')
self.assertEqual(falcon.HTTP_200, self.srmock.status)
self.assertEqual({}, json.loads(body[0]))
# Verify bad cluster name returns the proper result
manager.get.side_effect = Exception
body = self.simulate_request(
'/api/v0/cluster/bogus/hosts/10.2.0.2', method='DELETE')
self.assertEqual(falcon.HTTP_404, self.srmock.status)
self.assertEqual({}, json.loads(body[0]))
def test_cluster_deploy_retrieve(self):
"""
Verify retrieving a cluster deploy.
"""
with mock.patch('cherrypy.engine.publish') as _publish:
manager = mock.MagicMock(StoreHandlerManager)
_publish.return_value = [manager]
test_cluster_deploy = make_new(CLUSTER_DEPLOY)
# Verify if the cluster deploy exists the data is returned
manager.get.return_value = test_cluster_deploy
body = self.simulate_request('/api/v0/cluster/development/deploy')
self.assertEqual(falcon.HTTP_200, self.srmock.status)
self.assertEqual(json.loads(test_cluster_deploy.to_json()), json.loads(body[0]))
# Verify no cluster deploy returns the proper result
manager.reset_mock()
manager.get.side_effect = (
test_cluster_deploy,
Exception)
body = self.simulate_request('/api/v0/cluster/development/deploy')
self.assertEqual(falcon.HTTP_204, self.srmock.status)
self.assertEqual([], body) # Empty data
def test_host_creds_retrieve(self):
"""
Verify retrieving Host Credentials.
"""
with mock.patch('cherrypy.engine.publish') as _publish:
manager = mock.MagicMock(StoreHandlerManager)
_publish.return_value = [manager]
# Verify if the host exists the data is returned
manager.get.return_value = make_new(HOST)
body = self.simulate_request('/api/v0/host/10.2.0.2/creds')
# datasource's get should have been called once
self.assertEqual(self.srmock.status, falcon.HTTP_200)
self.assertEqual(
json.loads(HOST_CREDS_JSON),
json.loads(body[0]))
# Verify no host returns the proper result
manager.reset_mock()
manager.get.side_effect = Exception
body = self.simulate_request('/api/v0/host/10.9.9.9/creds')
self.assertEqual(self.srmock.status, falcon.HTTP_404)
self.assertEqual({}, json.loads(body[0]))
def test_host_status_retrieve_host_only(self):
"""
Verify retrieving Host status when it is in a host only cluster.
"""
with mock.patch('cherrypy.engine.publish') as _publish:
manager = mock.MagicMock(StoreHandlerManager)
_publish.return_value = [manager]
test_host = make_new(HOST)
test_cluster = make_new(CLUSTER)
test_cluster.hostset = [test_host.address]
# Verify if the host exists the data is returned
manager.get.side_effect = (
test_host,
test_cluster)
body = self.simulate_request('/api/v0/host/10.2.0.2/status')
self.assertEqual(self.srmock.status, falcon.HTTP_200)
result = json.loads(body[0])
self.assertEquals(C.CLUSTER_TYPE_HOST, result['type'])
self.assertEquals('available', result['host']['status'])
self.assertEquals({}, result['container_manager'])
def test_networks_listing(self):
"""
Verify listing Networks.
"""
with mock.patch('cherrypy.engine.publish') as _publish:
manager = mock.MagicMock(StoreHandlerManager)
_publish.return_value = [manager]
return_value = networks.Networks(
networks=[networks.Network.new(name=self.network_name)])
manager.list.return_value = return_value
body = self.simulate_request('/api/v0/networks')
self.assertEqual(falcon.HTTP_200, self.srmock.status)
self.assertEqual(
[self.network_name],
json.loads(body[0]))
def on_delete(self, req, resp, name):
"""
Handles the deletion of a Cluster.
:param req: Request instance that will be passed through.
:type req: falcon.Request
:param resp: Response instance that will be passed through.
:type resp: falcon.Response
:param name: The name of the Cluster being deleted.
:type name: str
"""
resp.body = '{}'
try:
store_manager = cherrypy.engine.publish('get-store-manager')[0]
store_manager.delete(Cluster.new(name=name))
resp.status = falcon.HTTP_200
self.logger.info(
'Deleted cluster {0} per request.'.format(name))
except:
self.logger.info(
'Deleting for non-existent cluster {0} requested.'.format(
name))
resp.status = falcon.HTTP_404
def on_delete(self, req, resp, name, address):
"""
Handles DELETE requests for individual hosts in a Cluster.
This removes a single host from the cluster, idempotently.
:param req: Request instance that will be passed through.
:type req: falcon.Request
:param resp: Response instance that will be passed through.
:type resp: falcon.Response
:param name: The name of the Cluster being requested.
:type name: str
:param address: The address of the Host being requested.
:type address: str
"""
try:
util.etcd_cluster_remove_host(name, address)
resp.status = falcon.HTTP_200
except KeyError:
resp.status = falcon.HTTP_404
def on_get(self, req, resp):
"""
Handles GET requests for Hosts.
:param req: Request instance that will be passed through.
:type req: falcon.Request
:param resp: Response instance that will be passed through.
:type resp: falcon.Response
"""
try:
store_manager = cherrypy.engine.publish('get-store-manager')[0]
hosts = store_manager.list(Hosts(hosts=[]))
if len(hosts.hosts) == 0:
raise Exception()
resp.status = falcon.HTTP_200
req.context['model'] = hosts
except Exception:
# This was originally a "no content" but I think a 404 makes
# more sense if there are no hosts
self.logger.warn(
'Store does not have any hosts. Returning [] and 404.')
resp.status = falcon.HTTP_404
req.context['model'] = None
return
def on_get(self, req, resp):
"""
Handles GET requests for Networks.
:param req: Request instance that will be passed through.
:type req: falcon.Request
:param resp: Response instance that will be passed through.
:type resp: falcon.Response
"""
try:
store_manager = cherrypy.engine.publish('get-store-manager')[0]
networks = store_manager.list(Networks(networks=[]))
if len(networks.networks) == 0:
raise Exception()
resp.status = falcon.HTTP_200
resp.body = json.dumps([
network.name for network in networks.networks])
except Exception:
self.logger.warn(
'Store does not have any networks. Returning [] and 404.')
resp.status = falcon.HTTP_404
req.context['model'] = None
return
def on_delete(self, req, resp, name):
"""
Handles the Deletion of a Network.
:param req: Request instance that will be passed through.
:type req: falcon.Request
:param resp: Response instance that will be passed through.
:type resp: falcon.Response
:param name: The friendly name of the network.
:type address: str
"""
resp.body = '{}'
store_manager = cherrypy.engine.publish('get-store-manager')[0]
try:
store_manager.delete(Network.new(name=name))
resp.status = falcon.HTTP_200
except Exception as error:
self.logger.warn('{}: {}'.format(type(error), error))
resp.status = falcon.HTTP_404
def on_get(self, request, response, room_alias=None):
"""Called when a GET request is sent to /rooms/{room_alias}"""
response.body = "{}"
if self.handler(room_alias):
response.status = falcon.HTTP_200
self.api.create_room(alias=room_alias)
else:
response.status = falcon.HTTP_404
def on_get(self, request, response, user_id=None):
"""Responds to GET request for users."""
response.body = "{}"
if self.handler(user_id):
response.status = falcon.HTTP_200
self.api.register(utils.mxid2localpart(user_id))
else:
response.status = falcon.HTTP_404
