805 lines
34 KiB
Python
805 lines
34 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import json
|
|
import sys
|
|
import types
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
from imagefind import api as api_module
|
|
from imagefind.config import Settings
|
|
from imagefind.database import DatabaseTransientError, utcnow
|
|
from imagefind.main import create_app
|
|
|
|
|
|
def make_app(tmp_path: Path):
|
|
settings = Settings(data_dir=tmp_path / "data", embedding_backend="hash", upload_reserve_gb=0)
|
|
settings.prepare()
|
|
app = create_app(settings)
|
|
app.state.services.auth.setup("original administrator password")
|
|
_, token = app.state.services.auth.create_api_token("test")
|
|
return app, {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def seed_videos(app) -> None:
|
|
now = utcnow()
|
|
with app.state.services.db.transaction() as conn:
|
|
conn.execute(
|
|
"INSERT INTO sources(id,kind,name,config_json,created_at,updated_at) VALUES(?,?,?,?,?,?)",
|
|
("source", "local", "Local", json.dumps({"path": "/media"}), now, now),
|
|
)
|
|
for index in range(3):
|
|
conn.execute(
|
|
"INSERT INTO videos(id,source_id,source_key,display_name,location,size_bytes,fingerprint,"
|
|
"duration_ms,status,available,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,'indexed',1,?,?)",
|
|
(
|
|
f"video-{index}",
|
|
"source",
|
|
f"video-{index}.mp4",
|
|
f"Video {index}",
|
|
f"/media/video-{index}.mp4",
|
|
100 + index,
|
|
f"fingerprint-{index}",
|
|
(index + 1) * 10_000,
|
|
now,
|
|
now,
|
|
),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO video_metadata(video_id,series,updated_at) VALUES('video-0','Series A',?),"
|
|
"('video-1','series a',?)",
|
|
(now, now),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO video_state(video_id,liked,favorited,progress_ms,completed,last_played_at,updated_at) "
|
|
"VALUES('video-0',1,0,3000,0,?,?),('video-1',0,1,10000,1,?,?)",
|
|
(now, now, now, now),
|
|
)
|
|
|
|
|
|
def test_video_list_batches_tag_items_in_one_read_connection(tmp_path: Path):
|
|
app, headers = make_app(tmp_path)
|
|
seed_videos(app)
|
|
service = app.state.services
|
|
now = utcnow()
|
|
with service.db.transaction() as conn:
|
|
conn.execute(
|
|
"INSERT INTO tag_groups(id,name,selection_mode,sort_order,created_at,updated_at) "
|
|
"VALUES('batch-group','类型','multi',1,?,?)",
|
|
(now, now),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO tags(id,group_id,name,created_at,updated_at) "
|
|
"VALUES('batch-tag','batch-group','测试标签',?,?)",
|
|
(now, now),
|
|
)
|
|
conn.executemany(
|
|
"INSERT INTO video_tags(video_id,tag_id) VALUES(?,'batch-tag')",
|
|
[(f"video-{index}",) for index in range(3)],
|
|
)
|
|
|
|
original_read = service.db.read
|
|
read_calls = []
|
|
|
|
@contextlib.contextmanager
|
|
def counted_read():
|
|
read_calls.append(1)
|
|
with original_read() as connection:
|
|
yield connection
|
|
|
|
service.db.read = counted_read
|
|
|
|
async def scenario() -> None:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/videos?limit=500", headers=headers)
|
|
assert response.status_code == 200
|
|
assert len(response.json()) == 3
|
|
assert all(item["tag_items"][0]["name"] == "测试标签" for item in response.json())
|
|
|
|
asyncio.run(scenario())
|
|
# The freshly issued API token is served by the bounded authentication
|
|
# cache, while one WAL read batches the complete video list/tag payload.
|
|
assert len(read_calls) == 1
|
|
|
|
|
|
def test_home_feed_groups_recent_collections_categories_and_unorganized(tmp_path: Path):
|
|
app, headers = make_app(tmp_path)
|
|
seed_videos(app)
|
|
now = utcnow()
|
|
with app.state.services.db.transaction() as conn:
|
|
conn.execute(
|
|
"INSERT INTO tag_groups(id,name,selection_mode,sort_order,created_at,updated_at) "
|
|
"VALUES('home-group','类型','multi',1,?,?)",
|
|
(now, now),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO tags(id,group_id,name,created_at,updated_at) VALUES"
|
|
"('home-tag-a','home-group','剧情',?,?),('home-tag-b','home-group','纪录',?,?),"
|
|
"('home-tag-small','home-group','单片',?,?)",
|
|
(now, now, now, now, now, now),
|
|
)
|
|
conn.executemany(
|
|
"INSERT INTO video_tags(video_id,tag_id) VALUES(?,?)",
|
|
[
|
|
("video-0", "home-tag-a"),
|
|
("video-1", "home-tag-a"),
|
|
("video-0", "home-tag-b"),
|
|
("video-1", "home-tag-b"),
|
|
("video-0", "home-tag-small"),
|
|
],
|
|
)
|
|
|
|
async def scenario() -> None:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
assert (await client.get("/api/v1/home")).status_code == 401
|
|
collection = await client.post(
|
|
"/api/v1/collections", headers=headers, json={"name": "首页合集", "video_ids": ["video-0"]}
|
|
)
|
|
assert collection.status_code == 201
|
|
response = await client.get("/api/v1/home?item_limit=4&tag_limit=1", headers=headers)
|
|
assert response.status_code == 200
|
|
feed = response.json()
|
|
assert feed["video_count"] == 3
|
|
assert [item["id"] for item in feed["recent"]["items"]] == ["video-0", "video-1", "video-2"]
|
|
assert feed["collections"]["items"][0]["name"] == "首页合集"
|
|
assert len(feed["categories"]) == 1
|
|
assert feed["categories"][0]["title"] == "剧情"
|
|
assert feed["categories"][0]["total"] == 2
|
|
assert feed["unorganized"]["total"] == 1
|
|
assert [item["id"] for item in feed["unorganized"]["items"]] == ["video-2"]
|
|
|
|
with app.state.services.db.transaction() as conn:
|
|
conn.execute("DELETE FROM video_tags")
|
|
conn.execute("DELETE FROM video_metadata")
|
|
conn.execute("DELETE FROM collection_items")
|
|
conn.execute("DELETE FROM collection_videos")
|
|
conn.execute("DELETE FROM collections")
|
|
deduplicated = (await client.get("/api/v1/home", headers=headers)).json()
|
|
assert deduplicated["categories"] == []
|
|
assert deduplicated["collections"]["items"] == []
|
|
assert deduplicated["unorganized"] is None
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_profile_validation_stats_filters_and_password_revocation(tmp_path: Path):
|
|
app, headers = make_app(tmp_path)
|
|
seed_videos(app)
|
|
|
|
async def scenario() -> None:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
profile = await client.get("/api/v1/profile", headers=headers)
|
|
assert profile.status_code == 200
|
|
assert profile.json()["nickname"] == "管理员"
|
|
assert profile.json()["avatar"]["text"] == "管"
|
|
assert profile.json()["avatar"]["accent"].startswith("#")
|
|
assert profile.json()["counts"] == {
|
|
"favorites": 1,
|
|
"likes": 1,
|
|
"history": 2,
|
|
"continue_watching": 1,
|
|
}
|
|
invalid = await client.patch("/api/v1/profile", headers=headers, json={"nickname": " "})
|
|
assert invalid.status_code == 400
|
|
changed = await client.patch("/api/v1/profile", headers=headers, json={"nickname": "NAS 管理员"})
|
|
assert changed.json()["avatar"]["text"] == "NA"
|
|
|
|
assert len((await client.get("/api/v1/videos?liked=true", headers=headers)).json()) == 1
|
|
assert len((await client.get("/api/v1/videos?favorite=true", headers=headers)).json()) == 1
|
|
assert len((await client.get("/api/v1/videos?continue_only=true", headers=headers)).json()) == 1
|
|
history = (await client.get("/api/v1/videos?played_only=true&sort=last_played", headers=headers)).json()
|
|
assert len(history) == 2
|
|
|
|
first = await client.post(
|
|
"/api/v1/auth/login", json={"password": "original administrator password"}
|
|
)
|
|
assert first.status_code == 200
|
|
old_cookie = first.cookies.get("imagefind_session")
|
|
wrong = await client.patch(
|
|
"/api/v1/profile/password",
|
|
headers=headers,
|
|
json={"current_password": "wrong", "new_password": "replacement administrator password"},
|
|
)
|
|
assert wrong.status_code == 400
|
|
changed_password = await client.patch(
|
|
"/api/v1/profile/password",
|
|
headers=headers,
|
|
json={
|
|
"current_password": "original administrator password",
|
|
"new_password": "replacement administrator password",
|
|
},
|
|
)
|
|
assert changed_password.status_code == 200
|
|
assert changed_password.json()["reauthenticate"] is True
|
|
client.cookies.set("imagefind_session", old_cookie)
|
|
revoked = await client.get("/api/v1/auth/me")
|
|
assert revoked.status_code == 401
|
|
assert (
|
|
await client.post("/api/v1/auth/login", json={"password": "replacement administrator password"})
|
|
).status_code == 200
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_bulk_video_state_actions_are_atomic_and_preserve_unrelated_state(tmp_path: Path):
|
|
app, headers = make_app(tmp_path)
|
|
seed_videos(app)
|
|
|
|
async def scenario() -> None:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
missing = await client.patch(
|
|
"/api/v1/videos/state/bulk",
|
|
headers=headers,
|
|
json={"video_ids": ["video-0", "missing"], "action": "unlike"},
|
|
)
|
|
assert missing.status_code == 404
|
|
assert len((await client.get("/api/v1/videos?liked=true", headers=headers)).json()) == 1
|
|
|
|
assert (
|
|
await client.patch(
|
|
"/api/v1/videos/state/bulk",
|
|
headers=headers,
|
|
json={"video_ids": ["video-0", "video-0"], "action": "unlike"},
|
|
)
|
|
).json() == {"updated": 1, "action": "unlike"}
|
|
await client.patch(
|
|
"/api/v1/videos/state/bulk",
|
|
headers=headers,
|
|
json={"video_ids": ["video-1"], "action": "unfavorite"},
|
|
)
|
|
await client.patch(
|
|
"/api/v1/videos/state/bulk",
|
|
headers=headers,
|
|
json={"video_ids": ["video-0", "video-1"], "action": "clear_history"},
|
|
)
|
|
assert (await client.get("/api/v1/videos?liked=true", headers=headers)).json() == []
|
|
assert (await client.get("/api/v1/videos?favorite=true", headers=headers)).json() == []
|
|
assert (await client.get("/api/v1/videos?played_only=true", headers=headers)).json() == []
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_jobs_support_stable_numbered_pagination_and_legacy_limit(tmp_path: Path):
|
|
app, headers = make_app(tmp_path)
|
|
diagnostic_job = ""
|
|
for index in range(23):
|
|
diagnostic_job = app.state.services.jobs.enqueue("scan_source", {"index": index})
|
|
app.state.services.jobs.set_diagnostics(
|
|
diagnostic_job,
|
|
{
|
|
"requested_device": "GPU.0",
|
|
"actual_device": "CPU",
|
|
"fallback_scope": "job",
|
|
"fallback_reason": "low_quality_result",
|
|
"private_value": "must not be exposed",
|
|
},
|
|
)
|
|
|
|
async def scenario() -> None:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
first = (await client.get("/api/v1/jobs?page=1&page_size=10", headers=headers)).json()
|
|
last = (await client.get("/api/v1/jobs?page=3&page_size=10", headers=headers)).json()
|
|
assert (first["page"], first["pages"], first["total"], len(first["items"])) == (1, 3, 23, 10)
|
|
assert (last["page"], len(last["items"])) == (3, 3)
|
|
assert not ({item["id"] for item in first["items"]} & {item["id"] for item in last["items"]})
|
|
diagnosed = next(item for item in first["items"] if item["id"] == diagnostic_job)
|
|
assert diagnosed["inference_diagnostics"] == {
|
|
"requested_device": "GPU.0",
|
|
"actual_device": "CPU",
|
|
"fallback_scope": "job",
|
|
"fallback_reason": "low_quality_result",
|
|
}
|
|
assert "diagnostics_json" not in diagnosed
|
|
legacy = (await client.get("/api/v1/jobs?limit=4", headers=headers)).json()
|
|
assert isinstance(legacy, list) and len(legacy) == 4
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_job_diagnostics_are_best_effort_on_transient_database_conflict(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
caplog,
|
|
):
|
|
app, _headers = make_app(tmp_path)
|
|
job_id = app.state.services.jobs.enqueue("scan_source", {"index": 1})
|
|
|
|
def conflict(*_args, **_kwargs):
|
|
raise DatabaseTransientError("temporary conflict")
|
|
|
|
monkeypatch.setattr(app.state.services.db, "write_with_retry", conflict)
|
|
app.state.services.jobs.set_diagnostics(job_id, {"actual_device": "GPU.0"})
|
|
|
|
assert "skipped diagnostics update" in caplog.text
|
|
|
|
|
|
def test_model_status_exposes_audio_circuit_breaker_state(tmp_path: Path):
|
|
app, headers = make_app(tmp_path)
|
|
accelerator = app.state.services.accelerator
|
|
accelerator.record_transient_failure("audio", "stall one", stage="inference_stall")
|
|
accelerator.record_transient_failure("audio", "stall two", stage="inference_stall")
|
|
accelerator.record_transient_failure("audio", "stall three", stage="inference_stall")
|
|
|
|
async def scenario() -> None:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/models", headers=headers)
|
|
assert response.status_code == 200
|
|
audio = response.json()["accelerator"]["components"]["audio"]
|
|
assert audio["circuit_state"] == "open"
|
|
assert audio["fallback_scope"] == "component"
|
|
assert audio["failure_count"] == 3
|
|
assert audio["retry_at"]
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_upload_list_returns_last_snapshot_when_refresh_exceeds_budget(tmp_path: Path, monkeypatch):
|
|
app, headers = make_app(tmp_path)
|
|
cached = {"id": "cached-upload", "status": "receiving", "stage": "receiving"}
|
|
app.state.services.uploads._list_cache = [cached]
|
|
monkeypatch.setattr(api_module, "API_READ_DEADLINE_SECONDS", 0.01)
|
|
|
|
async def delayed(function, /, *args, **kwargs):
|
|
await asyncio.sleep(0.05)
|
|
return function(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(api_module, "_background_api", delayed)
|
|
|
|
async def scenario() -> None:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/uploads", headers=headers)
|
|
assert response.status_code == 200
|
|
assert response.json() == [cached]
|
|
# Let the shielded refresh finish so the test loop closes cleanly.
|
|
await asyncio.sleep(0.06)
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_upload_page_returns_matching_snapshot_when_refresh_exceeds_budget(tmp_path: Path, monkeypatch):
|
|
app, headers = make_app(tmp_path)
|
|
cached = {
|
|
"items": [{"id": "cached-page-upload", "status": "completed", "stage": "done"}],
|
|
"status_items": [{"id": "cached-live-upload", "status": "receiving", "stage": "receiving"}],
|
|
"active_count": 1,
|
|
"failed_count": 0,
|
|
"page": 2,
|
|
"page_size": 10,
|
|
"total": 11,
|
|
"pages": 2,
|
|
}
|
|
app.state.services.uploads._page_cache[(2, 10)] = cached
|
|
monkeypatch.setattr(api_module, "API_READ_DEADLINE_SECONDS", 0.01)
|
|
|
|
async def delayed(function, /, *args, **kwargs):
|
|
await asyncio.sleep(0.05)
|
|
return function(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(api_module, "_background_api", delayed)
|
|
|
|
async def scenario() -> None:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/uploads?page=2&page_size=10", headers=headers)
|
|
assert response.status_code == 200
|
|
assert response.json() == cached
|
|
await asyncio.sleep(0.06)
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_failed_jobs_can_be_retried_without_overwriting_history(tmp_path: Path):
|
|
app, headers = make_app(tmp_path)
|
|
failed_id = app.state.services.jobs.enqueue(
|
|
"scan_source",
|
|
{"source_id": "source-retry"},
|
|
dedupe_key="scan:source-retry",
|
|
)
|
|
with app.state.services.db.transaction() as connection:
|
|
connection.execute(
|
|
"UPDATE jobs SET status='failed',progress=.35,error='temporary failure',finished_at=? WHERE id=?",
|
|
(utcnow(), failed_id),
|
|
)
|
|
|
|
async def scenario() -> None:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post(f"/api/v1/jobs/{failed_id}/retry", headers=headers, json={})
|
|
assert response.status_code == 202
|
|
retried_id = response.json()["job_id"]
|
|
assert response.json()["retried_from"] == failed_id
|
|
duplicate = await client.post(f"/api/v1/jobs/{failed_id}/retry", headers=headers, json={})
|
|
assert duplicate.json()["job_id"] == retried_id
|
|
active = await client.post(f"/api/v1/jobs/{retried_id}/retry", headers=headers, json={})
|
|
assert active.status_code == 409
|
|
missing = await client.post("/api/v1/jobs/missing/retry", headers=headers, json={})
|
|
assert missing.status_code == 404
|
|
|
|
with app.state.services.db.read() as connection:
|
|
failed = connection.execute("SELECT status,error FROM jobs WHERE id=?", (failed_id,)).fetchone()
|
|
retried = connection.execute(
|
|
"SELECT kind,payload_json,dedupe_key,status,attempts FROM jobs WHERE id=?",
|
|
(retried_id,),
|
|
).fetchone()
|
|
assert dict(failed) == {"status": "failed", "error": "temporary failure"}
|
|
assert retried["kind"] == "scan_source"
|
|
assert json.loads(retried["payload_json"]) == {"source_id": "source-retry"}
|
|
assert retried["dedupe_key"] == "scan:source-retry"
|
|
assert (retried["status"], retried["attempts"]) == ("queued", 0)
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_failed_jobs_can_be_retried_in_bulk_by_lane_without_replaying_superseded_history(
|
|
tmp_path: Path,
|
|
):
|
|
app, headers = make_app(tmp_path)
|
|
jobs = app.state.services.jobs
|
|
scan_failed = jobs.enqueue(
|
|
"scan_source", {"source_id": "source-bulk"}, dedupe_key="scan:source-bulk"
|
|
)
|
|
audio_failed = jobs.enqueue(
|
|
"transcribe_audio", {"video_id": "video-bulk"}, dedupe_key="audio:video-bulk"
|
|
)
|
|
superseded_failed = jobs.enqueue(
|
|
"index_video", {"video_id": "video-superseded"}, dedupe_key="index:video-superseded"
|
|
)
|
|
with app.state.services.db.transaction() as connection:
|
|
connection.execute(
|
|
"UPDATE jobs SET status='failed',error='temporary failure',finished_at=? "
|
|
"WHERE id IN (?,?,?)",
|
|
(utcnow(), scan_failed, audio_failed, superseded_failed),
|
|
)
|
|
superseding_job = jobs.enqueue(
|
|
"index_video", {"video_id": "video-superseded"}, dedupe_key="index:video-superseded"
|
|
)
|
|
|
|
async def scenario() -> None:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
scan_page = await client.get(
|
|
"/api/v1/jobs?page=1&page_size=10&lane=scan", headers=headers
|
|
)
|
|
assert scan_page.status_code == 200
|
|
assert scan_page.json()["retryable_failed_count"] == 1
|
|
|
|
retried_scan = await client.post(
|
|
"/api/v1/jobs/retry-failed?lane=scan", headers=headers, json={}
|
|
)
|
|
assert retried_scan.status_code == 202
|
|
assert retried_scan.json()["retried"] == 1
|
|
assert retried_scan.json()["lane"] == "scan"
|
|
|
|
duplicate = await client.post(
|
|
"/api/v1/jobs/retry-failed?lane=scan", headers=headers, json={}
|
|
)
|
|
assert duplicate.status_code == 202
|
|
assert duplicate.json()["retried"] == 0
|
|
|
|
all_page = await client.get("/api/v1/jobs?page=1&page_size=10", headers=headers)
|
|
assert all_page.json()["retryable_failed_count"] == 1
|
|
retried_audio = await client.post("/api/v1/jobs/retry-failed", headers=headers, json={})
|
|
assert retried_audio.status_code == 202
|
|
assert retried_audio.json()["retried"] == 1
|
|
|
|
with app.state.services.db.read() as connection:
|
|
old_statuses = {
|
|
row["id"]: row["status"]
|
|
for row in connection.execute(
|
|
"SELECT id,status FROM jobs WHERE id IN (?,?,?)",
|
|
(scan_failed, audio_failed, superseded_failed),
|
|
).fetchall()
|
|
}
|
|
active = connection.execute(
|
|
"SELECT id,kind,dedupe_key FROM jobs WHERE status='queued' ORDER BY created_at"
|
|
).fetchall()
|
|
assert old_statuses == {
|
|
scan_failed: "failed",
|
|
audio_failed: "failed",
|
|
superseded_failed: "failed",
|
|
}
|
|
assert [row["dedupe_key"] for row in active].count("index:video-superseded") == 1
|
|
assert {row["kind"] for row in active} >= {
|
|
"scan_source",
|
|
"transcribe_audio",
|
|
"index_video",
|
|
}
|
|
assert superseding_job in {row["id"] for row in active}
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_series_aggregate_merge_bulk_and_filter(tmp_path: Path):
|
|
app, headers = make_app(tmp_path)
|
|
seed_videos(app)
|
|
|
|
async def scenario() -> None:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
rows = (await client.get("/api/v1/series", headers=headers)).json()
|
|
assert rows[0]["video_count"] == 2
|
|
assert rows[0]["duration_ms"] == 30_000
|
|
renamed = await client.patch(
|
|
"/api/v1/series", headers=headers, json={"name": "SERIES A", "new_name": "Merged"}
|
|
)
|
|
assert renamed.json()["updated"] == 2
|
|
assigned = await client.post(
|
|
"/api/v1/videos/series/bulk",
|
|
headers=headers,
|
|
json={"video_ids": ["video-2"], "series": "Merged"},
|
|
)
|
|
assert assigned.json()["updated"] == 1
|
|
assert len((await client.get("/api/v1/videos?series=Merged", headers=headers)).json()) == 3
|
|
removed = await client.post(
|
|
"/api/v1/videos/series/bulk",
|
|
headers=headers,
|
|
json={"video_ids": ["video-2"], "series": None},
|
|
)
|
|
assert removed.json()["series"] is None
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_collections_support_single_membership_order_cover_and_dissolve(tmp_path: Path):
|
|
app, headers = make_app(tmp_path)
|
|
seed_videos(app)
|
|
|
|
async def scenario() -> None:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
first = await client.post(
|
|
"/api/v1/collections",
|
|
headers=headers,
|
|
json={"name": "第一季", "description": "按顺序播放", "video_ids": ["video-0", "video-1"]},
|
|
)
|
|
assert first.status_code == 201
|
|
first_id = first.json()["id"]
|
|
second = await client.post(
|
|
"/api/v1/collections",
|
|
headers=headers,
|
|
json={"name": "临时合集", "video_ids": ["video-2"]},
|
|
)
|
|
second_id = second.json()["id"]
|
|
moved = await client.post(
|
|
f"/api/v1/collections/{first_id}/videos",
|
|
headers=headers,
|
|
json={"video_ids": ["video-2"]},
|
|
)
|
|
assert moved.json()["updated"] == 1
|
|
assert (await client.get(f"/api/v1/collections/{second_id}", headers=headers)).json()[
|
|
"video_count"
|
|
] == 0
|
|
|
|
detail = (await client.get(f"/api/v1/collections/{first_id}", headers=headers)).json()
|
|
assert [video["id"] for video in detail["videos"]] == ["video-0", "video-1", "video-2"]
|
|
reordered = await client.patch(
|
|
f"/api/v1/collections/{first_id}/videos/order",
|
|
headers=headers,
|
|
json={"video_ids": ["video-2", "video-0", "video-1"]},
|
|
)
|
|
assert reordered.json()["updated"] == 3
|
|
invalid = await client.patch(
|
|
f"/api/v1/collections/{first_id}/videos/order",
|
|
headers=headers,
|
|
json={"video_ids": ["video-0"]},
|
|
)
|
|
assert invalid.status_code == 400
|
|
changed = await client.patch(
|
|
f"/api/v1/collections/{first_id}",
|
|
headers=headers,
|
|
json={"description": "更新后的简介", "cover_video_id": "video-1"},
|
|
)
|
|
assert changed.json()["cover_video_id"] == "video-1"
|
|
filtered = (
|
|
await client.get(
|
|
f"/api/v1/videos?collection_id={first_id}&sort=collection",
|
|
headers=headers,
|
|
)
|
|
).json()
|
|
assert [video["id"] for video in filtered] == ["video-2", "video-0", "video-1"]
|
|
assert all(video["collection_id"] == first_id for video in filtered)
|
|
|
|
dissolved = await client.delete(f"/api/v1/collections/{first_id}", headers=headers)
|
|
assert dissolved.json()["detached_videos"] == 3
|
|
assert len((await client.get("/api/v1/videos?limit=20", headers=headers)).json()) == 3
|
|
with app.state.services.db.read() as conn:
|
|
assert conn.execute("SELECT count(*) FROM collection_videos").fetchone()[0] == 0
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_proxy_is_encrypted_redacted_and_excluded_from_model_status(tmp_path: Path):
|
|
app, headers = make_app(tmp_path)
|
|
|
|
async def scenario() -> None:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.patch(
|
|
"/api/v1/models/config",
|
|
headers=headers,
|
|
json={
|
|
"hf_endpoint": "https://hf-mirror.example",
|
|
"proxy_url": "http://proxy.example:8080",
|
|
"proxy_username": "proxy-user",
|
|
"proxy_password": "proxy-secret",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["proxy"] == {
|
|
"enabled": True,
|
|
"url": "http://proxy.example:8080",
|
|
"username": "proxy-user",
|
|
"has_password": True,
|
|
}
|
|
status = (await client.get("/api/v1/models", headers=headers)).json()
|
|
assert "proxy-secret" not in json.dumps(status)
|
|
assert "password" not in status["proxy"]
|
|
assert status["accelerator"]["policy"] == "gpu_preferred"
|
|
assert status["accelerator"]["cpu_threads"] == 1
|
|
assert set(status["accelerator"]["components"]) == {"visual", "ocr", "faces", "audio"}
|
|
assert status["pip"]["index_url"] == "https://pypi.tuna.tsinghua.edu.cn/simple"
|
|
|
|
disabled = await client.patch(
|
|
"/api/v1/models/config",
|
|
headers=headers,
|
|
json={"proxy_enabled": False},
|
|
)
|
|
assert disabled.json()["proxy"]["enabled"] is False
|
|
assert disabled.json()["proxy"]["has_password"] is True
|
|
assert app.state.services.models._httpx_proxy() is None
|
|
assert app.state.services.models._hf_proxies() is None
|
|
|
|
sources = await client.patch(
|
|
"/api/v1/models/config",
|
|
headers=headers,
|
|
json={
|
|
"pip_index_url": "https://pypi.example/simple/",
|
|
"pytorch_index_url": "https://torch.example/cpu/",
|
|
},
|
|
)
|
|
assert sources.json()["pip"] == {
|
|
"index_url": "https://pypi.example/simple",
|
|
"pytorch_index_url": "https://torch.example/cpu",
|
|
}
|
|
rejected = await client.patch(
|
|
"/api/v1/models/config",
|
|
headers=headers,
|
|
json={"pip_index_url": "https://user:secret@pypi.example/simple"},
|
|
)
|
|
assert rejected.status_code == 400
|
|
|
|
asyncio.run(scenario())
|
|
stored = app.state.services.db.setting("model_proxy_config")
|
|
assert "proxy-secret" not in json.dumps(stored)
|
|
assert app.state.services.models.proxy_config(include_password=True)["password"] == "proxy-secret"
|
|
|
|
|
|
def test_huggingface_downloads_ignore_inherited_proxy_when_switch_is_off_for_non_mirror_endpoint(
|
|
tmp_path: Path, monkeypatch
|
|
):
|
|
app, _ = make_app(tmp_path)
|
|
manager = app.state.services.models
|
|
factories = []
|
|
|
|
class Session:
|
|
def __init__(self):
|
|
self.trust_env = True
|
|
self.proxies = {}
|
|
|
|
monkeypatch.setitem(sys.modules, "requests", types.SimpleNamespace(Session=Session))
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"huggingface_hub",
|
|
types.SimpleNamespace(
|
|
configure_http_backend=lambda *, backend_factory: factories.append(backend_factory)
|
|
),
|
|
)
|
|
manager.settings.model_hf_endpoint = "https://huggingface.co"
|
|
manager.set_proxy_config("http://proxy.example:8080", enabled=False)
|
|
manager._configure_hf_http_backend()
|
|
direct = factories[-1]()
|
|
assert direct.trust_env is False
|
|
assert direct.proxies == {}
|
|
|
|
manager.set_proxy_config("http://proxy.example:8080", enabled=True)
|
|
manager._configure_hf_http_backend()
|
|
proxied = factories[-1]()
|
|
assert proxied.trust_env is False
|
|
assert proxied.proxies == {
|
|
"http": "http://proxy.example:8080",
|
|
"https": "http://proxy.example:8080",
|
|
}
|
|
|
|
|
|
def test_hf_mirror_downloads_stay_direct_even_when_model_proxy_is_enabled(tmp_path: Path, monkeypatch):
|
|
app, _ = make_app(tmp_path)
|
|
manager = app.state.services.models
|
|
factories = []
|
|
|
|
class Session:
|
|
def __init__(self):
|
|
self.trust_env = True
|
|
self.proxies = {}
|
|
|
|
monkeypatch.setitem(sys.modules, "requests", types.SimpleNamespace(Session=Session))
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"huggingface_hub",
|
|
types.SimpleNamespace(configure_http_backend=lambda *, backend_factory: factories.append(backend_factory)),
|
|
)
|
|
manager.set_proxy_config("http://proxy.example:8080", enabled=True)
|
|
manager.settings.model_hf_endpoint = "https://hf-mirror.com"
|
|
|
|
assert manager._hf_proxies() is None
|
|
manager._configure_hf_http_backend()
|
|
direct = factories[-1]()
|
|
assert direct.trust_env is False
|
|
assert direct.proxies == {}
|
|
|
|
|
|
def test_model_uninstall_preserves_indexes_and_reports_usage(tmp_path: Path):
|
|
app, headers = make_app(tmp_path)
|
|
seed_videos(app)
|
|
root = app.state.services.settings.models_dir
|
|
(root / "ocr").mkdir(parents=True)
|
|
for name in ("det.onnx", "rec.onnx", "cls.onnx"):
|
|
(root / "ocr" / name).write_bytes(b"model-contents")
|
|
(root / "manifest.json").write_text(
|
|
json.dumps({"version": "v1", "components": {"ocr": {"version": "ocr-v1"}}}), encoding="utf-8"
|
|
)
|
|
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('frame','video-0',0,0,1,'thumb',?,?)",
|
|
(b"existing-vector", utcnow()),
|
|
)
|
|
app.state.services.accelerator.mark_ready("ocr", "CPU")
|
|
assert app.state.services.accelerator.status()["components"]["ocr"]["state"] == "ready"
|
|
|
|
async def scenario() -> None:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
deleted = await client.delete("/api/v1/models/ocr", headers=headers)
|
|
assert deleted.status_code == 200
|
|
assert deleted.json()["freed_bytes"] > 0
|
|
usage = (await client.get("/api/v1/storage/usage?refresh=true", headers=headers)).json()
|
|
assert set(usage["categories"]) == {
|
|
"database",
|
|
"ai_index",
|
|
"models",
|
|
"thumbnails",
|
|
"preview_cache",
|
|
"remote_cache",
|
|
"upload_staging",
|
|
"other",
|
|
}
|
|
assert usage["categories"]["ai_index"]["mode"] == "pgvector"
|
|
|
|
asyncio.run(scenario())
|
|
assert not (root / "ocr").exists()
|
|
assert app.state.services.accelerator.status()["components"]["ocr"]["state"] == "not_loaded"
|
|
with app.state.services.db.read() as conn:
|
|
assert conn.execute("SELECT vector_blob FROM frames WHERE id='frame'").fetchone()[0] == b"existing-vector"
|
|
|
|
|
|
def test_model_uninstall_conflicts_with_install_job(tmp_path: Path):
|
|
app, headers = make_app(tmp_path)
|
|
app.state.services.jobs.enqueue("install_models", {}, dedupe_key="install-models")
|
|
|
|
async def scenario() -> None:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.delete("/api/v1/models/all", headers=headers)
|
|
assert response.status_code == 409
|
|
|
|
asyncio.run(scenario())
|