from openapi_core.contrib.requests import RequestsOpenAPIResponse, RequestsOpenAPIRequest
from openapi_core.validation.response.exceptions import InvalidData, DataValidationError
from behave import then
from .utils import check_expected_values_in_response
from .utils.custom_exceptions import get_status_exception, check_forbidden_keyword, \
MissingSchema, DefaultSchemaIsUsed, ForbiddenKeyword, SchemaError, MissingEndpoint
@then('the response status code should equal {expected_http_status_code}')
[docs]
def status_code_validation(context, expected_http_status_code):
"""Validate that the response has the expected HTTP status code.
:param context: The Behave context
:param expected_http_status_code: Expected HTTP status code as string
:raises MissingEndpoint: If 404 wasn't expected but was received with the standard message
"""
context.request.expected_status_code = expected_http_status_code
if expected_http_status_code != "404" \
and "The requested URL was not found on the server." in context.request.response.text:
raise MissingEndpoint(context.request, None)
context.request.api.check_status(
context.request,
get_status_exception(expected_http_status_code, str(context.request.response.status_code)))
@then('the response should contain')
[docs]
def check_response_from_table(context):
"""Check that the response contains the expected key-value pairs from a table.
Expects a Gherkin table with 'key' and 'value' columns, and validates each
key-value pair against the response JSON.
:param context: The Behave context containing a table of keys and values
"""
try:
for row in context.table:
key = row['key'].strip()
value = row['value'].strip()
result, expected_value, response_value = check_expected_values_in_response(context, key, value)
context.request.api.check_extra(
context.request,
result,
f'In response body, the field "{key}" is supposed to be "{expected_value}" but is "{response_value}"')
except (TypeError, KeyError) as e:
context.request.api.check_extra(
context.request,
False,
f'In response body, the field "{key}" is supposed to be "{value}" but got {e}')
@then('the response should be empty')
[docs]
def check_empty_response(context):
"""Check that the response body is empty (empty JSON object or array).
:param context: The Behave context
"""
response = context.request.response.json()
context.request.api.check_extra(
context.request,
not response,
f'In response body, response is supposed to be empty but is "{response}"')
@then('the response should not be empty')
[docs]
def check_not_empty_response(context):
"""Check that the response body is not empty (has content).
:param context: The Behave context
"""
response = context.request.response.json()
context.request.api.check_extra(
context.request,
response,
'In response body, response is not supposed to be empty')
@then('the response json matches defined schema')
[docs]
def response_match_defined_schema(context):
"""Validate that the response JSON matches the defined schema.
Performs schema validation against the defined schema for the current endpoint,
checking for various schema-related issues including default schemas, forbidden
keywords, and data validation errors.
:param context: The Behave context
"""
exceptions = []
schema = "Not found"
try:
schema = context.request.api.get_schema_from_request(context.request)
if "Default" in schema or "default" in schema:
exceptions.append(DefaultSchemaIsUsed(context.request, schema))
forbidden_keyword = check_forbidden_keyword(context.request.response.json())
if forbidden_keyword:
exceptions.append(
ForbiddenKeyword(
context.request,
schema,
[forbidden_keyword]))
json_schema_exception = context.request.api.get_json_schema_exception(context.request)
if json_schema_exception:
exceptions.append(json_schema_exception)
try:
context.request.api.openapi.validate_response(
RequestsOpenAPIRequest(context.request.response.request),
RequestsOpenAPIResponse(context.request.response))
except InvalidData as e:
exceptions.append(context.request.api.get_invalid_data_exception(context.request, e))
except KeyError:
exceptions.append(MissingSchema(context.request, schema))
except DataValidationError as e:
exceptions.append(SchemaError(
context.request, schema,
[f"Cannot validate schema because of {e.__cause__}"]))
context.request.api.check_schema(
context.request,
exceptions)