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_fileuploadsearch_valid(self):
global active_mock_response
active_mock_response = 200, '{"status": "OK", "method": "uploadsearch", "limit": 30, "result": [{"im_name": "19609426340"}, {"im_name": "40623053246"}, {"im_name": "41229714662"}, {"im_name": "19356394967"}, {"im_name": "39882808162"}, {"im_name": "25301792601"}, {"im_name": "38213377507"}, {"im_name": "42894441677"}, {"im_name": "38475333633"}, {"im_name": "39012157235"}], "error": [], "total": 10, "page": 1}'
httpretty.register_uri(httpretty.POST, self.uploadsearch_entpoint, body=request_callback)
image_path = 'tests/fixtures/upload.jpg'
resp = self.api.uploadsearch(image_path=image_path)
self.assertEqual(request_callback.request.method, 'POST')
if (sys.version_info > (3, 0)):
content_type = dict(request_callback.request.headers._headers)['Content-Type'].split(';')[0].strip()
else:
content_type = request_callback.request.headers.__dict__['type']
self.assertEqual(content_type, 'multipart/form-data')
self.assertEqual(resp['status'], 'OK')
offset = uri.split("offset=")[1].split("&")[0]
if offset == "0":
body = mozilla_events_1
else:
body = mozilla_events_2
HTTPServer.requests_http.append(httpretty.last_request())
return (200, headers, body)
httpretty.register_uri(httpretty.GET,
MOZILLA_REPS_API_EVENTS,
responses=[
httpretty.Response(body=request_callback)
])
httpretty.register_uri(httpretty.GET,
MOZILLA_REPS_API_USERS,
responses=[
httpretty.Response(body=request_callback)
])
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_check_alive(self):
httpretty.enable()
httpretty.register_uri(httpretty.GET, SERVICE_URL, body="LIVE")
self.assertTrue(check_alive('SocketPortMonitor', SERVICE_HOST, port='8888'))
httpretty.disable()
httpretty.reset()
def test_iss_tle_data(self):
httpretty.register_uri(httpretty.POST,
'https://www.space-track.org/ajaxauth/login',
body="")
url = 'https://www.space-track.org/basicspacedata/query/class/' \
'tle_latest/NORAD_CAT_ID/25544/ORDINAL/1/'
httpretty.register_uri(httpretty.GET, url,
body=fixture('spacetrack_tle.json'))
username = os.getenv('SPACETRACK_USERNAME', '')
password = os.getenv('SPACETRACK_PASSWORD', '')
tle_getter = tle_data.TLE_getter(username, password)
iss_tle_data = tle_getter.get_data("25544")
self.assertEqual(iss_tle_data.tle_line0, "ISS (ZARYA)")
def test_issue_comments(self):
"""Test issue comments API call"""
issue_comments = read_file('data/github/github_issue_comments_2')
rate_limit = read_file('data/github/rate_limit')
httpretty.register_uri(httpretty.GET,
GITHUB_RATE_LIMIT,
body=rate_limit,
status=200,
forcing_headers={
'X-RateLimit-Remaining': '20',
'X-RateLimit-Reset': '15'
})
httpretty.register_uri(httpretty.GET,
GITHUB_ISSUE_2_COMMENTS_URL,
body=issue_comments,
status=200,
forcing_headers={
'X-RateLimit-Remaining': '20',
'X-RateLimit-Reset': '15'
})
client = GitHubClient("zhquan_example", "repo", "aaa")
pull_requested_reviewers_raw = [rev for rev in client.issue_comments(2)]
self.assertEqual(pull_requested_reviewers_raw[0], issue_comments)
def test_get_author_id():
"""Test that verifies that an author's id is properly returned."""
client = goodreads.Client(client_id="123abc")
base_url = "https://www.goodreads.com/"
api_call = "api/author_url/Chuck+Palahniuk?key=123abc"
url = base_url + api_call
sample_response = open('test/fixtures/get_author_id_response.xml')
body = sample_response.read()
httpretty.register_uri(httpretty.GET, url, body=body, status=200)
author_id = client.get_author_id('Chuck Palahniuk')
eq_('2546', author_id)
"result": [
{"sys_id": self.mock_incident["sys_id"]},
{"sys_id": self.mock_incident["sys_id"]},
]
}
)
httpretty.register_uri(
httpretty.GET,
"http://%s/%s" % (self.mock_connection["host"], self.mock_incident["path"]),
body=json_body_get,
status=200,
content_type="application/json",
)
json_body_delete = json.dumps({"success": True})
httpretty.register_uri(
httpretty.DELETE,
"http://%s/%s/%s"
% (
self.mock_connection["host"],
self.mock_incident["path"],
self.mock_incident["sys_id"],
),
body=json_body_delete,
status=204,
content_type="application/json",
)
r = self.client.query(
table="incident", query={"number": self.mock_incident["number"]}
)
self.assertRaises(MultipleResults, r.delete)
def test_associations(self):
httpretty.register_uri(httpretty.GET,
self.base_href + "/tenants/current",
location=self.tenant_href,
status=302)
httpretty.register_uri(httpretty.GET,
self.tenant_href,
body=json.dumps(self.tenant_body),
content_type="application/json")
httpretty.register_uri(httpretty.GET,
self.dir_href,
body=json.dumps(self.dir_body),
content_type="application/json")
httpretty.register_uri(httpretty.GET,
self.acc_href,
body=json.dumps(self.acc_body),
content_type="application/json")
account = self.client.accounts.get(self.acc_href)