242 lines
9.3 KiB
Python
242 lines
9.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise ImageFind's WebDAV server through its public direct listener."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import quote
|
|
|
|
import httpx
|
|
|
|
|
|
def call(client: httpx.Client, method: str, path: str, **kwargs: Any) -> httpx.Response:
|
|
response = client.request(method, path, **kwargs)
|
|
if response.is_error:
|
|
raise RuntimeError(f"{method} {path} -> {response.status_code}: {response.text[:1000]}")
|
|
return response
|
|
|
|
|
|
def digest(data: bytes) -> str:
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def encoded(parts: list[str]) -> str:
|
|
return "/webdav/" + "/".join(quote(part, safe="") for part in parts)
|
|
|
|
|
|
def videos(client: httpx.Client) -> list[dict[str, Any]]:
|
|
return call(client, "GET", "/api/v1/videos?limit=500").json()
|
|
|
|
|
|
def visible_video(client: httpx.Client, filename: str) -> dict[str, Any] | None:
|
|
return next((item for item in videos(client) if item.get("source_key", "").endswith("/" + filename)), None)
|
|
|
|
|
|
def wait_upload(client: httpx.Client, upload_id: str, timeout: int = 240) -> dict[str, Any]:
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
rows = call(client, "GET", "/api/v1/uploads?limit=500").json()
|
|
upload = next((item for item in rows if item["id"] == upload_id), None)
|
|
if upload and upload["status"] in {"completed", "failed", "cancelled"}:
|
|
if upload["status"] != "completed":
|
|
raise AssertionError(f"WebDAV upload failed: {upload.get('message') or upload.get('error')}")
|
|
return upload
|
|
time.sleep(1)
|
|
raise TimeoutError(f"WebDAV upload did not complete: {upload_id}")
|
|
|
|
|
|
def wait_video(client: httpx.Client, filename: str, timeout: int = 90) -> dict[str, Any]:
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
if video := visible_video(client, filename):
|
|
return video
|
|
time.sleep(1)
|
|
raise TimeoutError(f"WebDAV video did not appear: {filename}")
|
|
|
|
|
|
def upload_id(response: httpx.Response) -> str:
|
|
value = response.headers.get("x-imagefind-upload-id")
|
|
if not value:
|
|
raise AssertionError(f"WebDAV response omitted upload id: {dict(response.headers)}")
|
|
return value
|
|
|
|
|
|
def collection_video_path(nodes: list[dict[str, Any]], video_id: str, parents: list[str] | None = None):
|
|
parents = list(parents or [])
|
|
for node in nodes:
|
|
if node.get("kind") == "group":
|
|
found = collection_video_path(node.get("children") or [], video_id, [*parents, node["name"]])
|
|
if found is not None:
|
|
return found
|
|
elif node.get("video_id") == video_id:
|
|
return parents
|
|
return None
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--base-url", required=True)
|
|
parser.add_argument("--token-file", type=Path, required=True)
|
|
parser.add_argument("--run-dir", type=Path, required=True)
|
|
parser.add_argument("--case-id", default="v045")
|
|
args = parser.parse_args()
|
|
|
|
state = json.loads((args.run_dir / "state.json").read_text())
|
|
token = args.token_file.read_text().strip()
|
|
prefix = f"{state['run_id']}-{args.case_id.strip()}-dav"
|
|
payload = (args.run_dir / "e2e-positive.mp4").read_bytes()
|
|
checksum = digest(payload)
|
|
api = httpx.Client(
|
|
base_url=args.base_url.rstrip("/"),
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
timeout=httpx.Timeout(60, connect=10),
|
|
)
|
|
dav = httpx.Client(
|
|
base_url=args.base_url.rstrip("/"),
|
|
auth=httpx.BasicAuth("imagefind", token),
|
|
timeout=httpx.Timeout(120, connect=10, write=120, read=120),
|
|
)
|
|
|
|
config = call(api, "GET", "/api/v1/webdav/config").json()
|
|
if not config.get("enabled") or config.get("source_id") != state["source_id"]:
|
|
raise AssertionError(f"unexpected WebDAV config: {config}")
|
|
options = call(dav, "OPTIONS", "/webdav/")
|
|
if "PROPFIND" not in options.headers.get("allow", ""):
|
|
raise AssertionError("WebDAV OPTIONS omitted PROPFIND")
|
|
root = call(dav, "PROPFIND", "/webdav/", headers={"Depth": "1"})
|
|
if "multistatus" not in root.text:
|
|
raise AssertionError("WebDAV root did not return a multistatus document")
|
|
|
|
root_name = f"{prefix}-root.mp4"
|
|
root_response = call(
|
|
dav,
|
|
"PUT",
|
|
encoded([root_name]),
|
|
content=payload,
|
|
headers={"X-Content-SHA256": checksum},
|
|
)
|
|
root_task = wait_upload(api, upload_id(root_response))
|
|
root_video = wait_video(api, root_name)
|
|
|
|
collection_name = f"{prefix}-collection"
|
|
groups = [f"level-{number:02d}" for number in range(1, 6)]
|
|
path: list[str] = [collection_name]
|
|
for part in [collection_name, *groups]:
|
|
if part != collection_name:
|
|
path.append(part)
|
|
response = dav.request("MKCOL", encoded(path))
|
|
if response.status_code not in {201, 405}:
|
|
raise RuntimeError(f"MKCOL {path} -> {response.status_code}: {response.text[:1000]}")
|
|
|
|
resume_name = f"{prefix}-resume.mp4"
|
|
resume_path = encoded([collection_name, *groups, resume_name])
|
|
midpoint = len(payload) // 2
|
|
first = call(
|
|
dav,
|
|
"PUT",
|
|
resume_path,
|
|
content=payload[:midpoint],
|
|
headers={
|
|
"Content-Range": f"bytes 0-{midpoint - 1}/{len(payload)}",
|
|
"X-Content-SHA256": checksum,
|
|
},
|
|
)
|
|
if first.status_code != 204 or int(first.headers.get("upload-offset", 0)) != midpoint:
|
|
raise AssertionError(f"unexpected partial PUT response: {first.status_code} {dict(first.headers)}")
|
|
if visible_video(api, resume_name):
|
|
raise AssertionError("partial WebDAV upload leaked into /videos")
|
|
staged = call(dav, "HEAD", resume_path)
|
|
if int(staged.headers.get("upload-offset", 0)) != midpoint:
|
|
raise AssertionError(f"HEAD did not expose resume offset: {dict(staged.headers)}")
|
|
second = call(
|
|
dav,
|
|
"PUT",
|
|
resume_path,
|
|
content=payload[midpoint:],
|
|
headers={
|
|
"Content-Range": f"bytes {midpoint}-{len(payload) - 1}/{len(payload)}",
|
|
"X-Content-SHA256": checksum,
|
|
},
|
|
)
|
|
resume_task = wait_upload(api, upload_id(second))
|
|
resume_video = wait_video(api, resume_name)
|
|
|
|
part_name = f"{prefix}-move.mp4.part"
|
|
final_name = f"{prefix}-move.mp4"
|
|
part_path = encoded([collection_name, *groups, part_name])
|
|
final_path = encoded([collection_name, *groups, final_name])
|
|
temporary = call(
|
|
dav,
|
|
"PUT",
|
|
part_path,
|
|
content=payload,
|
|
headers={"X-Content-SHA256": checksum},
|
|
)
|
|
if temporary.headers.get("x-imagefind-upload-id"):
|
|
raise AssertionError("temporary WebDAV name was committed before MOVE")
|
|
moved = call(
|
|
dav,
|
|
"MOVE",
|
|
part_path,
|
|
headers={"Destination": args.base_url.rstrip("/") + final_path, "Overwrite": "F"},
|
|
)
|
|
move_task = wait_upload(api, upload_id(moved))
|
|
move_video = wait_video(api, final_name)
|
|
|
|
started = time.perf_counter()
|
|
duplicate = call(
|
|
dav,
|
|
"PUT",
|
|
resume_path,
|
|
content=payload,
|
|
headers={"X-Content-SHA256": checksum, "Expect": "100-continue"},
|
|
)
|
|
dedupe_ms = round((time.perf_counter() - started) * 1000, 1)
|
|
if duplicate.status_code != 204 or duplicate.headers.get("x-imagefind-deduplicated") != "true":
|
|
raise AssertionError(f"same-path duplicate was not acknowledged: {dict(duplicate.headers)}")
|
|
if duplicate.headers.get("x-imagefind-upload-id") != resume_task["id"]:
|
|
raise AssertionError("deduplicated PUT did not reference the original upload task")
|
|
|
|
ranged = call(dav, "GET", final_path, headers={"Range": "bytes=0-63"})
|
|
if ranged.status_code != 206 or ranged.content != payload[:64]:
|
|
raise AssertionError("WebDAV Range read returned unexpected bytes")
|
|
|
|
collections = call(api, "GET", "/api/v1/collections").json()
|
|
collection = next((item for item in collections if item["name"] == collection_name), None)
|
|
if not collection:
|
|
raise AssertionError("WebDAV directory did not create a collection")
|
|
detail = call(api, "GET", f"/api/v1/collections/{collection['id']}").json()
|
|
for video in (resume_video, move_video):
|
|
actual_path = collection_video_path(detail.get("items") or [], video["id"])
|
|
if actual_path != groups:
|
|
raise AssertionError(f"collection hierarchy mismatch: {video['id']} -> {actual_path}")
|
|
if any(item["id"] == root_video["id"] for item in detail.get("videos") or []):
|
|
raise AssertionError("root WebDAV upload unexpectedly joined a collection")
|
|
|
|
report = {
|
|
"root_upload": {"id": root_task["id"], "video_id": root_video["id"]},
|
|
"resumable_upload": {
|
|
"id": resume_task["id"],
|
|
"video_id": resume_video["id"],
|
|
"first_offset": midpoint,
|
|
"final_size": len(payload),
|
|
},
|
|
"temporary_move": {"id": move_task["id"], "video_id": move_video["id"]},
|
|
"deduplicated": True,
|
|
"dedupe_latency_ms": dedupe_ms,
|
|
"range_read": len(ranged.content),
|
|
"collection_id": collection["id"],
|
|
"collection_depth": len(groups),
|
|
}
|
|
(args.run_dir / "webdav-report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n")
|
|
print(json.dumps(report, ensure_ascii=False, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|