375 lines
24 KiB
Python
375 lines
24 KiB
Python
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from fastapi.testclient import TestClient
|
|
from PIL import Image
|
|
from httpx import ReadTimeout, RemoteProtocolError
|
|
from supabase import PostgrestAPIError
|
|
from dropbox.exceptions import ApiError, AuthError
|
|
from dropbox.files import DeleteError, LookupError
|
|
|
|
from backend import main
|
|
|
|
|
|
class ArticleMediaTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.directory = TemporaryDirectory()
|
|
self.addCleanup(self.directory.cleanup)
|
|
root = Path(self.directory.name)
|
|
(root / "posts.json").write_text("[]", encoding="utf-8")
|
|
for mocked in (
|
|
patch.object(main, "DATA_DIR", root),
|
|
patch.object(main, "POSTS_PATH", root / "posts.json"),
|
|
patch.object(main, "UPLOAD_DIR", root / "uploads"),
|
|
patch.object(main, "get_supabase", return_value=None),
|
|
patch.dict(os.environ, {"JOURNAL_ADMIN_PASSWORD": "test-password", "JOURNAL_TOKEN_SECRET": "test-secret", "JOURNAL_MEDIA_STORAGE": "local"}),
|
|
):
|
|
mocked.start()
|
|
self.addCleanup(mocked.stop)
|
|
main.read_data.cache_clear()
|
|
self.addCleanup(main.read_data.cache_clear)
|
|
self.client = TestClient(main.app)
|
|
token = self.client.post("/api/auth/login", json={"password": "test-password"}).json()["access_token"]
|
|
self.headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
def upload(self, data, purpose="attachment", name="photo.png"):
|
|
return self.client.post("/api/uploads", params={"name": name, "purpose": purpose}, content=data, headers=self.headers)
|
|
|
|
def test_media_survives_publish_and_can_be_retrieved(self):
|
|
image = io.BytesIO()
|
|
Image.new("RGB", (20, 20), "green").save(image, format="PNG")
|
|
banner_response = self.upload(image.getvalue(), "banner")
|
|
self.assertEqual(banner_response.status_code, 201)
|
|
banner = banner_response.json()
|
|
attachment = self.upload(b"Supporting notes", name="notes.txt").json()
|
|
document = {"type": "doc", "content": [
|
|
{"type": "heading", "attrs": {"level": 2}, "content": [{"type": "text", "text": "First section"}]},
|
|
{"type": "paragraph", "content": [{"type": "text", "text": "an opening paragraph with enough text to publish this article."}]},
|
|
{"type": "heading", "attrs": {"level": 2}, "content": [{"type": "text", "text": "Second section"}]},
|
|
{"type": "paragraph", "content": [{"type": "text", "text": "the next section keeps its separate heading and body."}]},
|
|
]}
|
|
response = self.client.post("/api/posts", headers=self.headers, json={
|
|
"title": "An article with media", "excerpt": "A useful introduction to this test article.",
|
|
"published_at": "2026-09-12", "read_time": 3, "tags": ["AI"],
|
|
"content": document, "banner": banner, "attachments": [attachment],
|
|
})
|
|
self.assertEqual(response.status_code, 201, response.text)
|
|
post = self.client.get("/api/posts/an-article-with-media").json()
|
|
self.assertEqual(post["content"], document)
|
|
self.assertEqual(post["attachments"], [attachment])
|
|
self.assertEqual(self.client.get("/api/posts").json()[0]["banner"], banner)
|
|
self.assertEqual(json.loads(main.POSTS_PATH.read_text())[0]["banner"], banner)
|
|
photo = self.client.get(banner["url"])
|
|
self.assertEqual(photo.content, image.getvalue())
|
|
self.assertEqual(photo.headers["content-type"], "image/png")
|
|
file = self.client.get(attachment["url"])
|
|
self.assertIn("attachment", file.headers["content-disposition"])
|
|
self.assertEqual(file.content, b"Supporting notes")
|
|
|
|
def test_invalid_uploads_and_unauthenticated_requests(self):
|
|
self.assertEqual(self.client.post("/api/uploads?name=a.txt", content=b"test").status_code, 401)
|
|
self.assertEqual(self.upload(b"not a photo", "banner").status_code, 422)
|
|
self.assertEqual(self.upload(b"").status_code, 422)
|
|
self.assertEqual(self.upload(b"x" * (8 * 1024 * 1024 + 1), "banner").status_code, 413)
|
|
self.assertEqual(self.client.get("/api/uploads/invalid").status_code, 404)
|
|
unsafe = self.upload(b"<script>alert(1)</script>", name="page.html").json()
|
|
downloaded = self.client.get(unsafe["url"])
|
|
self.assertEqual(downloaded.headers["content-type"], "application/octet-stream")
|
|
self.assertEqual(downloaded.headers["x-content-type-options"], "nosniff")
|
|
self.assertIn("attachment", downloaded.headers["content-disposition"])
|
|
|
|
def article_payload(self):
|
|
return {
|
|
"title": "An editable article", "excerpt": "An introduction with enough detail for readers.",
|
|
"published_at": "2026-09-12", "read_time": 3, "tags": ["Custom topic"],
|
|
"content": {"type": "doc", "content": [{"type": "paragraph", "content": [
|
|
{"type": "text", "text": "An original article body with enough text to validate and publish."}
|
|
]}]},
|
|
}
|
|
|
|
def test_admin_role_and_protected_crud(self):
|
|
self.assertEqual(self.client.post("/api/auth/login", json={"password": "wrong"}).status_code, 401)
|
|
session = self.client.get("/api/auth/session", headers=self.headers)
|
|
self.assertEqual(session.json(), {"authenticated": True, "role": "admin"})
|
|
payload = self.article_payload()
|
|
self.assertEqual(self.client.post("/api/posts", json=payload).status_code, 401)
|
|
created = self.client.post("/api/posts", json=payload, headers=self.headers).json()
|
|
path = f'/api/posts/{created["slug"]}'
|
|
for headers in ({}, {"Authorization": "Bearer 1.invalid"}):
|
|
self.assertEqual(self.client.put(path, json=payload, headers=headers).status_code, 401)
|
|
self.assertEqual(self.client.delete(path, headers=headers).status_code, 401)
|
|
self.assertEqual(self.client.get(path).json()["title"], payload["title"])
|
|
|
|
payload["title"] = "The updated article title"
|
|
payload["tags"] = ["Updated topic"]
|
|
payload["attachments"] = [self.upload(b"updated notes", name="notes.txt").json()]
|
|
updated = self.client.put(path, json=payload, headers=self.headers)
|
|
self.assertEqual(updated.status_code, 200, updated.text)
|
|
self.assertEqual(updated.json()["slug"], created["slug"])
|
|
self.assertEqual(self.client.get(path).json()["attachments"], payload["attachments"])
|
|
self.assertEqual(self.client.get("/api/posts").json()[0]["title"], payload["title"])
|
|
invalid = {**payload, "content": {"type": "doc", "content": []}}
|
|
self.assertEqual(self.client.put(path, json=invalid, headers=self.headers).status_code, 422)
|
|
self.assertEqual(self.client.put("/api/posts/missing", json=payload, headers=self.headers).status_code, 404)
|
|
self.assertEqual(self.client.delete(path, headers=self.headers).status_code, 200)
|
|
self.assertEqual(self.client.get(path).status_code, 404)
|
|
self.assertEqual(self.client.get("/api/posts").json(), [])
|
|
self.assertEqual(json.loads(main.POSTS_PATH.read_text()), [])
|
|
self.assertEqual(self.client.delete(path, headers=self.headers).status_code, 404)
|
|
|
|
def test_supabase_update_delete_are_scoped_and_report_failures(self):
|
|
client = MagicMock()
|
|
table = client.table.return_value
|
|
payload = self.article_payload()
|
|
table.update.return_value.eq.return_value.execute.return_value.data = [{**payload, "slug": "original-url"}]
|
|
table.select.return_value.eq.return_value.limit.return_value.retry.return_value.execute.return_value.data = [{**payload, "slug": "original-url"}]
|
|
table.select.return_value.order.return_value.retry.return_value.execute.return_value.data = []
|
|
table.delete.return_value.eq.return_value.execute.return_value.data = [{"slug": "original-url"}]
|
|
with patch.object(main, "get_supabase", return_value=client):
|
|
updated = self.client.put("/api/posts/original-url", json=payload, headers=self.headers)
|
|
self.assertEqual(updated.status_code, 200)
|
|
table.update.return_value.eq.assert_called_once_with("slug", "original-url")
|
|
self.assertEqual(self.client.delete("/api/posts/original-url", headers=self.headers).status_code, 200)
|
|
table.delete.return_value.eq.assert_called_once_with("slug", "original-url")
|
|
table.update.return_value.eq.return_value.execute.return_value.data = []
|
|
self.assertEqual(self.client.put("/api/posts/missing", json=payload, headers=self.headers).status_code, 404)
|
|
table.delete.return_value.eq.return_value.execute.return_value.data = []
|
|
self.assertEqual(self.client.delete("/api/posts/missing", headers=self.headers).status_code, 404)
|
|
table.update.return_value.eq.return_value.execute.side_effect = RuntimeError("offline")
|
|
self.assertEqual(self.client.put("/api/posts/original-url", json=payload, headers=self.headers).status_code, 502)
|
|
table.delete.return_value.eq.return_value.execute.side_effect = RuntimeError("offline")
|
|
self.assertEqual(self.client.delete("/api/posts/original-url", headers=self.headers).status_code, 502)
|
|
|
|
def test_journal_reads_retry_transient_errors_but_remain_public(self):
|
|
client = MagicMock()
|
|
query = MagicMock()
|
|
query.retry.return_value = query
|
|
client.table.return_value.select.return_value.order.return_value = query
|
|
client.table.return_value.select.return_value.eq.return_value.limit.return_value = query
|
|
article = {**self.article_payload(), "slug": "test-article", "accent": "mint"}
|
|
response = MagicMock(data=[article])
|
|
with patch.object(main, "get_supabase", return_value=client), patch.object(main.time, "sleep"):
|
|
for path in ("/api/posts", "/api/posts/test-article"):
|
|
for failure in (
|
|
RemoteProtocolError("connection ended"),
|
|
ReadTimeout("read timed out"),
|
|
PostgrestAPIError({"code": "503", "message": "unavailable"}),
|
|
):
|
|
query.execute.reset_mock()
|
|
query.execute.side_effect = [failure, response]
|
|
result = self.client.get(path) # No admin token: reading stays public.
|
|
self.assertEqual(result.status_code, 200, result.text)
|
|
self.assertEqual(query.execute.call_count, 2)
|
|
query.execute.reset_mock()
|
|
query.execute.side_effect = ReadTimeout("offline")
|
|
self.assertEqual(self.client.get(path).status_code, 502)
|
|
self.assertEqual(query.execute.call_count, 2)
|
|
query.execute.reset_mock()
|
|
query.execute.side_effect = PostgrestAPIError({"code": "42P01", "message": "missing table"})
|
|
self.assertEqual(self.client.get(path).status_code, 502)
|
|
self.assertEqual(query.execute.call_count, 1)
|
|
query.execute.side_effect = None
|
|
query.execute.return_value = MagicMock(data=[])
|
|
self.assertEqual(self.client.get("/api/posts").json(), [])
|
|
self.assertEqual(self.client.get("/api/posts/missing").status_code, 404)
|
|
|
|
def test_dropbox_upload_publish_and_download_use_file_id(self):
|
|
image = io.BytesIO()
|
|
Image.new("RGB", (20, 20), "blue").save(image, format="PNG")
|
|
data = image.getvalue()
|
|
dropbox = MagicMock()
|
|
dropbox.files_upload.return_value.id = "id:test_photo"
|
|
stream = MagicMock()
|
|
stream.iter_content.return_value = [data]
|
|
dropbox.files_download.return_value = (MagicMock(size=len(data)), stream)
|
|
with patch.dict(os.environ, {"JOURNAL_MEDIA_STORAGE": "dropbox"}), patch.object(main.dropbox_storage, "get_dropbox", return_value=dropbox):
|
|
result = self.upload(data, "banner")
|
|
self.assertEqual(result.status_code, 201, result.text)
|
|
media = result.json()
|
|
self.assertEqual(media["dropbox_file_id"], "id:test_photo")
|
|
self.assertEqual(media["storage"], "dropbox")
|
|
upload_id = media["url"].rsplit("/", 1)[-1]
|
|
self.assertFalse((main.UPLOAD_DIR / upload_id).exists())
|
|
payload = {**self.article_payload(), "banner": media, "attachments": [media]}
|
|
article = self.client.post("/api/posts", headers=self.headers, json=payload)
|
|
self.assertEqual(article.status_code, 201, article.text)
|
|
self.assertEqual(article.json()["banner"]["dropbox_file_id"], "id:test_photo")
|
|
saved = json.loads(main.POSTS_PATH.read_text())[0]
|
|
self.assertEqual(saved["attachments"][0]["dropbox_file_id"], "id:test_photo")
|
|
downloaded = self.client.get(media["url"])
|
|
self.assertEqual(downloaded.content, data)
|
|
self.assertEqual(downloaded.headers["content-type"], "image/png")
|
|
dropbox.files_download.assert_called_once_with("id:test_photo")
|
|
stream.close.assert_called_once()
|
|
payload["banner"]["dropbox_file_id"] = "id:someone_else"
|
|
self.assertEqual(self.client.put("/api/posts/an-editable-article", headers=self.headers, json=payload).status_code, 422)
|
|
|
|
def test_dropbox_media_metadata_survives_without_local_files_in_supabase(self):
|
|
client = MagicMock()
|
|
media_table = MagicMock()
|
|
article_table = MagicMock()
|
|
client.table.side_effect = lambda name: media_table if name == "article_media" else article_table
|
|
metadata = {"url": "/api/uploads/" + "a" * 32, "name": "image.png", "size": 10,
|
|
"media_type": "image/png", "storage": "dropbox", "dropbox_file_id": "id:remote_photo"}
|
|
media_table.upsert.return_value.retry.return_value.execute.return_value.data = [{"upload_id": "a" * 32, "metadata": metadata}]
|
|
media_query = media_table.select.return_value.eq.return_value.limit.return_value
|
|
media_query.retry.return_value.execute.return_value.data = [{"metadata": metadata}]
|
|
article_table.select.return_value.eq.return_value.limit.return_value.execute.return_value.data = []
|
|
article_table.insert.side_effect = lambda record: MagicMock(execute=lambda: MagicMock(data=[record]))
|
|
with patch.object(main, "get_supabase", return_value=client):
|
|
main.save_media_metadata("a" * 32, metadata)
|
|
media_table.upsert.assert_called_once_with([{"upload_id": "a" * 32, "metadata": metadata}], on_conflict="upload_id")
|
|
self.assertEqual(main.uploaded_media("a" * 32), metadata)
|
|
self.assertFalse(main.UPLOAD_DIR.exists())
|
|
result = self.client.post("/api/posts", headers=self.headers, json={**self.article_payload(), "banner": metadata})
|
|
self.assertEqual(result.status_code, 201, result.text)
|
|
self.assertEqual(article_table.insert.call_args.args[0]["banner"]["dropbox_file_id"], "id:remote_photo")
|
|
|
|
def test_dropbox_auth_failure_does_not_expire_journal_session(self):
|
|
dropbox = MagicMock()
|
|
dropbox.files_upload.side_effect = AuthError("test-request", "expired_access_token")
|
|
with patch.dict(os.environ, {"JOURNAL_MEDIA_STORAGE": "dropbox"}), patch.object(main.dropbox_storage, "get_dropbox", return_value=dropbox):
|
|
self.assertEqual(self.upload(b"notes", name="notes.txt").status_code, 503)
|
|
self.assertEqual(self.client.get("/api/auth/session", headers=self.headers).status_code, 200)
|
|
self.assertFalse(main.UPLOAD_DIR.exists())
|
|
|
|
def test_dropbox_migration_preserves_urls_originals_and_can_resume(self):
|
|
from backend.migrate_media_dropbox import migrate
|
|
media = self.upload(b"original notes", name="notes.txt").json()
|
|
result = self.client.post("/api/posts", headers=self.headers, json={**self.article_payload(), "attachments": [media]})
|
|
self.assertEqual(result.status_code, 201)
|
|
dropbox = MagicMock()
|
|
dropbox.files_upload.return_value.id = "id:migrated_file"
|
|
with patch.object(main.dropbox_storage, "get_dropbox", return_value=dropbox):
|
|
migrate(apply=False)
|
|
dropbox.files_upload.assert_not_called()
|
|
migrate(apply=True)
|
|
migrate(apply=True)
|
|
dropbox.files_upload.assert_called_once()
|
|
post = self.client.get("/api/posts/an-editable-article").json()
|
|
migrated = post["attachments"][0]
|
|
self.assertEqual(migrated["url"], media["url"])
|
|
self.assertEqual(migrated["dropbox_file_id"], "id:migrated_file")
|
|
self.assertEqual((main.UPLOAD_DIR / media["url"].rsplit("/", 1)[-1]).read_bytes(), b"original notes")
|
|
|
|
def test_edit_open_before_migration_keeps_old_media_and_new_uploads(self):
|
|
from backend.migrate_media_dropbox import migrate
|
|
image = io.BytesIO()
|
|
Image.new("RGB", (20, 20), "green").save(image, format="PNG")
|
|
banner = self.upload(image.getvalue(), "banner").json()
|
|
attachment = self.upload(b"original notes", name="notes.txt").json()
|
|
draft = {**self.article_payload(), "banner": banner, "attachments": [attachment]}
|
|
created = self.client.post("/api/posts", headers=self.headers, json=draft)
|
|
self.assertEqual(created.status_code, 201)
|
|
path = "/api/posts/an-editable-article"
|
|
dropbox = MagicMock()
|
|
dropbox.files_upload.side_effect = [
|
|
MagicMock(id="id:migrated_banner"), MagicMock(id="id:migrated_attachment"),
|
|
MagicMock(id="id:new_attachment"),
|
|
]
|
|
with patch.object(main.dropbox_storage, "get_dropbox", return_value=dropbox):
|
|
migrate(apply=True)
|
|
with patch.dict(os.environ, {"JOURNAL_MEDIA_STORAGE": "dropbox"}):
|
|
uploaded = self.upload(b"new notes", name="new.txt")
|
|
self.assertEqual(uploaded.status_code, 201)
|
|
draft["attachments"].append(uploaded.json())
|
|
# The browser still has the pre-migration metadata for the old files.
|
|
draft["title"] = "An edited article after migration"
|
|
saved = self.client.put(path, headers=self.headers, json=draft)
|
|
self.assertEqual(saved.status_code, 200, saved.text)
|
|
result = self.client.get(path).json()
|
|
self.assertEqual(result["title"], draft["title"])
|
|
self.assertEqual(result["banner"]["dropbox_file_id"], "id:migrated_banner")
|
|
self.assertEqual([m["dropbox_file_id"] for m in result["attachments"]],
|
|
["id:migrated_attachment", "id:new_attachment"])
|
|
self.assertEqual(result["banner"]["url"], banner["url"])
|
|
self.assertEqual(result["attachments"][0]["url"], attachment["url"])
|
|
# Compatibility must not accept altered file details or a substituted ID.
|
|
for changes in ({"size": 999}, {"name": "different.txt"},
|
|
{"media_type": "image/png"}, {"dropbox_file_id": "id:other"},
|
|
{"storage": "dropbox"}):
|
|
with self.subTest(changes=changes):
|
|
invalid = {**draft, "attachments": [{**attachment, **changes}]}
|
|
rejected = self.client.put(path, headers=self.headers, json=invalid)
|
|
self.assertEqual(rejected.status_code, 422)
|
|
|
|
def test_removal_deletes_dropbox_files_after_save_and_preserves_shared_media(self):
|
|
dropbox = MagicMock()
|
|
dropbox.files_upload.side_effect = [MagicMock(id=f"id:file_{i}") for i in range(4)]
|
|
image = io.BytesIO()
|
|
Image.new("RGB", (8, 8), "blue").save(image, format="PNG")
|
|
with patch.dict(os.environ, {"JOURNAL_MEDIA_STORAGE": "dropbox"}), patch.object(main.dropbox_storage, "get_dropbox", return_value=dropbox):
|
|
banner = self.upload(image.getvalue(), "banner").json()
|
|
attachment = self.upload(b"shared notes").json()
|
|
replacement = self.upload(image.getvalue(), "banner").json()
|
|
discarded = self.upload(b"removed before publishing").json()
|
|
payload = {**self.article_payload(), "banner": banner, "attachments": [attachment]}
|
|
self.assertEqual(self.client.post("/api/posts", headers=self.headers, json=payload).status_code, 201)
|
|
second = {**self.article_payload(), "title": "Another article with shared notes", "attachments": [attachment]}
|
|
self.assertEqual(self.client.post("/api/posts", headers=self.headers, json=second).status_code, 201)
|
|
edited = {**payload, "banner": replacement, "attachments": [], "discarded_uploads": [discarded["url"]]}
|
|
self.assertEqual(self.client.put("/api/posts/an-editable-article", headers=self.headers, json=edited).status_code, 200)
|
|
deleted = [call.args[0] for call in dropbox.files_delete_v2.call_args_list]
|
|
self.assertCountEqual(deleted, [banner["dropbox_file_id"], discarded["dropbox_file_id"]])
|
|
self.assertEqual(self.client.get(banner["url"]).status_code, 404)
|
|
self.assertEqual(main.uploaded_media(attachment["url"].rsplit("/", 1)[-1]), attachment)
|
|
self.assertEqual(self.client.delete("/api/posts/another-article-with-shared-notes", headers=self.headers).status_code, 200)
|
|
self.assertEqual(self.client.delete("/api/posts/an-editable-article", headers=self.headers).status_code, 200)
|
|
self.assertCountEqual([call.args[0] for call in dropbox.files_delete_v2.call_args_list],
|
|
[m["dropbox_file_id"] for m in [banner, attachment, replacement, discarded]])
|
|
self.assertEqual(main.media_cleanup.pending(), [])
|
|
|
|
def test_failed_file_deletion_is_durable_and_retryable_after_article_deleted(self):
|
|
dropbox = MagicMock()
|
|
dropbox.files_upload.return_value.id = "id:retry_delete"
|
|
dropbox.files_delete_v2.side_effect = AuthError("request", None)
|
|
with patch.dict(os.environ, {"JOURNAL_MEDIA_STORAGE": "dropbox"}), patch.object(main.dropbox_storage, "get_dropbox", return_value=dropbox):
|
|
media = self.upload(b"retry notes").json()
|
|
payload = {**self.article_payload(), "attachments": [media]}
|
|
self.client.post("/api/posts", headers=self.headers, json=payload)
|
|
response = self.client.delete("/api/posts/an-editable-article", headers=self.headers)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(len(main.media_cleanup.pending()), 1)
|
|
self.assertEqual(self.client.get("/api/auth/session", headers=self.headers).status_code, 200)
|
|
self.assertEqual(main.uploaded_media(media["url"].rsplit("/", 1)[-1]), media)
|
|
# Dropbox already removed the file, e.g. before a database outage.
|
|
dropbox.files_delete_v2.side_effect = ApiError("request", DeleteError.path_lookup(LookupError.not_found), None, None)
|
|
main.media_cleanup.cleanup_pending()
|
|
self.assertEqual(main.media_cleanup.pending(), [])
|
|
self.assertEqual(self.client.get(media["url"]).status_code, 404)
|
|
|
|
def test_failed_article_save_does_not_delete_referenced_file(self):
|
|
media = self.upload(b"must survive failed save").json()
|
|
payload = {**self.article_payload(), "attachments": [media]}
|
|
self.client.post("/api/posts", headers=self.headers, json=payload)
|
|
with patch.object(main, "save_local_posts", side_effect=OSError("disk unavailable")):
|
|
with self.assertRaises(OSError):
|
|
self.client.put("/api/posts/an-editable-article", headers=self.headers, json=self.article_payload())
|
|
main.read_data.cache_clear()
|
|
main.media_cleanup.cleanup_pending()
|
|
self.assertEqual(self.client.get(media["url"]).content, b"must survive failed save")
|
|
self.assertEqual(len(main.media_cleanup.pending()), 1)
|
|
self.assertEqual(self.client.put("/api/posts/an-editable-article", headers=self.headers, json=self.article_payload()).status_code, 200)
|
|
self.assertEqual(self.client.get(media["url"]).status_code, 404)
|
|
self.assertEqual(main.media_cleanup.pending(), [])
|
|
|
|
def test_removing_new_upload_before_publish_cleans_it_after_success(self):
|
|
media = self.upload(b"discarded draft upload").json()
|
|
payload = {**self.article_payload(), "discarded_uploads": [media["url"]]}
|
|
invalid = {**payload, "content": {"type": "doc", "content": []}}
|
|
self.assertEqual(self.client.post("/api/posts", headers=self.headers, json=invalid).status_code, 422)
|
|
self.assertEqual(self.client.get(media["url"]).status_code, 200)
|
|
result = self.client.post("/api/posts", headers=self.headers, json=payload)
|
|
self.assertEqual(result.status_code, 201)
|
|
self.assertNotIn("discarded_uploads", result.json())
|
|
self.assertEqual(self.client.get(media["url"]).status_code, 404)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|