FabledCurator/tests/test_discord_poster_cleanup.py

"""Removing a Discord source's posts by people outside its poster list (#4486).

Real Postgres (db_sync): the predicates read `raw_metadata->>'author'`, which
only the database can evaluate. The test that matters runs preview and apply
on one fixture and asserts they agree AND that the rows went (snippet #3087).
"""

import pytest
from sqlalchemy import func, select

from backend.app.models import (
    Artist,
    ImageProvenance,
    ImageRecord,
    Post,
    PostAttachment,
    Source,
)
from backend.app.services import discord_poster_cleanup as cleanup

pytestmark = pytest.mark.integration


def _image(db, artist, tmp_path, name, post):
    f = tmp_path / f"{name}.png"
    f.write_bytes(b"x")
    img = ImageRecord(
        artist_id=artist.id, path=str(f), sha256=name.ljust(64, "0"),
        size_bytes=1, mime="image/png", origin="downloaded", primary_post_id=post.id,
    )
    db.add(img)
    db.flush()
    return img


def _post(db, source, mid, raw, **extra):
    p = Post(
        source_id=source.id, artist_id=source.artist_id, external_post_id=mid,
        raw_metadata=raw, **extra,
    )
    db.add(p)
    db.flush()
    return p


def _fixture(db, tmp_path, authors=("Todding",)):
    """Todding's post, two of Jake's (one sharing an image with Todding's), an
    old post with no poster, and a drop that absorbed Todding's and Jake's."""
    artist = Artist(name="Todding", slug="todding-cleanup")
    db.add(artist)
    db.flush()
    src = Source(
        artist_id=artist.id, platform="discord",
        url="https://discord.com/channels/1/2",
        config_overrides={"discord_authors": list(authors)} if authors else None,
    )
    db.add(src)
    db.flush()
    todd = _post(db, src, "10", {"author": "todding", "author_name": "Todding", "author_id": "1"})
    jake1 = _post(db, src, "11", {"author": "jakeboii", "author_id": "8"})
    jake2 = _post(db, src, "12", {"author": "jakeboii", "author_id": "8"})
    old = _post(db, src, "13", {"category": "discord"})
    drop = _post(db, src, "fc-drop:10", None, synthesized_by="discord_drop")
    todd.absorbed_by_post_id = drop.id
    jake1.absorbed_by_post_id = drop.id

    art = _image(db, artist, tmp_path, "art", todd)
    meme = _image(db, artist, tmp_path, "meme", jake1)
    shared = _image(db, artist, tmp_path, "shared", jake2)
    legacy = _image(db, artist, tmp_path, "legacy", old)
    db.add_all([
        ImageProvenance(image_record_id=shared.id, post_id=todd.id, source_id=src.id),
        ImageProvenance(image_record_id=art.id, post_id=drop.id, source_id=src.id),
        ImageProvenance(image_record_id=meme.id, post_id=drop.id, source_id=src.id),
        PostAttachment(
            post_id=jake2.id, artist_id=artist.id, sha256="z" * 64,
            path="/store/z/f.zip", original_filename="f.zip", ext=".zip", size_bytes=1,
        ),
    ])
    db.commit()
    return src, {
        "todd": todd.id, "jake1": jake1.id, "jake2": jake2.id, "old": old.id,
        "drop": drop.id, "art": art.id, "meme": meme.id, "shared": shared.id,
        "legacy": legacy.id,
    }


def test_preview_matches_apply_and_the_rows_go(db_sync, tmp_path):
    src, ids = _fixture(db_sync, tmp_path)

    projected = cleanup.preview(db_sync, source_id=src.id)
    assert projected["authors"] == ["todding"]
    assert projected["posters"] == [{"poster": "jakeboii", "posts": 2}]
    assert projected["unknown_posts"] == 1

    done = cleanup.apply(db_sync, source_id=src.id, images_root=tmp_path)
    assert projected["posts"] == done["posts_deleted"] == 2
    assert projected["images"] == done["images_deleted"] == 1  # the meme only
    assert projected["attachments"] == done["attachments_deleted"] == 1
    assert projected["drops"] == done["drops_deleted"] == 1

    post_ids = set(db_sync.execute(select(Post.id).where(Post.source_id == src.id)).scalars())
    assert post_ids == {ids["todd"], ids["old"]}
    image_ids = set(db_sync.execute(
        select(ImageRecord.id).where(ImageRecord.id.in_(
            [ids["art"], ids["meme"], ids["shared"], ids["legacy"]]))
    ).scalars())
    # The meme was only Jake's. The shared piece is also on Todding's post, and
    # the drop's link to the art was never what kept it.
    assert image_ids == {ids["art"], ids["shared"], ids["legacy"]}
    assert not (tmp_path / "meme.png").exists()
    # Todding's message is back in the feed, for the sweep to regroup.
    assert db_sync.get(Post, ids["todd"]).absorbed_by_post_id is None
    assert db_sync.execute(select(func.count(PostAttachment.id)).where(
        PostAttachment.sha256 == "z" * 64)).scalar_one() == 0


def test_a_poster_matches_by_display_name_or_id_too(db_sync, tmp_path):
    src, _ = _fixture(db_sync, tmp_path, authors=("8",))
    projected = cleanup.preview(db_sync, source_id=src.id)
    assert projected["posters"] == [{"poster": "Todding", "posts": 1}]


def test_no_list_means_nothing_to_remove(db_sync, tmp_path):
    src, _ = _fixture(db_sync, tmp_path, authors=())
    with pytest.raises(cleanup.PosterCleanupError):
        cleanup.preview(db_sync, source_id=src.id)


def test_the_token_names_what_the_preview_showed():
    assert cleanup.confirm_token({"posts": 2, "images": 1}) == "remove-2-posts-1-images"