pigway-frame-flow/tests/test_audit_fixes.py

import shutil

import pytest


def test_image_write_permission_can_be_granted_without_admin(client):
    from io import BytesIO

    from PIL import Image

    data = BytesIO()
    Image.new("RGB", (30, 20), "green").save(data, "PNG")
    picture = client.post(
        "/api/v1/source-libraries/manual-uploads/upload",
        files={"file": ("fixture.png", data.getvalue())},
    ).json()
    scopes = client.get("/api/v1/client-scopes").json()
    assert "images:write" in scopes
    response = client.post(
        "/api/v1/clients",
        json={
            "client_id": "image-writer",
            "name": "Writer",
            "permissions": ["images:write"],
        },
    )
    assert response.status_code == 201
    client.headers["Authorization"] = "Bearer " + response.json()["token"]
    # Authorization succeeds and resource lookup fails; administrative routes stay forbidden.
    assert (
        client.put("/api/v1/assets/missing/star", json={"starred": True}).status_code
        == 404
    )
    assert (
        client.put(
            "/api/v1/source-images/" + picture["id"] + "/star", json={"starred": True}
        ).status_code
        == 200
    )
    assert (
        client.put(
            "/api/v1/source-images/" + picture["id"] + "/exclusion",
            json={"excluded": True},
        ).status_code
        == 409
    )
    assert client.get("/api/v1/clients").status_code == 403


def test_nested_transaction_rollback_includes_execute(service):
    with pytest.raises(RuntimeError):
        with service.db.transaction():
            service.db.execute("INSERT INTO app_config VALUES('outer','{}')")
            with service.db.transaction():
                service.db.execute("INSERT INTO app_config VALUES('inner','{}')")
            raise RuntimeError("injected failure")
    assert (
        service.db.one("SELECT key FROM app_config WHERE key IN ('outer','inner')")
        is None
    )


def test_low_disk_space_is_reported_without_enabling_cleanup(client, monkeypatch):
    monkeypatch.setattr(
        shutil,
        "disk_usage",
        lambda _path: shutil._ntuple_diskusage(100 * 1024**3, 99 * 1024**3, 1024**3),
    )
    data = client.get("/api/v1/storage").json()
    assert data["disk"]["low_space"] is True
    assert data["disk"]["free_bytes"] == 1024**3
    assert data["policy"]["enabled"] is False


def test_unsupported_image_plugin_is_never_invoked(monkeypatch):
    from PIL import Image

    from pigway_frame_flow.errors import FlowError
    from pigway_frame_flow.renderers.images import convert

    Image.init()
    invoked = []

    def vulnerable_plugin(*_args):
        invoked.append(True)
        raise RuntimeError("Unsupported plugin must not parse input")

    monkeypatch.setitem(Image.OPEN, "XBM", (vulnerable_plugin, lambda _prefix: True))
    with pytest.raises(FlowError):
        convert(b"#define image_width 1\n#define image_height 1\n")
    assert not invoked


def test_promotion_sql_failure_keeps_original_cache(service, monkeypatch):
    asset = service.assets.put(b"fixture", "cached.jpg", "cache")
    original = service.db.execute

    def fail(query, args=()):
        if query.startswith("UPDATE assets SET kind"):
            raise RuntimeError("injected promotion SQL failure")
        return original(query, args)

    monkeypatch.setattr(service.db, "execute", fail)
    with pytest.raises(RuntimeError):
        service.assets.promote(asset["asset_id"])
    assert service.assets.path(asset["asset_id"]).parent.name == "cache"
    assert service.assets.path(asset["asset_id"]).read_bytes() == b"fixture"
    assert not list((service.settings.data_dir / "images").iterdir())
    assert not service.db.all("SELECT * FROM asset_file_moves")


def test_promotion_restart_finishes_committed_move(service):
    from pigway_frame_flow.renderers.images import Assets

    asset = service.assets.put(b"fixture", "cached.jpg", "cache")
    source = service.assets.path(asset["asset_id"])
    target = service.settings.data_dir / "images" / source.name
    target.write_bytes(source.read_bytes())
    service.db.execute(
        "INSERT INTO asset_file_moves VALUES(?,?,?)",
        (asset["asset_id"], "cache", "images"),
    )
    service.db.execute(
        "UPDATE assets SET kind='images' WHERE id=?", (asset["asset_id"],)
    )
    Assets(service.db, service.settings)
    assert not source.exists()
    assert service.assets.path(asset["asset_id"]).read_bytes() == b"fixture"
    assert not service.db.all("SELECT * FROM asset_file_moves")