add image flow
This commit is contained in:
+272
-9
@@ -1,32 +1,62 @@
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
|
||||
import app.db as db_module
|
||||
from PIL import Image
|
||||
|
||||
from app.models import Box, Item, SubItem
|
||||
|
||||
|
||||
def create_box(client, name="Box A", note="Packed", room="Bedroom", status="ready"):
|
||||
return client.post(
|
||||
"/boxes",
|
||||
data={"name": name, "note": note, "room": room, "status": status},
|
||||
follow_redirects=False,
|
||||
)
|
||||
def create_box(client, name="Box A", note="Packed", room="Bedroom", status="ready", image=None):
|
||||
data = {"name": name, "note": note, "room": room, "status": status}
|
||||
files = {"image_file": image} if image is not None else None
|
||||
return client.post("/boxes", data=data, files=files, follow_redirects=False)
|
||||
|
||||
|
||||
def create_item(client, box_id, name="Item A", note="Note", quantity="2", is_container=False):
|
||||
def create_item(
|
||||
client,
|
||||
box_id,
|
||||
name="Item A",
|
||||
note="Note",
|
||||
quantity="2",
|
||||
is_container=False,
|
||||
image=None,
|
||||
):
|
||||
data = {"name": name, "note": note, "quantity": quantity}
|
||||
if is_container:
|
||||
data["is_container"] = "on"
|
||||
return client.post(f"/boxes/{box_id}/items", data=data, follow_redirects=False)
|
||||
files = {"image_file": image} if image is not None else None
|
||||
return client.post(f"/boxes/{box_id}/items", data=data, files=files, follow_redirects=False)
|
||||
|
||||
|
||||
def create_subitem(client, item_id, name="SubItem A", note="Small", quantity="3"):
|
||||
def create_subitem(client, item_id, name="SubItem A", note="Small", quantity="3", image=None):
|
||||
files = {"image_file": image} if image is not None else None
|
||||
return client.post(
|
||||
f"/items/{item_id}/subitems",
|
||||
data={"name": name, "note": note, "quantity": quantity},
|
||||
files=files,
|
||||
follow_redirects=False,
|
||||
)
|
||||
|
||||
|
||||
def make_image_upload(
|
||||
filename="sample.png",
|
||||
image_format="PNG",
|
||||
size=(2000, 1000),
|
||||
color=(255, 0, 0),
|
||||
mode="RGB",
|
||||
):
|
||||
image = Image.new(mode, size, color)
|
||||
output = BytesIO()
|
||||
image.save(output, format=image_format)
|
||||
return (filename, output.getvalue(), "image/png")
|
||||
|
||||
|
||||
def read_jpeg_size(image_bytes):
|
||||
with Image.open(BytesIO(image_bytes)) as image:
|
||||
return image.format, image.size
|
||||
|
||||
|
||||
def test_app_uses_isolated_sqlite_database(client):
|
||||
assert db_module.engine is not None
|
||||
|
||||
@@ -299,3 +329,236 @@ def test_post_redirects_are_reasonable(client, db_session):
|
||||
|
||||
assert item_response.headers["location"] == f"/items/{item.id}"
|
||||
assert subitem_response.headers["location"] == f"/items/{item.id}"
|
||||
|
||||
|
||||
def test_can_upload_image_for_box_and_process_it(client, db_session):
|
||||
response = create_box(client, name="Photo Box", image=make_image_upload())
|
||||
|
||||
assert response.status_code == 303
|
||||
|
||||
box = db_session.query(Box).one()
|
||||
assert box.image_blob is not None
|
||||
assert box.image_mime_type == "image/jpeg"
|
||||
assert box.image_width == 1600
|
||||
assert box.image_height == 800
|
||||
|
||||
image_format, image_size = read_jpeg_size(box.image_blob)
|
||||
assert image_format == "JPEG"
|
||||
assert image_size == (1600, 800)
|
||||
|
||||
|
||||
def test_can_upload_image_for_item(client, db_session):
|
||||
box = Box(name="Main Box")
|
||||
db_session.add(box)
|
||||
db_session.commit()
|
||||
|
||||
response = create_item(client, box.id, name="Book", image=make_image_upload())
|
||||
|
||||
assert response.status_code == 303
|
||||
|
||||
item = db_session.query(Item).one()
|
||||
assert item.image_blob is not None
|
||||
assert item.image_mime_type == "image/jpeg"
|
||||
|
||||
|
||||
def test_can_upload_image_for_subitem(client, db_session):
|
||||
box = Box(name="Main Box")
|
||||
item = Item(name="Pouch", box=box, is_container=True)
|
||||
db_session.add_all([box, item])
|
||||
db_session.commit()
|
||||
|
||||
response = create_subitem(client, item.id, name="Cable", image=make_image_upload())
|
||||
|
||||
assert response.status_code == 303
|
||||
|
||||
subitem = db_session.query(SubItem).one()
|
||||
assert subitem.image_blob is not None
|
||||
assert subitem.image_mime_type == "image/jpeg"
|
||||
|
||||
|
||||
def test_box_image_route_returns_jpeg(client, db_session):
|
||||
box = Box(name="Photo Box")
|
||||
db_session.add(box)
|
||||
db_session.commit()
|
||||
|
||||
client.post(
|
||||
f"/boxes/{box.id}/update",
|
||||
data={"name": box.name, "note": "", "room": "", "status": ""},
|
||||
files={"image_file": make_image_upload()},
|
||||
follow_redirects=False,
|
||||
)
|
||||
db_session.expire_all()
|
||||
|
||||
response = client.get(f"/boxes/{box.id}/image")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.headers["content-type"] == "image/jpeg"
|
||||
|
||||
|
||||
def test_item_image_route_returns_jpeg(client, db_session):
|
||||
box = Box(name="Main Box")
|
||||
item = Item(name="Lamp", box=box, is_container=False)
|
||||
db_session.add_all([box, item])
|
||||
db_session.commit()
|
||||
|
||||
client.post(
|
||||
f"/items/{item.id}/update",
|
||||
data={"name": item.name, "note": "", "quantity": "", "is_container": ""},
|
||||
files={"image_file": make_image_upload()},
|
||||
follow_redirects=False,
|
||||
)
|
||||
|
||||
response = client.get(f"/items/{item.id}/image")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.headers["content-type"] == "image/jpeg"
|
||||
|
||||
|
||||
def test_subitem_image_route_returns_jpeg(client, db_session):
|
||||
box = Box(name="Main Box")
|
||||
item = Item(name="Pouch", box=box, is_container=True)
|
||||
subitem = SubItem(name="Cable", parent_item=item)
|
||||
db_session.add_all([box, item, subitem])
|
||||
db_session.commit()
|
||||
|
||||
client.post(
|
||||
f"/subitems/{subitem.id}/update",
|
||||
data={"name": subitem.name, "note": "", "quantity": ""},
|
||||
files={"image_file": make_image_upload()},
|
||||
follow_redirects=False,
|
||||
)
|
||||
|
||||
response = client.get(f"/subitems/{subitem.id}/image")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.headers["content-type"] == "image/jpeg"
|
||||
|
||||
|
||||
def test_image_routes_return_404_when_no_image(client, db_session):
|
||||
box = Box(name="Main Box")
|
||||
item = Item(name="Lamp", box=box, is_container=False)
|
||||
subitem = SubItem(name="Cable", parent_item=Item(name="Pouch", box=box, is_container=True))
|
||||
db_session.add_all([box, item, subitem.parent_item, subitem])
|
||||
db_session.commit()
|
||||
|
||||
assert client.get(f"/boxes/{box.id}/image").status_code == 404
|
||||
assert client.get(f"/items/{item.id}/image").status_code == 404
|
||||
assert client.get(f"/subitems/{subitem.id}/image").status_code == 404
|
||||
|
||||
|
||||
def test_image_routes_return_404_for_missing_objects(client):
|
||||
assert client.get("/boxes/9999/image").status_code == 404
|
||||
assert client.get("/items/9999/image").status_code == 404
|
||||
assert client.get("/subitems/9999/image").status_code == 404
|
||||
|
||||
|
||||
def test_can_replace_existing_image(client, db_session):
|
||||
box = Box(name="Photo Box")
|
||||
db_session.add(box)
|
||||
db_session.commit()
|
||||
|
||||
client.post(
|
||||
f"/boxes/{box.id}/update",
|
||||
data={"name": box.name, "note": "", "room": "", "status": ""},
|
||||
files={"image_file": make_image_upload(color=(255, 0, 0))},
|
||||
follow_redirects=False,
|
||||
)
|
||||
db_session.expire_all()
|
||||
first_blob = db_session.get(Box, box.id).image_blob
|
||||
|
||||
client.post(
|
||||
f"/boxes/{box.id}/update",
|
||||
data={"name": box.name, "note": "", "room": "", "status": ""},
|
||||
files={"image_file": make_image_upload(color=(0, 255, 0))},
|
||||
follow_redirects=False,
|
||||
)
|
||||
db_session.expire_all()
|
||||
updated_box = db_session.get(Box, box.id)
|
||||
|
||||
assert updated_box.image_blob is not None
|
||||
assert updated_box.image_blob != first_blob
|
||||
|
||||
|
||||
def test_can_delete_existing_image_and_clear_metadata(client, db_session):
|
||||
box = Box(name="Photo Box")
|
||||
db_session.add(box)
|
||||
db_session.commit()
|
||||
|
||||
client.post(
|
||||
f"/boxes/{box.id}/update",
|
||||
data={"name": box.name, "note": "", "room": "", "status": ""},
|
||||
files={"image_file": make_image_upload()},
|
||||
follow_redirects=False,
|
||||
)
|
||||
|
||||
response = client.post(f"/boxes/{box.id}/image/delete", follow_redirects=False)
|
||||
|
||||
assert response.status_code == 303
|
||||
db_session.expire_all()
|
||||
updated_box = db_session.get(Box, box.id)
|
||||
assert updated_box.image_blob is None
|
||||
assert updated_box.image_mime_type is None
|
||||
assert updated_box.image_width is None
|
||||
assert updated_box.image_height is None
|
||||
assert client.get(f"/boxes/{box.id}/image").status_code == 404
|
||||
|
||||
|
||||
def test_deleting_image_when_none_exists_is_safe(client, db_session):
|
||||
item = Item(name="No Image Item", box=Box(name="Main Box"), is_container=False)
|
||||
db_session.add(item)
|
||||
db_session.commit()
|
||||
|
||||
response = client.post(f"/items/{item.id}/image/delete", follow_redirects=False)
|
||||
|
||||
assert response.status_code == 303
|
||||
|
||||
|
||||
def test_uploading_non_image_file_returns_400_and_does_not_write_data(client, db_session):
|
||||
box = Box(name="Main Box")
|
||||
db_session.add(box)
|
||||
db_session.commit()
|
||||
|
||||
response = client.post(
|
||||
f"/items/{box.id}/subitems",
|
||||
data={"name": "Should Fail", "note": "", "quantity": ""},
|
||||
files={"image_file": ("bad.txt", b"not an image", "text/plain")},
|
||||
follow_redirects=False,
|
||||
)
|
||||
|
||||
assert response.status_code in {400, 404}
|
||||
|
||||
item = Item(name="Pouch", box=box, is_container=True)
|
||||
db_session.add(item)
|
||||
db_session.commit()
|
||||
|
||||
response = create_subitem(
|
||||
client,
|
||||
item.id,
|
||||
name="Should Fail",
|
||||
image=("bad.txt", b"not an image", "text/plain"),
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert response.json()["detail"] == "上传的文件不是合法图片"
|
||||
assert db_session.query(SubItem).count() == 0
|
||||
|
||||
|
||||
def test_broken_image_processing_returns_400_and_keeps_image_fields_empty(client, db_session):
|
||||
box = Box(name="Main Box")
|
||||
db_session.add(box)
|
||||
db_session.commit()
|
||||
|
||||
response = client.post(
|
||||
f"/boxes/{box.id}/update",
|
||||
data={"name": box.name, "note": "", "room": "", "status": ""},
|
||||
files={"image_file": ("broken.jpg", b"\xff\xd8\xff\xdbbroken", "image/jpeg")},
|
||||
follow_redirects=False,
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
db_session.expire_all()
|
||||
updated_box = db_session.get(Box, box.id)
|
||||
assert updated_box.image_blob is None
|
||||
assert updated_box.image_mime_type is None
|
||||
assert updated_box.image_width is None
|
||||
assert updated_box.image_height is None
|
||||
|
||||
Reference in New Issue
Block a user