def test_get_http_status(self):
assert falcon.get_http_status(404) == falcon.HTTP_404
assert falcon.get_http_status(404.3) == falcon.HTTP_404
assert falcon.get_http_status('404.3') == falcon.HTTP_404
assert falcon.get_http_status(404.9) == falcon.HTTP_404
assert falcon.get_http_status('404') == falcon.HTTP_404
assert falcon.get_http_status(123) == '123 Unknown'
with pytest.raises(ValueError):
falcon.get_http_status('not_a_number')
with pytest.raises(ValueError):
falcon.get_http_status(0)
with pytest.raises(ValueError):
falcon.get_http_status(0)
with pytest.raises(ValueError):
falcon.get_http_status(99)
with pytest.raises(ValueError):
falcon.get_http_status(-404.3)
with pytest.raises(ValueError):
falcon.get_http_status('-404')
with pytest.raises(ValueError):
falcon.get_http_status('-404.3')
assert falcon.get_http_status(123, 'Go Away') == '123 Go Away'
python类HTTP_404的实例源码
def get_album_art(song_id: hug.types.text, response=None):
try:
api = music_api_names['offline_api']
except KeyError:
response.status = falcon.HTTP_400
return "Not in offline mode"
db = sqlite3.connect(api._db_path)
try:
cursor = db.execute("SELECT albumArt FROM albumArts WHERE songId=?", [song_id])
result = cursor.fetchone()
if not result:
response.status = falcon.HTTP_404
return None
return BytesIO(result[0])
finally:
db.close()
def on_get(self, req, resp, design_id):
"""Method Handler for GET design singleton.
:param req: Falcon request object
:param resp: Falcon response object
:param design_id: UUID of the design resource
"""
source = req.params.get('source', 'designed')
try:
design = None
if source == 'compiled':
design = self.orchestrator.get_effective_site(design_id)
elif source == 'designed':
design = self.orchestrator.get_described_site(design_id)
resp.body = json.dumps(design.obj_to_simple())
except errors.DesignError:
self.error(req.context, "Design %s not found" % design_id)
self.return_error(
resp,
falcon.HTTP_404,
message="Design %s not found" % design_id,
retry=False)
def on_get(self, req, resp, task_id):
"""Handler for GET method."""
try:
task = self.state_manager.get_task(uuid.UUID(task_id))
if task is None:
self.info(req.context, "Task %s does not exist" % task_id)
self.return_error(
resp,
falcon.HTTP_404,
message="Task %s does not exist" % task_id,
retry=False)
return
resp.body = json.dumps(task.to_dict())
resp.status = falcon.HTTP_200
except Exception as ex:
self.error(req.context, "Unknown error: %s" % (str(ex)))
self.return_error(
resp, falcon.HTTP_500, message="Unknown error", retry=False)
def on_get(self, req, resp):
q = req.get_param('q')
resp.content_type = 'text/plain; charset=utf-8'
if not q:
resp.body = 'Bad Request'
resp.status = falcon.HTTP_400
return
try:
zenline = zenlines[q]
except KeyError:
resp.body = 'Not Found'
resp.status = falcon.HTTP_404
return
resp.body = zenline
resp.status = falcon.HTTP_200
def on_get(self, req, resp, project=None, repo=None):
if project:
if not self.db.check_project_exists(project):
resp.body = json.dumps({'Error': 'Project key {} not found!'.format(project)})
resp.status = falcon.HTTP_404
return
latest = req.get_param_as_bool(name='latest', required=False)
weeks_ago = req.get_param_as_int(name='weeks_ago', required=False)
target_timestamp = get_epoch_time_of_weeks_ago(weeks=weeks_ago)
if project and repo:
results = self.db.get_results_for_repo(reponame=repo, timestamp=target_timestamp)
elif project:
if latest:
results = self.db.get_latest_results_for_project(project=project)
else:
results = self.db.get_results_for_project(project=project, timestamp=target_timestamp)
else:
if latest:
results = self.db.get_latest_results()
else:
results = self.db.get_all_results(timestamp=target_timestamp)
resp.body = json.dumps({'results': format_results(results)})
resp.status = falcon.HTTP_200
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_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_network_delete(self):
"""
Verify deleting a network.
"""
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/network/development', method='DELETE')
# Get is called to verify network 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/network/development', method='DELETE')
self.assertEquals(falcon.HTTP_404, self.srmock.status)
self.assertEquals('{}', body[0])
def on_get(self, req, resp, name):
"""
Handles GET requests for Cluster 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
:param name: The name of the Cluster being requested.
:type name: str
"""
try:
store_manager = cherrypy.engine.publish('get-store-manager')[0]
cluster = store_manager.get(Cluster.new(name=name))
except:
resp.status = falcon.HTTP_404
return
resp.body = json.dumps(cluster.hostset)
resp.status = falcon.HTTP_200
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_get(self, req, resp, name):
"""
Handles retrieval of an existing 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
"""
try:
store_manager = cherrypy.engine.publish('get-store-manager')[0]
network = store_manager.get(Network.new(name=name))
resp.status = falcon.HTTP_200
req.context['model'] = network
except:
resp.status = falcon.HTTP_404
return
def test_second_hook_raise_HTTPNotFound(self):
bad_headers = {
'Content-Type': 'application/json',
'URL-METHODS': 'GET, PUT',
}
result = self.simulate_post(headers=bad_headers, body=self.body)
self.assertEqual(falcon.HTTP_404, result.status)
def test_wrong_router(self):
result = self.simulate_get(path='/some/wrong/path', body=self.body)
self.assertEqual(falcon.HTTP_404, result.status)
def test_skip_process_resource(self):
global context
app = falcon.API(middleware=[RequestTimeMiddleware()])
app.add_route('/', MiddlewareClassResource())
client = testing.TestClient(app)
response = client.simulate_request(path='/404')
assert response.status == falcon.HTTP_404
assert 'start_time' in context
assert 'mid_time' not in context
assert 'end_time' in context
assert not context['req_succeeded']
def test_404_without_body(self, client):
client.app.add_route('/404', NotFoundResource())
response = client.simulate_request(path='/404')
assert response.status == falcon.HTTP_404
assert not response.content
def test_404_with_body(self, client):
client.app.add_route('/404', NotFoundResourceWithBody())
response = client.simulate_request(path='/404')
assert response.status == falcon.HTTP_404
assert response.content
expected_body = {
u'title': u'404 Not Found',
u'description': u'Not Found'
}
assert response.json == expected_body
def on_get(self, req, resp, design_id, kind, name):
source = req.params.get('source', 'designed')
try:
design = None
if source == 'compiled':
design = self.orchestrator.get_effective_site(design_id)
elif source == 'designed':
design = self.orchestrator.get_described_site(design_id)
part = None
if kind == 'Site':
part = design.get_site()
elif kind == 'Network':
part = design.get_network(name)
elif kind == 'NetworkLink':
part = design.get_network_link(name)
elif kind == 'HardwareProfile':
part = design.get_hardware_profile(name)
elif kind == 'HostProfile':
part = design.get_host_profile(name)
elif kind == 'BaremetalNode':
part = design.get_baremetal_node(name)
else:
self.error(req.context, "Kind %s unknown" % kind)
self.return_error(
resp,
falcon.HTTP_404,
message="Kind %s unknown" % kind,
retry=False)
return
resp.body = json.dumps(part.obj_to_simple())
except errors.DesignError as dex:
self.error(req.context, str(dex))
self.return_error(
resp, falcon.HTTP_404, message=str(dex), retry=False)
except Exception as exc:
self.error(req.context, str(exc))
self.return_error(
resp.falcon.HTTP_500, message=str(exc), retry=False)
def not_found(message="The requested resource does not exist"):
raise falcon.HTTPNotFound(description=message, code=falcon.HTTP_404)
def on_get(self, req, resp, account_id):
"""Handles GET requests"""
watched = Actions.get_watched(account_id)
if watched is None:
resp.status = falcon.HTTP_404
return
servers = Actions.get_servers(account_id)
resp.status = falcon.HTTP_200 # This is the default status
json_resp = {'account_id': account_id, 'watched': watched, 'servers': servers}
resp.body = json.dumps(json_resp)
def on_get(self, req, resp, server_id):
"""Handles GET requests"""
accounts = Actions.get_accounts(server_id)
if accounts is None:
resp.status = falcon.HTTP_404
return
json_response = {'server_id': server_id, 'accounts': accounts}
resp.status = falcon.HTTP_200 # This is the default status
resp.body = json.dumps(json_response)
def on_put(self, req, resp, server_id):
"""Handles PUT requests"""
resp.status = falcon.HTTP_404 # Disabled!
return
try:
raw_json = req.stream.read()
except Exception as ex:
raise falcon.HTTPError(falcon.HTTP_400,
'Error',
ex.message)
try:
result_json = json.loads(raw_json, encoding='utf-8')
except ValueError:
raise falcon.HTTPError(falcon.HTTP_400,
'Malformed JSON',
'Could not decode the request body. The '
'JSON was incorrect.')
accounts = Actions.get_accounts(server_id)
if accounts is None:
resp.status = falcon.HTTP_404
return
accounts = result_json['accounts']
for account_id in accounts:
Actions.add_account(server_id, account_id)
account.Actions.add_server(account_id, server_id)
resp.status = falcon.HTTP_200 # This is the default status
jsonresp = {'server_id': server_id, 'accounts': Actions.get_accounts(server_id)}
resp.body = json.dumps(jsonresp)
def on_get(self, req, resp, project=None):
if project:
if not self.db.check_project_exists(project):
resp.body = json.dumps({'Error': 'Project key {} not found!'.format(project)})
resp.status = falcon.HTTP_404
return
if project:
repos = self.db.get_repos_for_project(project)
body = {'repositories': repos}
else:
projects = self.db.get_projects()
body = {'projects': projects}
resp.body = json.dumps(body)
resp.status = falcon.HTTP_200
def test_clusters_listing_with_no_etcd_result(self):
"""
Verify listing Clusters handles no etcd result properly.
"""
with mock.patch('cherrypy.engine.publish') as _publish:
_publish.return_value = [[[], etcd.EtcdKeyNotFound()]]
body = self.simulate_request('/api/v0/clusters')
self.assertEqual(self.srmock.status, falcon.HTTP_404)
self.assertEqual('{}', body[0])
def test_cluster_retrieve(self):
"""
Verify retrieving a cluster.
"""
with mock.patch('cherrypy.engine.publish') as _publish:
manager = mock.MagicMock(StoreHandlerManager)
_publish.return_value = [manager]
test_cluster = make_new(CLUSTER_WITH_HOST)
# Verify if the cluster exists the data is returned
manager.get.return_value = test_cluster
manager.list.return_value = make_new(HOSTS)
body = self.simulate_request('/api/v0/cluster/development')
self.assertEqual(self.srmock.status, falcon.HTTP_200)
self.assertEqual(
json.loads(test_cluster.to_json_with_hosts()),
json.loads(body[0]))
# Verify no cluster returns the proper result
manager.get.reset_mock()
manager.get.side_effect = Exception
body = self.simulate_request('/api/v0/cluster/bogus')
self.assertEqual(falcon.HTTP_404, self.srmock.status)
self.assertEqual({}, json.loads(body[0]))
def test_hosts_listing_with_no_hosts(self):
"""
Verify listing Hosts when no hosts exists.
"""
with mock.patch('cherrypy.engine.publish') as _publish:
_publish.return_value = Hosts(hosts=[])
body = self.simulate_request('/api/v0/hosts')
# datasource's get should have been called once
self.assertEqual(self.srmock.status, falcon.HTTP_404)
self.assertEqual({}, json.loads(body[0]))
