peter
09ea0a8b4f
These accepted any JSON type on both POST and PATCH, silently mangled
into Python repr() strings at save time (a dict became the literal text
"{'a': 1}", unparseable as JSON) - and the immediate response echoed the
raw pre-save value, while a later GET returned the mangled one, an
internally inconsistent API.
Added validate_string_fields(), rejecting non-string values with 400
before they reach save(), matching #005's approach for tags. Verified live
and with new tests.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
457 lines
19 KiB
Python
457 lines
19 KiB
Python
import json
|
|
import shutil
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
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.authtoken.models import Token
|
|
from rest_framework.test import APITestCase
|
|
|
|
from artifacts.models import 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="daybook-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 = Token.objects.create(user=self.user)
|
|
self.other_user = User.objects.create_user(username="bob", password="pw")
|
|
self.other_token = Token.objects.create(user=self.other_user)
|
|
|
|
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_requires_auth(self):
|
|
response = self.client.get(reverse("capabilities-api"))
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
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)
|
|
|
|
# -- 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_session_login(self):
|
|
response = self.client.get(reverse("artifact-api"))
|
|
self.assertEqual(response.status_code, 302) # redirected to login
|
|
|
|
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_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_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_removes_artifact_and_files(self):
|
|
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)
|
|
self.assertFalse(Artifact.objects.filter(pk=artifact_id).exists())
|
|
self.assertFalse(stored_path.exists())
|
|
self.assertFalse(stored_path.with_suffix(".md").exists())
|
|
|
|
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_twice_second_call_404s(self):
|
|
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, 404)
|
|
|
|
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="daybook-data">') + len('id="daybook-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
|