我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用httpretty.GET。
def setUp(self): httpretty.enable() httpretty.register_uri( httpretty.GET, "http://test.gf/users", body=ujson.dumps([ sam_profile, jack_profile, ]), content_type="application/json" ) httpretty.register_uri( httpretty.GET, "http://test.gf/blog/sam", body=ujson.dumps(sams_articles), content_type="application/json" ) httpretty.register_uri( httpretty.GET, "http://test.gf/blog/jack", body=ujson.dumps(jacks_articles), content_type="application/json" )
def test_weibo_request(self): """ :return: """ body = """ {"next_cursor": 4098746105417334, "hasvisible": false, "uve_blank": -1, "statuses": ["4098748692646689", "4098748689403910", "4098748688918202", "4098747640117551", "4098747434341769", "4098747430814353", "4098747430145399", "4098747052681634", "4098746357066702", "4098746322855226", "4098746297959121", "4098746251989195", "4098746226771596", "4098746205413046", "4098746180805829", "4098746175979920", "4098746172045575", "4098746172045403", "4098746160243131", "4098746139219122"], "ad": [], "advertises": [], "interval": 0, "previous_cursor": 0, "total_number": 1322} """ httpretty.register_uri( httpretty.GET, "https://api.weibo.com/2/statuses/friends_timeline/ids.json", body=body, status=200, content_type='text/json' ) self.assertDictEqual(self.client.get("statuses/friends_timeline/ids.json"), json.loads(body))
def test_weibo_api_error(self): """ :return: """ body = """ { "request": "/statuses/home_timeline.json", "error_code": "20502", "error": "Need you follow uid." } """ httpretty.register_uri( httpretty.GET, "https://api.weibo.com/2/statuses/home_timeline.json", body=body, status=200, content_type='text/json' ) self.assertRaises(WeiboAPIError, self.client.get, "statuses/home_timeline.json")
def test_project_edit(connection, project, project_payload): """Verifies project renaming.""" httpretty.register_uri( httpretty.POST, '{}/project/{}/edit'.format( matchlight.MATCHLIGHT_API_URL_V2, project.upload_token, ), body='{}', status=200) new_name = 'Test Project 1' connection.projects.edit(project, new_name) assert project.name == new_name new_name = 'Test Project 2' httpretty.register_uri( httpretty.GET, '{}/project/{}'.format( matchlight.MATCHLIGHT_API_URL_V2, project.upload_token, ), body=json.dumps(project_payload), content_type='application/json', status=200) project = connection.projects.edit(project.upload_token, new_name) assert project.name == new_name
def test_project_get(connection, project_payload, project): """Verifies project retrieval.""" httpretty.register_uri( httpretty.GET, '{}/project/{}'.format( matchlight.MATCHLIGHT_API_URL_V2, project.upload_token), responses=[ httpretty.Response(body=json.dumps(project_payload), content_type='application/json', status=200), httpretty.Response(body='{}', content_type='application/json', status=404), httpretty.Response(body='{}', content_type='application/json', status=500), ]) project_ = connection.projects.get(project.upload_token) assert project.upload_token == project_.upload_token project_ = connection.projects.get(project.upload_token) assert project_ is None with pytest.raises(matchlight.error.ConnectionError): connection.projects.get(project.upload_token)
def test_alert_filter_record(connection, alert, alert_payload, document_record): """Verifies alert filtering on 'record_id'.""" httpretty.register_uri( httpretty.GET, '{}/alerts?record_id_filter={}&limit=50'.format( matchlight.MATCHLIGHT_API_URL_V2, document_record.id ), body=json.dumps({'alerts': [alert_payload]}), content_type='application/json', status=200 ) alerts = connection.alerts.filter(limit=50, record=document_record) assert len(alerts) == 1 assert alerts[0].id == alert_payload['id']
def test_alert_details(connection, alert, alert_details_pii_payload): """Verifies alert get details responses.""" httpretty.register_uri( httpretty.GET, '{}/alert/{}/details'.format( matchlight.MATCHLIGHT_API_URL_V2, alert.id ), body=json.dumps(alert_details_pii_payload), content_type='application/json', status=200 ) details_ = connection.alerts.get_details(alert) assert details_ == alert_details_pii_payload details_ = connection.alerts.get_details(alert.id) assert details_ == alert_details_pii_payload
def test_usa_jobs_updater_deduplicated_items(): httpretty.register_uri( httpretty.GET, USAJobsUpdater.base_url, body=callback ) updater = USAJobsUpdater( auth_key=FAKE_KEY, key_email=FAKE_EMAIL, ) postings = updater.deduplicated_postings() assert len(postings.keys()) == 3 assert sorted(postings.keys()) == [5, 6, 7] assert postings == { 5: {'k': 'v'}, 6: {'k': 'v2'}, 7: {'k': 'v3'}, }
def test_place_ua(): httpretty.register_uri( httpretty.GET, URL, body=RESPONSE, content_type='text/csv' ) results = place_ua.__wrapped__(city_cleaner) assert results == { 'LA': {'abbeville': '00037', 'ama': '62677'}, 'MD': {'aberdeen': '00199'}, 'NM': {'placitas': '01171'}, 'MO': {'st louis': '77770', 'grain valley': '43912'}, 'NC': {'winston-salem': '96670', 'winston': '96670', 'salem': '96670'}, 'ID': {'boise': '08785', 'boise city': '08785'} }
def test_ua_cbsa(): httpretty.register_uri( httpretty.GET, URL, body=RESPONSE, content_type='text/csv' ) results = ua_cbsa.__wrapped__() assert results == { '00037': [ ('10020', 'Abbeville, LA Micro Area'), ('35340', 'New Iberia, LA Micro Area'), ], '00091': [('48140', 'Wausau, WI Metro Area')] }
def test_get_doc_with_no_doc_name(self): httpretty.register_uri( httpretty.GET, "http://example.com/api/resource/SomeDoc/", body='{"data": { "f1": "v1","f2": "v2"}}', content_type="application/json" ) res = self.frappe.get_doc( "SomeDoc", filters=[["Note", "title", "LIKE", "S%"]], fields=["name", "foo"]) self.assertEquals(res, {'f1': 'v1', 'f2': 'v2'}) request = httpretty.last_request() url = urlparse.urlparse(request.path) query_dict = urlparse.parse_qs(url.query) self.assertEquals(query_dict['fields'], [u'["name", "foo"]']) self.assertEquals(query_dict['filters'], [u'[["Note", "title", "LIKE", "S%"]]'])
def test_httpretty_bypasses_a_unregistered_request(context): "httpretty should bypass a unregistered request by disabling it" httpretty.register_uri( httpretty.GET, "http://localhost:{0}/go-for-bubbles/".format(context.http_port), body="glub glub") fd = urllib2.urlopen('http://localhost:{0}/go-for-bubbles/'.format(context.http_port)) got1 = fd.read() fd.close() expect(got1).to.equal(b'glub glub') fd = urllib2.urlopen('http://localhost:{0}/come-again/'.format(context.http_port)) got2 = fd.read() fd.close() expect(got2).to.equal(b'<- HELLO WORLD ->') core.POTENTIAL_HTTP_PORTS.remove(context.http_port)
def test_disallow_net_connect_1(context): """ When allow_net_connect = False, a request that otherwise would have worked results in UnmockedError. """ httpretty.register_uri(httpretty.GET, "http://falcao.it/foo/", body="BAR") def foo(): fd = None try: fd = urllib2.urlopen('http://localhost:{0}/go-for-bubbles/'.format(context.http_port)) finally: if fd: fd.close() foo.should.throw(httpretty.UnmockedError)
def mock_user_info_response(self, status=200, username=None): """ Mock the user info endpoint response of the OAuth2 provider. """ username = username or USER_INFO['username'] data = { 'family_name': USER_INFO['last_name'], 'preferred_username': username, 'given_name': USER_INFO['first_name'], 'email': USER_INFO['email'], } httpretty.register_uri( httpretty.GET, OAUTH2_USER_INFO_URL, body=json.dumps(data), content_type='application/json', status=status )
def test_items(self): dwarves = [{'name': 'Dori'}, {'name': 'Nori'}, {'name': 'Ori'}, {'name': 'Bifur'}, {'name': 'Bofur'}, {'name': 'Bombur'}] def request_callback(request, uri, headers): if uri.endswith('dwarves') or uri.endswith('?page=1'): items = dwarves[:3] page_id = 1 elif uri.endswith('?page=2'): items = dwarves[3:] page_id = 2 else: items = [] page_id = uri.split('=')[-1] return 200, headers, json.dumps({'page': page_id, 'dwarves': items, 'total_pages': 2}) self.register_url(self.GET, 'dwarves', body=request_callback) api = QuizletAPI() self.assertEqual(list(api.dwarves.items()), dwarves)
def test_request_timeout(): timeout = 0.1 http_scheme = 'http' host = 'coordinator' port = 8080 url = http_scheme + '://' + host + ':' + str(port) + constants.URL_STATEMENT_PATH def long_call(request, uri, headers): time.sleep(timeout * 2) return (200, headers, "delayed success") httpretty.enable() for method in [httpretty.POST, httpretty.GET]: httpretty.register_uri(method, url, body=long_call) # timeout without retry for request_timeout in [timeout, (timeout, timeout)]: req = PrestoRequest( host=host, port=port, user='test', http_scheme=http_scheme, max_attempts=1, request_timeout=request_timeout, ) with pytest.raises(requests.exceptions.Timeout): req.get(url) with pytest.raises(requests.exceptions.Timeout): req.post('select 1') httpretty.disable() httpretty.reset()
def setUp(self): with open("line_file.json", "w") as f: f.write(JSON_LINE_CONTENT) with open("array_file.json", "w") as f: f.write(JSON_ARRAY_CONTENT) with open("object_file.json", "w") as f: f.write(JSON_OBJECT_CONTENT) with open("block_file.json", "w") as f: f.write(JSON_BLOCK_CONTENT) httpretty.enable() httpretty.register_uri( httpretty.GET, "http://test.gf/line", body=JSON_LINE_CONTENT, content_type="text/plain" ) httpretty.register_uri( httpretty.GET, "http://test.gf/array", body=JSON_ARRAY_CONTENT, content_type="application/json" ) httpretty.register_uri( httpretty.GET, "http://test.gf/object", body=JSON_OBJECT_CONTENT, content_type="application/json" )
def setUp(self): with open("test.csv", "w") as f: f.write(CSV_CONTENT) httpretty.enable() httpretty.register_uri( httpretty.GET, "http://test.gf/csv", body=CSV_CONTENT, content_type="text/plain" )
def test_req(self): http_server_fixture = HttpServerFixture() self.useFixture(http_server_fixture) context = self.workflow_context() result = Req( "GET", "http://test.gf/users", parser=_default_parser )(context, []) self.assertEquals(result, [sam_profile, jack_profile])
def test_graylog_api_search(self): httpretty.register_uri( httpretty.GET, "http://dummyhost:80/api/search/universal/absolute", body=self.generate_search_result(), content_type="application/json" ) # More of some dummy tests now sr = api.SearchRange("10 minutes ago", arrow.now()) q = api.SearchQuery(sr) result = self.api.search(q) self.assertEquals(len(result.messages), 1) self.assertEquals("*", result.query) q = api.SearchQuery(sr, fields=["level", "module", "message", "timestamp"], sort="level", ascending=True) qq = q.copy_with_range(sr) result = self.api.search(qq) self.assertEquals(len(result.messages), 1) self.assertEquals("*", result.query, ) q = api.SearchQuery(sr, fields=["level", "module", "message", "timestamp"], sort="level", ascending=False) result = self.api.search(q) self.assertEquals(len(result.messages), 1) self.assertEquals("*", result.query) result = self.api.search(q, fetch_all=True) self.assertEquals(len(result.messages), 1) self.assertEquals("*", result.query)
def test_to_many_results(self): httpretty.register_uri( httpretty.GET, "http://dummyhost:80/api/search/universal/absolute", body=self.generate_search_result(1000000), content_type="application/json" ) sr = api.SearchRange("10 minutes ago", arrow.now()) q = api.SearchQuery(sr) with self.assertRaises(RuntimeError): self.api.search(q, fetch_all=True)
def test_userinfo(self): httpretty.register_uri( httpretty.GET, "http://dummyhost:80/api/users/dummy", body='{"someuser" : "info" }', content_type="application/json" ) result = self.api.user_info() self.assertEquals({"someuser": "info"}, result)
def test_streams(self): httpretty.register_uri( httpretty.GET, "http://dummyhost:80/api/streams/enabled", body='[{"somestream" : "a" }]', content_type="application/json" ) result = self.api.streams() self.assertEquals([{"somestream": "a"}], result)
def test_saved_searches(self): httpretty.register_uri( httpretty.GET, "http://dummyhost:80/api/search/saved", body='{"searches" : [{"query": {"query": "level:INFO", "fields":"timestamp,message"}}]}', content_type="application/json" ) saved_searches = self.api.get_saved_queries() self.assertEquals("level:INFO", saved_searches['searches'][0]['query']['query']) self.assertEquals("timestamp,message", saved_searches['searches'][0]['query']['fields'])
def register_endpoint(endpoint, body, status=200, method=httpretty.GET): """Mock GitHub API response.""" httpretty.register_uri( method, 'https://api.github.com%s' % endpoint, body=json.dumps(body), status=status, )
def register_local_endpoint(endpoint, body, status=200, method=httpretty.GET): """Mock GitHub API response.""" httpretty.register_uri( method, 'https://api.github.com%s' % endpoint, body=json.dumps(body), status=status, ) # # Fixture generators #
def tfs_server_mock(): for method in (httpretty.GET, httpretty.POST, httpretty.PATCH): httpretty.register_uri(method, re.compile(r"http://tfs.tfs.ru(.*)"), body=request_callback_get, content_type="application/json")
def test_download(self): # Onedrive requires the path in the URL, we use the chunk's uid as # path. httpretty.register_uri( httpretty.GET, OnedriveAPIClient.DOWNLOAD_URL[1].format(path=self._get_path()), body=TEST_CHUNK_BODY) self.assertEqual(TEST_CHUNK_BODY, self.client.download(self.chunk))
def test_download(self): # Box requires the file id in the URL, the file_id is assigned by Box, # and therefore is stored in ChunkStorage.attrs. httpretty.register_uri( httpretty.GET, BoxAPIClient.DOWNLOAD_URL[1].format(file_id='abc123'), body=TEST_CHUNK_BODY) self.assertEqual(TEST_CHUNK_BODY, self.client.download(self.chunk))
def test_download(self): # GDrive requires the file id in the URL, the file_id is assigned by # Google, and therefore is stored in ChunkStorage.attrs. httpretty.register_uri( httpretty.GET, GDriveAPIClient.DOWNLOAD_URL[1].format(file_id='abc123'), body=TEST_CHUNK_BODY) self.assertEqual(TEST_CHUNK_BODY, self.client.download(self.chunk))
def setUp(self): super().setUp() httpretty.register_uri( METHODS.get(self.oauth2_client.USER_PROFILE_URL[0]), self.oauth2_client.USER_PROFILE_URL[1], body=json.dumps(USER_PROFILE_GOOGLE), content_type='application/json') httpretty.register_uri( METHODS.get(self.oauth2_client.USER_STORAGE_URL[0]), self.oauth2_client.USER_STORAGE_URL[1], body=json.dumps(USER_PROFILE_GOOGLE), content_type='application/json') httpretty.register_uri(httpretty.GET, self.oauth2_client.CREATE_URL, body=json.dumps(CREATE_GOOGLE), content_type='application/json')
def test_feed_download_output(mock_open, connection, feed, start_time, end_time, feed_download_url, feed_report_csv): """Verifies feed download writing to a file.""" mock_open.return_value = mock.MagicMock(spec=io.IOBase) httpretty.register_uri( httpretty.POST, '{}/feed/{}/prepare'.format( matchlight.MATCHLIGHT_API_URL_V2, feed.name), body=json.dumps({'feed_response_id': 1}), content_type='application/json', status=200) httpretty.register_uri( httpretty.POST, '{}/feed/{}/link'.format( matchlight.MATCHLIGHT_API_URL_V2, feed.name), responses=[ httpretty.Response( body=json.dumps({'status': 'pending'}), content_type='application/json', status=200), httpretty.Response( body=json.dumps({ 'status': 'ready', 'url': feed_download_url, }), content_type='application/json', status=200), ], ) body = '\n'.join(feed_report_csv).encode('utf-8') httpretty.register_uri( httpretty.GET, feed_download_url, content_type='text/csv', body=body) connection.feeds.download( feed, start_time, end_time, save_path='/tmp/output') file_handle = mock_open.return_value.__enter__.return_value file_handle.write.assert_called_once_with(body)
def test_feed_iteration(connection, feed): """Verifies feed iteration.""" httpretty.register_uri( httpretty.GET, '{}/feeds'.format(matchlight.MATCHLIGHT_API_URL_V2), body=json.dumps({'feeds': [feed.details]}), content_type='application/json', status=200, ) feeds_iterable = iter(connection.feeds) assert next(feeds_iterable).details == feed.details with pytest.raises(StopIteration): next(feeds_iterable)
def test_feed_list(connection, feed): """Verifies feed listing.""" httpretty.register_uri( httpretty.GET, '{}/feeds'.format(matchlight.MATCHLIGHT_API_URL_V2), body=json.dumps({'feeds': [feed.details]}), content_type='application/json', status=200, ) feeds = connection.feeds.all() assert len(feeds) == 1 assert feeds[0].details == feed.details
def test_project_filter(connection, project_payload, project): """Verifies project listing and filtering by type.""" httpretty.register_uri( httpretty.GET, '{}/projects'.format( matchlight.MATCHLIGHT_API_URL_V2), body=json.dumps({'data': [project_payload]}), content_type='application/json', status=200) projects = connection.projects.filter() assert len(projects) == 1 assert projects[0].upload_token == project.upload_token httpretty.reset() project_list = [project_payload] for _ in six.moves.range(5): payload = project_payload.copy() for project_type in PROJECT_TYPES: if project_type == payload['project_type']: continue payload['project_type'] = project_type break project_list.append(payload) httpretty.register_uri( httpretty.GET, '{}/projects'.format( matchlight.MATCHLIGHT_API_URL_V2), body=json.dumps({'data': project_list}), content_type='application/json', status=200) projects = connection.projects.filter(project_type=project.project_type) assert len(projects) == 1 assert projects[0].project_type == project.project_type
def test_project_iteration(connection, project, project_payload): """Verifies project iteration.""" httpretty.register_uri( httpretty.GET, '{}/projects'.format(matchlight.MATCHLIGHT_API_URL_V2), body=json.dumps({'data': [project_payload]}), content_type='application/json', status=200, ) projects_iterable = iter(connection.projects) assert next(projects_iterable).details == project.details with pytest.raises(StopIteration): next(projects_iterable)
def test_alert_dates(connection, alert, alert_payload): """Verifies alert date objects are converted correctly.""" httpretty.register_uri( httpretty.GET, '{}/alerts?limit=50'.format( matchlight.MATCHLIGHT_API_URL_V2 ), body=json.dumps({'alerts': [alert_payload]}), content_type='application/json', status=200 ) alerts = connection.alerts.filter(limit=50) assert isinstance(alerts[0].date, datetime.datetime) assert isinstance(alerts[0].last_modified, datetime.datetime)
def test_alert_filter(connection, alert, alert_payload): """Verifies alert listing and filtering.""" httpretty.register_uri( httpretty.GET, '{}/alerts?limit=50'.format( matchlight.MATCHLIGHT_API_URL_V2 ), body=json.dumps({'alerts': [alert_payload]}), content_type='application/json', status=200 ) alerts = connection.alerts.filter(limit=50) assert len(alerts) == 1 assert alerts[0].id == alert.id
def test_alert_filter_archived(connection, alert, alert_payload): """Verifies alert filtering on 'archived'.""" # Create opposite alert unarchived_payload = alert_payload.copy() unarchived_payload['archived'] = 'false' unarchived_payload['id'] = str(uuid.uuid4()) # Get archived alerts httpretty.register_uri( httpretty.GET, '{}/alerts?archived=1&limit=50'.format( matchlight.MATCHLIGHT_API_URL_V2 ), body=json.dumps({'alerts': [alert_payload]}), content_type='application/json', status=200 ) alerts = connection.alerts.filter(limit=50, archived=True) assert len(alerts) == 1 assert alerts[0].id == alert_payload['id'] # Get unarchived alerts httpretty.register_uri( httpretty.GET, '{}/alerts?archived=0&limit=50'.format( matchlight.MATCHLIGHT_API_URL_V2 ), body=json.dumps({'alerts': [unarchived_payload]}), content_type='application/json', status=200 ) alerts = connection.alerts.filter(limit=50, archived=False) assert len(alerts) == 1 assert alerts[0].id == unarchived_payload['id']
def test_alert_filter_project(connection, alert, alert_payload, project): """Verifies alert filtering on 'upload_token'.""" httpretty.register_uri( httpretty.GET, '{}/alerts?upload_token_filter={}&limit=50'.format( matchlight.MATCHLIGHT_API_URL_V2, project.upload_token ), body=json.dumps({'alerts': [alert_payload]}), content_type='application/json', status=200 ) alerts = connection.alerts.filter(limit=50, project=project) assert len(alerts) == 1 assert alerts[0].id == alert_payload['id']
def test_get_user_info(): server = BootstrapServer("http://test.test:5984") # Try to get user info -- Should fail try: server.get_user_info() assert False, "BootstrapServer.get_user_info should fail when not logged in" except RuntimeError as e: pass def on_get_session(request, uri, headers): credentials = request.headers.getheader("Authorization") if credentials.startswith("Basic "): username, password = b64decode(credentials[6:]).split(":") if username == "test" and password == "test": return 200, headers, '{"test": "test"}' return 401, headers, '{"test": "test"}' httpretty.register_uri( httpretty.GET, "http://test.test:5984/_session", body=on_get_session, content_type="application/json" ) server.log_in("test", "test") httpretty.register_uri( httpretty.HEAD, "http://test.test:5984/_users" ) httpretty.register_uri( httpretty.GET, "http://test.test:5984/_users/org.couchdb.user%3Atest", body='{"test": "test"}', content_type="application/json" ) res = server.get_user_info() assert server.get_user_info() == {"test": "test"} server.log_out() try: server.get_user_info() assert False, "Shouldn't be able to access the user's info when not logged in" except RuntimeError: pass
def test_get(self): """Function defined to test Plan get method.""" httpretty.register_uri( httpretty.GET, "https://api.paystack.co/plan/78", content_type='text/json', body='{"status": true, "contributors": true}', status=201, ) response = Plan.get(plan_id=78) self.assertEqual(response['status'], True)
def test_list(self): """Function defined to test paystackapi plan list method.""" httpretty.register_uri( httpretty.GET, "https://api.paystack.co/plan", content_type='text/json', body='{"status": true, "contributors": true}', status=201, ) response = Plan.list() self.assertEqual(response['status'], True)
def test_get(self): """Function defined to test Customer get method.""" httpretty.register_uri( httpretty.GET, "https://api.paystack.co/customer/4013", content_type='text/json', body='{"status": true, "contributors": true}', status=201, ) response = Customer.get(customer_id=4013) self.assertEqual(response['status'], True)
