Python responses 模块，DELETE 实例源码

def _setup_mocking(self):
        def _mock_callback(request):
            method = request.method.lower()
            url = request.path_url
            handler = getattr(self.app, method)
            r = handler(
                url,
                data=request.body,
                headers=dict(request.headers)
            )
            return (r.status_code, r.headers, r.data)

        pattern = re.compile("{}/(.*)".format(self.host))
        methods = [
            responses.GET,
            responses.POST,
            responses.PUT,
            responses.DELETE,
            responses.PATCH,
        ]
        for method in methods:
            responses.add_callback(
                method, pattern,
                callback=_mock_callback
            )

def test_delete_success(self):
        # WHEN `dpm delete` is invoked
        result = self.invoke(cli, ['delete'])

        # THEN 'delete ok' should be printed to stdout
        self.assertRegexpMatches(result.output, 'delete ok')
        # AND 2 requests should be sent
        self.assertEqual(
            [(x.request.method, x.request.url, jsonify(x.request))
             for x in responses.calls],
            [
                # POST authorization
                ('POST', 'https://example.com/api/auth/token',
                    {"username": "user", "secret": "access_token"}),
                # DELETE datapackage
                ('DELETE', 'https://example.com/api/package/user/some-datapackage', '')])
        # AND exit code should be 0
        self.assertEqual(result.exit_code, 0)

def test_purge_success(self):
        # WHEN `dpm purge` is invoked
        result = self.invoke(cli, ['purge'])

        # THEN 'purge ok' should be printed to stdout
        self.assertRegexpMatches(result.output, 'purge ok')
        # AND 2 requests should be sent
        self.assertEqual(
            [(x.request.method, x.request.url, jsonify(x.request))
             for x in responses.calls],
            [
                # POST authorization
                ('POST', 'https://example.com/api/auth/token',
                    {"username": "user", "secret": "access_token"}),
                # DELETE datapackage
                ('DELETE', 'https://example.com/api/package/user/some-datapackage/purge', '')])
        # AND exit code should be 0
        self.assertEqual(result.exit_code, 0)

def test_remove_failure(self):
        responses.add(
            responses.DELETE,
            'http://localhost:9999/rest/Content/1',
            content_type='application/json',
            body="Could not delete",
            status=400
        )

        slims = Slims("testSlims", "http://localhost:9999", "admin", "admin")
        record = Record({"pk": 1,
                         "tableName": "Content",
                         "columns": []},
                        slims.slims_api)

        self.assertRaises(Exception, record.remove)

def test_oauth_response_missing_keys(self):
        """
        A ``requests.RequestException`` is raised when the call for an OAuth2 access token returns no data.
        """
        responses.add(
            responses.POST,
            self.oauth_url,
            json={},
            status=200
        )
        responses.add(
            responses.DELETE,
            self.course_url,
            json={},
            status=200
        )

        degreed_api_client = DegreedAPIClient(self.enterprise_config)
        with self.assertRaises(requests.RequestException):
            degreed_api_client.delete_course_content({})

def test_endpoint_delete(self):
        with responses.RequestsMock() as rsps:
            rsps.add(responses.DELETE, MOCK_API_URL + '/users/1', status=204)

            generic_client.users.delete(1)

        with responses.RequestsMock() as rsps:
            rsps.add(responses.DELETE, MOCK_API_URL + '/users/1', status=404)

            with self.assertRaises(generic_client.ResourceNotFound):
                generic_client.users.delete(1)

        with responses.RequestsMock() as rsps:
            rsps.add(responses.DELETE, MOCK_API_URL + '/users/1', status=500)

            with self.assertRaises(generic_client.HTTPError):
                generic_client.users.delete(1)

