1078 lines
45 KiB
Python
1078 lines
45 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import sqlite3
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import httpx
|
|
from imagefind.config import Settings
|
|
from imagefind.database import Database, DatabaseTransientError, utcnow
|
|
from imagefind.jobs import JobCancelled, JobQueue
|
|
from imagefind.main import create_app
|
|
from imagefind.resources import ResourceGovernor
|
|
from imagefind.vectors import pack_vector
|
|
|
|
|
|
def _app(tmp_path: Path):
|
|
settings = Settings(
|
|
data_dir=tmp_path / "data",
|
|
embedding_backend="hash",
|
|
scan_interval_seconds=86400,
|
|
upload_reserve_gb=0,
|
|
)
|
|
settings.prepare()
|
|
app = create_app(settings)
|
|
media = tmp_path / "media"
|
|
media.mkdir()
|
|
source_id = app.state.services.sources.add_local("媒体", str(media))
|
|
_, token = app.state.services.auth.create_api_token("test")
|
|
return app, media, source_id, {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def test_resource_governor_hysteresis_and_manual_pause(tmp_path: Path, monkeypatch):
|
|
settings = Settings(data_dir=tmp_path / "data")
|
|
settings.prepare()
|
|
db = Database(settings.database_path)
|
|
db.initialize()
|
|
governor = ResourceGovernor(db, settings)
|
|
sample = {
|
|
"cpu_percent": 75.0,
|
|
"memory_total_bytes": 8 * 1024**3,
|
|
"memory_available_bytes": 4 * 1024**3,
|
|
"disk_available_bytes": 100 * 1024**3,
|
|
}
|
|
monkeypatch.setattr(governor, "sample", lambda: dict(sample))
|
|
assert governor.pressure_reason() is None
|
|
assert governor.pressure_reason() is None
|
|
assert "CPU" in governor.pressure_reason()
|
|
sample["cpu_percent"] = 60
|
|
assert governor.pressure_reason() is not None
|
|
sample["cpu_percent"] = 50
|
|
assert governor.pressure_reason() is None
|
|
governor.set_manual_pause(True)
|
|
assert governor.pressure_reason() == "管理员已暂停后台任务"
|
|
governor.set_manual_pause(False)
|
|
assert governor.pressure_reason() is None
|
|
|
|
|
|
def test_resource_samples_are_shared_across_concurrent_lane_polls(tmp_path: Path, monkeypatch):
|
|
settings = Settings(data_dir=tmp_path / "data", resource_sample_seconds=2)
|
|
settings.prepare()
|
|
db = Database(settings.database_path)
|
|
db.initialize()
|
|
governor = ResourceGovernor(db, settings)
|
|
calls = []
|
|
clock = iter((10.0, 10.2, 12.2))
|
|
|
|
def cpu_values():
|
|
calls.append(1)
|
|
return (1000 + len(calls) * 100, 500 + len(calls) * 50)
|
|
|
|
monkeypatch.setattr("imagefind.resources.time.monotonic", lambda: next(clock))
|
|
monkeypatch.setattr(governor, "_cpu_values", cpu_values)
|
|
monkeypatch.setattr(governor, "_memory_values", lambda: (8 * 1024**3, 4 * 1024**3))
|
|
monkeypatch.setattr(
|
|
"imagefind.resources.shutil.disk_usage",
|
|
lambda _path: SimpleNamespace(free=100 * 1024**3),
|
|
)
|
|
|
|
first = governor.sample()
|
|
cached = governor.sample()
|
|
refreshed = governor.sample()
|
|
|
|
assert cached == first
|
|
assert len(calls) == 2
|
|
assert refreshed["cpu_percent"] == 50.0
|
|
|
|
|
|
def test_resource_governor_uses_separate_admission_and_running_memory_floors(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
settings = Settings(data_dir=tmp_path / "data")
|
|
settings.prepare()
|
|
db = Database(settings.database_path)
|
|
db.initialize()
|
|
governor = ResourceGovernor(db, settings)
|
|
sample = {
|
|
"cpu_percent": 20.0,
|
|
"memory_total_bytes": 16 * 1024**3,
|
|
"memory_available_bytes": int(2.8 * 1024**3),
|
|
"disk_available_bytes": 100 * 1024**3,
|
|
}
|
|
monkeypatch.setattr(governor, "sample", lambda: dict(sample))
|
|
|
|
assert governor.pressure_reason() == "可用内存不足，等待系统恢复"
|
|
assert governor.pressure_reason(running=True) is None
|
|
|
|
sample["memory_available_bytes"] = 512 * 1024**2
|
|
assert governor.pressure_reason(running=True) == "可用内存不足，等待系统恢复"
|
|
|
|
|
|
def test_resource_governor_prioritizes_transfer_and_download_under_cpu_pressure(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
settings = Settings(data_dir=tmp_path / "data")
|
|
settings.prepare()
|
|
db = Database(settings.database_path)
|
|
db.initialize()
|
|
governor = ResourceGovernor(db, settings)
|
|
sample = {
|
|
"cpu_percent": 99.0,
|
|
"memory_total_bytes": 8 * 1024**3,
|
|
"memory_available_bytes": 6 * 1024**3,
|
|
"disk_available_bytes": 100 * 1024**3,
|
|
}
|
|
monkeypatch.setattr(governor, "sample", lambda: dict(sample))
|
|
|
|
for _ in range(4):
|
|
assert governor.pressure_reason(lane="transfer") is None
|
|
assert governor.pressure_reason(lane="download") is None
|
|
assert governor.pressure_reason(lane="ai") is None
|
|
assert governor.pressure_reason(lane="ai") is None
|
|
assert "CPU" in governor.pressure_reason(lane="ai")
|
|
assert governor.pressure_reason(lane="scan") is None
|
|
assert governor.pressure_reason(lane="scan") is None
|
|
assert "CPU" in governor.pressure_reason(lane="scan")
|
|
|
|
governor.set_manual_pause(True)
|
|
for lane in ("ai", "transfer", "download", "scan"):
|
|
assert governor.pressure_reason(lane=lane) == "管理员已暂停后台任务"
|
|
|
|
|
|
def test_running_ai_does_not_pause_itself_on_its_own_cpu_load(tmp_path: Path, monkeypatch):
|
|
settings = Settings(data_dir=tmp_path / "data")
|
|
settings.prepare()
|
|
db = Database(settings.database_path)
|
|
db.initialize()
|
|
governor = ResourceGovernor(db, settings)
|
|
sample = {
|
|
"cpu_percent": 99.0,
|
|
"memory_total_bytes": 8 * 1024**3,
|
|
"memory_available_bytes": 6 * 1024**3,
|
|
"disk_available_bytes": 100 * 1024**3,
|
|
}
|
|
monkeypatch.setattr(governor, "sample", lambda: dict(sample))
|
|
|
|
for _ in range(6):
|
|
assert governor.pressure_reason(lane="ai", running=True) is None
|
|
# Admission protection remains in force for the next heavy task.
|
|
assert governor.pressure_reason(lane="ai") is None
|
|
assert governor.pressure_reason(lane="ai") is None
|
|
assert "CPU" in governor.pressure_reason(lane="ai")
|
|
|
|
|
|
def test_resource_api_and_aria2_unavailable(tmp_path: Path, monkeypatch):
|
|
app, _, source_id, headers = _app(tmp_path)
|
|
monkeypatch.setattr(app.state.services.downloads, "executable", lambda: None)
|
|
|
|
async def scenario():
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
status = await client.get("/api/v1/system/resources", headers=headers)
|
|
assert status.status_code == 200
|
|
assert status.json()["profile"] == "balanced"
|
|
assert set(status.json()["lanes"]) == {"ai", "transfer", "download", "scan"}
|
|
assert status.json()["database"]["engine"] == "postgresql"
|
|
assert status.json()["database"]["journal_mode"] == "server"
|
|
assert status.json()["database"]["writer_queue_depth"] >= 0
|
|
paused = await client.patch(
|
|
"/api/v1/system/resources", headers=headers, json={"manual_pause": True}
|
|
)
|
|
assert paused.json()["manual_pause"] is True
|
|
invalid = await client.patch(
|
|
"/api/v1/system/resources",
|
|
headers=headers,
|
|
json={"cpu_pause_percent": 50, "cpu_resume_percent": 60},
|
|
)
|
|
assert invalid.status_code == 400
|
|
runtime = await client.get("/api/v1/downloads/runtime", headers=headers)
|
|
assert runtime.json()["available"] is False
|
|
created = await client.post(
|
|
"/api/v1/downloads",
|
|
headers=headers,
|
|
json={"url": "https://example.test/movie.mp4", "source_id": source_id},
|
|
)
|
|
assert created.status_code == 503
|
|
assert "aria2c" in created.json()["detail"]
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_four_lanes_claim_independently_and_old_process_leases_recover(tmp_path: Path):
|
|
app, _, source_id, headers = _app(tmp_path)
|
|
jobs = app.state.services.jobs
|
|
jobs.governor = None
|
|
expected = {
|
|
"ai": jobs.enqueue("index_video", {"video_id": "video"}),
|
|
"transfer": jobs.enqueue("transfer_upload", {"upload_id": "upload"}),
|
|
"download": jobs.enqueue("aria2_download", {"download_id": "download"}),
|
|
"scan": jobs.enqueue("scan_source", {"source_id": source_id}),
|
|
}
|
|
|
|
for lane, job_id in expected.items():
|
|
claimed = jobs._claim(lane=lane)
|
|
assert claimed and claimed[0] == job_id
|
|
assert jobs._claim(lane=lane) is None
|
|
|
|
restarted = JobQueue(app.state.services.db, governor=None)
|
|
assert restarted.recover_stale() == 4
|
|
with app.state.services.db.read() as conn:
|
|
states = {
|
|
row["id"]: row["status"]
|
|
for row in conn.execute("SELECT id,status FROM jobs WHERE id IN (?,?,?,?)", tuple(expected.values()))
|
|
}
|
|
assert states == {job_id: "queued" for job_id in expected.values()}
|
|
|
|
async def scenario():
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get(
|
|
"/api/v1/jobs?page=1&page_size=10&lane=transfer&status=queued",
|
|
headers=headers,
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["total"] == 1
|
|
assert response.json()["items"][0]["lane"] == "transfer"
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_locked_job_is_requeued_instead_of_failed(tmp_path: Path, monkeypatch):
|
|
settings = Settings(data_dir=tmp_path / "data", upload_reserve_gb=0)
|
|
settings.prepare()
|
|
db = Database(settings.database_path)
|
|
db.initialize()
|
|
jobs = JobQueue(db, poll_seconds=0.01, governor=None)
|
|
|
|
async def locked_handler(_job_id: str, _payload: dict) -> None:
|
|
raise DatabaseTransientError("serialization failure")
|
|
|
|
jobs.register("index_video", locked_handler)
|
|
job_id = jobs.enqueue("index_video", {"video_id": "video"})
|
|
|
|
claimed = jobs._claim(lane="ai")
|
|
assert claimed and claimed[0] == job_id
|
|
|
|
async def inline_to_thread(function, *args, **kwargs):
|
|
return function(*args, **kwargs)
|
|
|
|
# The test runner's Python 3.13 build does not wake its selector after a
|
|
# default-executor callback; execute the same worker path inline here.
|
|
monkeypatch.setattr("imagefind.jobs.asyncio.to_thread", inline_to_thread)
|
|
asyncio.run(jobs._execute_claimed(*claimed))
|
|
with db.read() as conn:
|
|
state = dict(
|
|
conn.execute(
|
|
"SELECT status,error,started_at,lease_owner,message,attempts FROM jobs WHERE id=?",
|
|
(job_id,),
|
|
).fetchone()
|
|
)
|
|
assert state["status"] == "queued"
|
|
assert state["error"] is None
|
|
assert state["started_at"] is None
|
|
assert state["lease_owner"] is None
|
|
assert "自动恢复" in state["message"]
|
|
assert jobs._progress_state == {}
|
|
assert jobs._pause_state == {}
|
|
|
|
|
|
def test_noncritical_job_heartbeats_ignore_transient_lock(tmp_path: Path, monkeypatch):
|
|
settings = Settings(data_dir=tmp_path / "data")
|
|
settings.prepare()
|
|
db = Database(settings.database_path)
|
|
db.initialize()
|
|
jobs = JobQueue(db, governor=None)
|
|
|
|
def locked(*_args, **_kwargs):
|
|
raise DatabaseTransientError("serialization failure")
|
|
|
|
monkeypatch.setattr(db, "write_with_retry", locked)
|
|
jobs.update("ephemeral", 0.25, "处理中")
|
|
jobs.set_pause_reason("ephemeral", "正在让行")
|
|
jobs.checkpoint("ephemeral", "心跳")
|
|
|
|
|
|
def test_aria2_daemon_uses_conservative_system_runtime_options(tmp_path: Path, monkeypatch):
|
|
app, _, _, _ = _app(tmp_path)
|
|
downloads = app.state.services.downloads
|
|
commands: list[list[str]] = []
|
|
|
|
class Process:
|
|
def poll(self):
|
|
return None
|
|
|
|
monkeypatch.setattr(downloads, "executable", lambda: "/usr/bin/aria2c")
|
|
monkeypatch.setattr(downloads, "_free_port", lambda: 6801)
|
|
monkeypatch.setattr(downloads, "_rpc", lambda method, parameters=None: {"version": "1.37"})
|
|
monkeypatch.setattr(
|
|
"imagefind.downloads.subprocess.Popen",
|
|
lambda command, **_kwargs: commands.append(command) or Process(),
|
|
)
|
|
|
|
downloads._start()
|
|
|
|
assert len(commands) == 1
|
|
command = commands[0]
|
|
assert command[0] == "/usr/bin/aria2c"
|
|
assert "--rpc-listen-all=false" in command
|
|
assert "--max-concurrent-downloads=1" in command
|
|
assert "--max-connection-per-server=4" in command
|
|
assert "--max-overall-download-limit=20M" in command
|
|
assert "--file-allocation=none" in command
|
|
|
|
|
|
def test_aria2_magnet_follows_payload_gid_and_preserves_safe_tree(tmp_path: Path, monkeypatch):
|
|
app, _, source_id, _ = _app(tmp_path)
|
|
services = app.state.services
|
|
services.storage.set_writable(source_id, True)
|
|
downloads = services.downloads
|
|
services.jobs.governor = None
|
|
monkeypatch.setattr(downloads, "_start", lambda: None)
|
|
task = downloads.create("magnet:?xt=urn:btih:0123456789abcdef", source_id, "incoming")
|
|
row = downloads._row(task["id"])
|
|
task_dir = Path(row["staging_path"])
|
|
episode = task_dir / "Show" / "Season 01" / "episode.mp4"
|
|
subtitle = task_dir / "Show" / "Season 01" / "episode.srt"
|
|
episode.parent.mkdir(parents=True)
|
|
episode.write_bytes(b"video")
|
|
subtitle.write_bytes(b"subtitle")
|
|
calls: list[tuple[str, list]] = []
|
|
statuses = iter(
|
|
[
|
|
{"gid": "metadata", "status": "complete", "followedBy": ["payload"]},
|
|
{
|
|
"gid": "payload",
|
|
"status": "complete",
|
|
"totalLength": "13",
|
|
"completedLength": "13",
|
|
"downloadSpeed": "0",
|
|
"files": [
|
|
{"path": str(episode), "selected": "true"},
|
|
{"path": str(subtitle), "selected": "true"},
|
|
],
|
|
},
|
|
]
|
|
)
|
|
|
|
def rpc(method: str, parameters=None):
|
|
calls.append((method, parameters or []))
|
|
if method == "aria2.addUri":
|
|
return "metadata"
|
|
if method == "aria2.tellStatus":
|
|
return next(statuses)
|
|
return None
|
|
|
|
transferred: list[SimpleNamespace] = []
|
|
|
|
def write_file(source, destination, filename, path, **_kwargs):
|
|
transferred.append(
|
|
SimpleNamespace(source=source, destination=destination, filename=filename, data=path.read_bytes())
|
|
)
|
|
return f"{destination}/{filename}"
|
|
|
|
monkeypatch.setattr(downloads, "_rpc", rpc)
|
|
monkeypatch.setattr(services.storage, "write_file", write_file)
|
|
downloads.run(row["job_id"], task["id"])
|
|
|
|
assert [call[0] for call in calls[:3]] == ["aria2.addUri", "aria2.tellStatus", "aria2.tellStatus"]
|
|
assert [item.destination for item in transferred] == [
|
|
"incoming/Show/Season 01",
|
|
"incoming/Show/Season 01",
|
|
]
|
|
assert [item.filename for item in transferred] == ["episode.mp4", "episode.srt"]
|
|
assert downloads.get(task["id"])["status"] == "completed"
|
|
assert not task_dir.exists()
|
|
|
|
|
|
def test_aria2_rejects_files_outside_its_private_staging_tree(tmp_path: Path):
|
|
app, _, _, _ = _app(tmp_path)
|
|
task_dir = app.state.services.settings.download_staging_dir / "safe-task"
|
|
task_dir.mkdir()
|
|
outside = tmp_path / "outside.mp4"
|
|
outside.write_bytes(b"video")
|
|
try:
|
|
app.state.services.downloads._safe_files(
|
|
task_dir, [{"path": str(outside), "selected": "true"}]
|
|
)
|
|
except ValueError as exc:
|
|
assert "暂存目录之外" in str(exc)
|
|
else:
|
|
raise AssertionError("outside aria2 paths must be rejected")
|
|
|
|
|
|
def test_aria2_failed_retry_is_atomic_even_while_old_job_is_finishing(tmp_path: Path, monkeypatch):
|
|
app, _, source_id, _ = _app(tmp_path)
|
|
services = app.state.services
|
|
services.storage.set_writable(source_id, True)
|
|
downloads = services.downloads
|
|
task = downloads.create("https://example.test/movie.mp4", source_id)
|
|
old_job_id = task["job_id"]
|
|
with services.db.transaction() as conn:
|
|
conn.execute(
|
|
"UPDATE downloads SET status='failed',gid='old-gid',error='network error' WHERE id=?",
|
|
(task["id"],),
|
|
)
|
|
conn.execute(
|
|
"UPDATE jobs SET status='running',started_at=?,lease_owner='test:download' WHERE id=?",
|
|
(utcnow(), old_job_id),
|
|
)
|
|
discarded = []
|
|
monkeypatch.setattr(
|
|
downloads,
|
|
"_discard_gid",
|
|
lambda gid, *, active: discarded.append((gid, active)),
|
|
)
|
|
|
|
retried = downloads.retry(task["id"])
|
|
|
|
assert retried["status"] == "queued"
|
|
assert retried["job_id"] != old_job_id
|
|
assert discarded == [("old-gid", False)]
|
|
with services.db.read() as conn:
|
|
old_job = conn.execute(
|
|
"SELECT status,dedupe_key FROM jobs WHERE id=?",
|
|
(old_job_id,),
|
|
).fetchone()
|
|
new_job = conn.execute(
|
|
"SELECT status,dedupe_key FROM jobs WHERE id=?",
|
|
(retried["job_id"],),
|
|
).fetchone()
|
|
assert old_job["status"] == "running"
|
|
assert old_job["dedupe_key"].endswith(f":finished:{old_job_id}")
|
|
assert dict(new_job) == {"status": "queued", "dedupe_key": f"download:{task['id']}"}
|
|
|
|
|
|
def test_aria2_retry_and_delete_ignore_rpc_transport_failure(tmp_path: Path, monkeypatch):
|
|
app, _, source_id, headers = _app(tmp_path)
|
|
services = app.state.services
|
|
services.storage.set_writable(source_id, True)
|
|
downloads = services.downloads
|
|
retry_task = downloads.create("https://example.test/retry.mp4", source_id)
|
|
delete_task = downloads.create("https://example.test/delete.mp4", source_id)
|
|
with services.db.transaction() as conn:
|
|
for task in (retry_task, delete_task):
|
|
conn.execute(
|
|
"UPDATE downloads SET status='failed',gid=?,error='network error' WHERE id=?",
|
|
(f"gid-{task['id']}", task["id"]),
|
|
)
|
|
conn.execute(
|
|
"UPDATE jobs SET status='failed',finished_at=?,error='network error' WHERE id=?",
|
|
(utcnow(), task["job_id"]),
|
|
)
|
|
|
|
monkeypatch.setattr(downloads, "_start", lambda: None)
|
|
|
|
def disconnected(*_args, **_kwargs):
|
|
raise httpx.ConnectError("aria2 disconnected")
|
|
|
|
monkeypatch.setattr(downloads, "_rpc", disconnected)
|
|
|
|
async def scenario():
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
retried = await client.post(
|
|
f"/api/v1/downloads/{retry_task['id']}/retry",
|
|
headers=headers,
|
|
)
|
|
assert retried.status_code == 202
|
|
assert retried.json()["status"] == "queued"
|
|
deleted = await client.delete(
|
|
f"/api/v1/downloads/{delete_task['id']}",
|
|
headers=headers,
|
|
)
|
|
assert deleted.status_code == 204
|
|
missing = await client.delete(
|
|
f"/api/v1/downloads/{delete_task['id']}",
|
|
headers=headers,
|
|
)
|
|
assert missing.status_code == 404
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_aria2_uses_result_cleanup_for_terminal_gid_and_force_remove_for_active_gid(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
app, _, source_id, _ = _app(tmp_path)
|
|
services = app.state.services
|
|
services.storage.set_writable(source_id, True)
|
|
downloads = services.downloads
|
|
terminal = downloads.create("https://example.test/terminal.mp4", source_id)
|
|
active = downloads.create("https://example.test/active.mp4", source_id)
|
|
with services.db.transaction() as conn:
|
|
conn.execute(
|
|
"UPDATE downloads SET status='completed',gid='terminal-gid' WHERE id=?",
|
|
(terminal["id"],),
|
|
)
|
|
conn.execute(
|
|
"UPDATE jobs SET status='completed',finished_at=? WHERE id=?",
|
|
(utcnow(), terminal["job_id"]),
|
|
)
|
|
conn.execute(
|
|
"UPDATE downloads SET status='downloading',gid='active-gid' WHERE id=?",
|
|
(active["id"],),
|
|
)
|
|
calls = []
|
|
monkeypatch.setattr(downloads, "_start", lambda: None)
|
|
monkeypatch.setattr(downloads, "_rpc", lambda method, parameters=None: calls.append((method, parameters)))
|
|
|
|
downloads.delete(terminal["id"])
|
|
downloads.delete(active["id"])
|
|
|
|
assert calls == [
|
|
("aria2.removeDownloadResult", ["terminal-gid"]),
|
|
("aria2.forceRemove", ["active-gid"]),
|
|
("aria2.removeDownloadResult", ["active-gid"]),
|
|
]
|
|
|
|
|
|
def test_job_queue_prioritizes_user_visible_work_and_cancels_video_batches(tmp_path: Path):
|
|
app, _, _, _ = _app(tmp_path)
|
|
jobs = app.state.services.jobs
|
|
jobs.governor = None
|
|
low = jobs.enqueue("transcribe_audio", {"video_id": "video-a"}, dedupe_key="audio:video-a")
|
|
batch = jobs.enqueue("suggest_tags", {"video_ids": ["video-b", "video-c"]}, dedupe_key="tags:batch")
|
|
scan = jobs.enqueue("scan_source", {"source_id": "source"})
|
|
upload = jobs.enqueue("transfer_upload", {"upload_id": "upload"})
|
|
model = jobs.enqueue("install_models", {"component": "audio"})
|
|
parse = jobs.enqueue("index_video", {"video_id": "video-d"})
|
|
|
|
assert jobs.cancel_for_video("video-c") == [batch]
|
|
with app.state.services.db.read() as conn:
|
|
assert conn.execute("SELECT status FROM jobs WHERE id=?", (batch,)).fetchone()[0] == "cancelled"
|
|
|
|
order = []
|
|
for _ in range(5):
|
|
claimed = jobs._claim()
|
|
assert claimed is not None
|
|
order.append(claimed[0])
|
|
with app.state.services.db.transaction() as conn:
|
|
conn.execute("UPDATE jobs SET status='completed' WHERE id=?", (claimed[0],))
|
|
|
|
assert order == [model, upload, scan, parse, low]
|
|
|
|
|
|
def test_deduplicated_queued_audio_job_is_promoted_without_restarting_running_work(tmp_path: Path):
|
|
app, _, _, _ = _app(tmp_path)
|
|
jobs = app.state.services.jobs
|
|
queued = jobs.enqueue(
|
|
"transcribe_audio", {"video_id": "video-a"}, dedupe_key="audio:video-a", priority=30
|
|
)
|
|
assert jobs.enqueue(
|
|
"transcribe_audio", {"video_id": "video-a"}, dedupe_key="audio:video-a", priority=0
|
|
) == queued
|
|
with app.state.services.db.read() as conn:
|
|
assert conn.execute("SELECT priority FROM jobs WHERE id=?", (queued,)).fetchone()[0] == 0
|
|
|
|
with app.state.services.db.transaction() as conn:
|
|
conn.execute("UPDATE jobs SET status='running',priority=20 WHERE id=?", (queued,))
|
|
assert jobs.enqueue(
|
|
"transcribe_audio", {"video_id": "video-a"}, dedupe_key="audio:video-a", priority=-5
|
|
) == queued
|
|
with app.state.services.db.read() as conn:
|
|
assert conn.execute("SELECT priority FROM jobs WHERE id=?", (queued,)).fetchone()[0] == 20
|
|
|
|
|
|
def test_current_video_audio_reuses_warm_cache_before_other_analysis(tmp_path: Path):
|
|
app, _, _, _ = _app(tmp_path)
|
|
jobs = app.state.services.jobs
|
|
jobs.governor = None
|
|
other_visual = jobs.enqueue("index_video", {"video_id": "video-b"})
|
|
warm_audio = jobs.enqueue("transcribe_audio", {"video_id": "video-a"}, priority=0)
|
|
|
|
claimed = jobs._claim(lane="ai")
|
|
|
|
assert claimed and claimed[0] == warm_audio
|
|
with app.state.services.db.read() as conn:
|
|
assert conn.execute("SELECT status FROM jobs WHERE id=?", (other_visual,)).fetchone()[0] == "queued"
|
|
|
|
|
|
def test_audio_jobs_yield_to_queued_model_maintenance(tmp_path: Path):
|
|
app, _, _, _ = _app(tmp_path)
|
|
jobs = app.state.services.jobs
|
|
audio = jobs.enqueue("transcribe_audio", {"video_id": "video-a"}, dedupe_key="audio:video-a")
|
|
jobs.enqueue("install_models", {"component": "audio"}, dedupe_key="install-model:audio")
|
|
with app.state.services.db.transaction() as conn:
|
|
conn.execute("UPDATE jobs SET status='running' WHERE id=?", (audio,))
|
|
|
|
class Governor:
|
|
def wait_sync(self, **_kwargs):
|
|
raise AssertionError("audio job should yield before waiting under resource pressure")
|
|
|
|
jobs.governor = Governor()
|
|
try:
|
|
jobs.checkpoint(audio)
|
|
except JobCancelled as exc:
|
|
assert "模型安装优先" in str(exc)
|
|
else:
|
|
raise AssertionError("audio job must yield to queued model installation")
|
|
|
|
|
|
def test_cancel_by_kind_cancels_queued_and_requests_running_jobs(tmp_path: Path):
|
|
app, _, _, _ = _app(tmp_path)
|
|
jobs = app.state.services.jobs
|
|
queued = jobs.enqueue("transcribe_audio", {"video_id": "video-a"})
|
|
running = jobs.enqueue("transcribe_audio", {"video_id": "video-b"})
|
|
other = jobs.enqueue("index_video", {"video_id": "video-c"})
|
|
with app.state.services.db.transaction() as conn:
|
|
conn.execute("UPDATE jobs SET status='running' WHERE id=?", (running,))
|
|
|
|
assert jobs.cancel_by_kind("transcribe_audio") == [queued, running]
|
|
|
|
with app.state.services.db.read() as conn:
|
|
rows = {
|
|
row["id"]: dict(row)
|
|
for row in conn.execute(
|
|
"SELECT id,status,cancel_requested,message FROM jobs WHERE id IN (?,?,?)",
|
|
(queued, running, other),
|
|
).fetchall()
|
|
}
|
|
assert rows[queued]["status"] == "cancelled"
|
|
assert rows[running]["status"] == "running"
|
|
assert rows[running]["cancel_requested"] == 1
|
|
assert rows[other]["status"] == "queued"
|
|
|
|
|
|
def test_running_job_cancellation_wins_over_failure_retry_and_success(tmp_path: Path):
|
|
database = sqlite3.connect(tmp_path / "job-finalization.sqlite3")
|
|
database.row_factory = sqlite3.Row
|
|
database.execute(
|
|
"CREATE TABLE jobs(id TEXT PRIMARY KEY,status TEXT,cancel_requested INTEGER,error TEXT,"
|
|
"finished_at TEXT,pause_reason TEXT,lease_owner TEXT,heartbeat_at TEXT,message TEXT,"
|
|
"run_after TEXT,started_at TEXT,progress REAL)"
|
|
)
|
|
for prefix, cancel_requested in (("cancel", 1), ("normal", 0)):
|
|
for outcome in ("failed", "retry", "completed"):
|
|
database.execute(
|
|
"INSERT INTO jobs(id,status,cancel_requested,message,progress) VALUES(?,?,?,?,?)",
|
|
(f"{prefix}-{outcome}", "running", cancel_requested, "处理中", 0.5),
|
|
)
|
|
|
|
JobQueue._finish_failed(database, "cancel-failed", "视频不存在")
|
|
JobQueue._finish_retry(database, "cancel-retry", "2099-01-01T00:00:00+00:00", "稍后重试")
|
|
JobQueue._finish_completed(database, "cancel-completed")
|
|
JobQueue._finish_failed(database, "normal-failed", "真实失败")
|
|
JobQueue._finish_retry(database, "normal-retry", "2099-01-01T00:00:00+00:00", "稍后重试")
|
|
JobQueue._finish_completed(database, "normal-completed")
|
|
|
|
rows = {row["id"]: dict(row) for row in database.execute("SELECT * FROM jobs").fetchall()}
|
|
for outcome in ("failed", "retry", "completed"):
|
|
row = rows[f"cancel-{outcome}"]
|
|
assert row["status"] == "cancelled"
|
|
assert row["error"] is None
|
|
assert row["message"] == "已取消"
|
|
assert rows["cancel-completed"]["progress"] == 0.5
|
|
assert rows["normal-failed"]["status"] == "failed"
|
|
assert rows["normal-failed"]["error"] == "真实失败"
|
|
assert rows["normal-retry"]["status"] == "queued"
|
|
assert rows["normal-retry"]["finished_at"] is None
|
|
assert rows["normal-retry"]["message"] == "稍后重试"
|
|
assert rows["normal-completed"]["status"] == "completed"
|
|
assert rows["normal-completed"]["progress"] == 1
|
|
|
|
|
|
def test_video_jobs_can_be_cancelled_inside_the_video_visibility_transaction(tmp_path: Path):
|
|
database = sqlite3.connect(tmp_path / "atomic-video-cancel.sqlite3")
|
|
database.row_factory = sqlite3.Row
|
|
database.execute(
|
|
"CREATE TABLE jobs(id TEXT PRIMARY KEY,status TEXT,cancel_requested INTEGER DEFAULT 0,"
|
|
"payload_json TEXT,finished_at TEXT,message TEXT)"
|
|
)
|
|
database.execute(
|
|
"CREATE TABLE job_resources(job_id TEXT,resource_type TEXT,resource_id TEXT,"
|
|
"PRIMARY KEY(job_id,resource_type,resource_id))"
|
|
)
|
|
rows = (
|
|
("queued", "queued", {"video_id": "video-a"}),
|
|
("running", "running", {"video_ids": ["video-a", "video-b"]}),
|
|
("other", "queued", {"video_id": "video-c"}),
|
|
)
|
|
database.executemany(
|
|
"INSERT INTO jobs(id,status,payload_json) VALUES(?,?,?)",
|
|
[(job_id, status, json.dumps(payload)) for job_id, status, payload in rows],
|
|
)
|
|
database.executemany(
|
|
"INSERT INTO job_resources(job_id,resource_type,resource_id) VALUES(?,?,?)",
|
|
[
|
|
("queued", "video", "video-a"),
|
|
("running", "video", "video-a"),
|
|
("running", "video", "video-b"),
|
|
("other", "video", "video-c"),
|
|
],
|
|
)
|
|
|
|
jobs = JobQueue(SimpleNamespace(), governor=None)
|
|
assert jobs.cancel_for_video("video-a", connection=database) == ["queued", "running"]
|
|
|
|
states = {
|
|
row["id"]: dict(row)
|
|
for row in database.execute("SELECT id,status,cancel_requested,message FROM jobs").fetchall()
|
|
}
|
|
assert states["queued"] == {
|
|
"id": "queued",
|
|
"status": "cancelled",
|
|
"cancel_requested": 1,
|
|
"message": "已取消",
|
|
}
|
|
assert states["running"] == {
|
|
"id": "running",
|
|
"status": "running",
|
|
"cancel_requested": 1,
|
|
"message": "正在安全停止",
|
|
}
|
|
assert states["other"]["status"] == "queued"
|
|
assert states["other"]["cancel_requested"] == 0
|
|
|
|
|
|
def test_video_delete_moves_source_to_recoverable_trash(tmp_path: Path):
|
|
app, media, source_id, headers = _app(tmp_path)
|
|
app.state.services.storage.set_writable(source_id, True)
|
|
source = media / "movie.mp4"
|
|
source.write_bytes(b"video")
|
|
app.state.services.scanner.refresh_path("refresh", source_id, "movie.mp4")
|
|
with app.state.services.db.read() as conn:
|
|
video_id = conn.execute("SELECT id FROM videos WHERE source_key='movie.mp4'").fetchone()["id"]
|
|
thumbnail = app.state.services.settings.thumbnails_dir / "delete-frame.jpg"
|
|
thumbnail.write_bytes(b"jpeg")
|
|
now = "2026-07-31T00:00:00+00:00"
|
|
with app.state.services.db.transaction() as conn:
|
|
conn.execute(
|
|
"INSERT INTO frames(id,video_id,timestamp_ms,segment_start_ms,segment_end_ms,thumbnail_path,"
|
|
"vector_blob,created_at) VALUES('delete-frame',?,0,0,1000,?,?,?)",
|
|
(video_id, str(thumbnail), pack_vector([1.0, 0.0]), now),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO text_entries(id,video_id,frame_id,kind,start_ms,end_ms,raw_text,tokens,created_at) "
|
|
"VALUES('delete-text',?,'delete-frame','ocr',0,1000,'删除文字','删除 文字',?)",
|
|
(video_id, now),
|
|
)
|
|
conn.execute("INSERT INTO text_fts(entry_id,tokens) VALUES('delete-text','删除 文字')")
|
|
conn.execute(
|
|
"INSERT INTO people(id,name,normalized_name,centroid_blob,face_count,created_at,updated_at) "
|
|
"VALUES('delete-person','未命名人物','','',1,?,?)",
|
|
(now, now),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO faces(id,frame_id,video_id,person_id,bbox_json,confidence,thumbnail_path,"
|
|
"vector_blob,created_at) VALUES('delete-face','delete-frame',?,'delete-person','[]',0.9,?,?,?)",
|
|
(video_id, str(thumbnail), pack_vector([1.0, 0.0]), now),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO video_state(video_id,liked,favorited,progress_ms,completed,updated_at) "
|
|
"VALUES(?,1,1,500,0,?)",
|
|
(video_id, now),
|
|
)
|
|
|
|
async def delete_request():
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.delete(f"/api/v1/videos/{video_id}", headers=headers)
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["source_deleted"] is True
|
|
assert payload["trash_id"]
|
|
assert payload["cancelled_jobs"] >= 1
|
|
assert "job_id" not in payload
|
|
return payload
|
|
|
|
payload = asyncio.run(delete_request())
|
|
assert not source.exists()
|
|
with app.state.services.db.read() as conn:
|
|
assert conn.execute("SELECT 1 FROM videos WHERE id=?", (video_id,)).fetchone() is None
|
|
assert conn.execute("SELECT 1 FROM text_fts WHERE entry_id='delete-text'").fetchone() is None
|
|
assert conn.execute("SELECT 1 FROM people WHERE id='delete-person'").fetchone() is None
|
|
trash = conn.execute("SELECT id,trash_key FROM trash WHERE id=?", (payload["trash_id"],)).fetchone()
|
|
tombstone = conn.execute(
|
|
"SELECT source_deleted FROM video_tombstones WHERE source_id=? AND source_key='movie.mp4'",
|
|
(source_id,),
|
|
).fetchone()
|
|
assert tombstone["source_deleted"] == 1
|
|
assert not thumbnail.exists()
|
|
assert (media / trash["trash_key"]).read_bytes() == b"video"
|
|
|
|
async def restore_request():
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
restored = await client.post(f"/api/v1/trash/{trash['id']}/restore", headers=headers)
|
|
assert restored.status_code == 200
|
|
assert source.read_bytes() == b"video"
|
|
assert restored.json()["job_id"]
|
|
|
|
asyncio.run(restore_request())
|
|
with app.state.services.db.read() as conn:
|
|
assert conn.execute(
|
|
"SELECT 1 FROM video_tombstones WHERE source_id=? AND source_key='movie.mp4'",
|
|
(source_id,),
|
|
).fetchone() is None
|
|
|
|
|
|
def test_video_delete_preserves_people_that_still_have_faces(tmp_path: Path):
|
|
app, media, source_id, headers = _app(tmp_path)
|
|
app.state.services.storage.set_writable(source_id, True)
|
|
for name in ("one.mp4", "two.mp4"):
|
|
(media / name).write_bytes(b"video")
|
|
app.state.services.scanner.refresh_path(f"refresh-{name}", source_id, name)
|
|
now = "2026-07-31T00:00:00+00:00"
|
|
with app.state.services.db.transaction() as conn:
|
|
one = conn.execute("SELECT id FROM videos WHERE source_key='one.mp4'").fetchone()[0]
|
|
two = conn.execute("SELECT id FROM videos WHERE source_key='two.mp4'").fetchone()[0]
|
|
for frame_id, video_id in (("frame-one", one), ("frame-two", two)):
|
|
thumbnail = app.state.services.settings.thumbnails_dir / f"{frame_id}.jpg"
|
|
conn.execute(
|
|
"INSERT INTO frames(id,video_id,timestamp_ms,segment_start_ms,segment_end_ms,thumbnail_path,"
|
|
"vector_blob,created_at) VALUES(?,?,0,0,1000,?,?,?)",
|
|
(frame_id, video_id, str(thumbnail), pack_vector([1, 0]), now),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO people(id,name,normalized_name,is_named,centroid_blob,face_count,created_at,updated_at) "
|
|
"VALUES('person','演员','演员',1,?,2,?,?)",
|
|
(pack_vector([1, 0]), now, now),
|
|
)
|
|
for face_id, frame_id, video_id, vector in (
|
|
("face-one", "frame-one", one, [1.0, 0.0]),
|
|
("face-two", "frame-two", two, [0.0, 1.0]),
|
|
):
|
|
conn.execute(
|
|
"INSERT INTO faces(id,frame_id,video_id,person_id,bbox_json,confidence,thumbnail_path,"
|
|
"vector_blob,created_at) VALUES(?,?,?,'person','[]',0.9,NULL,?,?)",
|
|
(face_id, frame_id, video_id, pack_vector(vector), now),
|
|
)
|
|
|
|
async def scenario():
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.delete(f"/api/v1/videos/{one}", headers=headers)
|
|
assert response.status_code == 200
|
|
hidden = await client.patch(
|
|
"/api/v1/people/person",
|
|
headers=headers,
|
|
json={"name": "演员", "hidden": True},
|
|
)
|
|
assert hidden.status_code == 200
|
|
|
|
asyncio.run(scenario())
|
|
with app.state.services.db.read() as conn:
|
|
person = conn.execute("SELECT face_count,centroid_blob,hidden FROM people WHERE id='person'").fetchone()
|
|
remaining = conn.execute("SELECT count(*) FROM faces WHERE person_id='person'").fetchone()[0]
|
|
assert person["face_count"] == 1
|
|
assert person["hidden"] == 1
|
|
assert remaining == 1
|
|
|
|
|
|
def test_video_delete_rolls_back_visibility_when_trash_move_fails(tmp_path: Path, monkeypatch):
|
|
app, media, source_id, headers = _app(tmp_path)
|
|
app.state.services.storage.set_writable(source_id, True)
|
|
source = media / "movie.mp4"
|
|
source.write_bytes(b"video")
|
|
app.state.services.scanner.refresh_path("refresh", source_id, "movie.mp4")
|
|
with app.state.services.db.read() as conn:
|
|
video_id = conn.execute("SELECT id FROM videos WHERE source_key='movie.mp4'").fetchone()[0]
|
|
|
|
def fail_trash(_source_id, _key):
|
|
raise OSError("remote offline")
|
|
|
|
monkeypatch.setattr(app.state.services.storage, "trash", fail_trash)
|
|
monkeypatch.setattr(
|
|
app.state.services.jobs,
|
|
"cancel_for_video",
|
|
lambda *_args, **_kwargs: (_ for _ in ()).throw(
|
|
AssertionError("jobs must not be cancelled before trash succeeds")
|
|
),
|
|
)
|
|
|
|
async def scenario():
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.delete(f"/api/v1/videos/{video_id}", headers=headers)
|
|
assert response.status_code == 502
|
|
|
|
asyncio.run(scenario())
|
|
with app.state.services.db.read() as conn:
|
|
row = conn.execute("SELECT available,status FROM videos WHERE id=?", (video_id,)).fetchone()
|
|
tombstone = conn.execute(
|
|
"SELECT 1 FROM video_tombstones WHERE source_id=? AND source_key='movie.mp4'",
|
|
(source_id,),
|
|
).fetchone()
|
|
assert dict(row) == {"available": 1, "status": "pending"}
|
|
assert tombstone is None
|
|
assert source.is_file()
|
|
|
|
|
|
def test_video_delete_restores_cancelled_jobs_when_later_stage_fails(tmp_path: Path, monkeypatch):
|
|
app, media, source_id, headers = _app(tmp_path)
|
|
app.state.services.storage.set_writable(source_id, True)
|
|
source = media / "movie.mp4"
|
|
source.write_bytes(b"video")
|
|
app.state.services.scanner.refresh_path("refresh", source_id, "movie.mp4")
|
|
with app.state.services.db.read() as conn:
|
|
video_id = conn.execute("SELECT id FROM videos WHERE source_key='movie.mp4'").fetchone()[0]
|
|
job_id = conn.execute(
|
|
"SELECT j.id FROM jobs j JOIN job_resources r ON r.job_id=j.id "
|
|
"WHERE r.resource_type='video' AND r.resource_id=? AND j.status='queued' LIMIT 1",
|
|
(video_id,),
|
|
).fetchone()[0]
|
|
|
|
monkeypatch.setattr(
|
|
app.state.services.vectors,
|
|
"delete_video",
|
|
lambda _video_id: (_ for _ in ()).throw(RuntimeError("vector backend unavailable")),
|
|
)
|
|
|
|
async def scenario():
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.delete(f"/api/v1/videos/{video_id}", headers=headers)
|
|
assert response.status_code == 409
|
|
|
|
asyncio.run(scenario())
|
|
with app.state.services.db.read() as conn:
|
|
video = conn.execute("SELECT available,status FROM videos WHERE id=?", (video_id,)).fetchone()
|
|
job = conn.execute(
|
|
"SELECT status,cancel_requested,finished_at FROM jobs WHERE id=?", (job_id,)
|
|
).fetchone()
|
|
trash = conn.execute("SELECT 1 FROM trash WHERE original_key='movie.mp4'").fetchone()
|
|
assert dict(video) == {"available": 1, "status": "pending"}
|
|
assert dict(job) == {"status": "queued", "cancel_requested": 0, "finished_at": None}
|
|
assert trash is None
|
|
assert source.read_bytes() == b"video"
|
|
|
|
|
|
def test_library_only_delete_tombstone_prevents_scan_resurrection(tmp_path: Path):
|
|
app, media, source_id, headers = _app(tmp_path)
|
|
source = media / "ghost.mp4"
|
|
source.write_bytes(b"video remains on remote storage")
|
|
app.state.services.scanner.refresh_path("refresh", source_id, "ghost.mp4")
|
|
with app.state.services.db.read() as conn:
|
|
video_id = conn.execute("SELECT id FROM videos WHERE source_key='ghost.mp4'").fetchone()[0]
|
|
|
|
async def scenario():
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.delete(
|
|
f"/api/v1/videos/{video_id}?delete_source=false",
|
|
headers=headers,
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["source_deleted"] is False
|
|
assert response.json()["trash_id"] is None
|
|
|
|
asyncio.run(scenario())
|
|
assert source.is_file()
|
|
app.state.services.scanner.scan("scan-after-delete", source_id)
|
|
with app.state.services.db.read() as conn:
|
|
assert conn.execute("SELECT 1 FROM videos WHERE source_key='ghost.mp4'").fetchone() is None
|
|
tombstone = conn.execute(
|
|
"SELECT video_id,source_deleted FROM video_tombstones "
|
|
"WHERE source_id=? AND source_key='ghost.mp4'",
|
|
(source_id,),
|
|
).fetchone()
|
|
assert dict(tombstone) == {"video_id": video_id, "source_deleted": 0}
|
|
|
|
# An explicit path refresh represents a deliberate re-upload/restore and
|
|
# is the supported way to clear the durable deletion marker.
|
|
app.state.services.scanner.refresh_path("explicit-refresh", source_id, "ghost.mp4")
|
|
with app.state.services.db.read() as conn:
|
|
assert conn.execute("SELECT 1 FROM videos WHERE source_key='ghost.mp4'").fetchone()
|
|
assert conn.execute(
|
|
"SELECT 1 FROM video_tombstones WHERE source_id=? AND source_key='ghost.mp4'",
|
|
(source_id,),
|
|
).fetchone() is None
|
|
|
|
|
|
def test_stale_visible_deleting_videos_are_hidden_on_recovery(tmp_path: Path):
|
|
app, media, source_id, _ = _app(tmp_path)
|
|
source = media / "movie.mp4"
|
|
source.write_bytes(b"video")
|
|
app.state.services.scanner.refresh_path("refresh", source_id, "movie.mp4")
|
|
with app.state.services.db.transaction() as conn:
|
|
video_id = conn.execute("SELECT id FROM videos WHERE source_key='movie.mp4'").fetchone()[0]
|
|
conn.execute("UPDATE videos SET status='deleting',available=1 WHERE id=?", (video_id,))
|
|
|
|
assert app.state.services.deletions.recover_stale_deleting() == 1
|
|
|
|
with app.state.services.db.read() as conn:
|
|
row = conn.execute("SELECT available,status,error FROM videos WHERE id=?", (video_id,)).fetchone()
|
|
tombstone = conn.execute(
|
|
"SELECT video_id FROM video_tombstones WHERE source_id=? AND source_key='movie.mp4'",
|
|
(source_id,),
|
|
).fetchone()
|
|
assert row["available"] == 0
|
|
assert row["status"] == "offline"
|
|
assert "删除任务未完成" in row["error"]
|
|
assert tombstone["video_id"] == video_id
|
|
|
|
|
|
def test_legacy_delete_jobs_are_migrated_synchronously(tmp_path: Path):
|
|
app, media, source_id, _ = _app(tmp_path)
|
|
app.state.services.storage.set_writable(source_id, True)
|
|
source = media / "movie.mp4"
|
|
source.write_bytes(b"video")
|
|
app.state.services.scanner.refresh_path("refresh", source_id, "movie.mp4")
|
|
with app.state.services.db.read() as conn:
|
|
video_id = conn.execute("SELECT id FROM videos WHERE source_key='movie.mp4'").fetchone()[0]
|
|
legacy = app.state.services.jobs.enqueue(
|
|
"delete_video", {"video_id": video_id}, dedupe_key=f"delete:{video_id}"
|
|
)
|
|
|
|
app.state.services._migrate_legacy_deletes()
|
|
|
|
with app.state.services.db.read() as conn:
|
|
job = conn.execute("SELECT status,progress,message FROM jobs WHERE id=?", (legacy,)).fetchone()
|
|
trash = conn.execute("SELECT trash_key FROM trash WHERE original_key='movie.mp4'").fetchone()
|
|
video = conn.execute("SELECT 1 FROM videos WHERE id=?", (video_id,)).fetchone()
|
|
assert job["status"] == "completed"
|
|
assert job["progress"] == 1
|
|
assert "视频已移入回收站" in job["message"]
|
|
assert trash is not None
|
|
assert video is None
|
|
assert (media / trash["trash_key"]).is_file()
|
|
|
|
|
|
def test_video_delete_rejects_read_only_source(tmp_path: Path):
|
|
app, media, source_id, headers = _app(tmp_path)
|
|
(media / "movie.mp4").write_bytes(b"video")
|
|
app.state.services.scanner.refresh_path("refresh", source_id, "movie.mp4")
|
|
with app.state.services.db.read() as conn:
|
|
video_id = conn.execute("SELECT id FROM videos").fetchone()["id"]
|
|
|
|
async def scenario():
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.delete(f"/api/v1/videos/{video_id}", headers=headers)
|
|
assert response.status_code == 403
|
|
assert (media / "movie.mp4").is_file()
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_visual_duplicate_cleanup_keeps_safetensors_and_updates_manifest(tmp_path: Path):
|
|
app, _, _, _ = _app(tmp_path)
|
|
manager = app.state.services.models
|
|
visual = manager.settings.models_dir / "visual" / "image"
|
|
visual.mkdir(parents=True)
|
|
(visual / "model.safetensors").write_bytes(b"safe")
|
|
duplicate = visual / "pytorch_model.bin"
|
|
duplicate.write_bytes(b"duplicate-weights")
|
|
manifest = {
|
|
"format_version": 2,
|
|
"version": "test",
|
|
"components": {"visual": {"version": "test"}},
|
|
"files": {
|
|
"visual/image/model.safetensors": "safe-hash",
|
|
"visual/image/pytorch_model.bin": "duplicate-hash",
|
|
},
|
|
}
|
|
(manager.settings.models_dir / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
|
|
result = manager.cleanup_visual_duplicates()
|
|
assert result["removed"] == 1
|
|
assert result["freed_bytes"] == len(b"duplicate-weights")
|
|
assert (visual / "model.safetensors").is_file()
|
|
assert not duplicate.exists()
|
|
updated = json.loads((manager.settings.models_dir / "manifest.json").read_text(encoding="utf-8"))
|
|
assert "visual/image/pytorch_model.bin" not in updated["files"]
|