Python requests 模块,PreparedRequest() 实例源码

我们从Python开源项目中,提取了以下48个代码示例,用于说明如何使用requests.PreparedRequest()

项目:deb-oslo.vmware    作者:openstack    | 项目源码 | 文件源码
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)
项目:ciscosparkapi    作者:CiscoDevNet    | 项目源码 | 文件源码
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))
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
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)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
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)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
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)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
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)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
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)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
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)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
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)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
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)
项目:delta-sdk-python    作者:Covata    | 项目源码 | 文件源码
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
项目:cloak-server    作者:encryptme    | 项目源码 | 文件源码
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)
项目:cloak-server    作者:encryptme    | 项目源码 | 文件源码
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
    #
项目:cloak-server    作者:encryptme    | 项目源码 | 文件源码
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
项目:ws-backend-community    作者:lavalamp-    | 项目源码 | 文件源码
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
            )
项目:ws-backend-community    作者:lavalamp-    | 项目源码 | 文件源码
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
项目:grakn-python    作者:graknlabs    | 项目源码 | 文件源码
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)
项目:lti    作者:pylti    | 项目源码 | 文件源码
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)
项目:lti    作者:pylti    | 项目源码 | 文件源码
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)
项目:lti    作者:pylti    | 项目源码 | 文件源码
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)
项目:ckanext-extractor    作者:stadt-karlsruhe    | 项目源码 | 文件源码
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
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
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)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
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)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
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)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
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)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
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)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
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)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
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)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
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)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
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)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
def test_prj_delete_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.delete(self.chan, url_prefix, auth, mock_session, send_opts)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
def test_get_bounding_box_success(self, mock_session, mock_resp):
        resolution = 0
        id = 44444
        bb_type = 'loose'

        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
        mock_session.send.return_value = mock_resp

        mock_resp.status_code = 200
        mock_resp.json.return_value = expected = {
            'x_range': [0, 10],
            'y_range': [0, 10],
            'z_range': [0, 10],
            't_range': [0, 10]
        }

        actual = self.vol.get_bounding_box(
            self.anno_chan, resolution, id, bb_type,
            url_prefix, auth, mock_session, send_opts)

        self.assertEqual(expected, actual)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
def test_get_ids_in_region_success(self, mock_session, mock_resp):
        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
        mock_session.send.return_value = mock_resp

        mock_resp.status_code = 200
        mock_resp.json.return_value = { 'ids': ['1', '10'] }

        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)

        expected = [1, 10]

        self.assertEqual(expected, actual)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
