Python responses 模块，GET 实例源码

我们从Python开源项目中，提取了以下48个代码示例，用于说明如何使用responses.GET。

项目：flash_services 作者：textbook | 项目源码 | 文件源码

def test_get_velocity_success(service):
    responses.add(
        responses.GET,
        'https://www.pivotaltracker.com/services/v5/projects/123/iterations'
        '/456?fields=%3Adefault%2Cvelocity%2Cstories',
        json={
            'velocity': 10,
            'stories': [
                {'current_state': 'foo', 'estimate': 5},
                {'current_state': 'foo'},
            ],
        }
    )

    result = service.details(456)

    assert result == {'velocity': 10, 'stories': {'foo': 5}}
    assert responses.calls[0].request.headers['X-TrackerToken'] == 'foobar'

项目：flash_services 作者：textbook | 项目源码 | 文件源码

def test_update_success(debug, service):
    service.current_iteration = 1
    service.project_version = 2
    service._cached = {'foo': 'bar'}
    responses.add(
        responses.GET,
        'https://www.pivotaltracker.com/services/v5/projects/123',
        json={'foo': 'bar', 'current_iteration_number': 0},
        adding_headers={'X-Tracker-Project-Version': '1'},
    )

    result = service.update()

    debug.assert_called_once_with('fetching Tracker project data')
    assert result == {'foo': 'bar'}
    assert responses.calls[0].request.headers['X-TrackerToken'] == 'foobar'

项目：flash_services 作者：textbook | 项目源码 | 文件源码

def test_update_details(debug, service):
    service.current_iteration = 1
    service.project_version = 1
    service._cached = {'foo': 'bar'}
    name = 'foo'
    responses.add(
        responses.GET,
        'https://www.pivotaltracker.com/services/v5/projects/123',
        json={'current_iteration_number': 1, 'name': name},
        adding_headers={'X-Tracker-Project-Version': '2'},
    )
    responses.add(
        responses.GET,
        'https://www.pivotaltracker.com/services/v5/projects/123/iterations'
        '/1?fields=%3Adefault%2Cvelocity%2Cstories',
        json={'velocity': 10, 'stories': []}
    )

    result = service.update()

    debug.assert_has_calls([
        mock.call('fetching Tracker project data'),
        mock.call('project updated, fetching iteration details'),
    ])
    assert result == dict(velocity=10, stories={}, name=name)

项目：flash_services 作者：textbook | 项目源码 | 文件源码

def test_update_enterprise_success(debug):
    responses.add(
        responses.GET,
        'http://dummy.url/repos/foo/bar/issues?state=all&access_token=foobar',
        headers={'User-Agent': 'bar'},
        json=[],
    )
    service = GitHubEnterpriseIssues(
        api_token='foobar',
        account='foo',
        repo='bar',
        root='http://dummy.url',
    )

    result = service.update()

    debug.assert_called_once_with('fetching GitHub issue data')
    assert result == {'issues': {}, 'name': 'foo/bar', 'health': 'neutral', 'halflife': None}

项目：flash_services 作者：textbook | 项目源码 | 文件源码

def test_format_build_missing_data(service):
    responses.add(
        responses.GET,
        'https://coveralls.io/foo/bar/baz.json?page=1',
        json=dict(builds=[{}]),
    )

    result = service.update()

    assert result['builds'][0] == dict(
        author='<no author>',
        committed='time not available',
        coverage=None,
        message_text=None,
        raw_coverage=None,
    )

项目：flash_services 作者：textbook | 项目源码 | 文件源码

def test_update_success(debug, service):
    responses.add(
        responses.GET,
        'https://api.github.com/repos/foo/bar/commits?access_token=foobar',
        json=[{'commit': {
            'author': {'name': 'alice'},
            'committer': {'name': 'bob', 'date': TWO_DAYS_AGO},
            'message': 'commit message',
        }}],
    )

    result = service.update()

    debug.assert_called_once_with('fetching GitHub project data')
    assert result == {'commits': [{
        'message': 'commit message',
        'author': 'alice [bob]',
        'committed': 'two days ago'
    }], 'name': 'foo/bar'}
    assert responses.calls[0].request.headers['User-Agent'] == 'bar'

项目：flash_services 作者：textbook | 项目源码 | 文件源码

def test_update_enterprise_success(debug):
    responses.add(
        responses.GET,
        'http://dummy.url/repos/foo/bar/commits?access_token=foobar',
        json=[{'commit': {
            'author': {'name': 'alice'},
            'committer': {'name': 'bob', 'date': TWO_DAYS_AGO},
            'message': 'commit message',
        }}],
    )

    service = GitHubEnterprise(api_token='foobar', account='foo', repo='bar',
                               root='http://dummy.url')

    result = service.update()

    debug.assert_called_once_with('fetching GitHub project data')
    assert result == {'commits': [{
        'message': 'commit message',
        'author': 'alice [bob]',
        'committed': 'two days ago'
    }], 'name': 'foo/bar'}
    assert responses.calls[0].request.headers['User-Agent'] == 'bar'

