Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_cloudsearch_add_single_fields(self):
"""
Check that a simple add document sends the actual document to AWS.
"""
document = DocumentServiceConnection(
endpoint="doc-demo-userdomain.us-east-1.cloudsearch.amazonaws.com")
document.add("1234", {"id": "1234", "title": "Title 1",
"category": ["cat_a", "cat_b", "cat_c"]})
document.commit()
args = json.loads(HTTPretty.last_request.body)[0]
self.assertEqual(args['fields']['category'], ['cat_a', 'cat_b',
'cat_c'])
self.assertEqual(args['fields']['id'], '1234')
self.assertEqual(args['fields']['title'], 'Title 1')
content_type="application/json")
httpretty.register_uri(httpretty.GET,
self.dir_href,
body=json.dumps(self.dir_body),
content_type="application/json")
expansion = Expansion('accounts', 'groups')
expansion.add_property('applications', offset=0, limit=5)
directory = self.client.directories.get(self.dir_href, expansion)
directory.name
# replace the comma inside parentheses to split easier
path = re.sub('(applications\%28.*)(\%2C)(.*\%29)',
'\g<1>;\g<3>', HTTPretty.last_request.path)
params = path.split('expand=')[1].split('%2C')
self.assertTrue(
sorted(params) ==
sorted(['applications%28offset%3A0;limit%3A5%29', 'accounts', 'groups'])
or
sorted(params) ==
sorted(['applications%28limit%3A5;offset%3A0%29', 'accounts', 'groups'])
)
def test_cloudsearch_result_fields_single(self):
search = SearchConnection(endpoint=HOSTNAME)
search.search(q='Test', return_fields=['author'])
args = self.get_args(HTTPretty.last_request.raw_requestline)
self.assertEqual(args['return'], ['author'])
def test_dates_are_in_datetime_format(self):
assert 'StartDate="%s"' % TEST_EVENT_LIST_START.strftime(EXCHANGE_DATETIME_FORMAT) in HTTPretty.last_request.body.decode('utf-8')
assert 'EndDate="%s"' % TEST_EVENT_LIST_END.strftime(EXCHANGE_DATETIME_FORMAT) in HTTPretty.last_request.body.decode('utf-8')
def test_methods(self):
client = Hammock(self.BASE_URL)
for method in ['GET', 'POST', 'PUT', 'DELETE', 'PATCH']:
HTTPretty.register_uri(getattr(HTTPretty, method), self.URL)
request = getattr(client, method)
resp = request('sample', 'path', 'to', 'resource')
self.assertEqual(HTTPretty.last_request.method, method)
body=json.dumps(self.tenant_body),
content_type="application/json")
httpretty.register_uri(httpretty.GET,
self.map_href,
body=json.dumps(self.dir_body),
content_type="application/json")
httpretty.register_uri(httpretty.DELETE,
self.map_href,
body='', status=204)
acc_store = self.client.account_store_mappings.get(self.map_href)
acc_store.delete()
self.assertEqual(HTTPretty.last_request.method, 'DELETE')
self.assertEqual(HTTPretty.last_request.path, self.map_path)
def test_does_not_have_last_request_by_default():
'HTTPretty.last_request is a dummy object by default'
HTTPretty.reset()
expect(HTTPretty.last_request.headers).to.be.empty
expect(HTTPretty.last_request.body).to.be.empty
self.strategy.set_request_data(location_query)
def _login(strategy, user):
strategy.session_set('username', user.username)
redirect = do_complete(self.strategy, user=self.user, login=_login)
url = self.strategy.build_absolute_uri('/password')
expect(redirect.url).to.equal(url)
HTTPretty.register_uri(HTTPretty.GET, redirect.url, status=200,
body='foobar')
HTTPretty.register_uri(HTTPretty.POST, redirect.url, status=200)
password = 'foobar'
requests.get(url)
requests.post(url, data={'password': password})
data = parse_qs(HTTPretty.last_request.body)
expect(data['password']).to.equal(password)
self.strategy.session_set('password', data['password'])
redirect = do_complete(self.strategy, user=self.user, login=_login)
expect(self.strategy.session_get('username')).to.equal(
self.expected_username
)
expect(redirect.url).to.equal(self.login_redirect_url)
def test_can_inspect_last_request(now):
"HTTPretty.last_request is a mimetools.Message request from last match"
HTTPretty.register_uri(HTTPretty.POST, "http://api.github.com/",
body='{"repositories": ["HTTPretty", "lettuce"]}')
headers, body = httplib2.Http().request(
'http://api.github.com', 'POST',
body='{"username": "gabrielfalcao"}',
headers={
'content-type': 'text/json',
},
)
expect(HTTPretty.last_request.method).to.equal('POST')
expect(HTTPretty.last_request.body).to.equal(
b'{"username": "gabrielfalcao"}',
)
expect(HTTPretty.last_request.headers['content-type']).to.equal(
'text/json',
)
expect(body).to.equal(b'{"repositories": ["HTTPretty", "lettuce"]}')