def test_list_groups_success(self, mock_session, mock_resp):
        expected =  ['g1', 'g2'] 
        mock_resp.status_code = 200
        mock_resp.json.return_value = { 'groups': 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.list_groups(
            None, url_prefix, auth, mock_session, send_opts)

        self.assertEqual(expected, actual)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
def test_create_group_success(self, mock_session):
        mock_session.prepare_request.return_value = PreparedRequest()
        fake_resp = Response()
        fake_resp.status_code = 201
        mock_session.send.return_value = fake_resp

        url_prefix = 'https://api.theboss.io'
        auth = 'mytoken'
        send_opts = {}

        self.prj.create_group(
            'mygroup', url_prefix, auth, mock_session, send_opts)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
def test_create_group_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_group(
            'mygroup', url_prefix, auth, mock_session, send_opts)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
def test_get_group_failure(self, mock_session):
        grp_name = 'mygroup'
        mock_session.prepare_request.return_value = PreparedRequest()
        fake_resp = Response()
        fake_resp.status_code = 404
        mock_session.send.return_value = fake_resp

        user = None
        url_prefix = 'https://api.theboss.io'
        auth = 'mytoken'
        send_opts = {}

        with self.assertRaises(HTTPError):
            self.prj.get_group(
                grp_name, user, url_prefix, auth, mock_session, send_opts)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
def test_delete_group_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

        user = None
        url_prefix = 'https://api.theboss.io'
        auth = 'mytoken'
        send_opts = {}

        self.prj.delete_group(
            'mygroup', url_prefix, auth, mock_session, send_opts)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
def test_list_group_members_failure(self, mock_session):
        fake_resp = Response()
        fake_resp.status_code = 403
        mock_session.prepare_request.return_value = PreparedRequest()
        mock_session.send.return_value = fake_resp

        group = 'fire'
        url_prefix = 'https://api.theboss.io'
        auth = 'mytoken'
        send_opts = {}

        with self.assertRaises(HTTPError):
            self.prj.list_group_members(
                group, url_prefix, auth, mock_session, send_opts)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
def test_add_group_member_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

        user = 'you'
        url_prefix = 'https://api.theboss.io'
        auth = 'mytoken'
        send_opts = {}

        self.prj.add_group_member(
            'mygroup', user, url_prefix, auth, mock_session, send_opts)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
def test_add_group_member_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

        user = 'you'
        url_prefix = 'https://api.theboss.io'
        auth = 'mytoken'
        send_opts = {}

        with self.assertRaises(HTTPError):
            self.prj.add_group_member(
                'mygroup', user, url_prefix, auth, mock_session, send_opts)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
def test_get_is_group_member_success(self, mock_session, mock_resp):
        mock_session.prepare_request.return_value = PreparedRequest()
        mock_resp.status_code = 200
        mock_resp.json.return_value = { 'result': True }
        mock_session.send.return_value = mock_resp

        user = 'you'
        url_prefix = 'https://api.theboss.io'
        auth = 'mytoken'
        send_opts = {}

        self.assertTrue(self.prj.get_is_group_member(
            'mygroup', user, url_prefix, auth, mock_session, send_opts))
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
def test_delete_group_member_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

        user = 'you'
        url_prefix = 'https://api.theboss.io'
        auth = 'mytoken'
        send_opts = {}

        self.prj.delete_group_member(
            'mygroup', user, url_prefix, auth, mock_session, send_opts)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
def test_delete_group_member_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

        user = 'you'
        url_prefix = 'https://api.theboss.io'
        auth = 'mytoken'
        send_opts = {}

        with self.assertRaises(HTTPError):
            self.prj.delete_group_member(
                'mygroup', user, url_prefix, auth, mock_session, send_opts)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
def test_list_group_maintainers_failure(self, mock_session):
        fake_resp = Response()
        fake_resp.status_code = 403
        mock_session.prepare_request.return_value = PreparedRequest()
        mock_session.send.return_value = fake_resp

        group = 'fire'
        url_prefix = 'https://api.theboss.io'
        auth = 'mytoken'
        send_opts = {}

        with self.assertRaises(HTTPError):
            self.prj.list_group_maintainers(
                group, url_prefix, auth, mock_session, send_opts)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
def test_add_group_maintainer_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

        user = 'you'
        url_prefix = 'https://api.theboss.io'
        auth = 'mytoken'
        send_opts = {}

        self.prj.add_group_maintainer(
            'mygroup', user, url_prefix, auth, mock_session, send_opts)
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
def test_get_is_group_maintainer_success(self, mock_session, mock_resp):
        mock_session.prepare_request.return_value = PreparedRequest()
        mock_resp.status_code = 200
        mock_resp.json.return_value = { 'result': True }
        mock_session.send.return_value = mock_resp

        user = 'you'
        url_prefix = 'https://api.theboss.io'
        auth = 'mytoken'
        send_opts = {}

        self.assertTrue(self.prj.get_is_group_maintainer(
            'mygroup', user, url_prefix, auth, mock_session, send_opts))
项目:intern    作者:jhuapl-boss    | 项目源码 | 文件源码
def test_get_is_group_maintainer_failure(self, mock_session):
        mock_session.prepare_request.return_value = PreparedRequest()
        fake_resp = Response()
        fake_resp.status_code = 404
        mock_session.send.return_value = fake_resp

        user = 'you'
        url_prefix = 'https://api.theboss.io'
        auth = 'mytoken'
        send_opts = {}

        with self.assertRaises(HTTPError):
            self.prj.get_is_group_maintainer(
                'mygroup', user, url_prefix, auth, mock_session, send_opts)