项目：flash_services 作者：textbook | 项目源码 | 文件源码

def test_estimated_formatting(service, url):
    response = dict(name='foo', builds=[
        dict(duration=0, description=None, timestamp=1481387964313, result=None),
        dict(duration=10000, description=None, timestamp=1481387964313, result='SUCCESS'),
        dict(duration=10000, description=None, timestamp=1481387964313, result='SUCCESS'),
        dict(duration=10000, description=None, timestamp=1481387964313, result='SUCCESS'),
        dict(duration=10000, description=None, timestamp=1481387964313, result='SUCCESS'),
        dict(duration=10000, description=None, timestamp=1481387964313, result='SUCCESS'),
    ])
    responses.add(responses.GET, url, json=response)

    result = service.update()

    assert len(result['builds']) == 4
    assert result['builds'][0] == dict(
        author='<no author>',
        duration=5,
        elapsed='five seconds left',
        message='<no message>',
        outcome='working',
        started_at=1481387964,
    )

项目：sentry-plugins 作者：getsentry | 项目源码 | 文件源码

def test_create_issue(self):
        responses.add(
            responses.GET,
            'https://getsentry.atlassian.net/rest/api/2/issue/createmeta',
            json=create_meta_response)
        responses.add(responses.POST, 'https://getsentry.atlassian.net/rest/api/2/issue', json={
            'key': 'SEN-1'
        })
        self.plugin.set_option('instance_url', 'https://getsentry.atlassian.net', self.project)
        group = self.create_group(message='Hello world', culprit='foo.bar')

        request = self.request.get('/')
        request.user = AnonymousUser()
        form_data = {
            'title': 'Hello',
            'description': 'Fix this.',
            'issuetype': 'bug',
            'project': 'SEN'
        }
        assert self.plugin.create_issue(request, group, form_data) == 'SEN-1'

项目：pymarsys 作者：transcovo | 项目源码 | 文件源码

def test_make_call(self):
        connection = AsyncConnection(
            TEST_USERNAME,
            TEST_SECRET,
            EMARSYS_URI
        )
        with aioresponses() as m:
            m.get(
                urljoin(EMARSYS_URI, 'api/v2/settings'),
                status=200,
                payload=EMARSYS_SETTINGS_RESPONSE
            )
            coroutine = connection.make_call('GET', 'api/v2/settings')
            loop = asyncio.get_event_loop()
            response = loop.run_until_complete(coroutine)
            assert response == EMARSYS_SETTINGS_RESPONSE

项目：pymarsys 作者：transcovo | 项目源码 | 文件源码

def test_list_choice(self):
        responses.add(
            responses.GET,
            urljoin(
                EMARSYS_URI, '{}/{}/{}'.format(
                    CONTACT_FIELDS_ENDPOINT,
                    5,
                    'choice'
                )
            ),
            json=EMARSYS_CONTACT_FIELDS_LIST_CHOICE_RESPONSE,
            status=200,
            content_type='application/json'
        )
        connection = SyncConnection(TEST_USERNAME, TEST_SECRET)
        contact_fields = ContactField(connection)

        response = contact_fields.list_choice(5)
        assert response == EMARSYS_CONTACT_FIELDS_LIST_CHOICE_RESPONSE

项目：instagram_private_api_extensions 作者：ping | 项目源码 | 文件源码

def test_downloader_conn_error(self):
        exception = ConnectionError()
        with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
            max_retry = 3
            for _ in range(max_retry + 1):
                rsps.add(responses.GET, self.TEST_MPD_URL, body=exception)

            dl = live.Downloader(
                mpd=self.TEST_MPD_URL,
                output_dir='output_connerror',
                duplicate_etag_retry=2,
                singlethreaded=True,
                max_connection_error_retry=max_retry)
            dl.run()
            dl.stream_id = '17875351285037717'
            output_file = 'output_connerror.mp4'
            dl.stitch(output_file, cleartempfiles=True)
            self.assertFalse(os.path.isfile(output_file), '{0!s} not generated'.format(output_file))

项目：jwplatform-py 作者：jwplayer | 项目源码 | 文件源码

