/
utils.py
127 lines (88 loc) · 3.78 KB
/
utils.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
import cgi
import traceback
import warnings
from contextlib import contextmanager
from functools import wraps
from typing import Any, Callable, Dict, Generator, List, Set, Tuple, Type, Union
from urllib.parse import urlsplit, urlunsplit
import yaml
from hypothesis.reporting import with_reporter
from werkzeug.wrappers import Response as BaseResponse
from werkzeug.wrappers.json import JSONMixin
from .types import Filter, NotSet
NOT_SET = NotSet()
def deprecated(func: Callable, message: str) -> Callable:
"""Emit a warning if the given function is used."""
@wraps(func) # pragma: no mutate
def inner(*args: Any, **kwargs: Any) -> Any:
warnings.warn(message, DeprecationWarning)
return func(*args, **kwargs)
return inner
def is_schemathesis_test(func: Callable) -> bool:
"""Check whether test is parametrized with schemathesis."""
try:
return hasattr(func, "_schemathesis_test")
except Exception:
return False
def get_base_url(uri: str) -> str:
"""Remove the path part off the given uri."""
parts = urlsplit(uri)[:2] + ("", "", "")
return urlunsplit(parts)
def force_tuple(item: Filter) -> Union[List, Set, Tuple]:
if not isinstance(item, (list, set, tuple)):
return (item,)
return item
def dict_true_values(**kwargs: Any) -> Dict[str, Any]:
"""Create dict with given kwargs while skipping items where bool(value) evaluates to False."""
return {key: value for key, value in kwargs.items() if bool(value)}
def dict_not_none_values(**kwargs: Any) -> Dict[str, Any]:
return {key: value for key, value in kwargs.items() if value is not None}
IGNORED_PATTERNS = (
"Falsifying example: ",
"You can add @seed",
"Failed to reproduce exception. Expected:",
"Flaky example!",
"Traceback (most recent call last):",
)
@contextmanager
def capture_hypothesis_output() -> Generator[List[str], None, None]:
"""Capture all output of Hypothesis into a list of strings.
It allows us to have more granular control over Schemathesis output.
Usage::
@given(i=st.integers())
def test(i):
assert 0
with capture_hypothesis_output() as output:
test() # hypothesis test
# output == ["Falsifying example: test(i=0)"]
"""
output = []
def get_output(value: str) -> None:
# Drop messages that could be confusing in the Schemathesis context
if value.startswith(IGNORED_PATTERNS):
return
output.append(value)
# the following context manager is untyped
with with_reporter(get_output): # type: ignore
yield output
def format_exception(error: Exception) -> str:
return "".join(traceback.format_exception_only(type(error), error))
def parse_content_type(content_type: str) -> Tuple[str, str]:
"""Parse Content Type and return main type and subtype."""
content_type, _ = cgi.parse_header(content_type)
main_type, sub_type = content_type.split("/", 1)
return main_type.lower(), sub_type.lower()
def are_content_types_equal(source: str, target: str) -> bool:
"""Check if two content types are the same excluding options."""
return parse_content_type(source) == parse_content_type(target)
def make_loader(*tags_to_remove: str) -> Type[yaml.SafeLoader]:
"""Create a YAML loader, that doesn't parse specific tokens into Python objects."""
cls: Type[yaml.SafeLoader] = type("YAMLLoader", (yaml.SafeLoader,), {})
cls.yaml_implicit_resolvers = {
key: [(tag, regexp) for tag, regexp in mapping if tag not in tags_to_remove]
for key, mapping in cls.yaml_implicit_resolvers.copy().items()
}
return cls
StringDatesYAMLLoader = make_loader("tag:yaml.org,2002:timestamp")
class WSGIResponse(BaseResponse, JSONMixin): # pylint: disable=too-many-ancestors
pass