def test_send_with_local_file_url(self, get_size_mock):
transport = service.RequestsTransport()
url = 'file:///foo'
request = requests.PreparedRequest()
request.url = url
data = b"Hello World"
get_size_mock.return_value = len(data)
def readinto_mock(buf):
buf[0:] = data
if six.PY3:
builtin_open = 'builtins.open'
open_mock = mock.MagicMock(name='file_handle',
spec=open)
import _io
file_spec = list(set(dir(_io.TextIOWrapper)).union(
set(dir(_io.BytesIO))))
else:
builtin_open = '__builtin__.open'
open_mock = mock.MagicMock(name='file_handle',
spec=file)
file_spec = file
file_handle = mock.MagicMock(spec=file_spec)
file_handle.write.return_value = None
file_handle.__enter__.return_value = file_handle
file_handle.readinto.side_effect = readinto_mock
open_mock.return_value = file_handle
with mock.patch(builtin_open, open_mock, create=True):
resp = transport.session.send(request)
self.assertEqual(data, resp.content)
python类PreparedRequest()的实例源码
def __init__(self, response):
assert isinstance(response, requests.Response)
# Extended exception data attributes
self.request = response.request
"""The :class:`requests.PreparedRequest` object that initiated the API
call."""
self.response = response
"""The :class:`requests.Response` object returned from the API call."""
# Error message
response_code = response.status_code
response_reason = " " + response.reason if response.reason else ""
description = SPARK_RESPONSE_CODES.get(response_code,
"Unknown Response Code")
detail = response_to_string(response)
super(SparkApiError, self).__init__("Response Code [{}]{} - {}\n{}"
"".format(response_code,
response_reason,
description,
detail))
def test_prj_create_success(self, mock_session, mock_resp):
mock_session.prepare_request.return_value = PreparedRequest()
mock_resp.status_code = 201
mock_resp.json.return_value = {
'type': 'image', 'name': 'chan', 'description': 'walker',
'experiment': 'bar', 'creator': 'me',
'default_time_sample': 2, 'datatype': 'uint16', 'base_resolution': 0
}
mock_session.send.return_value = mock_resp
url_prefix = 'https://api.theboss.io'
auth = 'mytoken'
send_opts = {}
actual = self.prj.create(self.chan, url_prefix, auth, mock_session, send_opts)
self.assertTrue(isinstance(actual, ChannelResource))
self.assertEqual('chan', actual.name)
self.assertEqual('foo', actual.coll_name)
self.assertEqual('bar', actual.exp_name)
self.assertEqual('image', actual.type)
def test_create_cutout_success(self, mock_session):
resolution = 0
x_range = [20, 40]
y_range = [50, 70]
z_range = [30, 50]
time_range = [10, 25]
data = numpy.random.randint(0, 3000, (15, 20, 20, 20), numpy.uint16)
url_prefix = 'https://api.theboss.io'
auth = 'mytoken'
mock_session.prepare_request.return_value = PreparedRequest()
fake_response = Response()
fake_response.status_code = 201
mock_session.send.return_value = fake_response
send_opts = {}
self.vol.create_cutout(
self.chan, resolution, x_range, y_range, z_range, time_range, data,
url_prefix, auth, mock_session, send_opts)
def test_create_cutout_failure(self, mock_session):
resolution = 0
x_range = [20, 40]
y_range = [50, 70]
z_range = [30, 50]
time_range = [10, 25]
data = numpy.random.randint(0, 3000, (15, 20, 20, 20), numpy.uint16)
url_prefix = 'https://api.theboss.io'
auth = 'mytoken'
mock_session.prepare_request.return_value = PreparedRequest()
fake_response = Response()
fake_response.status_code = 403
mock_session.send.return_value = fake_response
send_opts = {}
with self.assertRaises(HTTPError):
self.vol.create_cutout(
self.chan, resolution, x_range, y_range, z_range, time_range, data,
url_prefix, auth, mock_session, send_opts)
def test_get_cutout_failure(self, mock_session):
resolution = 0
x_range = [20, 40]
y_range = [50, 70]
z_range = [30, 50]
time_range = [10, 25]
id_list = []
url_prefix = 'https://api.theboss.io'
auth = 'mytoken'
fake_prepped_req = PreparedRequest()
fake_prepped_req.headers = {}
mock_session.prepare_request.return_value = fake_prepped_req
fake_response = Response()
fake_response.status_code = 403
mock_session.send.return_value = fake_response
send_opts = {}
with self.assertRaises(HTTPError):
actual = self.vol.get_cutout(
self.chan, resolution, x_range, y_range, z_range, time_range, id_list,
url_prefix, auth, mock_session, send_opts)
def test_get_ids_in_region_failure(self, mock_session):
resolution = 0
x_range = [0, 100]
y_range = [10, 50]
z_range = [20, 42]
t_range = [0, 1]
url_prefix = 'https://api.theboss.io'
auth = 'mytoken'
send_opts = {}
fake_prepped_req = PreparedRequest()
fake_prepped_req.headers = {}
mock_session.prepare_request.return_value = fake_prepped_req
fake_response = Response()
fake_response.status_code = 403
mock_session.send.return_value = fake_response
send_opts = {}
with self.assertRaises(HTTPError):
actual = self.vol.get_ids_in_region(
self.anno_chan, resolution, x_range, y_range, z_range, t_range,
url_prefix, auth, mock_session, send_opts)
def test_get_group_success(self, mock_session, mock_resp):
grp_name = 'mygroup'
mock_session.prepare_request.return_value = PreparedRequest()
mock_resp.status_code = 200
mock_resp.json.return_value = True
mock_session.send.return_value = mock_resp
user = None
url_prefix = 'https://api.theboss.io'
auth = 'mytoken'
send_opts = {}
actual = self.prj.get_group(
grp_name, user, url_prefix, auth, mock_session, send_opts)
self.assertTrue(actual)
def test_list_group_members_success(self, mock_session, mock_resp):
expected = ['john', 'mary']
mock_resp.status_code = 200
mock_resp.json.return_value = { 'members': expected }
mock_session.prepare_request.return_value = PreparedRequest()
mock_session.send.return_value = mock_resp
group = 'fire'
url_prefix = 'https://api.theboss.io'
auth = 'mytoken'
send_opts = {}
actual = self.prj.list_group_members(
group, url_prefix, auth, mock_session, send_opts)
self.assertEqual(expected, actual)
def test_list_group_maintainers_success(self, mock_session, mock_resp):
expected = ['john', 'mary']
mock_resp.status_code = 200
mock_resp.json.return_value = { 'maintainers': expected }
mock_session.prepare_request.return_value = PreparedRequest()
mock_session.send.return_value = mock_resp
group = 'fire'
url_prefix = 'https://api.theboss.io'
auth = 'mytoken'
send_opts = {}
actual = self.prj.list_group_maintainers(
group, url_prefix, auth, mock_session, send_opts)
self.assertEqual(expected, actual)
def signer(self, identity_id):
"""
Generates a request signer function for the
the authorizing identity.
>>> signer = api_client.signer(authorizing_identity)
:param str identity_id: the authorizing identity id
:return: the request signer function
:rtype: (:class:`PreparedRequest`) -> :class:`PreparedRequest`
"""
def sign_request(r):
# type: (requests.PreparedRequest) -> requests.PreparedRequest
signing_key = self.key_store.get_private_signing_key(identity_id)
r.headers = signer.get_updated_headers(
identity_id=identity_id,
method=r.method,
url=r.url,
headers=r.headers,
payload=r.body,
private_signing_key=signing_key)
return r
return sign_request
def _post_servers(self, request):
# type: (requests.PreparedRequest) -> requests.Response
data = parse_qs(force_text(request.body))
self.server_id = self._public_id('srv')
self.auth_token = ''.join(random.choice(mixed_alphabet) for i in xrange(20))
self.api_version = force_text(request.headers['X-Cloak-API-Version'])
self.name = data['name'][0]
self.target_id = data['target'][0]
# Make sure these exist
data['email'][0]
data['password'][0]
result = {
'server_id': self.server_id,
'auth_token': self.auth_token,
'server': self._server_result(),
}
return self._response(request, 201, result)
def _post_server_csr(self, request):
# type: (requests.PreparedRequest) -> requests.Response
data = parse_qs(force_text(request.body))
if self._authenticate(request):
self.csr = data['csr'][0]
self.pki_tag = ''.join(random.choice(mixed_alphabet) for i in xrange(16))
response = self._response(request, 202)
else:
response = self._response(request, 401)
return response
#
# Utils
#
def _response(self, request, status, result=None, headers={}):
# type: (requests.PreparedRequest, int, Any, Dict[str, str]) -> requests.Response
response = requests.Response()
response.status_code = status
response.url = request.url
if result is not None:
response.raw = io.BytesIO(json.dumps(result).encode('utf-8'))
response.encoding = 'utf-8'
else:
response.raw = io.BytesIO(b'')
response.encoding = 'latin1'
response.headers.update(headers)
return response
def decode(self, o):
"""
Decode the contents of the given JSON dictionary as an object used by Web Sight.
:param o: The JSON dictionary to process.
:return: The contents of the given JSON dictionary deserialized into an object.
"""
from lib.arin.response import BaseArinResponse
from ..geolocation import IpGeolocation
if "__class_type" not in o:
raise UnsupportedDeserializationError("No class type specified in JSON dictionary: %s" % o)
class_type = o["__class_type"]
deserialization_class = get_class_from_import_string(class_type)
if deserialization_class == PreparedRequest:
return self.__deserialize_requests_prepared_request(o)
elif deserialization_class == Response:
return self.__deserialize_requests_response(o)
elif issubclass(deserialization_class, BaseArinResponse):
return self.__deserialize_arin_response(o)
elif deserialization_class == IpGeolocation:
return self.__deserialize_ip_geolocation(o)
else:
raise UnsupportedDeserializationError(
"Class %s does not have a deserialization method."
% deserialization_class
)
def encode(self, o):
"""
Encode the contents of the given object into JSON.
:param o: The object to process.
:return: The contents of the given object in JSON format.
"""
from lib.arin.response import BaseArinResponse
from ..geolocation import IpGeolocation
if isinstance(o, PreparedRequest):
return self.__serialize_requests_prepared_request(o)
elif isinstance(o, Response):
return self.__serialize_requests_response(o)
elif isinstance(o, BaseArinResponse):
return self.__serialize_arin_response(o)
elif isinstance(o, IpGeolocation):
return self.__serialize_ip_geolocation(o)
else:
return super(WsSerializableJSONEncoder, self).encode(o)
# Protected Methods
# Private Methods
def __init__(self, status_code: int, response: Any):
self.headers: dict = None
self.body: str = None
self.path: str = None
self.params: dict = None
@urlmatch(netloc=mock_uri, path='^/kb/.*/graql$', method='POST')
def grakn_mock(url: SplitResult, request: PreparedRequest):
self.headers = request.headers
self.body = request.body
self.path = url.path
self.params = parse_qs(url.query)
return httmock.response(status_code, json.dumps(response))
self._httmock: HTTMock = HTTMock(grakn_mock)
def test_generate_launch_request(self):
launch_params = {
'lti_version': 'foo',
'lti_message_type': 'bar',
'resource_link_id': 'baz'
}
tc = ToolConsumer('client_key', 'client_secret',
launch_url='http://example.edu/',
params=launch_params)
launch_req = tc.generate_launch_request(nonce='abcd1234',
timestamp='1234567890')
self.assertIsInstance(launch_req, PreparedRequest)
got = parse_qs(unquote(launch_req.body.decode('utf-8')))
correct = launch_params.copy()
correct.update({
'oauth_nonce': 'abcd1234',
'oauth_timestamp': '1234567890',
'oauth_version': '1.0',
'oauth_signature_method': 'HMAC-SHA1',
'oauth_consumer_key': 'client_key',
'oauth_signature': 'u2xlj 1gF4y 6gKHNeiL9cN3tOI=',
})
self.assertEqual(got, correct)
def test_generate_launch_request(self):
launch_params = {
'lti_version': 'foo',
'lti_message_type': 'bar',
'resource_link_id': 'baz'
}
tc = ContentItemResponse('client_key', 'client_secret',
launch_url='http://example.edu/',
params=launch_params)
launch_req = tc.generate_launch_request(nonce='abcd1234',
timestamp='1234567890')
self.assertIsInstance(launch_req, PreparedRequest)
got = parse_qs(unquote(launch_req.body.decode('utf-8')))
correct = launch_params.copy()
correct.update({
'oauth_nonce': 'abcd1234',
'oauth_timestamp': '1234567890',
'oauth_version': '1.0',
'oauth_signature_method': 'HMAC-SHA1',
'oauth_consumer_key': 'client_key',
'oauth_signature': 'u2xlj 1gF4y 6gKHNeiL9cN3tOI=',
})
self.assertEqual(got, correct)
def test_register_proxy(self):
proxy = ToolProxy(params=test_params)
proxy.tc_profile = test_profile
signed_request = proxy.register_proxy({'tool_profile': 'A Real Tool Profile Goes here'})
self.assertIsInstance(signed_request, requests.PreparedRequest)
def extractor_before_request(self, request):
self.called += 1
assert_true(isinstance(request, requests.PreparedRequest))
request.url = 'http://test-url.example.com/file.pdf'
return request
def test_get_success(self, mock_session, mock_resp):
expected = ['default']
mock_resp.status_code = 200
mock_resp.json.return_value = expected
mock_session.prepare_request.return_value = PreparedRequest()
mock_session.send.return_value = mock_resp
url_prefix = 'https://api.theboss.io'
auth = 'mytoken'
send_opts = {}
actual = self.prj.get_user(
'johndoe', url_prefix, auth, mock_session, send_opts)
six.assertCountEqual(self, expected, actual)
def test_get_failure(self, mock_session):
mock_session.prepare_request.return_value = PreparedRequest()
fake_resp = Response()
fake_resp.status_code = 403
mock_session.send.return_value = fake_resp
url_prefix = 'https://api.theboss.io'
auth = 'mytoken'
send_opts = {}
with self.assertRaises(HTTPError):
self.prj.get_user(
'johndoe', url_prefix, auth, mock_session, send_opts)
def test_delete_success(self, mock_session):
mock_session.prepare_request.return_value = PreparedRequest()
fake_resp = Response()
fake_resp.status_code = 204
mock_session.send.return_value = fake_resp
url_prefix = 'https://api.theboss.io'
auth = 'mytoken'
send_opts = {}
self.prj.delete_user(
'johndoe', url_prefix, auth, mock_session, send_opts)
def test_prj_list_failure(self, mock_session, mock_resp):
mock_session.prepare_request.return_value = PreparedRequest()
mock_resp.status_code = 403
mock_resp.raise_for_status.side_effect = HTTPError()
mock_session.send.return_value = mock_resp
url_prefix = 'https://api.theboss.io'
auth = 'mytoken'
send_opts = {}
with self.assertRaises(HTTPError):
self.prj.list(self.chan, url_prefix, auth, mock_session, send_opts)
def test_prj_create_failure(self, mock_session):
mock_session.prepare_request.return_value = PreparedRequest()
fake_resp = Response()
fake_resp.status_code = 403
mock_session.send.return_value = fake_resp
url_prefix = 'https://api.theboss.io'
auth = 'mytoken'
send_opts = {}
with self.assertRaises(HTTPError):
self.prj.create(self.chan, url_prefix, auth, mock_session, send_opts)
def test_prj_get_success(self, mock_session, mock_resp):
chan_dict = {
'name': 'bar', 'description': 'none', 'experiment': 8,
'default_time_sample': 0, 'datatype': 'uint16',
'base_resolution': 0, 'linked_channel_layers': [], 'creator': 'me',
'type': 'image'
}
expected = ChannelResource(
chan_dict['name'], self.chan.coll_name, self.chan.exp_name, 'image')
expected.description = chan_dict['description']
expected.datatype = chan_dict['datatype']
expected.base_resolution = chan_dict['base_resolution']
expected.default_time_sample = chan_dict['default_time_sample']
expected.type = 'image'
mock_session.prepare_request.return_value = PreparedRequest()
mock_resp.json.return_value = chan_dict
mock_resp.status_code = 200
mock_session.send.return_value = mock_resp
url_prefix = 'https://api.theboss.io'
auth = 'mytoken'
send_opts = {}
actual = self.prj.get(self.chan, url_prefix, auth, mock_session, send_opts)
self.assertEqual(expected.name, actual.name)
self.assertEqual(expected.description, actual.description)
self.assertEqual(expected.exp_name, actual.exp_name)
self.assertEqual(expected.coll_name, actual.coll_name)
self.assertEqual(expected.default_time_sample, actual.default_time_sample)
self.assertEqual(expected.datatype, actual.datatype)
self.assertEqual(expected.base_resolution, actual.base_resolution)
self.assertEqual(expected.type, actual.type)
def test_prj_get_failure(self, mock_session, mock_resp):
mock_session.prepare_request.return_value = PreparedRequest()
mock_resp.status_code = 403
mock_resp.raise_for_status.side_effect = HTTPError()
mock_session.send.return_value = mock_resp
url_prefix = 'https://api.theboss.io'
auth = 'mytoken'
send_opts = {}
with self.assertRaises(HTTPError):
self.prj.get(self.chan, url_prefix, auth, mock_session, send_opts)
def test_prj_update_failure(self, mock_session):
mock_session.prepare_request.return_value = PreparedRequest()
fake_resp = Response()
fake_resp.status_code = 403
mock_session.send.return_value = fake_resp
url_prefix = 'https://api.theboss.io'
auth = 'mytoken'
send_opts = {}
with self.assertRaises(HTTPError):
self.prj.update(self.chan.name, self.chan, url_prefix, auth, mock_session, send_opts)
def test_prj_delete_success(self, mock_session):
mock_session.prepare_request.return_value = PreparedRequest()
fake_resp = Response()
fake_resp.status_code = 204
mock_session.send.return_value = fake_resp
url_prefix = 'https://api.theboss.io'
auth = 'mytoken'
send_opts = {}
self.prj.delete(self.chan, url_prefix, auth, mock_session, send_opts)
