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_server_http_error_string_list(mock_requests_send):
'[Requests] - Test if HTTP error that a JSON error string list is handled'
res = requests.Response()
res.status_code = 500
res.url = test_url
res.reason = 'Some Error'
res.headers = {'Content-Type': 'application/json'}
res.raw = io.BytesIO(b'{"errors": ["a", "b"]}')
err = requests.exceptions.HTTPError(response=res)
configure_mock_requests_send(mock_requests_send, err)
endpoint = RequestsEndpoint(test_url)
data = endpoint(graphql_query)
expected_data = {'errors': [{'message': 'a'}, {'message': 'b'}]}
expected_data.update({
'exception': err,
'status': 500,
'headers': {'Content-Type': 'application/json'},
})
eq_(data, expected_data)
check_mock_requests_send(mock_requests_send)
def test_operation_name(mock_requests_send):
'[Requests] - Test if operation name is passed to server'
configure_mock_requests_send(mock_requests_send, graphql_response_ok)
operation_name = 'xpto'
endpoint = RequestsEndpoint(test_url)
data = endpoint(graphql_query, operation_name=operation_name)
eq_(data, json.loads(graphql_response_ok))
check_mock_requests_send(mock_requests_send, operation_name=operation_name)
def test_server_http_error_list_message(mock_requests_send):
'[Requests] - Test HTTP JSON error with messages being a list'
res = requests.Response()
res.status_code = 500
res.url = test_url
res.reason = 'Some Error'
res.headers = {'Content-Type': 'application/json'}
res.raw = io.BytesIO(b'{"errors": [{"message": [1, 2]}]}')
err = requests.exceptions.HTTPError(response=res)
configure_mock_requests_send(mock_requests_send, err)
endpoint = RequestsEndpoint(test_url)
data = endpoint(graphql_query)
expected_data = {'errors': [{'message': '[1, 2]'}]}
expected_data.update({
'exception': err,
'status': 500,
'headers': {'Content-Type': 'application/json'},
})
eq_(data, expected_data)
check_mock_requests_send(mock_requests_send)
def test_server_http_non_conforming_json(mock_requests_send):
'[Requests] - Test HTTP error that is NOT conforming to GraphQL payload'
res = requests.Response()
res.status_code = 500
res.url = test_url
res.reason = 'Some Error'
res.headers = {'Content-Type': 'application/json'}
res.raw = io.BytesIO(b'{"message": "xpto"}')
err = requests.exceptions.HTTPError(response=res)
configure_mock_requests_send(mock_requests_send, err)
endpoint = RequestsEndpoint(test_url)
data = endpoint(graphql_query)
eq_(data, {
'errors': [{
'message': str(err),
'exception': err,
'status': 500,
'headers': {'Content-Type': 'application/json'},
'body': '{"message": "xpto"}',
}],
'data': None,
})
check_mock_requests_send(mock_requests_send)
def test_server_reported_error(mock_requests_send):
'[Requests] - Test if GraphQL errors reported with HTTP 200 is handled'
configure_mock_requests_send(mock_requests_send, graphql_response_error)
endpoint = RequestsEndpoint(test_url)
data = endpoint(graphql_query)
eq_(data, json.loads(graphql_response_error))
check_mock_requests_send(mock_requests_send)
def test_json_error(mock_requests_send):
'[Requests] - Test if broken server responses (invalid JSON) is handled'
configure_mock_requests_send(mock_requests_send,
graphql_response_json_error)
endpoint = RequestsEndpoint(test_url)
data = endpoint(graphql_query)
exc = get_json_exception(graphql_response_json_error)
got_exc = data['errors'][0].pop('exception')
assert isinstance(got_exc, json.JSONDecodeError), \
'{} is not json.JSONDecodeError'.format(type(got_exc))
eq_(data, {
'errors': [{
'message': str(exc),
'body': graphql_response_json_error,
}],
'data': None,
})
check_mock_requests_send(mock_requests_send)
def test_basic_bytes_query(mock_requests_send):
'[Requests] - Test if query with type bytes works'
configure_mock_requests_send(mock_requests_send, graphql_response_ok)
endpoint = RequestsEndpoint(test_url)
data = endpoint(graphql_query.encode('utf-8'))
eq_(data, json.loads(graphql_response_ok))
check_mock_requests_send(mock_requests_send)
def test_server_http_error(mock_requests_send):
'[Requests] - Test if HTTP error without JSON payload is handled'
res = requests.Response()
res.status_code = 500
res.url = test_url
res.reason = 'Some Error'
res.headers = {'Xpto': 'abc'}
res.raw = io.BytesIO(b'xpto')
err = requests.exceptions.HTTPError(response=res)
configure_mock_requests_send(mock_requests_send, err)
endpoint = RequestsEndpoint(test_url)
data = endpoint(graphql_query)
eq_(data, {
'errors': [{
'message': str(err),
'exception': err,
'status': 500,
'headers': {'Xpto': 'abc'},
'body': 'xpto',
}],
'data': None,
})
check_mock_requests_send(mock_requests_send)
def test_basic_auth(mock_requests_send):
'[Requests] - Test if basic usage with only auth works'
configure_mock_requests_send(mock_requests_send, graphql_response_ok)
endpoint = RequestsEndpoint(test_url,
auth=requests.auth.HTTPBasicAuth("user",
"password"))
data = endpoint(graphql_query)
eq_(data, json.loads(graphql_response_ok))
check_mock_requests_send(mock_requests_send)
eq_(str(endpoint),
'RequestsEndpoint('
+ 'url={}, '.format(test_url)
+ 'base_headers={}, timeout=None, method=POST, '
+ 'auth=)')
def test_get(mock_requests_send):
'[Requests] - Test if HTTP method GET request works'
configure_mock_requests_send(mock_requests_send, graphql_response_ok)
base_headers = {
'Xpto': 'abc',
}
extra_headers = {
'Extra': '123',
'Accept': extra_accept_header,
}
variables = {'repoOwner': 'owner', 'repoName': 'name'}
operation_name = 'xpto'
endpoint = RequestsEndpoint(test_url,
base_headers=base_headers,
method='GET')
data = endpoint(graphql_query,
extra_headers=extra_headers,
variables=variables,
operation_name=operation_name,
)
eq_(data, json.loads(graphql_response_ok))
check_mock_requests_send(mock_requests_send,
method='GET',
base_headers=base_headers,
extra_headers=extra_headers,
variables=variables,
operation_name=operation_name,
)
eq_(str(endpoint),