63 lines
2.3 KiB
Python
63 lines
2.3 KiB
Python
|
"""Attachment upload validation, metadata, download, and closure locking."""
|
||
|
|
from .util import create_ncr, hdr, to_closed
|
||
|
|
|
||
|
|
TINY_PNG = (
|
||
|
|
b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01"
|
||
|
|
b"\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00\x01"
|
||
|
|
b"\x00\x00\x05\x00\x01\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
async def test_upload_download_and_metadata(client, team):
|
||
|
|
ncr = await create_ncr(client, team)
|
||
|
|
r = await client.post(
|
||
|
|
f"/api/ncrs/{ncr['id']}/attachments",
|
||
|
|
files=[
|
||
|
|
("files", ("photo one.png", TINY_PNG, "image/png")),
|
||
|
|
("files", ("notes.txt", b"observed at station 4", "text/plain")),
|
||
|
|
],
|
||
|
|
headers=hdr(team["requester"]),
|
||
|
|
)
|
||
|
|
assert r.status_code == 201, r.text
|
||
|
|
items = r.json()
|
||
|
|
assert len(items) == 2
|
||
|
|
png = next(i for i in items if i["is_image"])
|
||
|
|
assert png["original_filename"] == "photo one.png"
|
||
|
|
assert png["uploaded_by"]["email"] == team["requester"]
|
||
|
|
assert png["size_bytes"] == len(TINY_PNG)
|
||
|
|
|
||
|
|
r = await client.get(
|
||
|
|
f"/api/attachments/{png['id']}/download", headers=hdr(team["ops"])
|
||
|
|
)
|
||
|
|
assert r.status_code == 200
|
||
|
|
assert r.content == TINY_PNG
|
||
|
|
|
||
|
|
# attachment add shows in detail + audit
|
||
|
|
detail = (await client.get(f"/api/ncrs/{ncr['id']}", headers=hdr(team["qc"]))).json()
|
||
|
|
assert len(detail["attachments"]) == 2
|
||
|
|
audit = (
|
||
|
|
await client.get(f"/api/ncrs/{ncr['id']}/audit", headers=hdr(team["qc"]))
|
||
|
|
).json()
|
||
|
|
assert sum(1 for a in audit["items"] if a["action"] == "attachment_add") == 2
|
||
|
|
|
||
|
|
|
||
|
|
async def test_disallowed_type_rejected(client, team):
|
||
|
|
ncr = await create_ncr(client, team)
|
||
|
|
r = await client.post(
|
||
|
|
f"/api/ncrs/{ncr['id']}/attachments",
|
||
|
|
files=[("files", ("malware.exe", b"MZ...", "application/octet-stream"))],
|
||
|
|
headers=hdr(team["requester"]),
|
||
|
|
)
|
||
|
|
assert r.status_code == 422
|
||
|
|
assert "not allowed" in r.json()["detail"]
|
||
|
|
|
||
|
|
|
||
|
|
async def test_attachments_locked_when_closed(client, team):
|
||
|
|
ncr = await to_closed(client, team, (await create_ncr(client, team))["id"])
|
||
|
|
r = await client.post(
|
||
|
|
f"/api/ncrs/{ncr['id']}/attachments",
|
||
|
|
files=[("files", ("late.png", TINY_PNG, "image/png"))],
|
||
|
|
headers=hdr(team["requester"]),
|
||
|
|
)
|
||
|
|
assert r.status_code == 409
|