def test_does_delete_first_for_nuke_call(self, api_responses, api_token):
        post_url = API_ENDPOINT.format(username=getpass.getuser())
        webapp_url = API_ENDPOINT.format(username=getpass.getuser()) + 'mydomain.com/'
        api_responses.add(responses.DELETE, webapp_url, status=200)
        api_responses.add(responses.POST, post_url, status=201, body=json.dumps({'status': 'OK'}))
        api_responses.add(responses.PATCH, webapp_url, status=200)

        Webapp('mydomain.com').create('2.7', '/virtualenv/path', '/project/path', nuke=True)

        delete = api_responses.calls[0]
        assert delete.request.method == 'DELETE'
        assert delete.request.url == webapp_url
        assert delete.request.headers['Authorization'] == f'Token {api_token}'

def test_ignores_404_from_delete_call_when_nuking(self, api_responses, api_token):
        post_url = API_ENDPOINT.format(username=getpass.getuser())
        webapp_url = API_ENDPOINT.format(username=getpass.getuser()) + 'mydomain.com/'
        api_responses.add(responses.DELETE, webapp_url, status=404)
        api_responses.add(responses.POST, post_url, status=201, body=json.dumps({'status': 'OK'}))
        api_responses.add(responses.PATCH, webapp_url, status=200)

        Webapp('mydomain.com').create('2.7', '/virtualenv/path', '/project/path', nuke=True)

def setUp(self):
        # GIVEN datapackage that can be treated as valid by the dpm
        self.valid_dp = datapackage.DataPackage({
                "name": "some-datapackage",
                "resources": [
                    {"name": "some-resource", "path": "./data/some_data.csv", }
                ]
            },
            default_base_path='.')
        patch('dpm.client.DataPackage', lambda *a: self.valid_dp).start()
        patch('dpm.client.exists', lambda *a: True).start()

        # AND the registry server that accepts any user
        responses.add(
            responses.POST, 'https://example.com/api/auth/token',
            json={'token': 'blabla'},
            status=200)
        # AND registry server accepts deletion of any datapackage
        responses.add(
            responses.DELETE, 'https://example.com/api/package/user/some-datapackage',
            json={'message': 'OK'},
            status=200)
        # AND registry server accepts purging of any datapackage
        responses.add(
            responses.DELETE, 'https://example.com/api/package/user/some-datapackage/purge',
            json={'message': 'OK'},
            status=200)

def setUp(self):
        # GIVEN datapackage that can be treated as valid by the dpm
        valid_dp = datapackage.DataPackage({
            "name": "some-datapackage",
            "resources": [
                {"path": "./data/some_data.csv", }
            ]
        })
        # AND client
        self.client = Client(dp1_path, self.config)
        self.client.datapackage = valid_dp

        # AND the registry server that accepts any user
        responses.add(
            responses.POST, 'http://127.0.0.1:5000/api/auth/token',
            json={'token': 'blabla'},
            status=200)
        # AND registry server accepts deletion of any datapackage
        responses.add(
            responses.DELETE, 'http://127.0.0.1:5000/api/package/user/some-datapackage',
            json={'message': 'OK'},
            status=200)
        # AND registry server accepts purging of any datapackage
        responses.add(
            responses.DELETE, 'http://127.0.0.1:5000/api/package/user/some-datapackage/purge',
            json={'message': 'OK'},
            status=200)

def test_delete_success(self):
        # WHEN delete() is invoked
        self.client.delete()

        # THEN 2 requests should be sent
        self.assertEqual(
            [(x.request.method, x.request.url, jsonify(x.request))
             for x in responses.calls],
            [
                # POST authorization
                ('POST', 'http://127.0.0.1:5000/api/auth/token',
                    {"username": "user", "secret": "access_token"}),
                # DELETE datapackage
                ('DELETE', 'http://127.0.0.1:5000/api/package/user/some-datapackage', '')])

def test_remove_success(self):
        responses.add(
            responses.DELETE,
            'http://localhost:9999/rest/Content/1',
            content_type='application/json',
        )

        slims = Slims("testSlims", "http://localhost:9999", "admin", "admin")
        record = Record({"pk": 1,
                         "tableName": "Content",
                         "columns": []},
                        slims.slims_api)

        record.remove()

