bvandeusen
def967a1a8
Removed the app/client fixtures duplicated across 36 test files (two variants: separate app + client(app), and a self-contained client() that called create_app inline) and the now-unused create_app imports. Both fixtures now live once in conftest.py. test_suggestions_bulk keeps its import (builds the app inline in two tests); test_health drops its local client + unused pytest_asyncio. Net -415 lines. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
412 lines
12 KiB
Python
412 lines
12 KiB
Python
"""FC-3k: /api/admin/* endpoint integration tests.
|
|
|
|
Monkeypatch `.delay()` for Tier-C tasks to capture dispatch payloads
|
|
without actually queuing. Tier-B/A endpoints run synchronously
|
|
through the real service.
|
|
"""
|
|
import hashlib
|
|
|
|
import pytest
|
|
|
|
from backend.app.models import Artist, ImageRecord, Tag, TagKind
|
|
from backend.app.models.tag import image_tag
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
# --- Tier-C: POST /artists/<slug>/cascade-delete --------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_artist_cascade_dry_run_returns_projection(client, db):
|
|
a = Artist(name="Aria", slug="aria")
|
|
db.add(a)
|
|
await db.commit()
|
|
|
|
resp = await client.post(
|
|
"/api/admin/artists/aria/cascade-delete",
|
|
json={"dry_run": True},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = await resp.get_json()
|
|
assert body["artist"]["slug"] == "aria"
|
|
assert "projected" in body
|
|
assert body["projected"]["images"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_artist_cascade_unknown_slug_404(client):
|
|
resp = await client.post(
|
|
"/api/admin/artists/no-such/cascade-delete",
|
|
json={"dry_run": True},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_artist_cascade_wrong_confirm_400(client, db):
|
|
a = Artist(name="Aria", slug="aria")
|
|
db.add(a)
|
|
await db.commit()
|
|
artist_id = a.id
|
|
|
|
resp = await client.post(
|
|
"/api/admin/artists/aria/cascade-delete",
|
|
json={"dry_run": False, "confirm": "wrong"},
|
|
)
|
|
assert resp.status_code == 400
|
|
body = await resp.get_json()
|
|
assert body["error"] == "confirm_mismatch"
|
|
assert body["expected"] == f"delete-artist-{artist_id}"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_artist_cascade_correct_confirm_dispatches(
|
|
client, db, monkeypatch,
|
|
):
|
|
a = Artist(name="Aria", slug="aria")
|
|
db.add(a)
|
|
await db.commit()
|
|
artist_id = a.id
|
|
|
|
dispatched = []
|
|
|
|
class _Result:
|
|
id = "fake-task-id"
|
|
|
|
def _delay(**kw):
|
|
dispatched.append(kw)
|
|
return _Result()
|
|
|
|
monkeypatch.setattr(
|
|
"backend.app.tasks.admin.delete_artist_cascade_task.delay", _delay,
|
|
)
|
|
|
|
resp = await client.post(
|
|
"/api/admin/artists/aria/cascade-delete",
|
|
json={
|
|
"dry_run": False,
|
|
"confirm": f"delete-artist-{artist_id}",
|
|
},
|
|
)
|
|
assert resp.status_code == 202
|
|
body = await resp.get_json()
|
|
assert body["task_id"] == "fake-task-id"
|
|
assert dispatched == [{"artist_id": artist_id}]
|
|
|
|
|
|
# --- Tier-C: POST /images/bulk-delete -------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_delete_dry_run_returns_counts(client, db, tmp_path):
|
|
a = Artist(name="B", slug="b")
|
|
db.add(a)
|
|
await db.flush()
|
|
i1 = ImageRecord(
|
|
artist_id=a.id, path=str(tmp_path / "1.jpg"),
|
|
sha256="1" * 64, size_bytes=10, mime="image/jpeg",
|
|
origin="imported_filesystem",
|
|
)
|
|
i2 = ImageRecord(
|
|
artist_id=a.id, path=str(tmp_path / "2.jpg"),
|
|
sha256="2" * 64, size_bytes=20, mime="image/jpeg",
|
|
origin="imported_filesystem",
|
|
)
|
|
db.add_all([i1, i2])
|
|
await db.commit()
|
|
|
|
resp = await client.post(
|
|
"/api/admin/images/bulk-delete",
|
|
json={
|
|
"image_ids": [i1.id, i2.id, 9_999_999],
|
|
"dry_run": True,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = await resp.get_json()
|
|
assert body["images_found"] == 2
|
|
assert body["bytes_on_disk"] == 30
|
|
assert body["missing_ids"] == [9_999_999]
|
|
# Dry-run hands the canonical Tier-C confirm token back so the
|
|
# frontend doesn't recompute SHA-256 client-side (crypto.subtle
|
|
# is Secure-Context-gated; FC runs over plain HTTP).
|
|
assert body["confirm_token"].startswith("delete-images-")
|
|
assert len(body["confirm_token"]) == len("delete-images-") + 8
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_delete_empty_list_400(client):
|
|
resp = await client.post(
|
|
"/api/admin/images/bulk-delete",
|
|
json={"image_ids": [], "dry_run": True},
|
|
)
|
|
assert resp.status_code == 400
|
|
body = await resp.get_json()
|
|
assert body["error"] == "invalid_image_ids"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_delete_non_int_400(client):
|
|
resp = await client.post(
|
|
"/api/admin/images/bulk-delete",
|
|
json={"image_ids": ["foo", 2], "dry_run": True},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_delete_wrong_confirm_400_with_expected_token(
|
|
client, db, tmp_path,
|
|
):
|
|
a = Artist(name="B", slug="b")
|
|
db.add(a)
|
|
await db.flush()
|
|
img = ImageRecord(
|
|
artist_id=a.id, path=str(tmp_path / "x.jpg"),
|
|
sha256="c" * 64, size_bytes=10, mime="image/jpeg",
|
|
origin="imported_filesystem",
|
|
)
|
|
db.add(img)
|
|
await db.commit()
|
|
img_id = img.id
|
|
|
|
resp = await client.post(
|
|
"/api/admin/images/bulk-delete",
|
|
json={
|
|
"image_ids": [img_id], "dry_run": False, "confirm": "nope",
|
|
},
|
|
)
|
|
assert resp.status_code == 400
|
|
body = await resp.get_json()
|
|
assert body["error"] == "confirm_mismatch"
|
|
assert body["expected"].startswith("delete-images-")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_delete_correct_confirm_dispatches(
|
|
client, db, tmp_path, monkeypatch,
|
|
):
|
|
a = Artist(name="B", slug="b")
|
|
db.add(a)
|
|
await db.flush()
|
|
img = ImageRecord(
|
|
artist_id=a.id, path=str(tmp_path / "x.jpg"),
|
|
sha256="d" * 64, size_bytes=10, mime="image/jpeg",
|
|
origin="imported_filesystem",
|
|
)
|
|
db.add(img)
|
|
await db.commit()
|
|
img_id = img.id
|
|
|
|
sha8 = hashlib.sha256(str(img_id).encode("utf-8")).hexdigest()[:8]
|
|
|
|
dispatched = []
|
|
|
|
class _Result:
|
|
id = "fake-bulk-id"
|
|
|
|
def _delay(**kw):
|
|
dispatched.append(kw)
|
|
return _Result()
|
|
|
|
monkeypatch.setattr(
|
|
"backend.app.tasks.admin.bulk_delete_images_task.delay", _delay,
|
|
)
|
|
|
|
resp = await client.post(
|
|
"/api/admin/images/bulk-delete",
|
|
json={
|
|
"image_ids": [img_id],
|
|
"dry_run": False,
|
|
"confirm": f"delete-images-{sha8}",
|
|
},
|
|
)
|
|
assert resp.status_code == 202
|
|
body = await resp.get_json()
|
|
assert body["task_id"] == "fake-bulk-id"
|
|
assert dispatched == [{"image_ids": [img_id]}]
|
|
|
|
|
|
# --- Tier-B: DELETE /tags/<id> + POST /tags/<dest>/merge -----------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tag_delete_200_after_cascade(client, db):
|
|
t = Tag(name="doomed-tag", kind=TagKind.general)
|
|
db.add(t)
|
|
await db.commit()
|
|
tag_id = t.id
|
|
|
|
resp = await client.delete(f"/api/admin/tags/{tag_id}")
|
|
assert resp.status_code == 200
|
|
body = await resp.get_json()
|
|
assert body["deleted"]["id"] == tag_id
|
|
assert body["associations_removed"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tag_delete_404_unknown(client):
|
|
resp = await client.delete("/api/admin/tags/9999999")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tag_merge_succeeds_for_same_kind(client, db):
|
|
src = Tag(name="src-tag", kind=TagKind.general)
|
|
dest = Tag(name="dest-tag", kind=TagKind.general)
|
|
db.add_all([src, dest])
|
|
await db.commit()
|
|
src_id = src.id
|
|
dest_id = dest.id
|
|
|
|
resp = await client.post(
|
|
f"/api/admin/tags/{dest_id}/merge",
|
|
json={"source_id": src_id},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = await resp.get_json()
|
|
assert body["result"]["target_id"] == dest_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tag_merge_rejects_self_merge(client, db):
|
|
t = Tag(name="x", kind=TagKind.general)
|
|
db.add(t)
|
|
await db.commit()
|
|
resp = await client.post(
|
|
f"/api/admin/tags/{t.id}/merge",
|
|
json={"source_id": t.id},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tag_merge_returns_400_on_kind_mismatch(client, db):
|
|
src = Tag(name="src", kind=TagKind.general)
|
|
dest = Tag(name="dest", kind=TagKind.artist)
|
|
db.add_all([src, dest])
|
|
await db.commit()
|
|
resp = await client.post(
|
|
f"/api/admin/tags/{dest.id}/merge",
|
|
json={"source_id": src.id},
|
|
)
|
|
assert resp.status_code == 400
|
|
body = await resp.get_json()
|
|
assert body["error"] == "tag_kind_mismatch"
|
|
|
|
|
|
# --- Tier-B helper: GET /tags/<id>/usage-count ----------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tag_usage_count_returns_zero_for_unused(client, db):
|
|
t = Tag(name="lonely", kind=TagKind.general)
|
|
db.add(t)
|
|
await db.commit()
|
|
resp = await client.get(f"/api/admin/tags/{t.id}/usage-count")
|
|
assert resp.status_code == 200
|
|
body = await resp.get_json()
|
|
assert body["count"] == 0
|
|
|
|
|
|
# --- Tier-A: POST /tags/prune-unused --------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prune_unused_dry_run_lists_unused(client, db, tmp_path):
|
|
a = Artist(name="P", slug="p")
|
|
db.add(a)
|
|
await db.flush()
|
|
img = ImageRecord(
|
|
artist_id=a.id, path=str(tmp_path / "p.jpg"),
|
|
sha256="e" * 64, size_bytes=10, mime="image/jpeg",
|
|
origin="imported_filesystem",
|
|
)
|
|
db.add(img)
|
|
used = Tag(name="kept", kind=TagKind.general)
|
|
unused = Tag(name="prune", kind=TagKind.general)
|
|
db.add_all([used, unused])
|
|
await db.flush()
|
|
await db.execute(image_tag.insert().values(
|
|
image_record_id=img.id, tag_id=used.id,
|
|
))
|
|
await db.commit()
|
|
|
|
resp = await client.post(
|
|
"/api/admin/tags/prune-unused", json={"dry_run": True},
|
|
)
|
|
body = await resp.get_json()
|
|
assert body["count"] == 1
|
|
assert "prune" in body["sample_names"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prune_unused_commit_deletes_and_returns_count(client, db):
|
|
db.add(Tag(name="byebye", kind=TagKind.general))
|
|
await db.commit()
|
|
|
|
resp = await client.post(
|
|
"/api/admin/tags/prune-unused", json={"dry_run": False},
|
|
)
|
|
body = await resp.get_json()
|
|
assert body["deleted"] >= 1
|
|
assert "byebye" in body["sample_names"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_purge_legacy_dry_run_counts_by_kind_and_prefix(client, db):
|
|
db.add_all([
|
|
Tag(name="BlenderKnight:Hannah_BJ_Loops", kind=TagKind.archive),
|
|
Tag(name="BlenderKnight:May Animation", kind=TagKind.post),
|
|
Tag(name="SomeArtist", kind=TagKind.artist),
|
|
# IR `source` kind fell back to general during migration —
|
|
# caught by the source:* name prefix, not by kind.
|
|
Tag(name="source:patreon", kind=TagKind.general),
|
|
Tag(name="blonde hair", kind=TagKind.general), # must survive
|
|
])
|
|
await db.commit()
|
|
|
|
resp = await client.post(
|
|
"/api/admin/tags/purge-legacy", json={"dry_run": True},
|
|
)
|
|
body = await resp.get_json()
|
|
assert body["count"] == 4
|
|
assert body["by_kind"] == {"archive": 1, "post": 1, "artist": 1}
|
|
assert body["by_prefix"] == {"source:*": 1}
|
|
assert "blonde hair" not in body["sample_names"]
|
|
assert "source:patreon" in body["sample_names"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_purge_legacy_commit_deletes_only_legacy(client, db):
|
|
from sqlalchemy import func, select
|
|
|
|
db.add_all([
|
|
Tag(name="arch1", kind=TagKind.archive),
|
|
Tag(name="source:fanbox", kind=TagKind.general),
|
|
Tag(name="keepme", kind=TagKind.general),
|
|
Tag(name="char1", kind=TagKind.character),
|
|
])
|
|
await db.commit()
|
|
|
|
resp = await client.post(
|
|
"/api/admin/tags/purge-legacy", json={"dry_run": False},
|
|
)
|
|
body = await resp.get_json()
|
|
assert body["deleted"] == 2 # arch1 + source:fanbox
|
|
|
|
# The plain general + character tags survive.
|
|
keep = (await db.execute(
|
|
select(Tag.name).where(Tag.name.in_(["keepme", "char1"]))
|
|
)).scalars().all()
|
|
assert set(keep) == {"keepme", "char1"}
|
|
# No archive/post/artist or source:* left.
|
|
gone = (await db.execute(
|
|
select(func.count()).select_from(Tag).where(
|
|
(Tag.kind.in_([TagKind.archive, TagKind.post, TagKind.artist]))
|
|
| (Tag.name.like("source:%"))
|
|
)
|
|
)).scalar_one()
|
|
assert gone == 0
|