Skip to content

Commit

Permalink
Add tests for the resolvePath API
Browse files Browse the repository at this point in the history
  • Loading branch information
krassowski committed Feb 1, 2024
1 parent bfb5bfe commit 35a73e7
Showing 1 changed file with 147 additions and 0 deletions.
147 changes: 147 additions & 0 deletions tests/services/kernels/test_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -183,6 +183,153 @@ async def test_main_kernel_handler(jp_fetch, jp_base_url, jp_serverapp, pending_
await pending_kernel_is_ready(kernel3["id"])


@pytest.mark.timeout(TEST_TIMEOUT)
async def test_resolve_path_kernel(jp_fetch, jp_serverapp, jp_root_dir):
query_path = "hello.py"
contents = "(irrelevant)"

# Put the file in a kernel subdirectory
kernel_root = jp_root_dir / "more" / "path" / "segments"
kernel_root.mkdir(parents=True)
kernel_path = kernel_root / query_path
kernel_path.write_text(contents)

# Create a kernel in temp path
r = await jp_fetch(
"api",
"kernels",
method="POST",
body=json.dumps(
{"name": NATIVE_KERNEL_NAME, "path": str(kernel_root.relative_to(jp_root_dir))}
),
)
kernel_id = json.loads(r.body.decode())["id"]

# Resolve the path
r = await jp_fetch(
"api", "resolvePath", params={"kernel": kernel_id, "path": query_path}, method="GET"
)

assert r.code == 200
resolution = json.loads(r.body.decode())
assert len(resolution["resolved"]) == 1
path = resolution["resolved"][0]
assert path["scope"] == "kernel"

# Should reveal the kernel path
assert path["path"] == str(kernel_path)


@pytest.mark.timeout(TEST_TIMEOUT)
async def test_resolve_path_server_traversal(jp_fetch, jp_serverapp, jp_root_dir):
query_path = "hello.py"
contents = "(irrelevant)"

# Put the file above the root dir
server_path = jp_root_dir / ".." / query_path
server_path.write_text(contents)

# Resolve the path
r = await jp_fetch("api", "resolvePath", params={"path": query_path}, method="GET")

assert r.code == 200
resolution = json.loads(r.body.decode())

# Should NOT disclose file existence on path traversal
assert len(resolution["resolved"]) == 0


@pytest.mark.timeout(TEST_TIMEOUT)
async def test_resolve_path_server(jp_fetch, jp_serverapp, jp_root_dir):
query_path = "hello.py"
contents = "(irrelevant)"

# Put the file in the root dir
server_path = jp_root_dir / query_path
server_path.write_text(contents)

# Resolve the path
r = await jp_fetch("api", "resolvePath", params={"path": query_path}, method="GET")

assert r.code == 200
resolution = json.loads(r.body.decode())
assert len(resolution["resolved"]) == 1
path = resolution["resolved"][0]
assert path["scope"] == "server"

# Should NOT reveal server path, just acknowledge the name
assert path["path"] == server_path.name


@pytest.mark.timeout(TEST_TIMEOUT)
async def test_resolve_path_server_and_kernel(jp_fetch, jp_serverapp, jp_root_dir):
query_path = "hello.py"
contents = "(irrelevant)"

# Put a file for server in the root and for kernel space in a subdirectory
kernel_root = jp_root_dir / "more" / "path" / "segments"
kernel_root.mkdir(parents=True)
kernel_path = kernel_root / query_path
kernel_path.write_text(contents)
server_path = jp_root_dir / query_path
server_path.write_text(contents)

# Create a kernel in temp path
r = await jp_fetch(
"api",
"kernels",
method="POST",
body=json.dumps(
{"name": NATIVE_KERNEL_NAME, "path": str(kernel_root.relative_to(jp_root_dir))}
),
)
kernel_id = json.loads(r.body.decode())["id"]

# Resolve the path
r = await jp_fetch(
"api", "resolvePath", params={"kernel": kernel_id, "path": query_path}, method="GET"
)
assert r.code == 200
resolution = json.loads(r.body.decode())

# Should present candidates for both server and kernel
assert len(resolution["resolved"]) == 2
assert {k["scope"] for k in resolution["resolved"]} == {"server", "kernel"}


@pytest.mark.timeout(TEST_TIMEOUT)
async def test_resolve_path_missing(jp_fetch, jp_serverapp, jp_root_dir):
query_path = "hello.py"
contents = "(irrelevant)"

# Put a file for server in the root and for kernel space in a subdirectory
kernel_root = jp_root_dir / "more" / "path" / "segments"
kernel_root.mkdir(parents=True)
kernel_path = kernel_root / query_path
kernel_path.write_text(contents)
server_path = jp_root_dir / query_path
server_path.write_text(contents)

# Create a kernel in temp path
r = await jp_fetch(
"api",
"kernels",
method="POST",
body=json.dumps(
{"name": NATIVE_KERNEL_NAME, "path": str(kernel_root.relative_to(jp_root_dir))}
),
)
kernel_id = json.loads(r.body.decode())["id"]

# Resolve the path
r = await jp_fetch(
"api", "resolvePath", params={"kernel": kernel_id, "path": "not_hello"}, method="GET"
)
assert r.code == 200
resolution = json.loads(r.body.decode())
assert len(resolution["resolved"]) == 0


@pytest.mark.timeout(TEST_TIMEOUT)
async def test_kernel_handler(jp_fetch, jp_serverapp, pending_kernel_is_ready):
# Create a kernel
Expand Down

0 comments on commit 35a73e7

Please sign in to comment.