def test_hide_greeting(self):
        exp="""
        {
            "fields": [
                "greeting"
            ]
        }
        """
        with MessengerAPIMock(subpath="messenger_profile", expected=exp,
                              method=MessengerAPIMock.DELETE, utest=self) as m:
            self.page.hide_greeting()

def test_hide_starting_button(self):
        exp="""
        {
            "fields": [
                "get_started"
            ]
        }
        """
        with MessengerAPIMock(subpath="messenger_profile", expected=exp,
                              method=MessengerAPIMock.DELETE, utest=self) as m:
            self.page.hide_starting_button()

def test_hide_persistent_menu(self):
        exp="""
        {
            "fields": [
                "persistent_menu"
            ]
        }
        """
        with MessengerAPIMock(subpath="messenger_profile", expected=exp,
                              method=MessengerAPIMock.DELETE, utest=self) as m:
            self.page.hide_persistent_menu()

def set_delete_object_response(self, id, code=200):
        """Sets mock DELETE /api/odlcs/<id> response.

        Args:
            id (int): Target ID.
            code (int): Status code to respond with.
        """
        self.rsps.add(
            responses.DELETE,
            "{}/api/odlcs/{:d}".format(self.url, id),
            status=code,
            body="Target deleted." if code == 200 else "")

def set_delete_object_image_response(self, id, code=200):
        """Sets mock DELETE /api/odlcs/<id>/image response.

        Args:
            id (int): Target ID.
            code (str): Status code to respond with.
        """
        self.rsps.add(
            responses.DELETE,
            "{}/api/odlcs/{:d}/image".format(self.url, id),
            body="Image deleted." if code == 200 else "",
            status=code)

def test_quit(driver):
    responses.add(
        responses.DELETE,
        'http://127.0.0.1:3456/wd/hub/session/2345',
        json={
            'status': 0,
            'sessionId': '2345',
            'value': ''
        })
    assert driver == driver.quit()

def test_close_window(driver):
    responses.add(
        responses.DELETE,
        'http://127.0.0.1:3456/wd/hub/session/2345/window',
        json={
            'status': 0,
            'sessionId': '2345',
            'value': ''
        })
    assert driver.close() == driver

def test_delete_confirmed(self, mock_input):
        owner = 'foo'
        package = 'bar'

        mock_input.return_value = '%s/%s' % (owner, package)

        delete_url = "%s/api/package/%s/%s/" % (command.get_registry_url(), owner, package)
        self.requests_mock.add(responses.DELETE, delete_url, json.dumps(dict()))

        command.delete('%s/%s' % (owner, package))

def test_delete_course_completion(self):
        """
        ``delete_course_completion`` should use the appropriate URLs for transmission.
        """
        responses.add(
            responses.POST,
            self.oauth_url,
            json=self.expected_token_response_body,
            status=200
        )
        responses.add(
            responses.DELETE,
            self.completion_status_url,
            json='{}',
            status=200
        )

        payload = {
            'orgCode': self.company_id,
            'completions': [{
                'employeeId': 'abc123',
                'id': "course-v1:ColumbiaX+DS101X+1T2016",
            }]
        }
        degreed_api_client = DegreedAPIClient(self.enterprise_config)
        output = degreed_api_client.delete_course_completion('fake-user', json.dumps(payload))

        assert output == (200, '"{}"')
        assert len(responses.calls) == 2
        assert responses.calls[0].request.url == self.oauth_url
        assert responses.calls[1].request.url == self.completion_status_url

def test_delete_course_content(self):
        """
        ``delete_course_content`` should use the appropriate URLs for transmission.
        """
        responses.add(
            responses.POST,
            self.oauth_url,
            json=self.expected_token_response_body,
            status=200
        )
        responses.add(
            responses.DELETE,
            self.course_url,
            json='{}',
            status=200
        )

        payload = {
            'orgCode': 'org-code',
            'providerCode': 'provider-code',
            'courses': [{
                'contentId': 'content-id',
            }],
        }
        degreed_api_client = DegreedAPIClient(self.enterprise_config)
        output = degreed_api_client.delete_course_content(payload)
        assert output == (200, '"{}"')
        assert len(responses.calls) == 2
        assert responses.calls[0].request.url == self.oauth_url
        assert responses.calls[1].request.url == self.course_url

