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
|
from __future__ import annotations
import sys
import asyncio
import selectors
from typing import Any
import pytest
pytest_plugins = (
"tests.fix_db",
"tests.fix_pq",
"tests.fix_dns",
"tests.fix_mypy",
"tests.fix_faker",
"tests.fix_proxy",
"tests.fix_psycopg",
"tests.fix_crdb",
"tests.fix_gc",
"tests.pool.fix_pool",
)
collect_ignore: list[str] = []
if sys.version_info[:2] < (3, 14):
collect_ignore.append("test_tstring.py")
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",
"gevent: the test requires the gevent module to be installed",
"dns: the test requires dnspython to run",
"postgis: the test requires the PostGIS extension to run",
"numpy: the test requires numpy module to be installed",
]
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 = [f"default selector: {selectors.DefaultSelector.__name__}"]
if (loop := config.getoption("--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.
if (cache := session.config.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)
asyncio_options: dict[str, Any] = {}
if sys.platform == "win32":
asyncio_options["loop_factory"] = lambda: asyncio.SelectorEventLoop(
selectors.SelectSelector()
)
@pytest.fixture(
params=[pytest.param(("asyncio", asyncio_options.copy()), id="asyncio")],
scope="session",
)
def anyio_backend(request):
backend, options = request.param
if request.config.option.loop == "uvloop":
options["use_uvloop"] = True
return backend, options
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)
|
