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
|
from __future__ import annotations
import pytest
from tornado.ioloop import IOLoop
from distributed.deploy.cluster import Cluster
from distributed.utils_test import gen_test
@gen_test()
async def test_eq():
async with Cluster(asynchronous=True, name="A") as clusterA, Cluster(
asynchronous=True, name="A2"
) as clusterA2, Cluster(asynchronous=True, name="B") as clusterB:
assert clusterA != "A"
assert not (clusterA == "A")
assert clusterA == clusterA
assert not (clusterA != clusterA)
assert clusterA != clusterA2
assert not (clusterA == clusterA2)
assert clusterA != clusterB
assert not (clusterA == clusterB)
@gen_test()
async def test_repr():
async with Cluster(asynchronous=True, name="A") as cluster:
assert cluster.scheduler_address == "<Not Connected>"
res = repr(cluster)
expected = "Cluster(A, '<Not Connected>', workers=0, threads=0, memory=0 B)"
assert res == expected
@gen_test()
async def test_logs_deprecated():
async with Cluster(asynchronous=True) as cluster:
with pytest.warns(FutureWarning, match="get_logs"):
cluster.logs()
@gen_test()
async def test_deprecated_loop_properties():
class ExampleCluster(Cluster):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.loop = self.io_loop = IOLoop.current()
with pytest.warns(DeprecationWarning) as warninfo:
async with ExampleCluster(asynchronous=True, loop=IOLoop.current()):
pass
assert [(w.category, *w.message.args) for w in warninfo] == [
(DeprecationWarning, "setting the loop property is deprecated")
]
|
