-
-
Notifications
You must be signed in to change notification settings - Fork 1.5k
/
test_multiplexer.py
142 lines (102 loc) · 3.43 KB
/
test_multiplexer.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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
from multiprocessing import Event
from os import environ, getpid
from typing import Any, Dict, Type, Union
from unittest.mock import Mock
import pytest
from sanic import Sanic
from sanic.worker.multiplexer import WorkerMultiplexer
from sanic.worker.state import WorkerState
@pytest.fixture
def monitor_publisher():
return Mock()
@pytest.fixture
def worker_state():
return {}
@pytest.fixture
def m(monitor_publisher, worker_state):
environ["SANIC_WORKER_NAME"] = "Test"
worker_state["Test"] = {}
yield WorkerMultiplexer(monitor_publisher, worker_state)
del environ["SANIC_WORKER_NAME"]
def test_has_multiplexer_default(app: Sanic):
event = Event()
@app.main_process_start
async def setup(app, _):
app.shared_ctx.event = event
@app.after_server_start
def stop(app):
if hasattr(app, "m") and isinstance(app.m, WorkerMultiplexer):
app.shared_ctx.event.set()
app.stop()
app.run()
assert event.is_set()
def test_not_have_multiplexer_single(app: Sanic):
event = Event()
@app.main_process_start
async def setup(app, _):
app.shared_ctx.event = event
@app.after_server_start
def stop(app):
if hasattr(app, "m") and isinstance(app.m, WorkerMultiplexer):
app.shared_ctx.event.set()
app.stop()
app.run(single_process=True)
assert not event.is_set()
def test_not_have_multiplexer_legacy(app: Sanic):
event = Event()
@app.main_process_start
async def setup(app, _):
app.shared_ctx.event = event
@app.after_server_start
def stop(app):
if hasattr(app, "m") and isinstance(app.m, WorkerMultiplexer):
app.shared_ctx.event.set()
app.stop()
app.run(legacy=True)
assert not event.is_set()
def test_ack(worker_state: Dict[str, Any], m: WorkerMultiplexer):
worker_state["Test"] = {"foo": "bar"}
m.ack()
assert worker_state["Test"] == {"foo": "bar", "state": "ACKED"}
def test_restart_self(monitor_publisher: Mock, m: WorkerMultiplexer):
m.restart()
monitor_publisher.send.assert_called_once_with("Test")
def test_restart_foo(monitor_publisher: Mock, m: WorkerMultiplexer):
m.restart("foo")
monitor_publisher.send.assert_called_once_with("foo")
def test_reload_alias(monitor_publisher: Mock, m: WorkerMultiplexer):
m.reload()
monitor_publisher.send.assert_called_once_with("Test")
def test_terminate(monitor_publisher: Mock, m: WorkerMultiplexer):
m.terminate()
monitor_publisher.send.assert_called_once_with("__TERMINATE__")
def test_properties(
monitor_publisher: Mock, worker_state: Dict[str, Any], m: WorkerMultiplexer
):
assert m.reload == m.restart
assert m.pid == getpid()
assert m.name == "Test"
assert m.workers == worker_state
assert m.state == worker_state["Test"]
assert isinstance(m.state, WorkerState)
@pytest.mark.parametrize(
"params,expected",
(
({}, "Test"),
({"name": "foo"}, "foo"),
({"all_workers": True}, "__ALL_PROCESSES__:"),
({"name": "foo", "all_workers": True}, ValueError),
),
)
def test_restart_params(
monitor_publisher: Mock,
m: WorkerMultiplexer,
params: Dict[str, Any],
expected: Union[str, Type[Exception]],
):
if isinstance(expected, str):
m.restart(**params)
monitor_publisher.send.assert_called_once_with(expected)
else:
with pytest.raises(expected):
m.restart(**params)