|
| 1 | +import os |
| 2 | +import socket |
| 3 | +import subprocess |
| 4 | +import sys |
| 5 | +import warnings |
1 | 6 | from pathlib import Path
|
2 | 7 | from unittest.mock import patch
|
3 | 8 |
|
4 | 9 | import pytest
|
5 | 10 | from traitlets.tests.utils import check_help_all_output
|
6 | 11 |
|
7 |
| -from jupyter_server.utils import is_namespace_package, url_escape, url_unescape |
| 12 | +from jupyter_server.utils import ( |
| 13 | + check_pid, |
| 14 | + check_version, |
| 15 | + is_namespace_package, |
| 16 | + path2url, |
| 17 | + run_sync_in_loop, |
| 18 | + samefile_simple, |
| 19 | + to_api_path, |
| 20 | + unix_socket_in_use, |
| 21 | + url2path, |
| 22 | + url_escape, |
| 23 | + url_unescape, |
| 24 | +) |
8 | 25 |
|
9 | 26 |
|
10 | 27 | def test_help_output():
|
@@ -59,3 +76,49 @@ def test_is_namespace_package_no_spec():
|
59 | 76 |
|
60 | 77 | assert is_namespace_package("dummy") is None
|
61 | 78 | mocked_spec.assert_called_once_with("dummy")
|
| 79 | + |
| 80 | + |
| 81 | +@pytest.mark.skipif(os.name == "nt", reason="Paths are annoying on Windows") |
| 82 | +def test_path_utils(tmp_path): |
| 83 | + path = str(tmp_path) |
| 84 | + assert os.path.basename(path2url(path)) == os.path.basename(path) |
| 85 | + |
| 86 | + url = path2url(path) |
| 87 | + assert path.endswith(url2path(url)) |
| 88 | + |
| 89 | + assert samefile_simple(path, path) |
| 90 | + |
| 91 | + assert to_api_path(path, os.path.dirname(path)) == os.path.basename(path) |
| 92 | + |
| 93 | + |
| 94 | +def test_check_version(): |
| 95 | + assert check_version("1.0.2", "1.0.1") |
| 96 | + assert not check_version("1.0.0", "1.0.1") |
| 97 | + assert check_version(1.0, "1.0.1") |
| 98 | + |
| 99 | + |
| 100 | +def test_check_pid(): |
| 101 | + proc = subprocess.Popen([sys.executable]) |
| 102 | + proc.kill() |
| 103 | + proc.wait() |
| 104 | + check_pid(proc.pid) |
| 105 | + |
| 106 | + |
| 107 | +async def test_run_sync_in_loop(): |
| 108 | + async def foo(): |
| 109 | + pass |
| 110 | + |
| 111 | + with warnings.catch_warnings(): |
| 112 | + warnings.simplefilter("ignore") |
| 113 | + await run_sync_in_loop(foo()) |
| 114 | + |
| 115 | + |
| 116 | +@pytest.mark.skipif(os.name != "posix", reason="Requires unix sockets") |
| 117 | +def test_unix_socket_in_use(tmp_path): |
| 118 | + root_tmp_dir = Path("/tmp").resolve() |
| 119 | + server_address = os.path.join(root_tmp_dir, os.path.basename(tmp_path)) |
| 120 | + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) |
| 121 | + sock.bind(server_address) |
| 122 | + sock.listen(0) |
| 123 | + assert unix_socket_in_use(server_address) |
| 124 | + sock.close() |
0 commit comments