def test_existing_resource():
    url_expr = re.compile(r'https?://api\.test\.tst/v1/videos/show\?.*'
                          'video_key=VideoKey.*')
    responses.add(
        responses.GET, url_expr,
        status=200,
        content_type='application/json',
        body='{"status": "ok", '
             '"rate_limit": {"reset": 1478929300, "limit": 50, "remaining": 47},'
             '"video": {"status": "ready", "expires_date": null, "description": null, '
             '"title": "Title", "views": 179, "tags": "", "sourceformat": null, '
             '"mediatype": "video", "upload_session_id": null, "custom": {}, '
             '"duration": "817.56", "sourceurl": null, "link": null, "author": null, '
             '"key": "VideoKey", "error": null, "date": 1464754765, '
             '"md5": "653bc15b6cba7319c2df9b5cf869b5b8", "sourcetype": "file", '
             '"size": "904237686"}}')

    jwp_client = jwplatform.Client('api_key', 'api_secret', host='api.test.tst')
    resp = jwp_client.videos.show(video_key='VideoKey')

    assert resp['status'] == 'ok'
    assert 'status' in resp['video']
    assert resp['video']['key'] == 'VideoKey'

项目：jwplatform-py 作者：jwplayer | 项目源码 | 文件源码

def test_nonexisting_resource():
    url_expr = re.compile(r'https?://api\.test\.tst/v1/videos/abcd/show\?.*'
                          'abcd_key=AbcdKey.*')
    responses.add(
        responses.GET, url_expr,
        status=404,
        content_type='application/json',
        body='{"status": "error", '
             '"message": "API method `/videos/abcd/show` not found", '
             '"code": "NotFound", "title": "Not Found"}')

    jwp_client = jwplatform.Client('api_key', 'api_secret', host='api.test.tst')

    with pytest.raises(jwplatform.errors.JWPlatformNotFoundError) as err:
        jwp_client.videos.abcd.show(abcd_key='AbcdKey')

    assert err.value.message == 'API method `/videos/abcd/show` not found'

项目：AtlassianBot 作者：gpailler | 项目源码 | 文件源码

def test_wrong_auth(bot):
    with controlled_responses() as rsps:
        rsps.rsps.add(
            responses.GET,
            'http://host/rest/api/2/issue/JIRA-1',
            status=401)

        message = get_message('JIRA-1')
        bot.display_issues(message)

        assert len(message.send_webapi.call_args_list) == 1
        args, kwargs = message.send_webapi.call_args_list[0]
        assert args[0] is ''
        assert json.loads(args[1]) == [{
                'color': 'danger',
                'fallback': 'Jira authentication error',
                'text': ':exclamation: Jira authentication error'}]

项目：AtlassianBot 作者：gpailler | 项目源码 | 文件源码

def __add_responses(self):
        for request in self.requests_get:
            body = None
            content_type = 'application/json'

            if 'content_type' in request:
                content_type = request['content_type']

            if 'text' in request:
                if content_type.startswith('image/'):
                    body = base64.b64decode(request['text'])
                else:
                    body = json.dumps(request['text'])

            self.rsps.add(
                responses.GET,
                request['url'],
                status=request['code'],
                body=body,
                content_type=content_type,
                match_querystring=True)

项目：drift 作者：dgnorth | 项目源码 | 文件源码

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
            )

项目：python-twitch-client 作者：tsifrer | 项目源码 | 文件源码

def test_get_by_slug():
    slug = 'OpenUglySnoodVoteNay'

    responses.add(
        responses.GET,
        '%sclips/%s' % (BASE_URL, slug),
        body=json.dumps(example_clip),
        status=200,
        content_type='application/json'
    )

    client = TwitchClient('client id')
    clip = client.clips.get_by_slug(slug)

    assert isinstance(clip, Clip)
    assert clip.broadcast_id == example_clip['broadcast_id']

项目：python-twitch-client 作者：tsifrer | 项目源码 | 文件源码

def test_get_top():
    params = {'limit': 1, 'period': 'month'}

    responses.add(
        responses.GET,
        '%sclips/top' % BASE_URL,
        body=json.dumps(example_clips),
        status=200,
        content_type='application/json'
    )

    client = TwitchClient('client id')
    clips = client.clips.get_top(**params)

    assert len(clips) == len(example_clips)
    assert isinstance(clips[0], Clip)
    assert clips[0].broadcast_id == example_clips['clips'][0]['broadcast_id']

项目：libmozdata 作者：mozilla | 项目源码 | 文件源码

def setup_versions(self, stable, devel, nightly, esr, esr_next=None):
        """
        Add a mock response for official versions
        """
        self.cleanup()
        body = {
            "FIREFOX_NIGHTLY": nightly,
            "FIREFOX_ESR": esr,
            "FIREFOX_ESR_NEXT": esr_next,
            "LATEST_FIREFOX_DEVEL_VERSION": devel,
            "LATEST_FIREFOX_OLDER_VERSION": "3.6.28",
            "LATEST_FIREFOX_RELEASED_DEVEL_VERSION": devel,
            "LATEST_FIREFOX_VERSION": stable,
        }
        local_mock = responses.RequestsMock()
        local_mock.add(responses.GET, versions.URL_VERSIONS, json=body)
        local_mock.start()
        yield local_mock
        local_mock.stop()