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')
python类GET的实例源码
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):
# 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_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 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)