276 lines
8.4 KiB
Python
276 lines
8.4 KiB
Python
"""WebDAV protocol tests: OPTIONS, PROPFIND, PROPPATCH, COPY, MOVE, LOCK, UNLOCK."""
|
|
import xml.etree.ElementTree as ET
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sanic import Sanic
|
|
|
|
from cista import config, watching
|
|
from cista.fileserver import bp as fileserver_bp
|
|
|
|
_DAV_NS = "DAV:"
|
|
_METHODS = ("MKCOL", "MOVE", "COPY", "PROPFIND")
|
|
|
|
|
|
@pytest.fixture
|
|
def setup_storage(tmp_path: Path):
|
|
config.config = config.Config(path=tmp_path, listen=":0", public=True)
|
|
watching.state.root = []
|
|
watching.rootpath = tmp_path
|
|
yield tmp_path
|
|
watching.state.root = []
|
|
|
|
|
|
@pytest_asyncio.fixture()
|
|
async def client(setup_storage: Path):
|
|
app = Sanic(f"files-dav-test-{uuid4().hex}", strict_slashes=True)
|
|
app.router.ALLOWED_METHODS = (*app.router.ALLOWED_METHODS, *_METHODS)
|
|
app.blueprint(fileserver_bp)
|
|
yield app.asgi_client
|
|
|
|
|
|
def _dav(tag: str) -> str:
|
|
return f"{{{_DAV_NS}}}{tag}"
|
|
|
|
|
|
def _parse_multistatus(body: bytes) -> list[ET.Element]:
|
|
root = ET.fromstring(body)
|
|
assert root.tag == _dav("multistatus")
|
|
return root.findall(_dav("response"))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# OPTIONS
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_options_advertises_dav_class(client):
|
|
_, res = await client.options("/files/")
|
|
|
|
assert res.status_code == 200
|
|
assert "1" in res.headers.get("dav", "")
|
|
assert "PROPFIND" in res.headers.get("allow", "")
|
|
assert "COPY" in res.headers.get("allow", "")
|
|
assert "MOVE" in res.headers.get("allow", "")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_options_without_trailing_slash(client):
|
|
"""WebDAV clients (e.g. Windows) send OPTIONS /files without trailing slash."""
|
|
_, res = await client.options("/files")
|
|
|
|
assert res.status_code == 200
|
|
assert "1" in res.headers.get("dav", "")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PROPFIND
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_propfind_root_depth0(client, setup_storage: Path):
|
|
_, res = await client.request("PROPFIND", "/files/", headers={"Depth": "0"})
|
|
|
|
assert res.status_code == 207
|
|
responses = _parse_multistatus(res.body)
|
|
assert len(responses) == 1
|
|
href = responses[0].findtext(_dav("href"))
|
|
assert href == "/files/"
|
|
rt = responses[0].find(f".//{_dav('resourcetype')}/{_dav('collection')}")
|
|
assert rt is not None, "Root should be a collection"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_propfind_root_depth1_lists_children(client, setup_storage: Path):
|
|
(setup_storage / "alpha.txt").write_text("a", encoding="utf-8")
|
|
(setup_storage / "beta").mkdir()
|
|
|
|
_, res = await client.request("PROPFIND", "/files/", headers={"Depth": "1"})
|
|
|
|
assert res.status_code == 207
|
|
responses = _parse_multistatus(res.body)
|
|
hrefs = [r.findtext(_dav("href")) for r in responses]
|
|
assert "/files/" in hrefs
|
|
assert "/files/alpha.txt" in hrefs
|
|
assert "/files/beta/" in hrefs
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_propfind_file_has_content_length(client, setup_storage: Path):
|
|
(setup_storage / "data.txt").write_text("hello", encoding="utf-8")
|
|
|
|
_, res = await client.request("PROPFIND", "/files/data.txt", headers={"Depth": "0"})
|
|
|
|
assert res.status_code == 207
|
|
responses = _parse_multistatus(res.body)
|
|
cl = responses[0].findtext(f".//{_dav('getcontentlength')}")
|
|
assert cl == "5"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_propfind_depth_infinity_rejected(client, setup_storage: Path):
|
|
_, res = await client.request(
|
|
"PROPFIND", "/files/", headers={"Depth": "infinity"}
|
|
)
|
|
assert res.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_propfind_missing_resource_returns_404(client):
|
|
_, res = await client.request("PROPFIND", "/files/no-such-file.txt")
|
|
assert res.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# COPY
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_copy_file_to_new_path(client, setup_storage: Path):
|
|
(setup_storage / "src.txt").write_text("copy me", encoding="utf-8")
|
|
|
|
_, res = await client.request(
|
|
"COPY",
|
|
"/files/src.txt",
|
|
headers={"Destination": "http://localhost/files/dst.txt"},
|
|
)
|
|
|
|
assert res.status_code == 201
|
|
assert (setup_storage / "src.txt").is_file()
|
|
assert (setup_storage / "dst.txt").read_text() == "copy me"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_copy_overwrites_existing_by_default(client, setup_storage: Path):
|
|
(setup_storage / "src.txt").write_text("new", encoding="utf-8")
|
|
(setup_storage / "dst.txt").write_text("old", encoding="utf-8")
|
|
|
|
_, res = await client.request(
|
|
"COPY",
|
|
"/files/src.txt",
|
|
headers={"Destination": "http://localhost/files/dst.txt"},
|
|
)
|
|
|
|
assert res.status_code == 204
|
|
assert (setup_storage / "dst.txt").read_text() == "new"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_copy_overwrite_false_returns_412(client, setup_storage: Path):
|
|
(setup_storage / "src.txt").write_text("x", encoding="utf-8")
|
|
(setup_storage / "dst.txt").write_text("y", encoding="utf-8")
|
|
|
|
_, res = await client.request(
|
|
"COPY",
|
|
"/files/src.txt",
|
|
headers={
|
|
"Destination": "http://localhost/files/dst.txt",
|
|
"Overwrite": "F",
|
|
},
|
|
)
|
|
|
|
assert res.status_code == 412
|
|
assert (setup_storage / "dst.txt").read_text() == "y"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_copy_directory_recursively(client, setup_storage: Path):
|
|
(setup_storage / "src").mkdir()
|
|
(setup_storage / "src" / "child.txt").write_text("child", encoding="utf-8")
|
|
|
|
_, res = await client.request(
|
|
"COPY",
|
|
"/files/src",
|
|
headers={"Destination": "http://localhost/files/dst"},
|
|
)
|
|
|
|
assert res.status_code == 201
|
|
assert (setup_storage / "dst" / "child.txt").read_text() == "child"
|
|
assert (setup_storage / "src" / "child.txt").is_file()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_copy_missing_parent_returns_409(client, setup_storage: Path):
|
|
(setup_storage / "src.txt").write_text("x", encoding="utf-8")
|
|
|
|
_, res = await client.request(
|
|
"COPY",
|
|
"/files/src.txt",
|
|
headers={"Destination": "http://localhost/files/nodir/dst.txt"},
|
|
)
|
|
|
|
assert res.status_code == 409
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# MOVE
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_move_renames_file(client, setup_storage: Path):
|
|
(setup_storage / "old.txt").write_text("data", encoding="utf-8")
|
|
|
|
_, res = await client.request(
|
|
"MOVE",
|
|
"/files/old.txt",
|
|
headers={"Destination": "http://localhost/files/new.txt"},
|
|
)
|
|
|
|
assert res.status_code == 201
|
|
assert not (setup_storage / "old.txt").exists()
|
|
assert (setup_storage / "new.txt").read_text() == "data"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_move_overwrites_existing(client, setup_storage: Path):
|
|
(setup_storage / "src.txt").write_text("src", encoding="utf-8")
|
|
(setup_storage / "dst.txt").write_text("dst", encoding="utf-8")
|
|
|
|
_, res = await client.request(
|
|
"MOVE",
|
|
"/files/src.txt",
|
|
headers={"Destination": "http://localhost/files/dst.txt"},
|
|
)
|
|
|
|
assert res.status_code == 204
|
|
assert not (setup_storage / "src.txt").exists()
|
|
assert (setup_storage / "dst.txt").read_text() == "src"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_move_overwrite_false_returns_412(client, setup_storage: Path):
|
|
(setup_storage / "src.txt").write_text("src", encoding="utf-8")
|
|
(setup_storage / "dst.txt").write_text("dst", encoding="utf-8")
|
|
|
|
_, res = await client.request(
|
|
"MOVE",
|
|
"/files/src.txt",
|
|
headers={
|
|
"Destination": "http://localhost/files/dst.txt",
|
|
"Overwrite": "F",
|
|
},
|
|
)
|
|
|
|
assert res.status_code == 412
|
|
assert (setup_storage / "src.txt").is_file()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_move_same_source_and_dest_is_noop(client, setup_storage: Path):
|
|
(setup_storage / "file.txt").write_text("x", encoding="utf-8")
|
|
|
|
_, res = await client.request(
|
|
"MOVE",
|
|
"/files/file.txt",
|
|
headers={"Destination": "http://localhost/files/file.txt"},
|
|
)
|
|
|
|
assert res.status_code == 204
|
|
assert (setup_storage / "file.txt").is_file()
|