我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用httpretty.last_request()。
def test_patch_saved_query_none_fields(mocked_url, mocked_rw_apikey, mocked_account_resource_id, capsys): test_saved_query_id = str(uuid.uuid4()) mocked_url.return_value = '', MOCK_API_URL mocked_rw_apikey.return_value = str(uuid.uuid4()) mocked_account_resource_id.return_value = str(uuid.uuid4()) httpretty.register_uri(httpretty.PATCH, MOCK_API_URL, status=200, content_type='application/json', body=json.dumps({"saved_query": SAVED_QUERY_RESPONSE})) api.update_saved_query(test_saved_query_id, name=None, statement="new_statement") out, err = capsys.readouterr() assert "Saved query with id %s updated" % test_saved_query_id in out body = json.loads(httpretty.last_request().body)['saved_query'] assert "name" not in body assert "statement" in body['leql']
def test_disable_api_key(mocked_url, mocked_owner_apikey, mocked_owner_apikey_id, mocked_account_resource_id, capsys): api_key_id = str(uuid.uuid4()) mocked_url.return_value = '', MOCK_API_URL mocked_owner_apikey.return_value = str(uuid.uuid4()) mocked_owner_apikey_id.return_value = str(uuid.uuid4()) mocked_account_resource_id.return_value = str(uuid.uuid4()) httpretty.register_uri(httpretty.PATCH, MOCK_API_URL, status=200, content_type='application/json', body=json.dumps({})) api.update(api_key_id, False) out, err = capsys.readouterr() assert {'apikey': {'active': False}} == json.loads(httpretty.last_request().body) assert not err assert 'Disabled api key with id: %s' % api_key_id in out
def test_enable_api_key(mocked_url, mocked_owner_apikey, mocked_owner_apikey_id, mocked_account_resource_id, capsys): api_key_id = str(uuid.uuid4()) mocked_url.return_value = '', MOCK_API_URL mocked_owner_apikey.return_value = str(uuid.uuid4()) mocked_owner_apikey_id.return_value = str(uuid.uuid4()) mocked_account_resource_id.return_value = str(uuid.uuid4()) httpretty.register_uri(httpretty.PATCH, MOCK_API_URL, status=200, content_type='application/json', body=json.dumps({})) api.update(api_key_id, True) out, err = capsys.readouterr() assert {'apikey': {'active': True}} == json.loads(httpretty.last_request().body) assert not err assert 'Enabled api key with id: %s' % api_key_id in out
def testCanDeleteOwnedDomain(self): httpretty.enable() httpretty.register_uri(httpretty.DELETE, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.') httpretty.register_uri(httpretty.DELETE, settings.NSMASTER_PDNS_API + '/zones/' + self.ownedDomains[1].name+ '.') url = reverse('domain-detail', args=(self.ownedDomains[1].pk,)) response = self.client.delete(url) self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT) self.assertEqual(httpretty.last_request().method, 'DELETE') self.assertEqual(httpretty.last_request().headers['Host'], 'nsmaster:8081') httpretty.reset() httpretty.register_uri(httpretty.DELETE, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.') httpretty.register_uri(httpretty.DELETE, settings.NSMASTER_PDNS_API + '/zones/' + self.ownedDomains[1].name+ '.') response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) self.assertTrue(isinstance(httpretty.last_request(), httpretty.core.HTTPrettyRequestEmpty))
def testCanDeleteOwnedDynDomain(self): httpretty.enable() httpretty.register_uri(httpretty.DELETE, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.') httpretty.register_uri(httpretty.DELETE, settings.NSMASTER_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.') url = reverse('domain-detail', args=(self.ownedDomains[1].pk,)) response = self.client.delete(url) self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT) # FIXME In this testing scenario, the parent domain dedyn.io does not # have the proper NS and DS records set up, so we cannot test their # deletion. httpretty.reset() httpretty.register_uri(httpretty.DELETE, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.') httpretty.register_uri(httpretty.DELETE, settings.NSMASTER_PDNS_API + '/zones/' + self.ownedDomains[1].name+ '.') response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) self.assertTrue(isinstance(httpretty.last_request(), httpretty.core.HTTPrettyRequestEmpty))
def test_providing_res_with_url_calls_datapusher_correctly(self): config['datapusher.url'] = 'http://datapusher.ckan.org' httpretty.HTTPretty.register_uri( httpretty.HTTPretty.POST, 'http://datapusher.ckan.org/job', content_type='application/json', body=json.dumps({'job_id': 'foo', 'job_key': 'bar'})) package = model.Package.get('annakarenina') tests.call_action_api( self.app, 'datastore_create', apikey=self.sysadmin_user.apikey, resource=dict(package_id=package.id, url='demo.ckan.org') ) assert len(package.resources) == 4, len(package.resources) resource = package.resources[3] data = json.loads(httpretty.last_request().body) assert data['metadata']['resource_id'] == resource.id, data assert not data['metadata'].get('ignore_hash'), data assert data['result_url'].endswith('/action/datapusher_hook'), data assert data['result_url'].startswith('http://'), data
def test_returns_latest_rates(self): expected_response = {'base': 'EUR', 'date': '2016-04-29', 'rates': {'GBP': 0.78025}} httpretty.register_uri(httpretty.GET, self.url, body=json.dumps(expected_response), content_type='application/json') client = Fixerio() response = client.latest() self.assertDictEqual(response, expected_response) request = httpretty.last_request() self.assertEqual(request.method, 'GET') self.assertEqual(request.path, self.path) self.assertEqual(request.querystring, {}) self.assertEqual(request.body, b'')
def test_returns_latest_rates_for_base_passed_in_constructor(self): base = 'USD' expected_response = {'base': base, 'date': '2016-05-13', 'rates': {'GBP': 0.69403}} httpretty.register_uri(httpretty.GET, self.url, body=json.dumps(expected_response), content_type='application/json') client = Fixerio(base=base) response = client.latest() self.assertDictEqual(response, expected_response) request = httpretty.last_request() self.assertEqual(request.method, 'GET') params = urlencode({'base': base}) expected_path = '{url}?{params}'.format(url=self.path, params=params) self.assertEqual(request.path, expected_path) self.assertEqual(request.querystring, {'base': [base]}) self.assertEqual(request.body, b'')
def test_returns_latest_rates_for_base_passed_in_method(self): base = 'USD' expected_response = {'base': base, 'date': '2016-05-13', 'rates': {'GBP': 0.69403}} httpretty.register_uri(httpretty.GET, self.url, body=json.dumps(expected_response), content_type='application/json') client = Fixerio() response = client.latest(base=base) self.assertDictEqual(response, expected_response) request = httpretty.last_request() self.assertEqual(request.method, 'GET') params = urlencode({'base': base}) expected_path = '{url}?{params}'.format(url=self.path, params=params) self.assertEqual(request.path, expected_path) self.assertEqual(request.querystring, {'base': [base]}) self.assertEqual(request.body, b'')
def test_returns_latest_rates_for_base_passed_in_method_if_both(self): base = 'USD' another_base = 'EUR' expected_response = {'base': base, 'date': '2016-05-13', 'rates': {'GBP': 0.69403}} httpretty.register_uri(httpretty.GET, self.url, body=json.dumps(expected_response), content_type='application/json') client = Fixerio(base=another_base) response = client.latest(base=base) self.assertDictEqual(response, expected_response) request = httpretty.last_request() self.assertEqual(request.method, 'GET') params = urlencode({'base': base}) expected_path = '{url}?{params}'.format(url=self.path, params=params) self.assertEqual(request.path, expected_path) self.assertEqual(request.querystring, {'base': [base]}) self.assertEqual(request.body, b'')
def test_returns_latest_rates_for_symbols_passed_in_constructor(self): symbols = ['USD', 'GBP'] expected_response = { "base": "EUR", "date": "2016-05-19", "rates": {"GBP": 0.76585, "USD": 1.1197} } httpretty.register_uri(httpretty.GET, self.url, body=json.dumps(expected_response), content_type='application/json') client = Fixerio(symbols=symbols) response = client.latest() self.assertDictEqual(response, expected_response) request = httpretty.last_request() self.assertEqual(request.method, 'GET') symbols_str = ','.join(symbols) params = urlencode({'symbols': symbols_str}) expected_path = '{url}?{params}'.format(url=self.path, params=params) self.assertEqual(request.path, expected_path) self.assertEqual(request.querystring, {'symbols': [symbols_str]}) self.assertEqual(request.body, b'')
def test_returns_latest_rates_for_symbols_passed_in_method_if_both(self): symbols = ['USD', 'GBP'] other_symbols = ['JPY', 'EUR'] expected_response = { "base": "EUR", "date": "2016-05-19", "rates": {"GBP": 0.76585, "USD": 1.1197} } httpretty.register_uri(httpretty.GET, self.url, body=json.dumps(expected_response), content_type='application/json') client = Fixerio(symbols=other_symbols) response = client.latest(symbols=symbols) self.assertDictEqual(response, expected_response) request = httpretty.last_request() self.assertEqual(request.method, 'GET') symbols_str = ','.join(symbols) params = urlencode({'symbols': symbols_str}) expected_path = '{url}?{params}'.format(url=self.path, params=params) self.assertEqual(request.path, expected_path) self.assertEqual(request.querystring, {'symbols': [symbols_str]}) self.assertEqual(request.body, b'')
def test_returns_latest_rates_for_secure_passed_in_constructor(self): expected_response = {'base': 'EUR', 'date': '2016-04-29', 'rates': {'GBP': 0.78025}} httpretty.register_uri(httpretty.GET, self.secure_url, body=json.dumps(expected_response), content_type='application/json') client = Fixerio(secure=True) response = client.latest() self.assertDictEqual(response, expected_response) request = httpretty.last_request() self.assertEqual(request.method, 'GET') self.assertEqual(request.path, self.path) self.assertEqual(request.querystring, {}) self.assertEqual(request.body, b'')
def test_returns_latest_rates_for_secure_passed_in_method(self): expected_response = {'base': 'EUR', 'date': '2016-04-29', 'rates': {'GBP': 0.78025}} httpretty.register_uri(httpretty.GET, self.secure_url, body=json.dumps(expected_response), content_type='application/json') client = Fixerio() response = client.latest(secure=True) self.assertDictEqual(response, expected_response) request = httpretty.last_request() self.assertEqual(request.method, 'GET') self.assertEqual(request.path, self.path) self.assertEqual(request.querystring, {}) self.assertEqual(request.body, b'')
def test_returns_latest_rates_for_secure_passed_in_method_if_both(self): expected_response = {'base': 'EUR', 'date': '2016-04-29', 'rates': {'GBP': 0.78025}} httpretty.register_uri(httpretty.GET, self.secure_url, body=json.dumps(expected_response), content_type='application/json') client = Fixerio(secure=False) response = client.latest(secure=True) self.assertDictEqual(response, expected_response) request = httpretty.last_request() self.assertEqual(request.method, 'GET') self.assertEqual(request.path, self.path) self.assertEqual(request.querystring, {}) self.assertEqual(request.body, b'')
def test_returns_historical_rates_for_base_passed_in_method(self): base = 'USD' expected_response = {'base': base, 'date': '2016-05-13', 'rates': {'GBP': 0.69403}} httpretty.register_uri(httpretty.GET, self.url, body=json.dumps(expected_response), content_type='application/json') client = Fixerio() response = client.historical_rates(date=self.date, base=base) self.assertDictEqual(response, expected_response) request = httpretty.last_request() self.assertEqual(request.method, 'GET') params = urlencode({'base': base}) expected_path = '{url}?{params}'.format(url=self.path, params=params) self.assertEqual(request.path, expected_path) self.assertEqual(request.querystring, {'base': [base]}) self.assertEqual(request.body, b'')
def test_returns_historical_rates_for_base_passed_if_both(self): base = 'USD' another_base = 'EUR' expected_response = {'base': base, 'date': '2016-05-13', 'rates': {'GBP': 0.69403}} httpretty.register_uri(httpretty.GET, self.url, body=json.dumps(expected_response), content_type='application/json') client = Fixerio(base=another_base) response = client.historical_rates(date=self.date, base=base) self.assertDictEqual(response, expected_response) request = httpretty.last_request() self.assertEqual(request.method, 'GET') params = urlencode({'base': base}) expected_path = '{url}?{params}'.format(url=self.path, params=params) self.assertEqual(request.path, expected_path) self.assertEqual(request.querystring, {'base': [base]}) self.assertEqual(request.body, b'')
def test_returns_historical_rates_for_symbols_passed_in_constructor(self): symbols = ['USD', 'GBP'] expected_response = {"base": "EUR", "date": "2000-01-03", "rates": {"GBP": 0.6246, "USD": 1.009}} httpretty.register_uri(httpretty.GET, self.url, body=json.dumps(expected_response), content_type='application/json') client = Fixerio(symbols=symbols) response = client.historical_rates(date=self.date) self.assertDictEqual(response, expected_response) request = httpretty.last_request() self.assertEqual(request.method, 'GET') symbols_str = ','.join(symbols) params = urlencode({'symbols': symbols_str}) expected_path = '{url}?{params}'.format(url=self.path, params=params) self.assertEqual(request.path, expected_path) self.assertEqual(request.querystring, {'symbols': [symbols_str]}) self.assertEqual(request.body, b'')
def test_returns_historical_rates_for_symbols_passed_in_method(self): symbols = ['USD', 'GBP'] expected_response = {"base": "EUR", "date": "2000-01-03", "rates": {"GBP": 0.6246, "USD": 1.009}} httpretty.register_uri(httpretty.GET, self.url, body=json.dumps(expected_response), content_type='application/json') client = Fixerio() response = client.historical_rates(date=self.date, symbols=symbols) self.assertDictEqual(response, expected_response) request = httpretty.last_request() self.assertEqual(request.method, 'GET') symbols_str = ','.join(symbols) params = urlencode({'symbols': symbols_str}) expected_path = '{url}?{params}'.format(url=self.path, params=params) self.assertEqual(request.path, expected_path) self.assertEqual(request.querystring, {'symbols': [symbols_str]}) self.assertEqual(request.body, b'')
def test_returns_historical_rates_for_symbols_passed_if_both(self): symbols = ['USD', 'GBP'] other_symbols = ['JPY', 'EUR'] expected_response = {"base": "EUR", "date": "2000-01-03", "rates": {"GBP": 0.6246, "USD": 1.009}} httpretty.register_uri(httpretty.GET, self.url, body=json.dumps(expected_response), content_type='application/json') client = Fixerio(symbols=other_symbols) response = client.historical_rates(date=self.date, symbols=symbols) self.assertDictEqual(response, expected_response) request = httpretty.last_request() self.assertEqual(request.method, 'GET') symbols_str = ','.join(symbols) params = urlencode({'symbols': symbols_str}) expected_path = '{url}?{params}'.format(url=self.path, params=params) self.assertEqual(request.path, expected_path) self.assertEqual(request.querystring, {'symbols': [symbols_str]}) self.assertEqual(request.body, b'')
def test_returns_historical_rates_for_secure_passed_in_method(self): expected_response = {'base': 'EUR', 'date': '2000-01-03', 'rates': {'GBP': 0.6246}} httpretty.register_uri(httpretty.GET, self.secure_url, body=json.dumps(expected_response), content_type='application/json') client = Fixerio() response = client.historical_rates(date=self.date, secure=True) self.assertDictEqual(response, expected_response) request = httpretty.last_request() self.assertEqual(request.method, 'GET') self.assertEqual(request.path, self.path) self.assertEqual(request.querystring, {}) self.assertEqual(request.body, b'')
def test_returns_historical_rates_for_base_passed_if_both(self): expected_response = {'base': 'EUR', 'date': '2000-01-03', 'rates': {'GBP': 0.6246}} httpretty.register_uri(httpretty.GET, self.secure_url, body=json.dumps(expected_response), content_type='application/json') client = Fixerio(secure=False) response = client.historical_rates(date=self.date, secure=True) self.assertDictEqual(response, expected_response) request = httpretty.last_request() self.assertEqual(request.method, 'GET') self.assertEqual(request.path, self.path) self.assertEqual(request.querystring, {}) self.assertEqual(request.body, b'')
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 setup_http_server(): remote_identities = read_file(REMOTE_IDENTITIES_FILE) http_requests = [] def request_callback(method, uri, headers): last_request = httpretty.last_request() if uri.startswith(REMOTE_IDENTITIES_FILE_URL): body = remote_identities http_requests.append(last_request) return (200, headers, body) httpretty.register_uri(httpretty.GET, REMOTE_IDENTITIES_FILE_URL, responses=[ httpretty.Response(body=request_callback) ]) return http_requests
def setup_http_server(): eclipse_projects = read_file(ECLIPSE_PROJECTS_FILE) http_requests = [] def request_callback(method, uri, headers): last_request = httpretty.last_request() if uri.startswith(ECLIPSE_PROJECTS_URL): body = eclipse_projects http_requests.append(last_request) return (200, headers, body) httpretty.register_uri(httpretty.GET, ECLIPSE_PROJECTS_URL, responses=[ httpretty.Response(body=request_callback) ]) return http_requests
def it_should_post_the_correct_body(self): expect(httpretty.last_request()).to(have_json([ { "eventId": "foo", "eventType": "settings", "data": { "$userStreamAcl": { "$r": ["bob"], "$w": [], "$d": [], "$mr": [], "$mw": [] }, "$systemStreamAcl": { "$r": ["$admins"], "$w": ["$admins"], "$d": ["$admins"], "$mr":[ "$admins"], "$mw": ["$admins"] } } }]))
def it_should_post_the_correct_body(self): expect(httpretty.last_request()).to(have_json([ { "eventId": "foo", "eventType": "settings", "data": { "$userStreamAcl": { "$r": ["$all"], "$w": ["$all"], "$d": ["$all"], "$mr": ["$all"], "$mw": ["$all"] }, "$systemStreamAcl": { "$r": ["fred"], "$w": ["fred"], "$d": [], "$mr":[], "$mw": [] } } }]))
def it_should_post_the_correct_body(self): expect(httpretty.last_request()).to(have_json([ { "eventId": "foo", "eventType": "settings", "data": { "$userStreamAcl": { "$r": ["$all", "giddy", "kevin"], "$w": ["ouro"], "$d": ["ouro"], "$mr": ["ouro"], "$mw": ["ouro"] }, "$systemStreamAcl": { "$r": ["$admins", "ouro"], "$w": ["$admins"], "$d": ["$admins"], "$mr":["$admins"], "$mw": ["$admins"] } } }]))
def it_should_post_the_correct_body(self): expect(httpretty.last_request()).to(have_json([ { "eventId": "foo", "eventType": "settings", "data": { "$userStreamAcl": { "$r": ["$all"], "$w": [], "$d": ["ouro"], "$mr": ["ouro"], "$mw": ["ouro"] }, "$systemStreamAcl": { "$r": ["$admins", "ouro"], "$w": ["$admins"], "$d": ["$admins"], "$mr":["$admins"], "$mw": ["$admins"] } } }]))
def it_should_post_the_correct_body(self): expect(httpretty.last_request()).to(have_json([ { "eventId": "foo", "eventType": "settings", "data": { "$userStreamAcl": { "$r": ["$all"], "$w": ["ouro"], "$d": ["ouro"], "$mr": ["ouro"], "$mw": ["ouro"] }, "$systemStreamAcl": { "$r": ["$admins", "ouro"], "$w": ["$admins", "giddy", "kevin"], "$d": ["$admins"], "$mr":["$admins"], "$mw": ["$admins"] } } }]))
def test_get_limited_result(self): """ Make sure fetching by dict type query works """ json_body = json.dumps({'result': [{'number': self.mock_incident['number']}]}) httpretty.register_uri(httpretty.GET, "http://%s/%s" % (self.mock_connection['host'], self.mock_incident['path']), body=json_body, status=200, content_type="application/json") r = self.client.query(table='incident', query={}) # Trigger a request by fetching next element from the generator next(r.get_all(limit=2)) # Get last request QS qs = httpretty.last_request().querystring # Make sure sysparm_limit equals limit self.assertEqual(int(qs['sysparm_limit'][0]), 2) # Make sure sysparm_suppress_pagination_header is True self.assertTrue(qs['sysparm_suppress_pagination_header'])
def test_get_multiple_with_offset(self): """ Make sure offset works properly """ json_body = json.dumps({'result': [{'sys_id': self.mock_incident['sys_id'], 'this': 'that'}, {'sys_id': self.mock_incident['sys_id'], 'this': 'that'}]}) httpretty.register_uri(httpretty.GET, "http://%s/%s" % (self.mock_connection['host'], self.mock_incident['path']), body=json_body, status=200, content_type="application/json") r = self.client.query(table='incident', query={'number': self.mock_incident['number']}) list(r.get_multiple(limit=100, offset=50)) qs = httpretty.last_request().querystring # Make sure sysparm_offset is set to the expected value self.assertEqual(int(qs['sysparm_offset'][0]), 50) # Make sure sysparm_limit is set to the expected value self.assertEqual(int(qs['sysparm_limit'][0]), 100)
def testCantDeleteOtherDomains(self): httpretty.enable() httpretty.register_uri(httpretty.DELETE, settings.NSLORD_PDNS_API + '/zones/' + self.otherDomains[1].name + '.') httpretty.register_uri(httpretty.DELETE, settings.NSMASTER_PDNS_API + '/zones/' + self.otherDomains[1].name+ '.') url = reverse('domain-detail', args=(self.otherDomains[1].pk,)) response = self.client.delete(url) self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT) self.assertTrue(isinstance(httpretty.last_request(), httpretty.core.HTTPrettyRequestEmpty)) self.assertTrue(Domain.objects.filter(pk=self.otherDomains[1].pk).exists())
def testCantDeleteOtherDynDomains(self): httpretty.enable() httpretty.register_uri(httpretty.DELETE, settings.NSLORD_PDNS_API + '/zones/' + self.otherDomains[1].name + '.') httpretty.register_uri(httpretty.DELETE, settings.NSMASTER_PDNS_API + '/zones/' + self.otherDomains[1].name+ '.') url = reverse('domain-detail', args=(self.otherDomains[1].pk,)) response = self.client.delete(url) self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT) self.assertTrue(isinstance(httpretty.last_request(), httpretty.core.HTTPrettyRequestEmpty)) self.assertTrue(Domain.objects.filter(pk=self.otherDomains[1].pk).exists())
def testPostCausesPdnsAPICall(self): httpretty.enable() httpretty.register_uri(httpretty.PATCH, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.') httpretty.register_uri(httpretty.PUT, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + './notify') url = reverse('rrsets', args=(self.ownedDomains[1].name,)) data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'} self.client.post(url, json.dumps(data), content_type='application/json') result = json.loads(httpretty.httpretty.latest_requests[-2].parsed_body) self.assertEqual(result['rrsets'][0]['name'], self.ownedDomains[1].name + '.') self.assertEqual(result['rrsets'][0]['records'][0]['content'], '1.2.3.4') self.assertEqual(httpretty.last_request().method, 'PUT')
def test_harvest_before_download_null_url_stops_gather_stage(self): plugin = p.get_plugin('test_rdf_harvester') source_url = 'http://return.none' # Mock the GET request to get the file httpretty.register_uri(httpretty.GET, source_url, body=self.rdf_content, content_type=self.rdf_content_type) # The harvester will try to do a HEAD request first so we need to mock # this as well httpretty.register_uri(httpretty.HEAD, source_url, status=405, content_type=self.rdf_content_type) harvest_source = self._create_harvest_source(source_url) self._create_harvest_job(harvest_source['id']) self._run_jobs(harvest_source['id']) self._gather_queue(1) eq_(plugin.calls['before_download'], 1) # Run the jobs to mark the previous one as Finished self._run_jobs() # Check that the file was not requested assert 'return.none' not in httpretty.last_request().headers['host'] # Get the harvest source with the udpated status harvest_source = h.call_action('harvest_source_show', id=harvest_source['id']) last_job_status = harvest_source['status']['last_job'] eq_(last_job_status['status'], 'Finished') eq_(last_job_status['stats']['added'], 0)
def test_harvest_before_download_errors_get_stored(self): plugin = p.get_plugin('test_rdf_harvester') source_url = 'http://return.errors' # Mock the GET request to get the file httpretty.register_uri(httpretty.GET, source_url, body=self.rdf_content, content_type=self.rdf_content_type) # The harvester will try to do a HEAD request first so we need to mock # this as well httpretty.register_uri(httpretty.HEAD, source_url, status=405, content_type=self.rdf_content_type) harvest_source = self._create_harvest_source(source_url) self._create_harvest_job(harvest_source['id']) self._run_jobs(harvest_source['id']) self._gather_queue(1) eq_(plugin.calls['before_download'], 1) # Run the jobs to mark the previous one as Finished self._run_jobs() # Check that the file was not requested assert 'return.errors' not in httpretty.last_request().headers['host'] # Get the harvest source with the udpated status harvest_source = h.call_action('harvest_source_show', id=harvest_source['id']) last_job_status = harvest_source['status']['last_job'] eq_('Error 1', last_job_status['gather_error_summary'][0][0]) eq_('Error 2', last_job_status['gather_error_summary'][1][0])
def test_harvest_after_download_empty_content_stops_gather_stage(self): plugin = p.get_plugin('test_rdf_harvester') source_url = 'http://return.empty.content' # Mock the GET request to get the file httpretty.register_uri(httpretty.GET, source_url, body='return.empty.content', content_type=self.rdf_content_type) # The harvester will try to do a HEAD request first so we need to mock # this as well httpretty.register_uri(httpretty.HEAD, source_url, status=405, content_type=self.rdf_content_type) harvest_source = self._create_harvest_source(source_url) self._create_harvest_job(harvest_source['id']) self._run_jobs(harvest_source['id']) self._gather_queue(1) eq_(plugin.calls['after_download'], 1) # Run the jobs to mark the previous one as Finished self._run_jobs() # Check that the file was requested assert ('return.empty.content' in httpretty.last_request().headers['host']) # Get the harvest source with the udpated status harvest_source = h.call_action('harvest_source_show', id=harvest_source['id']) last_job_status = harvest_source['status']['last_job'] eq_(last_job_status['status'], 'Finished') eq_(last_job_status['stats']['added'], 0)
def test_harvest_after_download_errors_get_stored(self): plugin = p.get_plugin('test_rdf_harvester') source_url = 'http://return.content.errors' # Mock the GET request to get the file httpretty.register_uri(httpretty.GET, source_url, body='return.errors', content_type=self.rdf_content_type) # The harvester will try to do a HEAD request first so we need to mock # this as well httpretty.register_uri(httpretty.HEAD, source_url, status=405, content_type=self.rdf_content_type) harvest_source = self._create_harvest_source(source_url) self._create_harvest_job(harvest_source['id']) self._run_jobs(harvest_source['id']) self._gather_queue(1) eq_(plugin.calls['after_download'], 1) # Run the jobs to mark the previous one as Finished self._run_jobs() # Check that the file was requested assert ('return.content.errors' in httpretty.last_request().headers['host']) # Get the harvest source with the udpated status harvest_source = h.call_action('harvest_source_show', id=harvest_source['id']) last_job_status = harvest_source['status']['last_job'] eq_('Error 1', last_job_status['gather_error_summary'][0][0]) eq_('Error 2', last_job_status['gather_error_summary'][1][0])
def test_returns_historical_rates(self): expected_response = {'base': 'EUR', 'date': '2000-01-03', 'rates': {'GBP': 0.6246}} httpretty.register_uri(httpretty.GET, self.url, body=json.dumps(expected_response), content_type='application/json') client = Fixerio() response = client.historical_rates(date=self.date.isoformat()) self.assertDictEqual(response, expected_response) request = httpretty.last_request() self.assertEqual(request.method, 'GET') self.assertEqual(request.path, self.path) self.assertEqual(request.querystring, {}) self.assertEqual(request.body, b'') client = Fixerio() response = client.historical_rates(date=self.date) self.assertDictEqual(response, expected_response) request = httpretty.last_request() self.assertEqual(request.method, 'GET') self.assertEqual(request.path, self.path) self.assertEqual(request.querystring, {}) self.assertEqual(request.body, b'')
def test_coreRespond(): httpretty.register_uri(httpretty.POST, 'https://localhost/callback', body="") bc3cb.respond.respond("I'm trying it! ????", 'https://localhost/callback') # The response will be encoded when we get it back from httpretty. assert httpretty.last_request().body.decode('utf-8') == "content=I'm trying it! ????"
def test_about_from_index(self): """Test traveling from index to about page""" link = 'http://startupfairy.com/' httpretty.register_uri(httpretty.GET, link) response = requests.get(link) self.assertEqual(200, response.status_code) link = 'http://startupfairy.com/about' httpretty.register_uri(httpretty.GET, link, body='[{"title": "About | Startup Fairy"}]', content_type="application/json") response = requests.get(link) self.assertEqual(200, response.status_code) #self.assertEqual('', httpretty.last_request().body)
def it_should_post_metadata(self): expect(httpretty.last_request()).to(have_json([{ "eventId": "foo", "eventType": "settings" }]))
def it_should_post_metadata(self): expect(httpretty.last_request()).to(have_json([{ "eventId": "foo", "eventType": "settings", "data": { "$acl": { "$r": ['devs', 'ops', '$admins'], "$w": ['ops', '$admins'], "$d": ['$admins'], "$mr": ['$all'], "$mw": ['$admins'] } } }]))
def it_should_post_the_correct_acl(self): expect(httpretty.last_request()).to(have_json([{ "eventId": "foo", "eventType": "settings", "data": { "$acl": { "$r": ["devs", "ops", "qa"], "$w": ["ops"] } } }]))
def it_should_post_the_correct_body(self): expect(httpretty.last_request()).to(have_json([{ "eventId": "foo", "eventType": "settings", "data": { "$acl": { "$w": [], "$r": ["greg"] } } }]))
def it_should_post_the_correct_body(self): expect(httpretty.last_request()).to(have_json([{ "eventId": "foo", "eventType": "settings", "data": { "$acl": { "$w": ['fred', 'greg'], "$r": ['fred', "greg", 'john'], "$d": ["$admins", 'fred'], "$mw": ["$admins"], "$mr": ["$admins", 'fred'] } } }]))
def it_should_delete_the_user(self): expect(httpretty.last_request()).to( have_posted_to("/users/foo", method=httpretty.DELETE))
def it_should_have_the_correct_request_body(self): expect(httpretty.last_request()).to( have_json({ "loginName": "bob", "fullName": "bob", "password": "password", "groups": [] }))
def it_should_have_the_correct_request_body(self): expect(httpretty.last_request()).to( have_json({ "loginName": "foo", "fullName": "Ramrod McTavish", "password": "bar", "groups": ["devs"]}))
