devplacepy/tests/unit/attachments.py

# retoor <retoor@molodetz.nl>

from devplacepy import attachments as att
from devplacepy.database import get_table
from devplacepy.utils import generate_uid


def _make_attachment(directory, stored_name, deleted_at=None):
    uid = generate_uid()
    get_table("attachments").insert(
        {
            "uid": uid,
            "deleted_at": deleted_at,
            "deleted_by": None,
            "directory": directory,
            "stored_name": stored_name,
            "target_type": "post",
            "target_uid": generate_uid(),
        }
    )
    return uid


def _write_blob(base, directory, stored_name, content=b"data"):
    file_dir = base / directory
    file_dir.mkdir(parents=True, exist_ok=True)
    (file_dir / stored_name).write_bytes(content)
    return file_dir / stored_name


def test_purge_soft_deleted_attachments_removes_row_and_blob(local_db, tmp_path, monkeypatch):
    monkeypatch.setattr(att, "ATTACHMENTS_DIR", tmp_path)
    directory = "ab/cd"
    stored_name = f"{generate_uid()}.png"
    _write_blob(tmp_path, directory, stored_name, b"x" * 100)
    uid = _make_attachment(directory, stored_name, deleted_at="2020-01-01T00:00:00+00:00")

    removed, freed = att.purge_soft_deleted_attachments()

    assert removed >= 1
    assert freed == 100
    assert get_table("attachments").find_one(uid=uid) is None
    assert not (tmp_path / directory / stored_name).exists()


def test_purge_soft_deleted_attachments_dry_run_changes_nothing(local_db, tmp_path, monkeypatch):
    monkeypatch.setattr(att, "ATTACHMENTS_DIR", tmp_path)
    directory = "ab/cd"
    stored_name = f"{generate_uid()}.png"
    _write_blob(tmp_path, directory, stored_name, b"x" * 50)
    uid = _make_attachment(directory, stored_name, deleted_at="2020-01-01T00:00:00+00:00")

    removed, freed = att.purge_soft_deleted_attachments(dry_run=True)

    assert removed >= 1
    assert freed == 50
    assert get_table("attachments").find_one(uid=uid) is not None
    assert (tmp_path / directory / stored_name).exists()


def test_purge_soft_deleted_attachments_ignores_live_rows(local_db, tmp_path, monkeypatch):
    monkeypatch.setattr(att, "ATTACHMENTS_DIR", tmp_path)
    directory = "ab/cd"
    stored_name = f"{generate_uid()}.png"
    _write_blob(tmp_path, directory, stored_name)
    uid = _make_attachment(directory, stored_name, deleted_at=None)

    att.purge_soft_deleted_attachments()

    assert get_table("attachments").find_one(uid=uid) is not None
    assert (tmp_path / directory / stored_name).exists()


def test_purge_soft_deleted_attachments_also_removes_thumbnail(local_db, tmp_path, monkeypatch):
    monkeypatch.setattr(att, "ATTACHMENTS_DIR", tmp_path)
    directory = "ab/cd"
    stem = generate_uid()
    stored_name = f"{stem}.jpg"
    _write_blob(tmp_path, directory, stored_name, b"x" * 20)
    _write_blob(tmp_path, directory, f"{stem}_thumb.jpg", b"y" * 5)
    _make_attachment(directory, stored_name, deleted_at="2020-01-01T00:00:00+00:00")

    removed, freed = att.purge_soft_deleted_attachments()

    assert removed >= 1
    assert freed == 25
    assert not (tmp_path / directory / f"{stem}_thumb.jpg").exists()


def test_sweep_orphan_attachment_blobs_removes_unreferenced_file(local_db, tmp_path, monkeypatch):
    monkeypatch.setattr(att, "ATTACHMENTS_DIR", tmp_path)
    directory = "ab/cd"
    orphan_name = f"{generate_uid()}.o"
    _write_blob(tmp_path, directory, orphan_name, b"y" * 30)

    removed, freed = att.sweep_orphan_attachment_blobs()

    assert removed == 1
    assert freed == 30
    assert not (tmp_path / directory / orphan_name).exists()


def test_sweep_orphan_attachment_blobs_keeps_referenced_file_and_its_thumbnail(
    local_db, tmp_path, monkeypatch
):
    monkeypatch.setattr(att, "ATTACHMENTS_DIR", tmp_path)
    directory = "ab/cd"
    stem = generate_uid()
    stored_name = f"{stem}.jpg"
    thumb_name = f"{stem}_thumb.jpg"
    _write_blob(tmp_path, directory, stored_name)
    _write_blob(tmp_path, directory, thumb_name)
    _make_attachment(directory, stored_name, deleted_at=None)

    removed, freed = att.sweep_orphan_attachment_blobs()

    assert removed == 0
    assert freed == 0
    assert (tmp_path / directory / stored_name).exists()
    assert (tmp_path / directory / thumb_name).exists()


def test_sweep_orphan_attachment_blobs_keeps_blob_referenced_only_by_soft_deleted_row(
    local_db, tmp_path, monkeypatch
):
    monkeypatch.setattr(att, "ATTACHMENTS_DIR", tmp_path)
    directory = "ab/cd"
    stored_name = f"{generate_uid()}.png"
    _write_blob(tmp_path, directory, stored_name)
    _make_attachment(directory, stored_name, deleted_at="2020-01-01T00:00:00+00:00")

    removed, freed = att.sweep_orphan_attachment_blobs()

    assert removed == 0
    assert freed == 0
    assert (tmp_path / directory / stored_name).exists()


def test_sweep_orphan_attachment_blobs_dry_run_changes_nothing(local_db, tmp_path, monkeypatch):
    monkeypatch.setattr(att, "ATTACHMENTS_DIR", tmp_path)
    directory = "ab/cd"
    orphan_name = f"{generate_uid()}.o"
    _write_blob(tmp_path, directory, orphan_name, b"z" * 10)

    removed, freed = att.sweep_orphan_attachment_blobs(dry_run=True)

    assert removed == 1
    assert freed == 10
    assert (tmp_path / directory / orphan_name).exists()