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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
|
import os
import stat
import subprocess
import sys
import time
import pytest
from notebook import DEFAULT_NOTEBOOK_PORT
from .launchnotebook import UNIXSocketNotebookTestBase
from ..utils import urlencode_unix_socket, urlencode_unix_socket_path
pytestmark = pytest.mark.integration_tests
@pytest.mark.skipif(sys.platform == "win32", reason="do not run on windows")
def test_shutdown_sock_server_integration():
sock = UNIXSocketNotebookTestBase.sock
url = urlencode_unix_socket(sock).encode()
encoded_sock_path = urlencode_unix_socket_path(sock)
p = subprocess.Popen(
['jupyter-notebook', f'--sock={sock}', '--sock-mode=0700'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
complete = False
for line in iter(p.stderr.readline, b''):
print(line.decode())
if url in line:
complete = True
break
assert complete, 'did not find socket URL in stdout when launching notebook'
assert encoded_sock_path.encode() in subprocess.check_output(['jupyter-notebook', 'list'])
# Ensure umask is properly applied.
assert stat.S_IMODE(os.lstat(sock).st_mode) == 0o700
try:
subprocess.check_output(['jupyter-notebook', 'stop'], stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
assert 'There is currently no server running on' in e.output.decode()
else:
raise AssertionError('expected stop command to fail due to target mis-match')
assert encoded_sock_path.encode() in subprocess.check_output(['jupyter-notebook', 'list'])
subprocess.check_output(['jupyter-notebook', 'stop', sock])
assert encoded_sock_path.encode() not in subprocess.check_output(['jupyter-notebook', 'list'])
p.wait()
def test_sock_server_validate_sockmode_type():
try:
subprocess.check_output(
['jupyter-notebook', '--sock=/tmp/nonexistent', '--sock-mode=badbadbad'],
stderr=subprocess.STDOUT
)
except subprocess.CalledProcessError as e:
assert 'badbadbad' in e.output.decode()
else:
raise AssertionError('expected execution to fail due to validation of --sock-mode param')
def test_sock_server_validate_sockmode_accessible():
try:
subprocess.check_output(
['jupyter-notebook', '--sock=/tmp/nonexistent', '--sock-mode=0444'],
stderr=subprocess.STDOUT
)
except subprocess.CalledProcessError as e:
assert '0444' in e.output.decode()
else:
raise AssertionError('expected execution to fail due to validation of --sock-mode param')
def _ensure_stopped(check_msg='There are no running servers'):
try:
subprocess.check_output(
['jupyter-notebook', 'stop'],
stderr=subprocess.STDOUT
)
except subprocess.CalledProcessError as e:
assert check_msg in e.output.decode()
else:
raise AssertionError('expected all servers to be stopped')
@pytest.mark.skipif(not bool(os.environ.get('RUN_NB_INTEGRATION_TESTS', False)), reason="for local testing")
def test_stop_multi_integration():
"""Tests lifecycle behavior for mixed-mode server types w/ default ports.
Mostly suitable for local dev testing due to reliance on default port binding.
"""
TEST_PORT = '9797'
MSG_TMPL = 'Shutting down server on {}...'
_ensure_stopped()
# Default port.
p1 = subprocess.Popen(
['jupyter-notebook', '--no-browser']
)
# Unix socket.
sock = UNIXSocketNotebookTestBase.sock
p2 = subprocess.Popen(
['jupyter-notebook', f'--sock={sock}']
)
# Specified port
p3 = subprocess.Popen(
['jupyter-notebook', '--no-browser', f'--port={TEST_PORT}']
)
time.sleep(3)
assert MSG_TMPL.format(DEFAULT_NOTEBOOK_PORT) in subprocess.check_output(
['jupyter-notebook', 'stop']
).decode()
_ensure_stopped('There is currently no server running on 8888')
assert MSG_TMPL.format(sock) in subprocess.check_output(
['jupyter-notebook', 'stop', sock]
).decode()
assert MSG_TMPL.format(TEST_PORT) in subprocess.check_output(
['jupyter-notebook', 'stop', TEST_PORT]
).decode()
_ensure_stopped()
p1.wait()
p2.wait()
p3.wait()
@pytest.mark.skipif(sys.platform == "win32", reason="do not run on windows")
def test_launch_socket_collision():
"""Tests UNIX socket in-use detection for lifecycle correctness."""
sock = UNIXSocketNotebookTestBase.sock
check_msg = f'socket {sock} is already in use'
_ensure_stopped()
# Start a server.
cmd = ['jupyter-notebook', f'--sock={sock}']
p1 = subprocess.Popen(cmd)
time.sleep(3)
# Try to start a server bound to the same UNIX socket.
try:
subprocess.check_output(cmd, stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
assert check_msg in e.output.decode()
else:
raise AssertionError(f'expected error, instead got {e.output.decode()}')
# Stop the background server, ensure it's stopped and wait on the process to exit.
subprocess.check_call(['jupyter-notebook', 'stop', sock])
_ensure_stopped()
p1.wait()
|