def test_expired_token(self):
        """
        If our token expires after some call, make sure to get it again.

        Make a call, have the token expire after waiting some time (technically no time since time is frozen),
        and make a call again and notice 2 OAuth calls in total are required.
        """
        responses.add(
            responses.POST,
            self.oauth_url,
            json={"expires_in": 0, "access_token": self.access_token},
            status=200
        )
        responses.add(
            responses.DELETE,
            self.course_url,
            json='{}',
            status=200
        )

        payload = {
            'orgCode': 'org-code',
            'providerCode': 'provider-code',
            'courses': [{
                'contentId': 'content-id',
            }],
        }
        degreed_api_client = DegreedAPIClient(self.enterprise_config)
        output1 = degreed_api_client.delete_course_content(payload)
        output2 = degreed_api_client.delete_course_content(payload)
        assert output1 == output2 == (200, '"{}"')
        assert len(responses.calls) == 4
        assert responses.calls[0].request.url == self.oauth_url
        assert responses.calls[1].request.url == self.course_url
        assert responses.calls[2].request.url == self.oauth_url
        assert responses.calls[3].request.url == self.course_url

def test_existing_token_is_valid(self):
        """
        On a second call in the same session, if the token isn't expired we shouldn't need to do another OAuth2 call.
        """
        responses.add(
            responses.POST,
            self.oauth_url,
            json=self.expected_token_response_body,
            status=200
        )
        responses.add(
            responses.DELETE,
            self.course_url,
            json='{}',
            status=200
        )

        payload = {
            'orgCode': 'org-code',
            'providerCode': 'provider-code',
            'courses': [{
                'contentId': 'content-id',
            }],
        }
        degreed_api_client = DegreedAPIClient(self.enterprise_config)
        output1 = degreed_api_client.delete_course_content(payload)
        output2 = degreed_api_client.delete_course_content(payload)
        assert output1 == output2 == (200, '"{}"')
        assert len(responses.calls) == 3
        assert responses.calls[0].request.url == self.oauth_url
        assert responses.calls[1].request.url == self.course_url
        assert responses.calls[2].request.url == self.course_url

def test_methods(self):
        """
        When the HTTP method-specific methods are called, the correct request
        method is used.
        """
        client = ContainerHttpClient('127.0.0.1', '45678')

        responses.add(responses.GET, 'http://127.0.0.1:45678/', status=200)
        responses.add(
            responses.OPTIONS, 'http://127.0.0.1:45678/foo', status=201)
        responses.add(responses.HEAD, 'http://127.0.0.1:45678/bar', status=403)
        responses.add(responses.POST, 'http://127.0.0.1:45678/baz', status=404)
        responses.add(responses.PUT, 'http://127.0.0.1:45678/test', status=418)
        responses.add(
            responses.PATCH, 'http://127.0.0.1:45678/a/b/c', status=501)
        responses.add(
            responses.DELETE, 'http://127.0.0.1:45678/d/e/f', status=503)

        get_response = client.get()
        options_response = client.options('/foo')
        head_response = client.head('/bar')
        post_response = client.post('/baz')
        put_response = client.put('/test')
        patch_response = client.patch('/a/b/c')
        delete_response = client.delete('/d/e/f')

        self.assertEqual(get_response.status_code, 200)
        self.assertEqual(options_response.status_code, 201)
        self.assertEqual(head_response.status_code, 403)
        self.assertEqual(post_response.status_code, 404)
        self.assertEqual(put_response.status_code, 418)
        self.assertEqual(patch_response.status_code, 501)
        self.assertEqual(delete_response.status_code, 503)

        self.assertEqual(len(responses.calls), 7)

