1from __future__ import annotations
2
3import contextlib
4import gc
5import os
6import threading
7from functools import partial
8
9import pytest
10
11from .utils import ExpectEvent, Helper, P, StartWatching, TestEventQueue
12
13
14@pytest.fixture
15def p(tmpdir, *args):
16    """
17    Convenience function to join the temporary directory path
18    with the provided arguments.
19    """
20    return partial(os.path.join, tmpdir)
21
22
23@pytest.fixture(autouse=True)
24def _no_thread_leaks():
25    """
26    Fail on thread leak.
27    We do not use pytest-threadleak because it is not reliable.
28    """
29    old_thread_count = threading.active_count()
30    yield
31    gc.collect()  # Clear the stuff from other function-level fixtures
32    assert threading.active_count() == old_thread_count  # Only previously existing threads
33
34
35@pytest.fixture(autouse=True)
36def _no_warnings(recwarn):
37    """Fail on warning."""
38
39    yield
40
41    warnings = []
42    for warning in recwarn:  # pragma: no cover
43        message = str(warning.message)
44        filename = warning.filename
45        if (
46            "Not importing directory" in message
47            or "Using or importing the ABCs" in message
48            or "dns.hash module will be removed in future versions" in message
49            or "is still running" in message
50            or "eventlet" in filename
51        ):
52            continue
53        warnings.append(f"{warning.filename}:{warning.lineno} {warning.message}")
54    assert not warnings, warnings
55
56
57@pytest.fixture(name="helper")
58def helper_fixture(tmpdir):
59    with contextlib.closing(Helper(tmp=os.fspath(tmpdir))) as helper:
60        yield helper
61
62
63@pytest.fixture(name="p")
64def p_fixture(helper: Helper) -> P:
65    return helper.joinpath
66
67
68@pytest.fixture(name="event_queue")
69def event_queue_fixture(helper: Helper) -> TestEventQueue:
70    return helper.event_queue
71
72
73@pytest.fixture(name="start_watching")
74def start_watching_fixture(helper: Helper) -> StartWatching:
75    return helper.start_watching
76
77
78@pytest.fixture(name="expect_event")
79def expect_event_fixture(helper: Helper) -> ExpectEvent:
80    return helper.expect_event
81