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'])
python类last_request()的实例源码
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"]}))
