287 lines
10 KiB
Python
287 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise synchronous video deletion and the live recycle-bin workflow."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
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(path: Path) -> str:
|
|
value = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
while block := handle.read(1024 * 1024):
|
|
value.update(block)
|
|
return value.hexdigest()
|
|
|
|
|
|
def videos(client: httpx.Client) -> list[dict[str, Any]]:
|
|
return call(client, "GET", "/api/v1/videos?limit=500").json()
|
|
|
|
|
|
def find_video(client: httpx.Client, source_id: str, filename: str) -> dict[str, Any] | None:
|
|
return next(
|
|
(
|
|
item
|
|
for item in videos(client)
|
|
if item.get("source_id") == source_id
|
|
and item.get("source_key", "").rsplit("/", 1)[-1] == filename
|
|
),
|
|
None,
|
|
)
|
|
|
|
|
|
def upload(
|
|
client: httpx.Client,
|
|
fixture: Path,
|
|
source_id: str,
|
|
filename: str,
|
|
title: str,
|
|
) -> dict[str, Any]:
|
|
visible = find_video(client, source_id, filename)
|
|
if visible:
|
|
return visible
|
|
existing = call(client, "GET", "/api/v1/uploads?limit=500").json()
|
|
task = next(
|
|
(
|
|
item
|
|
for item in existing
|
|
if item.get("filename") == filename
|
|
and item.get("status") in {"receiving", "queued", "transferring", "indexing"}
|
|
),
|
|
None,
|
|
)
|
|
if task is None:
|
|
task = call(
|
|
client,
|
|
"POST",
|
|
"/api/v1/uploads",
|
|
json={
|
|
"source_id": source_id,
|
|
"relative_path": "ingest",
|
|
"filename": filename,
|
|
"title": title,
|
|
"size_bytes": fixture.stat().st_size,
|
|
"sha256": digest(fixture),
|
|
"conflict": "skip",
|
|
},
|
|
).json()
|
|
upload_id = task["id"]
|
|
if task.get("status") != "completed":
|
|
chunk_size = int(task["chunk_size"])
|
|
received = {int(value) for value in task.get("received_chunks", task.get("received", []))}
|
|
with fixture.open("rb") as handle:
|
|
for index in range(int(task["total_chunks"])):
|
|
block = handle.read(chunk_size)
|
|
if index in received:
|
|
continue
|
|
call(
|
|
client,
|
|
"PUT",
|
|
f"/api/v1/uploads/{upload_id}/chunks/{index}",
|
|
content=block,
|
|
headers={"X-Chunk-SHA256": hashlib.sha256(block).hexdigest()},
|
|
)
|
|
call(client, "POST", f"/api/v1/uploads/{upload_id}/complete")
|
|
deadline = time.monotonic() + 180
|
|
while time.monotonic() < deadline:
|
|
tasks = call(client, "GET", "/api/v1/uploads?limit=500").json()
|
|
task = next(item for item in tasks if item["id"] == upload_id)
|
|
if task["status"] == "completed":
|
|
break
|
|
if task["status"] in {"failed", "cancelled"}:
|
|
raise AssertionError(f"upload failed: {task.get('message') or task.get('error')}")
|
|
time.sleep(1)
|
|
else:
|
|
raise TimeoutError(f"upload did not complete: {filename}")
|
|
deadline = time.monotonic() + 60
|
|
while time.monotonic() < deadline:
|
|
video = find_video(client, source_id, filename)
|
|
if video:
|
|
return video
|
|
time.sleep(1)
|
|
raise TimeoutError(f"completed upload did not appear in /videos: {filename}")
|
|
|
|
|
|
def purge_source_key(client: httpx.Client, source_id: str, key: str) -> None:
|
|
try:
|
|
result = call(
|
|
client,
|
|
"POST",
|
|
"/api/v1/files/trash",
|
|
json={"source_id": source_id, "keys": [key]},
|
|
).json()
|
|
except RuntimeError as exc:
|
|
if "404" in str(exc):
|
|
return
|
|
raise
|
|
for trash_id in result.get("ids", []):
|
|
call(client, "DELETE", f"/api/v1/trash/{trash_id}")
|
|
|
|
|
|
def search_ids(client: httpx.Client, text: str) -> list[str]:
|
|
result = call(
|
|
client,
|
|
"POST",
|
|
"/api/v1/search",
|
|
json={"text": text, "recognition_types": ["metadata"]},
|
|
).json()
|
|
return [item["video_id"] for item in result.get("items", [])]
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--base-url", required=True)
|
|
auth = parser.add_mutually_exclusive_group(required=True)
|
|
auth.add_argument("--token-file", type=Path)
|
|
auth.add_argument("--password")
|
|
parser.add_argument("--run-dir", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
|
|
state = json.loads((args.run_dir / "state.json").read_text())
|
|
headers = {}
|
|
if args.token_file is not None:
|
|
headers["Authorization"] = f"Bearer {args.token_file.read_text().strip()}"
|
|
client = httpx.Client(
|
|
base_url=args.base_url.rstrip("/"),
|
|
headers=headers,
|
|
timeout=httpx.Timeout(60, connect=10),
|
|
)
|
|
if args.password is not None:
|
|
login = call(
|
|
client,
|
|
"POST",
|
|
"/api/v1/auth/login",
|
|
json={"password": args.password, "remember_device": False},
|
|
).json()
|
|
csrf = str(login.get("csrf_token") or "")
|
|
if not csrf:
|
|
raise AssertionError("login response omitted CSRF token")
|
|
client.headers["X-CSRF-Token"] = csrf
|
|
source_id = state["source_id"]
|
|
run_id = state["run_id"]
|
|
cases = [
|
|
{
|
|
"mode": "trash",
|
|
"fixture": args.run_dir / "e2e-positive.mp4",
|
|
"filename": f"{run_id}-delete-trash.mp4",
|
|
"title": "删除回收验收 481",
|
|
"key": f"ingest/{run_id}-delete-trash.mp4",
|
|
},
|
|
{
|
|
"mode": "tombstone",
|
|
"fixture": args.run_dir / "e2e-negative.mp4",
|
|
"filename": f"{run_id}-delete-only.mp4",
|
|
"title": "仅移出验收 592",
|
|
"key": f"ingest/{run_id}-delete-only.mp4",
|
|
},
|
|
]
|
|
report: dict[str, Any] = {}
|
|
try:
|
|
trash_case = cases[0]
|
|
first = upload(
|
|
client,
|
|
trash_case["fixture"],
|
|
source_id,
|
|
trash_case["filename"],
|
|
trash_case["title"],
|
|
)
|
|
deleted = call(client, "DELETE", f"/api/v1/videos/{first['id']}?delete_source=true").json()
|
|
if not deleted.get("source_deleted") or not deleted.get("trash_id"):
|
|
raise AssertionError(f"source deletion result mismatch: {deleted}")
|
|
if find_video(client, source_id, trash_case["filename"]):
|
|
raise AssertionError("source-deleted video remained visible")
|
|
if first["id"] in search_ids(client, "删除回收验收 481"):
|
|
raise AssertionError("source-deleted video remained in metadata search")
|
|
trash_rows = call(client, "GET", f"/api/v1/trash?source_id={source_id}").json()
|
|
trash_item = next((item for item in trash_rows if item["id"] == deleted["trash_id"]), None)
|
|
if not trash_item:
|
|
raise AssertionError("deleted source was not listed in recycle bin")
|
|
restored = call(client, "POST", f"/api/v1/trash/{trash_item['id']}/restore", json={}).json()
|
|
deadline = time.monotonic() + 90
|
|
restored_video = None
|
|
while time.monotonic() < deadline:
|
|
restored_video = find_video(client, source_id, trash_case["filename"])
|
|
if restored_video:
|
|
break
|
|
time.sleep(1)
|
|
if not restored_video:
|
|
raise AssertionError("restored source did not return to /videos")
|
|
removed_again = call(
|
|
client,
|
|
"DELETE",
|
|
f"/api/v1/videos/{restored_video['id']}?delete_source=true",
|
|
).json()
|
|
call(client, "DELETE", f"/api/v1/trash/{removed_again['trash_id']}")
|
|
report["trash"] = {
|
|
"deleted": deleted,
|
|
"restored_key": restored["key"],
|
|
"restored_video_id": restored_video["id"],
|
|
"purged": True,
|
|
}
|
|
|
|
tombstone_case = cases[1]
|
|
second = upload(
|
|
client,
|
|
tombstone_case["fixture"],
|
|
source_id,
|
|
tombstone_case["filename"],
|
|
tombstone_case["title"],
|
|
)
|
|
removed = call(client, "DELETE", f"/api/v1/videos/{second['id']}?delete_source=false").json()
|
|
if removed.get("source_deleted") or removed.get("trash_id"):
|
|
raise AssertionError(f"library-only deletion result mismatch: {removed}")
|
|
scan = call(client, "POST", f"/api/v1/sources/{source_id}/scan", json={}).json()
|
|
time.sleep(5)
|
|
if find_video(client, source_id, tombstone_case["filename"]):
|
|
raise AssertionError("tombstoned video reappeared after source scan")
|
|
if second["id"] in search_ids(client, "仅移出验收 592"):
|
|
raise AssertionError("tombstoned video remained in metadata search")
|
|
purge_source_key(client, source_id, tombstone_case["key"])
|
|
report["tombstone"] = {
|
|
"deleted": removed,
|
|
"scan_job_id": scan.get("job_id"),
|
|
"remained_hidden_after_scan": True,
|
|
"source_cleaned": True,
|
|
}
|
|
finally:
|
|
for case in cases:
|
|
video = find_video(client, source_id, case["filename"])
|
|
if video:
|
|
try:
|
|
deletion = call(
|
|
client,
|
|
"DELETE",
|
|
f"/api/v1/videos/{video['id']}?delete_source=true",
|
|
).json()
|
|
if deletion.get("trash_id"):
|
|
call(client, "DELETE", f"/api/v1/trash/{deletion['trash_id']}")
|
|
except Exception:
|
|
pass
|
|
else:
|
|
try:
|
|
purge_source_key(client, source_id, case["key"])
|
|
except Exception:
|
|
pass
|
|
(args.run_dir / "delete-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()
|