forked from dask/distributed
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_versions.py
126 lines (91 loc) · 3.69 KB
/
test_versions.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
import re
import sys
import pytest
from distributed.versions import get_versions, error_message
from distributed import Client, Worker
from distributed.utils_test import gen_cluster
# if one of the nodes reports this version, there's a mismatch
mismatched_version = get_versions()
mismatched_version["packages"]["distributed"] = "0.0.0.dev0"
# for really old versions, the `package` key is missing - version is UNKNOWN
key_err_version = {}
# if no key is available for one package, we assume it's MISSING
missing_version = get_versions()
del missing_version["packages"]["distributed"]
# if a node doesn't report any version info, we treat them as UNKNOWN
# the happens if the node is pre-32cb96e, i.e. <=2.9.1
unknown_version = None
@pytest.fixture
def kwargs_matching():
return dict(
scheduler=get_versions(),
workers={f"worker-{i}": get_versions() for i in range(3)},
client=get_versions(),
)
def test_versions_match(kwargs_matching):
assert error_message(**kwargs_matching) == ""
@pytest.fixture(params=["client", "scheduler", "worker-1"])
def node(request):
"""Node affected by version mismatch."""
return request.param
@pytest.fixture(params=["MISMATCHED", "MISSING", "KEY_ERROR", "NONE"])
def effect(request):
"""Specify type of mismatch."""
return request.param
@pytest.fixture
def kwargs_not_matching(kwargs_matching, node, effect):
affected_version = {
"MISMATCHED": mismatched_version,
"MISSING": missing_version,
"KEY_ERROR": key_err_version,
"NONE": unknown_version,
}[effect]
kwargs = kwargs_matching
if node in kwargs["workers"]:
kwargs["workers"][node] = affected_version
else:
kwargs[node] = affected_version
return kwargs
@pytest.fixture
def pattern(effect):
"""Pattern to match in the right-hand column."""
return {
"MISMATCHED": r"0\.0\.0\.dev0",
"MISSING": "MISSING",
"KEY_ERROR": "UNKNOWN",
"NONE": "UNKNOWN",
}[effect]
def test_version_mismatch(node, effect, kwargs_not_matching, pattern):
msg = error_message(**kwargs_not_matching)
assert "Mismatched versions found" in msg
assert "distributed" in msg
assert re.search(node + r"\s+\|\s+" + pattern, msg)
def test_scheduler_mismatched_irrelevant_package(kwargs_matching):
"""An irrelevant package on the scheduler can have any version."""
kwargs_matching["scheduler"]["packages"]["numpy"] = "0.0.0"
assert "numpy" in kwargs_matching["client"]["packages"]
assert error_message(**kwargs_matching) == ""
def test_scheduler_additional_irrelevant_package(kwargs_matching):
"""An irrelevant package on the scheduler does not need to be present elsewhere."""
kwargs_matching["scheduler"]["packages"]["pyspark"] = "0.0.0"
assert error_message(**kwargs_matching) == ""
@gen_cluster()
async def test_version_warning_in_cluster(s, a, b):
s.workers[a.address].versions["packages"]["dask"] = "0.0.0"
with pytest.warns(None) as record:
async with Client(s.address, asynchronous=True) as client:
pass
assert record
assert any("dask" in str(r.message) for r in record)
assert any("0.0.0" in str(r.message) for r in record)
assert any(a.address in str(r.message) for r in record)
async with Worker(s.address) as w:
assert any("This Worker" in line.message for line in w.logs)
assert any("dask" in line.message for line in w.logs)
assert any(
"0.0.0" in line.message and a.address in line.message for line in w.logs
)
def test_python_version():
required = get_versions()["packages"]
assert "python" in required
assert required["python"] == ".".join(map(str, sys.version_info))