live-hub-py/tests/test_cleanup_orphan_file_assets.py

"""协作台账孤儿图清理脚本：只删「过期 + 没人引用」的图，且删对象成功才删台账行。

脚本本体在 ``scripts/cleanup_orphan_file_assets.py``（一次性运维脚本，不入包），
所以这里用 importlib 按路径加载。
"""

from __future__ import annotations

import importlib.util
from datetime import UTC, datetime, timedelta
from pathlib import Path
from types import ModuleType

import pytest

from web.backend.models import FileAsset, HuyaExternalOrder
from web.backend.services import file_storage

_SCRIPT_PATH = (
    Path(__file__).resolve().parents[1] / "scripts" / "cleanup_orphan_file_assets.py"
)


def load_script() -> ModuleType:
    spec = importlib.util.spec_from_file_location(
        "cleanup_orphan_file_assets", _SCRIPT_PATH
    )
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


@pytest.fixture(scope="module")
def cleanup_script() -> ModuleType:
    return load_script()


class FakeStorage:
    """记录被删的 key；``fail_keys`` 里的 key 删除时抛错。"""

    def __init__(self, fail_keys: set[str] | None = None):
        self.deleted: list[str] = []
        self.fail_keys = fail_keys or set()

    def delete_object(self, key: str) -> None:
        if key in self.fail_keys:
            raise RuntimeError(f"删不掉 {key}")
        self.deleted.append(key)


def make_asset(
    db,
    key: str,
    *,
    scene: str = "collab",
    days_old: int = 40,
    size: int = 1024,
    status: str = "active",
    thumb: str = "",
    medium: str = "",
) -> FileAsset:
    asset = FileAsset(
        object_key=key,
        scene=scene,
        size_bytes=size,
        original_filename=key.rsplit("/", 1)[-1],
        thumbnail_object_key=thumb,
        medium_object_key=medium,
        status=status,
        created_at=datetime.now(UTC) - timedelta(days=days_old),
    )
    db.add(asset)
    db.commit()
    return asset


def make_order(db, order_no: str, keys: list[str]) -> HuyaExternalOrder:
    order = HuyaExternalOrder(
        order_no=order_no,
        shop="佳佳",
        bind_images=list(keys),
        home_images=[],
        status="pending",
        ck_status="unverified",
        version=0,
    )
    db.add(order)
    db.commit()
    return order


def test_find_orphans_skips_referenced_and_fresh_assets(db, cleanup_script):
    referenced = make_asset(
        db, "collab/2026/09/01/ref.png", thumb="collab/2026/09/01/ref.thumb.jpg"
    )
    make_order(db, "A1", [referenced.object_key])
    make_asset(db, "collab/2026/09/02/fresh.png", days_old=3)
    orphan = make_asset(db, "collab/2026/09/03/orphan.png")

    orphans = cleanup_script.find_orphans(db, days=30)

    assert [asset.object_key for asset in orphans] == [orphan.object_key]


def test_find_orphans_ignores_other_scenes_and_inactive_rows(db, cleanup_script):
    make_asset(db, "misc/2026/09/01/other.png", scene="misc")
    make_asset(db, "collab/2026/09/01/gone.png", status="deleted")

    assert cleanup_script.find_orphans(db, days=30) == []


def test_dry_run_writes_nothing(db, cleanup_script):
    orphan = make_asset(db, "collab/2026/09/01/orphan.png")
    storage = FakeStorage()

    result = cleanup_script.run(db, apply=False, days=30)

    assert result["candidates"] == 1
    assert result["deleted"] == 0
    assert result["freed_bytes"] == 1024
    assert storage.deleted == []
    assert db.query(FileAsset).filter(FileAsset.id == orphan.id).count() == 1


def test_apply_deletes_original_and_variants_then_the_row(
    db, cleanup_script, monkeypatch
):
    make_asset(
        db,
        "collab/2026/09/01/orphan.png",
        thumb="collab/2026/09/01/orphan.thumb.webp",
        medium="collab/2026/09/01/orphan.medium.jpg",
    )
    storage = FakeStorage()
    monkeypatch.setattr(
        file_storage, "get_object_storage", lambda settings=None: storage
    )

    result = cleanup_script.run(db, apply=True, days=30)

    assert result["deleted"] == 1
    assert result["failed"] == 0
    assert storage.deleted == [
        "collab/2026/09/01/orphan.png",
        "collab/2026/09/01/orphan.thumb.webp",
        "collab/2026/09/01/orphan.medium.jpg",
    ]
    assert db.query(FileAsset).count() == 0


def test_failed_object_delete_keeps_the_ledger_row(db, cleanup_script, monkeypatch):
    """对象没删掉就绝不能删台账行 —— 否则再也没人知道这个对象存在。"""
    orphan = make_asset(db, "collab/2026/09/01/orphan.png")
    storage = FakeStorage(fail_keys={orphan.object_key})
    monkeypatch.setattr(
        file_storage, "get_object_storage", lambda settings=None: storage
    )

    result = cleanup_script.run(db, apply=True, days=30)

    assert result["deleted"] == 0
    assert result["failed"] == 1
    assert db.query(FileAsset).filter(FileAsset.id == orphan.id).count() == 1


def test_referenced_asset_survives_a_full_apply(db, cleanup_script, monkeypatch):
    referenced = make_asset(db, "collab/2026/09/01/ref.png")
    make_order(db, "A1", [referenced.object_key])
    make_asset(db, "collab/2026/09/01/orphan.png")
    storage = FakeStorage()
    monkeypatch.setattr(
        file_storage, "get_object_storage", lambda settings=None: storage
    )

    cleanup_script.run(db, apply=True, days=30)

    remaining = [asset.object_key for asset in db.query(FileAsset).all()]
    assert remaining == [referenced.object_key]
    assert storage.deleted == ["collab/2026/09/01/orphan.png"]