def test_get_run_by_id(self):
response = self.app.get('/runs/1234')
self.assertEqual(http_client.NOT_FOUND, response.status_code)
pb = 'tests/fixtures/playbooks/hello_world.yml'
response = self.app.post(
'/runs',
data=json.dumps(dict(playbook_path=pb,
inventory_file='localhosti,',
options={'connection': 'local'})),
content_type='application/json')
res_dict = json.loads(response.data)
run_id = res_dict['id']
self.assertEqual(http_client.OK, response.status_code)
response = self.app.get('/runs/{}'.format(run_id))
self.assertEqual(http_client.OK, response.status_code)
self._wait_for_run_complete(run_id)
python类NOT_FOUND的实例源码
def test_get_state(self):
"""Test getting an instance state."""
self.impl.get.return_value = {
'name': 'foo.bar#0000000001', 'host': 'baz1', 'state': 'running'
}
resp = self.client.get('/state/foo.bar#0000000001')
resp_json = b''.join(resp.response)
self.assertEqual(json.loads(resp_json.decode()), {
'name': 'foo.bar#0000000001', 'oom': None, 'signal': None,
'expires': None, 'when': None, 'host': 'baz1',
'state': 'running', 'exitcode': None
})
self.assertEqual(resp.status_code, http_client.OK)
self.impl.get.assert_called_with('foo.bar#0000000001')
self.impl.get.return_value = None
resp = self.client.get('/state/foo.bar#0000000002')
self.assertEqual(resp.status_code, http_client.NOT_FOUND)
def _handle_error(url, response):
"""Handle response status codes."""
handlers = {
http_client.NOT_FOUND: NotFoundError(
'Resource not found: {}'.format(url)
),
http_client.FOUND: AlreadyExistsError(
'Resource already exists: {}'.format(url)
),
http_client.FAILED_DEPENDENCY: ValidationError(response),
http_client.UNAUTHORIZED: NotAuthorizedError(response),
http_client.BAD_REQUEST: BadRequestError(response),
}
if response.status_code in handlers:
raise handlers[response.status_code]
def init(api, _cors, impl):
"""Configures REST handlers for allocation resource."""
namespace = webutils.namespace(
api, __name__, 'Local nodeinfo redirect API.'
)
@namespace.route('/<hostname>/<path:path>')
class _NodeRedirect(restplus.Resource):
"""Redirects to local nodeinfo endpoint."""
def get(self, hostname, path):
"""Returns list of local instances."""
hostport = impl.get(hostname)
if not hostport:
return 'Host not found.', http_client.NOT_FOUND
url = utils.encode_uri_parts(path)
return flask.redirect('http://%s/%s' % (hostport, url),
code=http_client.FOUND)
def catalog_1(self, *tokens):
"""Outputs the contents of the specified catalog file, using the
name in the request path, directly to the client."""
try:
name = tokens[0]
except IndexError:
raise cherrypy.HTTPError(http_client.FORBIDDEN,
_("Directory listing not allowed."))
try:
fpath = self.repo.catalog_1(name,
pub=self._get_req_pub())
except srepo.RepositoryError as e:
# Treat any remaining repository error as a 404, but
# log the error and include the real failure
# information.
cherrypy.log("Request failed: {0}".format(str(e)))
raise cherrypy.HTTPError(http_client.NOT_FOUND, str(e))
self.__set_response_expires("catalog", 86400, 86400)
return serve_file(fpath, "text/plain; charset=utf-8")
def file_0(self, *tokens):
"""Outputs the contents of the file, named by the SHA-1 hash
name in the request path, directly to the client."""
try:
fhash = tokens[0]
except IndexError:
fhash = None
try:
fpath = self.repo.file(fhash, pub=self._get_req_pub())
except srepo.RepositoryFileNotFoundError as e:
raise cherrypy.HTTPError(http_client.NOT_FOUND, str(e))
except srepo.RepositoryError as e:
# Treat any remaining repository error as a 404, but
# log the error and include the real failure
# information.
cherrypy.log("Request failed: {0}".format(str(e)))
raise cherrypy.HTTPError(http_client.NOT_FOUND, str(e))
self.__set_response_expires("file", 86400*365, 86400*365)
return serve_file(fpath, "application/data")
def test_face_root(self):
"""Verify that files outside of the package content web root
cannot be accessed, and that files inside can be."""
depot_url = self.dc.get_depot_url()
# Since /usr/share/lib/pkg/web/ is the content web root,
# any attempts to go outside that directory should fail
# with a 404 error.
try:
urlopen("{0}/../../../../bin/pkg".format(depot_url))
except HTTPError as e:
if e.code != http_client.NOT_FOUND:
raise
f = urlopen("{0}/robots.txt".format(depot_url))
self.assertTrue(len(f.read()))
f.close()
def _refresh(self, http_request):
"""Refreshes the access_token.
Skip all the storage hoops and just refresh using the API.
Args:
http_request: callable, a callable that matches the method
signature of httplib2.Http.request, used to make
the refresh request.
Raises:
HttpAccessTokenRefreshError: When the refresh fails.
"""
query = '?scope=%s' % urllib.parse.quote(self.scope, '')
uri = META.replace('{?scope}', query)
response, content = http_request(
uri, headers={'Metadata-Flavor': 'Google'})
content = _from_bytes(content)
if response.status == http_client.OK:
try:
token_content = json.loads(content)
except Exception as e:
raise HttpAccessTokenRefreshError(str(e),
status=response.status)
self.access_token = token_content['access_token']
else:
if response.status == http_client.NOT_FOUND:
content += (' This can occur if a VM was created'
' with no service account or scopes.')
raise HttpAccessTokenRefreshError(content, status=response.status)
def test_get_member_for_invalid_id(self):
# | GIVEN |
# setting a valid member identity
self.test_resource_collection.members_identities = ('1',)
self.conn.get.side_effect = exceptions.ResourceNotFoundError(
method='GET', url='http://foo.bar:8000/redfish/v1/Fakes/2',
response=mock.Mock(status_code=http_client.NOT_FOUND))
# | WHEN & THEN |
self.assertRaises(exceptions.ResourceNotFoundError,
self.test_resource_collection.get_member, '2')
self.conn.get.assert_called_once_with(path='Fakes/2')
def test_not_found_error(self):
self.request.return_value.status_code = http_client.NOT_FOUND
self.request.return_value.json.side_effect = ValueError('no json')
with self.assertRaisesRegex(exceptions.ResourceNotFoundError,
'Resource http://foo.bar not found') as cm:
self.conn._op('GET', 'http://foo.bar')
exc = cm.exception
self.assertEqual(http_client.NOT_FOUND, exc.status_code)
def raise_for_response(method, url, response):
"""Raise a correct error class, if needed."""
if response.status_code < http_client.BAD_REQUEST:
return
elif response.status_code == http_client.NOT_FOUND:
raise ResourceNotFoundError(method, url, response)
elif response.status_code == http_client.BAD_REQUEST:
raise BadRequestError(method, url, response)
elif response.status_code in (http_client.UNAUTHORIZED,
http_client.FORBIDDEN):
raise AccessError(method, url, response)
elif response.status_code >= http_client.INTERNAL_SERVER_ERROR:
raise ServerSideError(method, url, response)
else:
raise HTTPError(method, url, response)
def test_get_failure(self):
http = request_mock(
http_client.NOT_FOUND, 'text/html', '<p>Error</p>')
with self.assertRaises(http_client.HTTPException):
_metadata.get(http, PATH)
# Verify mocks.
self.assertEqual(http.requests, 1)
self.assertEqual(http.uri, EXPECTED_URL)
self.assertEqual(http.method, 'GET')
self.assertIsNone(http.body)
self.assertEqual(http.headers, _metadata.METADATA_HEADERS)
def _environment_check_gce_helper(self, status_ok=True,
server_software=''):
if status_ok:
headers = {'status': http_client.OK}
headers.update(client._GCE_HEADERS)
else:
headers = {'status': http_client.NOT_FOUND}
http = http_mock.HttpMock(headers=headers)
with mock.patch('oauth2client.client.os') as os_module:
os_module.environ = {client._SERVER_SOFTWARE: server_software}
with mock.patch('oauth2client.transport.get_http_object',
return_value=http) as new_http:
if server_software == '':
self.assertFalse(client._in_gae_environment())
else:
self.assertTrue(client._in_gae_environment())
if status_ok and server_software == '':
self.assertTrue(client._in_gce_environment())
else:
self.assertFalse(client._in_gce_environment())
# Verify mocks.
if server_software == '':
new_http.assert_called_once_with(
timeout=client.GCE_METADATA_TIMEOUT)
self.assertEqual(http.requests, 1)
self.assertEqual(http.uri, client._GCE_METADATA_URI)
self.assertEqual(http.method, 'GET')
self.assertIsNone(http.body)
self.assertEqual(http.headers, client._GCE_HEADERS)
else:
new_http.assert_not_called()
self.assertEqual(http.requests, 0)
def test_get_failure(self):
http = request_mock(
http_client.NOT_FOUND, 'text/html', '<p>Error</p>')
with self.assertRaises(http_client.HTTPException):
_metadata.get(http, PATH)
# Verify mocks.
self.assertEqual(http.requests, 1)
self.assertEqual(http.uri, EXPECTED_URL)
self.assertEqual(http.method, 'GET')
self.assertIsNone(http.body)
self.assertEqual(http.headers, _metadata.METADATA_HEADERS)
def _environment_check_gce_helper(self, status_ok=True,
server_software=''):
if status_ok:
headers = {'status': http_client.OK}
headers.update(client._GCE_HEADERS)
else:
headers = {'status': http_client.NOT_FOUND}
http = http_mock.HttpMock(headers=headers)
with mock.patch('oauth2client.client.os') as os_module:
os_module.environ = {client._SERVER_SOFTWARE: server_software}
with mock.patch('oauth2client.transport.get_http_object',
return_value=http) as new_http:
if server_software == '':
self.assertFalse(client._in_gae_environment())
else:
self.assertTrue(client._in_gae_environment())
if status_ok and server_software == '':
self.assertTrue(client._in_gce_environment())
else:
self.assertFalse(client._in_gce_environment())
# Verify mocks.
if server_software == '':
new_http.assert_called_once_with(
timeout=client.GCE_METADATA_TIMEOUT)
self.assertEqual(http.requests, 1)
self.assertEqual(http.uri, client._GCE_METADATA_URI)
self.assertEqual(http.method, 'GET')
self.assertIsNone(http.body)
self.assertEqual(http.headers, client._GCE_HEADERS)
else:
new_http.assert_not_called()
self.assertEqual(http.requests, 0)
def check_url(url):
# The url will be routed to PODMProxy resource if it didn't match
# others. So filter out these incorrect url.
# Only support proxy request of below podm API
resource = url.split("/")[0]
filterext = ["Chassis", "Services", "Managers", "Systems",
"EventService", "Nodes", "EthernetSwitches"]
if resource not in filterext:
flask.abort(http_client.NOT_FOUND)
def _do_switch(dpid, func, ret_func):
dpid = dpid_lib.str_to_dpid(dpid)
try:
ret = func(dpid)
except KeyError:
return Response(status=http_client.NOT_FOUND,
body='no dpid is found %s' %
dpid_lib.dpid_to_str(dpid))
return ret_func(ret)
def _do_key(dpid, key, func, ret_func):
dpid = dpid_lib.str_to_dpid(dpid)
try:
ret = func(dpid, key)
except KeyError:
return Response(status=http_client.NOT_FOUND,
body='no dpid/key is found %s %s' %
(dpid_lib.dpid_to_str(dpid), key))
return ret_func(ret)
def test_get_404(self, resp_mock):
"""Test treadmill.restclient.get NOT_FOUND (404)"""
resp_mock.return_value.status_code = http_client.NOT_FOUND
with self.assertRaises(restclient.NotFoundError):
restclient.get('http://foo.com', '/')
def test_app_log_failure(self):
"""Dummy tests for the case when logs cannot be found."""
self.impl.log.get.side_effect = get_log_failure
resp = self.client.get(
'/local-app/proid.app/uniq/service/service_name'
)
self.assertEqual(resp.status_code, http_client.NOT_FOUND)
resp = self.client.get('/local-app/proid.app/uniq/sys/component')
self.assertEqual(resp.status_code, http_client.NOT_FOUND)
def test_arch_get(self):
"""Dummy tests for returning application archives."""
self.impl.archive.get.return_value = __file__
resp = self.client.get('/archive/<app>/<uniq>/app')
self.assertEqual(resp.status_code, http_client.OK)
self.impl.archive.get.side_effect = err
resp = self.client.get('/archive/<app>/<uniq>/app')
self.assertEqual(resp.status_code, http_client.NOT_FOUND)
resp = self.client.get('/archive/<app>/<uniq>/sys')
self.assertEqual(resp.status_code, http_client.NOT_FOUND)
def test_redir_no_path_no_lb(self, *_args):
"""Test redirect with no path and no LB"""
resp = self.app.get('/redir/lb/type1/cell1')
self.assertEqual(resp.status_code, http_client.NOT_FOUND)
def test_update_not_found(self):
uuid = uuidutils.generate_uuid()
response = self.patch_json('/servers/%s' % uuid,
[{'path': '/metadata/a', 'value': 'b',
'op': 'add'}],
headers=self.headers,
expect_errors=True)
self.assertEqual(http_client.NOT_FOUND, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])
def get_all(self, flavor_uuid):
"""Retrieve a list of extra specs of the queried flavor."""
flavor = objects.Flavor.get(pecan.request.context,
flavor_uuid)
# public flavor to all projects
if flavor.is_public:
msg = _("Access list not available for public flavors.")
raise wsme.exc.ClientSideError(
msg, status_code=http_client.NOT_FOUND)
# private flavor to listed projects only
return _marshall_flavor_access(flavor)
def _refresh(self, http_request):
"""Refreshes the access_token.
Skip all the storage hoops and just refresh using the API.
Args:
http_request: callable, a callable that matches the method
signature of httplib2.Http.request, used to make
the refresh request.
Raises:
HttpAccessTokenRefreshError: When the refresh fails.
"""
response, content = http_request(
META, headers={'Metadata-Flavor': 'Google'})
content = _from_bytes(content)
if response.status == http_client.OK:
try:
token_content = json.loads(content)
except Exception as e:
raise HttpAccessTokenRefreshError(str(e),
status=response.status)
self.access_token = token_content['access_token']
else:
if response.status == http_client.NOT_FOUND:
content += (' This can occur if a VM was created'
' with no service account or scopes.')
raise HttpAccessTokenRefreshError(content, status=response.status)
def _refresh(self, http_request):
"""Refreshes the access_token.
Skip all the storage hoops and just refresh using the API.
Args:
http_request: callable, a callable that matches the method
signature of httplib2.Http.request, used to make
the refresh request.
Raises:
HttpAccessTokenRefreshError: When the refresh fails.
"""
response, content = http_request(
META, headers={'Metadata-Flavor': 'Google'})
content = _from_bytes(content)
if response.status == http_client.OK:
try:
token_content = json.loads(content)
except Exception as e:
raise HttpAccessTokenRefreshError(str(e),
status=response.status)
self.access_token = token_content['access_token']
else:
if response.status == http_client.NOT_FOUND:
content += (' This can occur if a VM was created'
' with no service account or scopes.')
raise HttpAccessTokenRefreshError(content, status=response.status)
def _refresh(self, http_request):
"""Refreshes the access_token.
Skip all the storage hoops and just refresh using the API.
Args:
http_request: callable, a callable that matches the method
signature of httplib2.Http.request, used to make
the refresh request.
Raises:
HttpAccessTokenRefreshError: When the refresh fails.
"""
response, content = http_request(
META, headers={'Metadata-Flavor': 'Google'})
content = _from_bytes(content)
if response.status == http_client.OK:
try:
token_content = json.loads(content)
except Exception as e:
raise HttpAccessTokenRefreshError(str(e),
status=response.status)
self.access_token = token_content['access_token']
else:
if response.status == http_client.NOT_FOUND:
content += (' This can occur if a VM was created'
' with no service account or scopes.')
raise HttpAccessTokenRefreshError(content, status=response.status)
test__metadata.py 文件源码
项目：google-auth-library-python
作者: GoogleCloudPlatform
项目源码
文件源码
阅读 23
收藏 0
点赞 0
评论 0
def test_get_failure():
request = make_request(
'Metadata error', status=http_client.NOT_FOUND)
with pytest.raises(exceptions.TransportError) as excinfo:
_metadata.get(request, PATH)
assert excinfo.match(r'Metadata error')
request.assert_called_once_with(
method='GET',
url=_metadata._METADATA_ROOT + PATH,
headers=_metadata._METADATA_HEADERS)
def test_default_error_code(self):
class FakeMasakariException(exception.MasakariException):
code = http.NOT_FOUND
exc = FakeMasakariException()
self.assertEqual(http.NOT_FOUND, exc.kwargs['code'])
def test_error_code_from_kwarg(self):
class FakeMasakariException(exception.MasakariException):
code = http.INTERNAL_SERVER_ERROR
exc = FakeMasakariException(code=http.NOT_FOUND)
self.assertEqual(exc.kwargs['code'], http.NOT_FOUND)
