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
|
from __future__ import annotations
from time import sleep
from distributed import Client
from distributed.metrics import time
from distributed.utils import open_port
from distributed.utils_test import (
get_cert,
new_config_file,
popen,
tls_only_config,
tls_security,
)
ca_file = get_cert("tls-ca-cert.pem")
cert = get_cert("tls-cert.pem")
key = get_cert("tls-key.pem")
keycert = get_cert("tls-key-cert.pem")
tls_args = ["--tls-ca-file", ca_file, "--tls-cert", keycert]
tls_args_2 = ["--tls-ca-file", ca_file, "--tls-cert", cert, "--tls-key", key]
def wait_for_cores(c, nthreads=1):
start = time()
while len(c.nthreads()) < 1:
sleep(0.1)
assert time() < start + 10
def test_basic(loop, requires_default_ports):
with popen(["dask", "scheduler", "--no-dashboard"] + tls_args) as s:
with popen(
["dask", "worker", "--no-dashboard", "tls://127.0.0.1:8786"] + tls_args
) as w:
with Client(
"tls://127.0.0.1:8786", loop=loop, security=tls_security()
) as c:
wait_for_cores(c)
def test_sni(loop):
port = open_port()
with popen(["dask-scheduler", "--no-dashboard", f"--port={port}"] + tls_args) as s:
with popen(
[
"dask-worker",
"--no-dashboard",
"--scheduler-sni",
"localhost",
f"tls://127.0.0.1:{port}",
]
+ tls_args
) as w:
with Client(
f"tls://127.0.0.1:{port}", loop=loop, security=tls_security()
) as c:
wait_for_cores(c)
def test_nanny(loop):
port = open_port()
with popen(
[
"dask",
"scheduler",
"--no-dashboard",
f"--port={port}",
]
+ tls_args
) as s:
with popen(
["dask", "worker", "--no-dashboard", "--nanny", f"tls://127.0.0.1:{port}"]
+ tls_args
) as w:
with Client(
f"tls://127.0.0.1:{port}", loop=loop, security=tls_security()
) as c:
wait_for_cores(c)
def test_separate_key_cert(loop):
port = open_port()
with popen(
[
"dask",
"scheduler",
"--no-dashboard",
f"--port={port}",
]
+ tls_args_2
) as s:
with popen(
["dask", "worker", "--no-dashboard", f"tls://127.0.0.1:{port}"] + tls_args_2
) as w:
with Client(
f"tls://127.0.0.1:{port}", loop=loop, security=tls_security()
) as c:
wait_for_cores(c)
def test_use_config_file(loop):
port = open_port()
with new_config_file(tls_only_config()):
with popen(
[
"dask",
"scheduler",
"--no-dashboard",
"--host",
"tls://",
"--port",
str(port),
]
) as s:
with popen(
["dask", "worker", "--no-dashboard", f"tls://127.0.0.1:{port}"]
) as w:
with Client(
f"tls://127.0.0.1:{port}", loop=loop, security=tls_security()
) as c:
wait_for_cores(c)
|
