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="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 = 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_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) 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_requires_auth(self): response = self.client.get(reverse("skill-api")) self.assertEqual(response.status_code, 403) 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: daybook-posting", body) self.assertIn("DAYBOOK_URL=http://testserver", body) self.assertNotIn("your-daybook-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: daybook-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_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="

hi

") 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="nested" ) self.assertEqual(response.status_code, 400) def test_create_rejects_html_full_without_html_tag(self): response = self.create_artifact(kind="html_full", content="

not a full document

") self.assertEqual(response.status_code, 400) def test_create_rejects_deck_without_section(self): response = self.create_artifact(kind="deck", content="

Not a deck slide

") 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="

fine

") 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('def', 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": ""}) 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 "), 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("", start) embedded = stored_html[start:end] self.assertEqual(json.loads(embedded)["value"], "") # -- 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\n\n\n\n[x](javascript:alert(1))", ).data["id"] html = self._stored_html(aid).lower() self.assertNotIn("", html) self.assertIn('class="codehilite"', html) self.assertIn('def', 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="", ).data["id"] # html_full is the deliberate verbatim escape hatch — its script stays. self.assertIn("", 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="x" ).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)