from abc import ABC, abstractmethod
[docs]
class AWrongStatus(AssertionError, ABC):
def __init__(self, expected_status: str, actual_status: str):
[docs]
self.expected_status = expected_status
[docs]
self.actual_status = actual_status
super().__init__(self.get_message())
@abstractmethod
[docs]
def get_message(self) -> str:
pass
def __call__(self):
if self.expected_status != self.actual_status:
raise self
[docs]
def is_raised(self) -> bool:
return self.expected_status != self.actual_status
[docs]
class ShouldHaveSucceeded(AWrongStatus):
[docs]
def get_message(self):
return f"Should have succeeded ({self.expected_status}) but failed ({self.actual_status})"
[docs]
class ShouldHaveFailed(AWrongStatus):
[docs]
def get_message(self):
return f"Should have failed ({self.expected_status}) but succeeded ({self.actual_status})"
[docs]
class WrongErrorCode(AWrongStatus):
[docs]
def get_message(self):
return f"{self.actual_status} instead of {self.expected_status}"
[docs]
class WrongSuccessCode(AWrongStatus):
[docs]
def get_message(self):
return f"{self.actual_status} instead of {self.expected_status}"
[docs]
class ServerError(AWrongStatus):
[docs]
def get_message(self):
return f"{self.actual_status} instead of {self.expected_status}"
[docs]
class ASchemaError(AssertionError, ABC):
def __init__(self, request, schema: str = None):
[docs]
self.endpoint = request.original_path_url
[docs]
self.verb = request.response.request.method
[docs]
self.status_code = request.response.status_code
super().__init__(self.get_message())
@abstractmethod
[docs]
def get_message(self) -> str:
pass
[docs]
class DefaultSchemaIsUsed(ASchemaError):
[docs]
def get_message(self):
return f"For {self.verb} - {self.endpoint} - {self.status_code}, default schema should not be used"
[docs]
class MissingEndpoint(ASchemaError):
[docs]
def get_message(self):
return f"For {self.verb} - {self.endpoint} - {self.status_code}, missing in {self.api.swagger_file}"
[docs]
class MissingSchema(ASchemaError):
[docs]
def get_message(self):
return f"For {self.verb} - {self.endpoint} - {self.status_code} - {self.status_code}, " +\
"missing in {self.api.swagger_file}"
[docs]
class UnrespectedSchema(ASchemaError):
def __init__(self, request, schema: str, errors: list[str]):
super().__init__(request, schema)
[docs]
def get_message(self):
return f"For {self.verb} - {self.endpoint} - {self.status_code}, schema {self.schema} is not respected:\n\t" +\
'\n\t'.join(self.errors)
[docs]
class InvalidFieldInSchema(ASchemaError):
def __init__(self, request, schema: str, fields: list[str]):
super().__init__(request, schema)
[docs]
def get_message(self):
return f"For {self.verb} - {self.endpoint} - {self.status_code}, schema {self.schema} invalid fields:\n" +\
'\n\t'.join(self.fields)
[docs]
class SchemaError(ASchemaError):
def __init__(self, request, schema: str, errors: list[str]):
super().__init__(request, schema)
[docs]
def get_message(self):
return f"For {self.verb} - {self.endpoint} - {self.status_code}, error(s) in schema {self.schema}:\n" +\
'\n\t'.join(self.errors)
[docs]
class ForbiddenKeyword(ASchemaError):
def __init__(self, request, schema: str, errors: list[str]):
super().__init__(request, schema)
[docs]
def get_message(self):
return f"For {self.verb} - {self.endpoint} - {self.status_code}, forbidden keyword:\n" +\
'\n\t'.join(self.errors)
# Checks
[docs]
FORBIDDEN_KEYWORDS = ["foreign key", "q_"]
[docs]
def get_status_exception(expected_status, actual_status) -> AWrongStatus:
assert len(actual_status) == 3
assert len(expected_status) == 3
if expected_status.startswith('2'):
if actual_status.startswith('2'):
return WrongSuccessCode(expected_status, actual_status)
elif actual_status.startswith('4') or actual_status.startswith('5'):
return ShouldHaveSucceeded(expected_status, actual_status)
elif expected_status.startswith('4'):
if actual_status.startswith('2'):
return ShouldHaveFailed(expected_status, actual_status)
elif actual_status.startswith('5'):
return ServerError(expected_status, actual_status)
elif actual_status.startswith('4'):
return WrongErrorCode(expected_status, actual_status)
raise NotImplementedError(f"expected_status: {expected_status} actual_status: {actual_status}")
[docs]
def check_forbidden_keyword(my_obj) -> str:
if isinstance(my_obj, dict):
for key, value in my_obj.items():
if key == "token":
continue
forbidden_value = check_forbidden_keyword(value)
if forbidden_value:
return forbidden_value
elif isinstance(my_obj, list):
for value in my_obj:
forbidden_value = check_forbidden_keyword(value)
if forbidden_value:
return forbidden_value
elif isinstance(my_obj, str):
for forbidden_keyword in FORBIDDEN_KEYWORDS:
if forbidden_keyword in my_obj:
return forbidden_keyword
return ""