130 lines
4.4 KiB
Python
130 lines
4.4 KiB
Python
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")
|