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
|
import sys
import asyncio
import selectors
from typing import List
pytest_plugins = (
"tests.fix_db",
"tests.fix_pq",
"tests.fix_mypy",
"tests.fix_faker",
"tests.fix_proxy",
"tests.fix_psycopg",
"tests.fix_crdb",
"tests.pool.fix_pool",
)
def pytest_configure(config):
markers = [
"slow: this test is kinda slow (skip with -m 'not slow')",
"flakey(reason): this test may fail unpredictably')",
# There are troubles on travis with these kind of tests and I cannot
# catch the exception for my life.
"subprocess: the test import psycopg after subprocess",
"timing: the test is timing based and can fail on cheese hardware",
"dns: the test requires dnspython to run",
"postgis: the test requires the PostGIS extension to run",
]
for marker in markers:
config.addinivalue_line("markers", marker)
def pytest_addoption(parser):
parser.addoption(
"--loop",
choices=["default", "uvloop"],
default="default",
help="The asyncio loop to use for async tests.",
)
def pytest_report_header(config):
rv = []
rv.append(f"default selector: {selectors.DefaultSelector.__name__}")
loop = config.getoption("--loop")
if loop != "default":
rv.append(f"asyncio loop: {loop}")
return rv
def pytest_sessionstart(session):
# Detect if there was a segfault in the previous run.
#
# In case of segfault, pytest doesn't get a chance to write failed tests
# in the cache. As a consequence, retries would find no test failed and
# assume that all tests passed in the previous run, making the whole test pass.
cache = session.config.cache
if cache.get("segfault", False):
session.warn(Warning("Previous run resulted in segfault! Not running any test"))
session.warn(Warning("(delete '.pytest_cache/v/segfault' to clear this state)"))
raise session.Failed
cache.set("segfault", True)
# Configure the async loop.
loop = session.config.getoption("--loop")
if loop == "uvloop":
import uvloop
uvloop.install()
else:
assert loop == "default"
if sys.platform == "win32":
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
allow_fail_messages: List[str] = []
def pytest_sessionfinish(session, exitstatus):
# Mark the test run successful (in the sense -weak- that we didn't segfault).
session.config.cache.set("segfault", False)
def pytest_terminal_summary(terminalreporter, exitstatus, config):
if allow_fail_messages:
terminalreporter.section("failed tests ignored")
for msg in allow_fail_messages:
terminalreporter.line(msg)
|