-
Notifications
You must be signed in to change notification settings - Fork 1.2k
Expand file tree
/
Copy pathconftest.py
More file actions
141 lines (102 loc) · 3.67 KB
/
Copy pathconftest.py
File metadata and controls
141 lines (102 loc) · 3.67 KB
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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
"""Fixtures shared by the whole suite.
Everything here is hermetic: temporary directories only, no listening sockets,
no network. Tests that genuinely need an externally launched visdom must carry
``@pytest.mark.server`` so CI can deselect them.
"""
from unittest.mock import patch
import pytest
from visdom.data_model.json_store import JSONStore
from visdom.server.app import Application
from visdom.utils import shared_utils
from testutils.fakes import FakeHandler, FakeSocket, SpyStore
@pytest.fixture
def env_path(tmp_path):
"""Disposable environment directory."""
return str(tmp_path)
@pytest.fixture
def store(env_path):
"""JSONStore backed by a temporary directory."""
return JSONStore(env_path)
@pytest.fixture
def spy_store(env_path):
"""JSONStore that records which backend methods the server calls."""
return SpyStore(env_path)
@pytest.fixture
def app(env_path):
"""Application instance with temporary persistence and no listener.
The port is only recorded on the instance; nothing binds it, so this is
safe to construct in parallel with other tests.
"""
return Application(port=8097, env_path=env_path)
@pytest.fixture
def app_factory(env_path):
"""Build Applications sharing one ``env_path``, for reload assertions."""
def build(**kwargs):
kwargs.setdefault("port", 8097)
kwargs.setdefault("env_path", env_path)
return Application(**kwargs)
return build
@pytest.fixture
def fake_socket():
return FakeSocket()
@pytest.fixture
def handler(env_path):
"""Duck-typed handler with its own empty state and store."""
return FakeHandler(env_path=env_path)
@pytest.fixture
def app_handler(app):
"""Handler sharing a real Application's state, storage and subscribers."""
return FakeHandler(
state=app.state,
storage=app.storage,
subs=app.subs,
sources=app.sources,
env_path=app.env_path,
)
@pytest.fixture
def offline_client():
"""Visdom client that never opens a connection.
With ``send=False`` the client's ``_send`` returns the payload instead of
performing I/O, so plot methods can be asserted as pure functions.
"""
import visdom
return visdom.Visdom(send=False, use_incoming_socket=False)
@pytest.fixture
def capture_send(offline_client):
"""Run a client call and return the payload it would have transmitted.
Usage::
sent = capture_send(lambda v: v.line(Y=[1, 2, 3]))
assert sent["payload"]["data"][0]["type"] == "scatter"
"""
def run(call, win_exists=None):
sent = {}
def capture(msg, endpoint="events", **_):
sent["payload"] = msg
sent["endpoint"] = endpoint
return "win_capture"
patches = [patch.object(offline_client, "_send", side_effect=capture)]
if win_exists is not None:
patches.append(
patch.object(offline_client, "win_exists", return_value=win_exists)
)
for p in patches:
p.start()
try:
call(offline_client)
finally:
for p in reversed(patches):
p.stop()
return sent
return run
@pytest.fixture(autouse=True)
def reset_warn_once():
"""Isolate ``shared_utils.warn_once`` between tests.
It dedupes against a module-level set, so without this a warning raised by
an earlier test silently suppresses the same warning in a later one, and
assertions on it pass or fail depending on execution order.
"""
previous = set(shared_utils._seen_warnings)
shared_utils._seen_warnings.clear()
yield
shared_utils._seen_warnings.clear()
shared_utils._seen_warnings.update(previous)