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
|
import sys
import pytest
# Skip this module if on Windows. Unix sockets are not available on Windows.
pytestmark = pytest.mark.skipif(
sys.platform.startswith("win"), reason="Unix sockets are not available on Windows."
)
import urllib
if not sys.platform.startswith("win"):
from tornado.netutil import bind_unix_socket
import jupyter_server.serverapp
from jupyter_server.utils import async_fetch, url_path_join, urlencode_unix_socket
@pytest.fixture
def jp_server_config(jp_unix_socket_file):
"""Configure the serverapp fixture with the unix socket."""
return {"ServerApp": {"sock": jp_unix_socket_file, "allow_remote_access": True}}
@pytest.fixture
def http_server_port(jp_unix_socket_file, jp_process_id):
"""Unix socket and process ID used by tornado's HTTP Server.
Overrides the http_server_port fixture from pytest-tornasync and replaces
it with a tuple: (unix socket, process id)
"""
return (bind_unix_socket(jp_unix_socket_file), jp_process_id)
@pytest.fixture
def jp_unix_socket_fetch(jp_unix_socket_file, jp_auth_header, jp_base_url, http_server, io_loop):
"""A fetch fixture for Jupyter Server tests that use the unix_serverapp fixture"""
async def client(*parts, headers=None, params=None, **kwargs):
# Handle URL strings
host_url = urlencode_unix_socket(jp_unix_socket_file)
path_url = url_path_join(jp_base_url, *parts)
params_url = urllib.parse.urlencode(params or {})
url = url_path_join(host_url, path_url + "?" + params_url)
r = await async_fetch(url, headers=headers or {}, io_loop=io_loop, **kwargs)
return r
return client
async def test_get_spec(jp_unix_socket_fetch):
# Handle URL strings
parts = ["api", "spec.yaml"]
# Make request and verify it succeeds.'
response = await jp_unix_socket_fetch(*parts)
assert response.code == 200
assert response.body is not None
async def test_list_running_servers(jp_unix_socket_file, http_server):
"""Test that a server running on unix sockets is discovered by the server list"""
servers = list(jupyter_server.serverapp.list_running_servers())
assert len(servers) >= 1
assert jp_unix_socket_file in {info["sock"] for info in servers}
|
