peter
4254164bef
Delete now moves an artifact to a recoverable Trash instead of hard-removing.
- Model: Artifact.deleted_at (nullable, indexed) = in Trash.
- Soft-delete everywhere: DELETE /api/artifacts/{id} sets deleted_at
(idempotent 204); session POST /a/{id}/delete for the feed's hover Delete
icon; POST /a/{id}/undelete restores; POST /feed/empty-trash purges.
- Trashed artifacts are off-quota (deleting frees quota) and eviction-protected
(never evicted early); hidden from feed, sources, counts, and all API reads
(list/detail/raw).
- Sidebar: "Deleted" and "Keep" system folders at the bottom (counts, hidden
when empty); ?folder=deleted / ?folder=keep views reuse the row rendering.
Deleted view has hover Undelete + an Empty Trash button. Keep = view over
is_kept. Live-poll disabled in folder views.
- Docs (api.md delete section, CLAUDE.md), 10 new/updated tests. 150 passing;
API soft-delete lifecycle verified live.
Purge is manual (Empty Trash) this release; auto purge-on-delete (background,
~3-day retention) is the planned fast-follow.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
749 lines
33 KiB
Python
749 lines
33 KiB
Python
import json
|
|
import shutil
|
|
import tempfile
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
|
|
import openpyxl
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import override_settings
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
from rest_framework.test import APITestCase
|
|
|
|
from daybook_project import __version__
|
|
from artifacts.models import ApiKey, Artifact
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
class ArtifactAPITestCase(APITestCase):
|
|
"""Exercises the ingest/list/detail API against a real (temp) filesystem
|
|
and a real (test) database, the way an agent posting over HTTP would."""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super().setUpClass()
|
|
cls.artifacts_root = Path(tempfile.mkdtemp(prefix="agentfeed-test-artifacts-"))
|
|
cls.settings_override = override_settings(ARTIFACTS_ROOT=cls.artifacts_root)
|
|
cls.settings_override.enable()
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.settings_override.disable()
|
|
shutil.rmtree(cls.artifacts_root, ignore_errors=True)
|
|
super().tearDownClass()
|
|
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(username="alice", password="pw")
|
|
self.token = ApiKey.objects.create(user=self.user, name="alice-key")
|
|
self.other_user = User.objects.create_user(username="bob", password="pw")
|
|
self.other_token = ApiKey.objects.create(user=self.other_user, name="bob-key")
|
|
|
|
def auth_headers(self, token):
|
|
return {"HTTP_AUTHORIZATION": f"Bearer {token.key}"}
|
|
|
|
def create_artifact(self, token=None, **overrides):
|
|
token = token or self.token
|
|
payload = {
|
|
"source": "test-agent",
|
|
"title": "Hello World",
|
|
"kind": "markdown",
|
|
"content": "# Hello\n\nBody text.",
|
|
}
|
|
payload.update(overrides)
|
|
return self.client.post(
|
|
reverse("artifact-api"), payload, format="json", **self.auth_headers(token)
|
|
)
|
|
|
|
# -- capabilities -------------------------------------------------
|
|
|
|
def test_capabilities_is_public(self):
|
|
# Capability/contract info only, no user data — public, no auth needed.
|
|
response = self.client.get(reverse("capabilities-api"))
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_capabilities_with_token(self):
|
|
response = self.client.get(reverse("capabilities-api"), **self.auth_headers(self.token))
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn("markdown", response.data["artifact_kinds"])
|
|
|
|
def test_capabilities_lists_supported_methods(self):
|
|
response = self.client.get(reverse("capabilities-api"), **self.auth_headers(self.token))
|
|
methods = response.data["methods"]
|
|
self.assertEqual(methods["/api/artifacts"], ["GET", "POST"])
|
|
self.assertEqual(methods["/api/artifacts/{id}"], ["GET", "PATCH", "DELETE"])
|
|
|
|
def test_capabilities_with_session(self):
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("capabilities-api"))
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_capabilities_advertises_skill(self):
|
|
response = self.client.get(reverse("capabilities-api"), **self.auth_headers(self.token))
|
|
self.assertEqual(response.data["methods"]["/api/skill"], ["GET"])
|
|
self.assertTrue(response.data["skill_url"].endswith("/api/skill"))
|
|
|
|
# -- skill ----------------------------------------------------------
|
|
|
|
def test_skill_is_public(self):
|
|
# Bootstrapping instructions, no secrets — public, no auth needed.
|
|
response = self.client.get(reverse("skill-api"))
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_skill_returns_markdown_with_origin_filled_in(self):
|
|
response = self.client.get(reverse("skill-api"), **self.auth_headers(self.token))
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response["Content-Type"], "text/markdown; charset=utf-8")
|
|
body = response.content.decode()
|
|
# Frontmatter is intact and the config placeholder is rewritten to this
|
|
# instance's origin (http://testserver under the test client).
|
|
self.assertIn("name: agentfeed-posting", body)
|
|
self.assertIn("AGENTFEED_URL=http://testserver", body)
|
|
self.assertNotIn("your-agentfeed-instance.example.com", body)
|
|
|
|
def test_skill_json_format(self):
|
|
response = self.client.get(
|
|
reverse("skill-api"), {"format": "json"}, **self.auth_headers(self.token)
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data["version"], __version__)
|
|
self.assertEqual(response.data["url"], "http://testserver")
|
|
self.assertIn("name: agentfeed-posting", response.data["skill"])
|
|
|
|
def test_skill_with_session(self):
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("skill-api"))
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
# -- create (POST) --------------------------------------------------
|
|
|
|
def test_create_markdown_artifact(self):
|
|
response = self.create_artifact(tags=["demo", "weekly"])
|
|
self.assertEqual(response.status_code, 201)
|
|
body = response.data
|
|
self.assertEqual(body["source"], "test-agent")
|
|
self.assertEqual(body["kind"], "markdown")
|
|
self.assertEqual(body["title"], "Hello World")
|
|
self.assertEqual([tag["name"] for tag in body["tags"]], ["demo", "weekly"])
|
|
|
|
artifact = Artifact.objects.get(pk=body["id"])
|
|
self.assertEqual(artifact.owner, self.user)
|
|
|
|
def test_create_artifact_writes_files_to_disk(self):
|
|
response = self.create_artifact()
|
|
artifact = Artifact.objects.get(pk=response.data["id"])
|
|
stored_path = self.artifacts_root / artifact.path
|
|
self.assertTrue(stored_path.exists())
|
|
# Markdown artifacts also preserve the original .md alongside the rendered .html
|
|
self.assertTrue(stored_path.with_suffix(".md").exists())
|
|
|
|
def test_create_artifact_requires_token(self):
|
|
response = self.client.post(
|
|
reverse("artifact-api"),
|
|
{"source": "test-agent", "title": "x", "kind": "markdown", "content": "hi"},
|
|
format="json",
|
|
)
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
def test_create_artifact_rejects_invalid_kind(self):
|
|
response = self.create_artifact(kind="pdf")
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_create_artifact_rejects_missing_fields(self):
|
|
response = self.client.post(
|
|
reverse("artifact-api"), {"kind": "markdown"}, format="json", **self.auth_headers(self.token)
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
# -- list (GET) -------------------------------------------------------
|
|
|
|
def test_list_requires_auth(self):
|
|
# No session and no token -> JSON 403, never a browser-login redirect.
|
|
response = self.client.get(reverse("artifact-api"))
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
def test_list_scoped_to_owner(self):
|
|
self.create_artifact()
|
|
self.create_artifact(token=self.other_token, title="Not mine")
|
|
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("artifact-api"))
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual([a["title"] for a in response.data], ["Hello World"])
|
|
|
|
def test_list_readable_with_bearer_token(self):
|
|
# Agents read back with the same token they post with; owner-scoped.
|
|
self.create_artifact()
|
|
self.create_artifact(token=self.other_token, title="Not mine")
|
|
|
|
response = self.client.get(reverse("artifact-api"), **self.auth_headers(self.token))
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual([a["title"] for a in response.data], ["Hello World"])
|
|
|
|
def test_list_filter_by_label(self):
|
|
self.create_artifact(tags=["urgent"])
|
|
self.create_artifact(title="Second", tags=["later"])
|
|
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("artifact-api"), {"label": "urgent"})
|
|
self.assertEqual(len(response.data), 1)
|
|
self.assertEqual(response.data[0]["title"], "Hello World")
|
|
|
|
def test_list_filter_by_kind(self):
|
|
self.create_artifact(kind="markdown")
|
|
self.create_artifact(title="Frag", kind="html_fragment", content="<p>hi</p>")
|
|
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("artifact-api"), {"kind": "html_fragment"})
|
|
self.assertEqual(len(response.data), 1)
|
|
self.assertEqual(response.data[0]["title"], "Frag")
|
|
|
|
def test_list_filter_by_query(self):
|
|
self.create_artifact(title="Weekly Report")
|
|
self.create_artifact(title="Random Notes")
|
|
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("artifact-api"), {"q": "weekly"})
|
|
self.assertEqual(len(response.data), 1)
|
|
self.assertEqual(response.data[0]["title"], "Weekly Report")
|
|
|
|
def test_list_filter_by_day(self):
|
|
self.create_artifact()
|
|
today = timezone.localdate().isoformat()
|
|
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("artifact-api"), {"day": today})
|
|
self.assertEqual(len(response.data), 1)
|
|
|
|
response = self.client.get(reverse("artifact-api"), {"day": "2000-01-01"})
|
|
self.assertEqual(len(response.data), 0)
|
|
|
|
# -- detail (GET/PATCH) -----------------------------------------------
|
|
|
|
def test_detail_get_scoped_to_owner(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
|
|
self.client.force_login(self.other_user)
|
|
response = self.client.get(reverse("artifact-detail-api", args=[artifact_id]))
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("artifact-detail-api", args=[artifact_id]))
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_detail_readable_with_bearer_token(self):
|
|
# The api_url returned by POST is fetchable with the posting token.
|
|
artifact_id = self.create_artifact().data["id"]
|
|
|
|
response = self.client.get(
|
|
reverse("artifact-detail-api", args=[artifact_id]), **self.auth_headers(self.token)
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data["id"], artifact_id)
|
|
|
|
# ...but only the owner's token; another user's token gets 404.
|
|
response = self.client.get(
|
|
reverse("artifact-detail-api", args=[artifact_id]), **self.auth_headers(self.other_token)
|
|
)
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
def test_detail_requires_auth(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
response = self.client.get(reverse("artifact-detail-api", args=[artifact_id]))
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
# -- raw content (GET {id}/raw) ---------------------------------------
|
|
|
|
def _raw_body(self, response):
|
|
return b"".join(response.streaming_content).decode()
|
|
|
|
def test_raw_returns_exact_content_with_bearer(self):
|
|
create = self.create_artifact(content="# Exact\n\nverify me")
|
|
artifact_id = create.data["id"]
|
|
# metadata detail advertises the raw endpoint
|
|
self.assertTrue(create.data["api_raw_url"].endswith(f"/api/artifacts/{artifact_id}/raw"))
|
|
|
|
response = self.client.get(
|
|
reverse("artifact-raw-api", args=[artifact_id]), **self.auth_headers(self.token)
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response["Content-Type"], "text/plain; charset=utf-8")
|
|
# Content round-trips faithfully; normalize newlines since the stored
|
|
# sidecar is written in the host's text mode (CRLF on Windows dev, LF on
|
|
# the Linux deploy).
|
|
self.assertEqual(self._raw_body(response).replace("\r\n", "\n"), "# Exact\n\nverify me")
|
|
|
|
def test_raw_requires_auth(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
response = self.client.get(reverse("artifact-raw-api", args=[artifact_id]))
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
def test_raw_owner_scoped(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
response = self.client.get(
|
|
reverse("artifact-raw-api", args=[artifact_id]), **self.auth_headers(self.other_token)
|
|
)
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
def test_raw_not_available_for_render_only_kind(self):
|
|
artifact_id = self.create_artifact(kind="html_fragment", content="<p>hi</p>").data["id"]
|
|
response = self.client.get(
|
|
reverse("artifact-raw-api", args=[artifact_id]), **self.auth_headers(self.token)
|
|
)
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
def test_patch_updates_title_and_tags(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
response = self.client.patch(
|
|
reverse("artifact-detail-api", args=[artifact_id]),
|
|
{"title": "Updated title", "tags": ["a", "b"]},
|
|
format="json",
|
|
**self.auth_headers(self.token),
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data["title"], "Updated title")
|
|
self.assertEqual([t["name"] for t in response.data["tags"]], ["a", "b"])
|
|
|
|
def test_patch_rejects_blank_title(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
response = self.client.patch(
|
|
reverse("artifact-detail-api", args=[artifact_id]),
|
|
{"title": ""},
|
|
format="json",
|
|
**self.auth_headers(self.token),
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_patch_requires_token(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
response = self.client.patch(
|
|
reverse("artifact-detail-api", args=[artifact_id]), {"title": "No token"}, format="json"
|
|
)
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
def test_patch_cannot_touch_other_users_artifact(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
response = self.client.patch(
|
|
reverse("artifact-detail-api", args=[artifact_id]),
|
|
{"title": "Hijacked"},
|
|
format="json",
|
|
**self.auth_headers(self.other_token),
|
|
)
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
# -- delete (DELETE) --------------------------------------------------
|
|
|
|
def test_delete_soft_deletes_to_trash(self):
|
|
# DELETE moves the artifact to Trash (recoverable), not a hard removal.
|
|
artifact_id = self.create_artifact().data["id"]
|
|
artifact = Artifact.objects.get(pk=artifact_id)
|
|
stored_path = self.artifacts_root / artifact.path
|
|
self.assertTrue(stored_path.exists())
|
|
|
|
response = self.client.delete(
|
|
reverse("artifact-detail-api", args=[artifact_id]), **self.auth_headers(self.token)
|
|
)
|
|
self.assertEqual(response.status_code, 204)
|
|
# Row still exists, marked deleted; files kept on disk until purge.
|
|
artifact.refresh_from_db()
|
|
self.assertIsNotNone(artifact.deleted_at)
|
|
self.assertTrue(stored_path.exists())
|
|
# But it's gone from the API's list/detail (hidden like the feed).
|
|
self.assertEqual(self.client.get(reverse("artifact-api"), **self.auth_headers(self.token)).data, [])
|
|
|
|
def test_delete_requires_token(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
response = self.client.delete(reverse("artifact-detail-api", args=[artifact_id]))
|
|
self.assertEqual(response.status_code, 403)
|
|
self.assertTrue(Artifact.objects.filter(pk=artifact_id).exists())
|
|
|
|
def test_delete_cannot_touch_other_users_artifact(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
response = self.client.delete(
|
|
reverse("artifact-detail-api", args=[artifact_id]), **self.auth_headers(self.other_token)
|
|
)
|
|
self.assertEqual(response.status_code, 404)
|
|
self.assertTrue(Artifact.objects.filter(pk=artifact_id).exists())
|
|
|
|
def test_delete_is_idempotent(self):
|
|
# Soft-delete is idempotent: deleting an already-trashed artifact is a no-op 204.
|
|
artifact_id = self.create_artifact().data["id"]
|
|
first = self.client.delete(
|
|
reverse("artifact-detail-api", args=[artifact_id]), **self.auth_headers(self.token)
|
|
)
|
|
self.assertEqual(first.status_code, 204)
|
|
|
|
second = self.client.delete(
|
|
reverse("artifact-detail-api", args=[artifact_id]), **self.auth_headers(self.token)
|
|
)
|
|
self.assertEqual(second.status_code, 204)
|
|
|
|
def test_deleted_artifact_is_gone_from_list_and_detail(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
self.client.delete(
|
|
reverse("artifact-detail-api", args=[artifact_id]), **self.auth_headers(self.token)
|
|
)
|
|
|
|
self.client.force_login(self.user)
|
|
list_response = self.client.get(reverse("artifact-api"))
|
|
self.assertEqual(list_response.data, [])
|
|
|
|
detail_response = self.client.get(reverse("artifact-detail-api", args=[artifact_id]))
|
|
self.assertEqual(detail_response.status_code, 404)
|
|
|
|
# -- field length limits (issue #003) ----------------------------------
|
|
|
|
def test_create_rejects_oversized_title(self):
|
|
response = self.create_artifact(title="A" * 201)
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertIn("title", response.data["detail"])
|
|
|
|
def test_create_rejects_oversized_source(self):
|
|
response = self.create_artifact(source="a" * 81)
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertIn("source", response.data["detail"])
|
|
|
|
def test_create_accepts_title_at_exact_limit(self):
|
|
response = self.create_artifact(title="A" * 200)
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
def test_patch_rejects_oversized_title(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
response = self.client.patch(
|
|
reverse("artifact-detail-api", args=[artifact_id]),
|
|
{"title": "A" * 201},
|
|
format="json",
|
|
**self.auth_headers(self.token),
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
# -- content-shape validation (issue #004) -----------------------------
|
|
|
|
def test_create_rejects_nested_html_in_html_fragment(self):
|
|
response = self.create_artifact(
|
|
kind="html_fragment", content="<html><body>nested</body></html>"
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_create_rejects_html_full_without_html_tag(self):
|
|
response = self.create_artifact(kind="html_full", content="<p>not a full document</p>")
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_create_rejects_deck_without_section(self):
|
|
response = self.create_artifact(kind="deck", content="<h1>Not a deck slide</h1>")
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_create_rejects_whitespace_only_markdown(self):
|
|
response = self.create_artifact(kind="markdown", content=" \n\t ")
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_create_accepts_well_formed_html_fragment(self):
|
|
response = self.create_artifact(kind="html_fragment", content="<p>fine</p>")
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
# -- PATCH validation (issue #005) -------------------------------------
|
|
|
|
def test_patch_rejects_non_string_non_list_tags(self):
|
|
artifact_id = self.create_artifact(tags=["important", "keep-me"]).data["id"]
|
|
response = self.client.patch(
|
|
reverse("artifact-detail-api", args=[artifact_id]),
|
|
{"tags": 12345},
|
|
format="json",
|
|
**self.auth_headers(self.token),
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
artifact = Artifact.objects.get(pk=artifact_id)
|
|
self.assertEqual(artifact.tags, ["important", "keep-me"])
|
|
|
|
def test_patch_rejects_immutable_fields(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
for field, value in [
|
|
("content", "new content"),
|
|
("source", "other-source"),
|
|
("kind", "deck"),
|
|
("day", "2020-01-01"),
|
|
]:
|
|
response = self.client.patch(
|
|
reverse("artifact-detail-api", args=[artifact_id]),
|
|
{field: value},
|
|
format="json",
|
|
**self.auth_headers(self.token),
|
|
)
|
|
self.assertEqual(response.status_code, 400, field)
|
|
self.assertIn(field, response.data["detail"])
|
|
|
|
# -- syntax highlighting -------------------------------------------------
|
|
|
|
def test_markdown_fenced_code_block_gets_syntax_highlighted(self):
|
|
response = self.create_artifact(
|
|
content="# Test\n\n```python\ndef hello():\n return \"world\"\n```"
|
|
)
|
|
artifact = Artifact.objects.get(pk=response.data["id"])
|
|
stored_html = (self.artifacts_root / artifact.path).read_text(encoding="utf-8")
|
|
|
|
self.assertIn('class="codehilite"', stored_html)
|
|
self.assertIn('<span class="k">def</span>', stored_html)
|
|
self.assertIn('href="http://testserver/static/themes/pygments.css"', stored_html)
|
|
|
|
# -- json kind ------------------------------------------------------------
|
|
|
|
def test_create_json_artifact(self):
|
|
response = self.create_artifact(kind="json", content='{"a": 1, "b": [1, 2, 3]}')
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
artifact = Artifact.objects.get(pk=response.data["id"])
|
|
stored_html = (self.artifacts_root / artifact.path).read_text(encoding="utf-8")
|
|
self.assertIn('renderjson/renderjson.js', stored_html)
|
|
self.assertIn('type="application/json"', stored_html)
|
|
|
|
# raw sidecar preserves the exact original JSON text
|
|
raw_path = (self.artifacts_root / artifact.path).with_suffix(".json")
|
|
self.assertEqual(raw_path.read_text(encoding="utf-8"), '{"a": 1, "b": [1, 2, 3]}')
|
|
|
|
def test_create_json_artifact_rejects_invalid_json(self):
|
|
response = self.create_artifact(kind="json", content="{not valid json")
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_json_artifact_escapes_script_close_tag_safely(self):
|
|
payload = json.dumps({"value": "</script><script>alert(1)</script>"})
|
|
response = self.create_artifact(kind="json", content=payload)
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
artifact = Artifact.objects.get(pk=response.data["id"])
|
|
stored_html = (self.artifacts_root / artifact.path).read_text(encoding="utf-8")
|
|
|
|
# The page has exactly 3 real <script> elements (renderjson lib, the
|
|
# embedded data block, the init snippet). If the injected payload's
|
|
# "</script" wasn't escaped, it would add a 4th closing tag here.
|
|
self.assertEqual(stored_html.count("</script>"), 3)
|
|
|
|
# The escaped content must still round-trip to the exact original value.
|
|
start = stored_html.index('id="agentfeed-data">') + len('id="agentfeed-data">')
|
|
end = stored_html.index("</script>", start)
|
|
embedded = stored_html[start:end]
|
|
self.assertEqual(json.loads(embedded)["value"], "</script><script>alert(1)</script>")
|
|
|
|
# -- csv kind ---------------------------------------------------------------
|
|
|
|
def test_create_csv_artifact(self):
|
|
response = self.create_artifact(kind="csv", content="name,score\nalice,10\nbob,7")
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
artifact = Artifact.objects.get(pk=response.data["id"])
|
|
stored_html = (self.artifacts_root / artifact.path).read_text(encoding="utf-8")
|
|
self.assertIn("gridjs/gridjs.umd.js", stored_html)
|
|
self.assertIn('"columns": ["name", "score"]', stored_html)
|
|
self.assertIn('["alice", "10"]', stored_html)
|
|
|
|
raw_path = (self.artifacts_root / artifact.path).with_suffix(".csv")
|
|
self.assertEqual(raw_path.read_text(encoding="utf-8"), "name,score\nalice,10\nbob,7")
|
|
|
|
def test_create_csv_artifact_with_header_only(self):
|
|
response = self.create_artifact(kind="csv", content="name,score")
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
# -- source_name/source_id/source_prompt type checks (issue #008) -------
|
|
|
|
def test_create_rejects_non_string_source_name(self):
|
|
response = self.create_artifact(source_name=12345)
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertIn("source_name", response.data["detail"])
|
|
|
|
def test_create_rejects_non_string_source_id(self):
|
|
response = self.create_artifact(source_id=["a", "b"])
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_create_rejects_non_string_source_prompt(self):
|
|
response = self.create_artifact(source_prompt={"nested": "object"})
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_patch_rejects_non_string_source_name(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
response = self.client.patch(
|
|
reverse("artifact-detail-api", args=[artifact_id]),
|
|
{"source_name": 12345},
|
|
format="json",
|
|
**self.auth_headers(self.token),
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
artifact = Artifact.objects.get(pk=artifact_id)
|
|
self.assertEqual(artifact.source_name, "test-agent") # unchanged, not mangled
|
|
|
|
# -- xlsx upload converts to csv kind ------------------------------------
|
|
|
|
def make_xlsx(self, rows, filename="sheet.xlsx"):
|
|
workbook = openpyxl.Workbook()
|
|
sheet = workbook.active
|
|
for row in rows:
|
|
sheet.append(row)
|
|
buffer = BytesIO()
|
|
workbook.save(buffer)
|
|
buffer.seek(0)
|
|
buffer.name = filename
|
|
return buffer
|
|
|
|
def test_xlsx_upload_converts_to_csv_kind(self):
|
|
xlsx = self.make_xlsx([["Item", "Q1", "Q2"], ["Cash", 1000, 1200]])
|
|
response = self.client.post(
|
|
reverse("artifact-api"),
|
|
{"source": "test-agent", "title": "Balance Sheet", "file": xlsx},
|
|
format="multipart",
|
|
**self.auth_headers(self.token),
|
|
)
|
|
self.assertEqual(response.status_code, 201)
|
|
self.assertEqual(response.data["kind"], "csv")
|
|
|
|
artifact = Artifact.objects.get(pk=response.data["id"])
|
|
raw_path = (self.artifacts_root / artifact.path).with_suffix(".csv")
|
|
# read_text() normalizes line endings, so this is OS-portable rather
|
|
# than asserting the exact on-disk bytes (which are "\r\n" on Windows).
|
|
self.assertEqual(raw_path.read_text(encoding="utf-8"), "Item,Q1,Q2\nCash,1000,1200\n")
|
|
|
|
stored_html = (self.artifacts_root / artifact.path).read_text(encoding="utf-8")
|
|
self.assertIn('"columns": ["Item", "Q1", "Q2"]', stored_html)
|
|
self.assertIn('["Cash", "1000", "1200"]', stored_html)
|
|
|
|
def test_xlsx_upload_does_not_mangle_fields(self):
|
|
"""Regression test: {**a_querydict} previously wrapped every value
|
|
(including title) in a list - QueryDict's __getitem__ override is
|
|
bypassed by the ** spread, exposing its internal list-per-key
|
|
storage directly. Separately, pre-setting None for absent optional
|
|
fields defeated create_text_artifact's own .get(field, default)
|
|
calls (.get only falls through when the key is missing, not None),
|
|
which previously caused a NOT NULL constraint failure on
|
|
source_prompt."""
|
|
xlsx = self.make_xlsx([["a", "b"]])
|
|
response = self.client.post(
|
|
reverse("artifact-api"),
|
|
{
|
|
"source": "test-agent",
|
|
"title": "Untitled Sheet",
|
|
"tags": "finance,q2",
|
|
"file": xlsx,
|
|
},
|
|
format="multipart",
|
|
**self.auth_headers(self.token),
|
|
)
|
|
self.assertEqual(response.status_code, 201)
|
|
self.assertEqual(response.data["title"], "Untitled Sheet") # not ["Untitled Sheet"]
|
|
self.assertEqual(response.data["source_prompt"], "") # defaulted, not None
|
|
self.assertEqual([t["name"] for t in response.data["tags"]], ["finance", "q2"])
|
|
|
|
def test_xlsx_upload_rejects_corrupt_file(self):
|
|
buffer = BytesIO(b"not a real xlsx file")
|
|
buffer.name = "broken.xlsx"
|
|
response = self.client.post(
|
|
reverse("artifact-api"),
|
|
{"source": "test-agent", "title": "Broken", "file": buffer},
|
|
format="multipart",
|
|
**self.auth_headers(self.token),
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
# -- XSS sanitization of themed kinds (issue #014) -----------------------
|
|
|
|
def _stored_html(self, artifact_id):
|
|
artifact = Artifact.objects.get(pk=artifact_id)
|
|
return (self.artifacts_root / artifact.path).read_text(encoding="utf-8")
|
|
|
|
def test_markdown_strips_script_and_event_handlers(self):
|
|
aid = self.create_artifact(
|
|
kind="markdown",
|
|
content="# Hi\n\n<script>alert(document.cookie)</script>\n\n<img src=x onerror=alert(1)>\n\n[x](javascript:alert(1))",
|
|
).data["id"]
|
|
html = self._stored_html(aid).lower()
|
|
self.assertNotIn("<script", html)
|
|
self.assertNotIn("onerror", html)
|
|
self.assertNotIn("javascript:", html)
|
|
|
|
def test_html_fragment_strips_script_and_svg_onload(self):
|
|
aid = self.create_artifact(
|
|
kind="html_fragment",
|
|
content="<p>hi</p><script>alert(1)</script><svg onload=alert(2)></svg>",
|
|
).data["id"]
|
|
html = self._stored_html(aid).lower()
|
|
self.assertNotIn("<script", html)
|
|
self.assertNotIn("onload", html)
|
|
self.assertNotIn("<svg", html)
|
|
|
|
def test_deck_strips_script(self):
|
|
aid = self.create_artifact(
|
|
kind="deck", content="<section>Slide<script>alert(1)</script></section>"
|
|
).data["id"]
|
|
self.assertNotIn("<script", self._stored_html(aid).lower())
|
|
|
|
def test_sanitization_preserves_legitimate_formatting(self):
|
|
aid = self.create_artifact(
|
|
kind="markdown",
|
|
content="# Title\n\n| a | b |\n|---|---|\n| 1 | 2 |\n\n```python\ndef f():\n return 1\n```\n\n",
|
|
).data["id"]
|
|
html = self._stored_html(aid)
|
|
self.assertIn("<table>", html)
|
|
self.assertIn('class="codehilite"', html)
|
|
self.assertIn('<span class="k">def</span>', html)
|
|
self.assertIn("/assets/ab12cd34ef56", html) # relative image URL survives
|
|
|
|
def test_html_full_is_not_sanitized(self):
|
|
aid = self.create_artifact(
|
|
kind="html_full",
|
|
content="<!doctype html><html><head></head><body><script>ok()</script></body></html>",
|
|
).data["id"]
|
|
# html_full is the deliberate verbatim escape hatch — its script stays.
|
|
self.assertIn("<script>ok()</script>", self._stored_html(aid))
|
|
|
|
def test_sanitized_kinds_served_with_script_blocking_csp(self):
|
|
aid = self.create_artifact(kind="markdown", content="# hi").data["id"]
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("artifact", args=[aid]))
|
|
self.assertIn("script-src 'none'", response.headers.get("Content-Security-Policy", ""))
|
|
|
|
def test_html_full_served_without_restrictive_csp(self):
|
|
aid = self.create_artifact(
|
|
kind="html_full", content="<!doctype html><html><body>x</body></html>"
|
|
).data["id"]
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("artifact", args=[aid]))
|
|
self.assertIsNone(response.headers.get("Content-Security-Policy"))
|
|
|
|
# -- deeply nested JSON -> 400 not 500 (issue #015) ----------------------
|
|
|
|
def test_deeply_nested_json_content_returns_400(self):
|
|
deep = "[" * 5000 + "]" * 5000 # valid JSON, just too deep to parse
|
|
response = self.create_artifact(kind="json", content=deep)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_deeply_nested_request_body_returns_400(self):
|
|
# Nesting in the request body itself crashes DRF's parser before the
|
|
# view runs; the custom exception handler must map it to 400.
|
|
nested = "null"
|
|
for _ in range(5000):
|
|
nested = '{"n":' + nested + "}"
|
|
body = (
|
|
'{"source":"t","kind":"markdown","title":"x","content":"hi",'
|
|
'"metadata":' + nested + "}"
|
|
)
|
|
response = self.client.post(
|
|
reverse("artifact-api"),
|
|
data=body,
|
|
content_type="application/json",
|
|
**self.auth_headers(self.token),
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_normal_depth_json_still_accepted(self):
|
|
response = self.create_artifact(kind="json", content='{"a": {"b": {"c": [1, 2, 3]}}}')
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
# -- ragged CSV rows normalized, no data lost (issue #016) ---------------
|
|
|
|
def test_ragged_csv_rows_preserve_all_cells(self):
|
|
aid = self.create_artifact(kind="csv", content="a,b,c\n1,2,3,4,5\n6,7").data["id"]
|
|
html = self._stored_html(aid)
|
|
# Long row's extra cells are kept via auto-named columns; short row padded.
|
|
self.assertIn('"columns": ["a", "b", "c", "Column 4", "Column 5"]', html)
|
|
self.assertIn('["1", "2", "3", "4", "5"]', html)
|
|
self.assertIn('["6", "7", "", "", ""]', html)
|
|
|