def setUp(self):
        super(DeleteTestCase, self).setUp()
        self.mock_cookie()

        self.service = 'MyService'
        self.guid = '001122'
        self.url = "{}/{}Set(guid'{}')".format(CDMSRestApi.CRM_REST_BASE_URL, self.service, self.guid)
        responses.add(
            responses.DELETE, self.url,
            status=204
        )

def test_resource_delete(self):
        with responses.RequestsMock() as rsps:
            rsps.add(responses.GET, MOCK_API_URL + '/users/1', json={
                'id': 1,
                'username': 'user1',
                'group': 'watchers',
            })

            user1 = generic_client.users.get(id=1)
            self.assertEqual(user1.username, 'user1')

        with responses.RequestsMock() as rsps:
            rsps.add(responses.DELETE, MOCK_API_URL + '/users/1', status=204)

            user1.delete()

def test_resource_delete_uuid(self):
        with responses.RequestsMock() as rsps:
            rsps.add(responses.GET, MOCK_API_URL + '/users/1', json={
                'uuid': 1,
                'username': 'user1',
                'group': 'watchers',
            })

            user1 = generic_client.users.get(uuid=1)
            self.assertEqual(user1.username, 'user1')

        with responses.RequestsMock() as rsps:
            rsps.add(responses.DELETE, MOCK_API_URL + '/users/1', status=204)

            user1.delete()

def test_resource_delete(self):
        with responses.RequestsMock() as rsps:
            rsps.add(responses.GET, MOCK_API_URL + '/users/1/', json={
                'id': 1,
                'username': 'user1',
                'group': 'watchers',
            })

            user1 = generic_client.users.get(id=1)
            self.assertEqual(user1.username, 'user1')

        with responses.RequestsMock() as rsps:
            rsps.add(responses.DELETE, MOCK_API_URL + '/users/1/', status=204)

            user1.delete()

def setup_recording(self, **kwargs):
        _logger.info("recording ...")
        self.responses.reset()
        all_requests_re = re.compile("http.*")
        methods = (responses.GET, responses.POST, responses.PUT,
                   responses.PATCH, responses.DELETE, responses.HEAD,
                   responses.OPTIONS)
        for http_method in methods:
            self.responses.add_callback(
                http_method, all_requests_re,
                match_querystring=False,
                callback=self.record())

def test_cookies(driver):
    responses.add(
        responses.GET,
        'http://127.0.0.1:3456/wd/hub/session/2345/cookie',
        json={
            'status': 0,
            'sessionId': '2345',
            'value': [
            {
                'name': '_ga',
                'value': 3213123
            }, {
                'name': 'first',
                'value': False
            }]
        })
    responses.add(
        responses.POST,
        'http://127.0.0.1:3456/wd/hub/session/2345/cookie',
        json={
            'status': 0,
            'sessionId': '2345',
            'value': ''
        })
    responses.add(
        responses.DELETE,
        'http://127.0.0.1:3456/wd/hub/session/2345/cookie',
        json={
            'status': 0,
            'sessionId': '2345',
            'value': ''
        })
    responses.add(
        responses.DELETE,
        'http://127.0.0.1:3456/wd/hub/session/2345/cookie/first',
        json={
            'status': 0,
            'sessionId': '2345',
            'value': ''
        })
    assert driver.cookies == [
        {
            'name': '_ga',
            'value': 3213123
        }, {
            'name': 'first',
            'value': False
        }]
    assert driver.get_cookie('first')['value'] == False
    assert driver.get_cookie('last') == None
    assert driver.delete_cookie('first') == driver
    assert driver.delete_all_cookies() == driver
    assert driver.add_cookie({'name': 'first', 'value': True}) == driver

    with pytest.raises(TypeError):
        driver.add_cookie('error')

    with pytest.raises(KeyError):
        driver.add_cookie({'name': 'first'})