mirror of
https://github.com/wahyd4/links.git
synced 2026-08-09 05:06:16 +10:00
207 lines
8.0 KiB
Python
207 lines
8.0 KiB
Python
"""
|
|
Integration tests for the Files REST API (/api/files) and download endpoint.
|
|
"""
|
|
|
|
import pytest
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
|
|
|
|
def _png():
|
|
data = (
|
|
b"\x89PNG\r\n\x1a\n"
|
|
b"\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01"
|
|
b"\x08\x02\x00\x00\x00\x90wS\xde"
|
|
b"\x00\x00\x00\x0cIDATx\x9cc\xf8\x0f\x00\x00\x01\x01\x00\x05\x18\xd8N"
|
|
b"\x00\x00\x00\x00IEND\xaeB`\x82"
|
|
)
|
|
return SimpleUploadedFile("test.png", data, content_type="image/png")
|
|
|
|
|
|
def _pdf():
|
|
data = (
|
|
b"%PDF-1.4\n1 0 obj<</Type/Catalog/Pages 2 0 R>>endobj\n"
|
|
b"2 0 obj<</Type/Pages/Kids[3 0 R]/Count 1>>endobj\n"
|
|
b"3 0 obj<</Type/Page/MediaBox[0 0 612 792]/Parent 2 0 R>>endobj\n"
|
|
b"xref\n0 4\ntrailer<</Size 4/Root 1 0 R>>\nstartxref\n0\n%%EOF"
|
|
)
|
|
return SimpleUploadedFile("document.pdf", data, content_type="application/pdf")
|
|
|
|
|
|
def _mp4_generic_content_type():
|
|
"""An .mp4 upload whose client sent a generic Content-Type.
|
|
|
|
Reproduces the iOS-uploader behaviour that caused mb4 files to be
|
|
persisted as "application/octet-stream" and therefore un-previewable
|
|
in the UI. The server must infer "video/mp4" from the filename instead.
|
|
"""
|
|
# minimal mp4 box header — content is irrelevant, only the extension
|
|
# and Content-Type matter for the mime-type inference path.
|
|
data = b"\x00\x00\x00\x20ftypisom\x00\x00\x02\x00isomiso2avc1mp41"
|
|
return SimpleUploadedFile(
|
|
"clip.mp4", data, content_type="application/octet-stream"
|
|
)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
class TestFilesAPI:
|
|
BASE = "/api/files"
|
|
|
|
def test_list_empty(self, api_client):
|
|
r = api_client.get(f"{self.BASE}")
|
|
assert r.status_code == 200
|
|
assert r.json()["count"] == 0
|
|
|
|
def test_upload_single_file(self, api_client):
|
|
r = api_client.post(f"{self.BASE}", {"files": _png()}, format="multipart")
|
|
assert r.status_code == 201
|
|
results = r.json()
|
|
assert len(results) == 1
|
|
assert results[0]["name"] == "test.png"
|
|
assert results[0]["mime_type"] == "image/png"
|
|
assert results[0]["is_public"] is False
|
|
assert "id" in results[0]
|
|
|
|
def test_upload_multiple_files(self, api_client):
|
|
r = api_client.post(
|
|
f"{self.BASE}",
|
|
{"files": [_png(), _pdf()]},
|
|
format="multipart",
|
|
)
|
|
assert r.status_code == 201
|
|
assert len(r.json()) == 2
|
|
|
|
def test_list_after_upload(self, api_client):
|
|
api_client.post(f"{self.BASE}", {"files": _png()}, format="multipart")
|
|
api_client.post(f"{self.BASE}", {"files": _pdf()}, format="multipart")
|
|
r = api_client.get(f"{self.BASE}")
|
|
assert r.json()["count"] == 2
|
|
|
|
def test_get_file_detail(self, api_client):
|
|
pk = api_client.post(
|
|
f"{self.BASE}", {"files": _png()}, format="multipart"
|
|
).json()[0]["id"]
|
|
r = api_client.get(f"{self.BASE}/{pk}")
|
|
assert r.status_code == 200
|
|
assert r.json()["name"] == "test.png"
|
|
|
|
def test_upload_no_files_returns_400(self, api_client):
|
|
r = api_client.post(f"{self.BASE}", {}, format="multipart")
|
|
assert r.status_code == 400
|
|
|
|
def test_upload_write_failure_returns_500_and_no_phantom_record(self, api_client, monkeypatch):
|
|
"""If the file can't be fully persisted to disk (e.g. storage failure),
|
|
no FileUpload DB record should be created — otherwise it points at a
|
|
file that will always 404."""
|
|
import os as _os
|
|
|
|
from links import file_views
|
|
|
|
def _boom(*args, **kwargs):
|
|
raise OSError("simulated storage failure")
|
|
|
|
monkeypatch.setattr(file_views.os, "fsync", _boom)
|
|
|
|
from links.models import FileUpload
|
|
|
|
r = api_client.post(f"{self.BASE}", {"files": _png()}, format="multipart")
|
|
assert r.status_code == 500
|
|
assert "errors" in r.json()
|
|
assert FileUpload.objects.count() == 0
|
|
# No leftover partial file should remain in the uploads folder.
|
|
assert list(_os.scandir(file_views._get_upload_folder())) == []
|
|
|
|
def test_delete_file(self, api_client):
|
|
pk = api_client.post(
|
|
f"{self.BASE}", {"files": _png()}, format="multipart"
|
|
).json()[0]["id"]
|
|
r = api_client.delete(f"{self.BASE}/{pk}")
|
|
assert r.status_code == 204
|
|
assert api_client.get(f"{self.BASE}/{pk}").status_code == 404
|
|
|
|
def test_download_url_works(self, client, api_client):
|
|
uploaded = api_client.post(
|
|
f"{self.BASE}", {"files": _png()}, format="multipart"
|
|
).json()[0]
|
|
pk, name = uploaded["id"], uploaded["name"]
|
|
r = client.get(f"/ui/files/{pk}-{name}")
|
|
assert r.status_code == 200
|
|
assert r["Content-Type"] == "image/png"
|
|
|
|
def test_download_increments_count(self, client, api_client):
|
|
uploaded = api_client.post(
|
|
f"{self.BASE}", {"files": _png()}, format="multipart"
|
|
).json()[0]
|
|
pk, name = uploaded["id"], uploaded["name"]
|
|
client.get(f"/ui/files/{pk}-{name}")
|
|
client.get(f"/ui/files/{pk}-{name}")
|
|
detail = api_client.get(f"{self.BASE}/{pk}").json()
|
|
assert detail["download_count"] == 2
|
|
|
|
def test_toggle_public(self, api_client):
|
|
pk = api_client.post(
|
|
f"{self.BASE}", {"files": _png()}, format="multipart"
|
|
).json()[0]["id"]
|
|
# Make public
|
|
r = api_client.post(f"{self.BASE}/{pk}/toggle-public")
|
|
assert r.status_code == 200
|
|
assert r.json()["is_public"] is True
|
|
assert r.json()["public_url"] is not None
|
|
# Make private again
|
|
r2 = api_client.post(f"{self.BASE}/{pk}/toggle-public")
|
|
assert r2.json()["is_public"] is False
|
|
assert r2.json()["public_url"] is None
|
|
|
|
def test_public_file_accessible(self, client, api_client):
|
|
pk = api_client.post(
|
|
f"{self.BASE}", {"files": _png()}, format="multipart"
|
|
).json()[0]["id"]
|
|
api_client.post(f"{self.BASE}/{pk}/toggle-public")
|
|
detail = api_client.get(f"{self.BASE}/{pk}").json()
|
|
r = client.get(detail["public_url"])
|
|
assert r.status_code == 200
|
|
|
|
def test_private_file_not_accessible_via_public_url(self, client, api_client):
|
|
pk = api_client.post(
|
|
f"{self.BASE}", {"files": _png()}, format="multipart"
|
|
).json()[0]["id"]
|
|
name = "test.png"
|
|
r = client.get(f"/public/files/{pk}-{name}")
|
|
assert r.status_code == 404
|
|
|
|
def test_set_expiry(self, api_client):
|
|
pk = api_client.post(
|
|
f"{self.BASE}", {"files": _png()}, format="multipart"
|
|
).json()[0]["id"]
|
|
r = api_client.post(
|
|
f"{self.BASE}/{pk}/set-expiry",
|
|
{"expires_at": "2099-12-31T00:00:00Z"},
|
|
format="json",
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json()["is_expired"] is False
|
|
|
|
def test_formatted_size_present(self, api_client):
|
|
uploaded = api_client.post(
|
|
f"{self.BASE}", {"files": _png()}, format="multipart"
|
|
).json()[0]
|
|
assert "formatted_size" in uploaded
|
|
assert "B" in uploaded["formatted_size"]
|
|
|
|
def test_upload_infers_mime_from_filename_when_client_sends_generic(self, api_client):
|
|
"""An .mp4 uploaded with Content-Type=application/octet-stream must
|
|
still be persisted as video/mp4 so the UI can render a <video>."""
|
|
uploaded = api_client.post(
|
|
f"{self.BASE}", {"files": _mp4_generic_content_type()}, format="multipart"
|
|
).json()[0]
|
|
assert uploaded["mime_type"] == "video/mp4"
|
|
|
|
def test_download_serves_inline_for_generic_content_type_mp4(self, client, api_client):
|
|
uploaded = api_client.post(
|
|
f"{self.BASE}", {"files": _mp4_generic_content_type()}, format="multipart"
|
|
).json()[0]
|
|
pk, name = uploaded["id"], uploaded["name"]
|
|
r = client.get(f"/ui/files/{pk}-{name}")
|
|
assert r.status_code == 200
|
|
assert r["Content-Type"] == "video/mp4"
|
|
assert (r["Content-Disposition"] or "").startswith("inline"), r["Content-Disposition"]
|