daybook/tests/views/test_trash.py

import shutil
import tempfile
from datetime import timedelta
from pathlib import Path

from django.contrib.auth import get_user_model
from django.test import TestCase, override_settings
from django.urls import reverse
from django.utils import timezone

from artifacts.models import Artifact
from artifacts.storage import purge_expired_trash, user_usage_bytes

User = get_user_model()


# Run the purge inline (no background thread) so its effects are deterministic.
@override_settings(AGENTFEED_ASYNC_PURGE=False)
class TrashKeepViewTestCase(TestCase):
    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        cls.artifacts_root = Path(tempfile.mkdtemp(prefix="agentfeed-test-trash-"))
        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.other = User.objects.create_user(username="bob", password="pw")
        self.client.force_login(self.user)
        self._n = 0

    def make(self, owner=None, title="Note", source="agent", is_kept=False, size_bytes=1):
        self._n += 1
        return Artifact.objects.create(
            id=f"trs{self._n:05d}",
            owner=owner or self.user,
            source=source,
            title=title,
            kind=Artifact.MARKDOWN,
            day=timezone.localdate(),
            path=f"{source}/trs{self._n:05d}.html",
            original_kind=Artifact.MARKDOWN,
            size_bytes=size_bytes,
            is_kept=is_kept,
        )

    def test_delete_soft_deletes(self):
        a = self.make()
        r = self.client.post(reverse("artifact-delete", args=[a.id]))
        self.assertEqual(r.status_code, 200)
        a.refresh_from_db()
        self.assertIsNotNone(a.deleted_at)

    def test_delete_requires_login(self):
        self.client.logout()
        a = self.make()
        r = self.client.post(reverse("artifact-delete", args=[a.id]))
        self.assertEqual(r.status_code, 302)

    def test_delete_scoped_to_owner(self):
        a = self.make(owner=self.other)
        r = self.client.post(reverse("artifact-delete", args=[a.id]))
        self.assertEqual(r.status_code, 404)
        a.refresh_from_db()
        self.assertIsNone(a.deleted_at)

    def test_deleted_hidden_from_feed_shown_in_trash(self):
        a = self.make(title="ThrowawayThing")
        self.client.post(reverse("artifact-delete", args=[a.id]))
        self.assertNotIn("ThrowawayThing", self.client.get("/").content.decode())
        self.assertIn("ThrowawayThing", self.client.get("/?folder=deleted").content.decode())

    def test_undelete_restores(self):
        a = self.make(title="ComeBack")
        self.client.post(reverse("artifact-delete", args=[a.id]))
        r = self.client.post(reverse("artifact-undelete", args=[a.id]))
        self.assertEqual(r.status_code, 200)
        a.refresh_from_db()
        self.assertIsNone(a.deleted_at)
        self.assertIn("ComeBack", self.client.get("/").content.decode())

    def test_empty_trash_purges(self):
        a = self.make()
        b = self.make()
        self.client.post(reverse("artifact-delete", args=[a.id]))
        self.client.post(reverse("artifact-delete", args=[b.id]))
        r = self.client.post(reverse("feed-empty-trash"))
        self.assertEqual(r.status_code, 200)
        self.assertEqual(r.json()["purged"], 2)
        self.assertFalse(Artifact.objects.filter(pk__in=[a.id, b.id]).exists())

    def test_deleted_is_off_quota(self):
        a = self.make(size_bytes=1000)
        self.assertEqual(user_usage_bytes(self.user), 1000)
        self.client.post(reverse("artifact-delete", args=[a.id]))
        self.assertEqual(user_usage_bytes(self.user), 0)

    def test_keep_folder_shows_only_kept(self):
        self.make(title="KeptOne", is_kept=True)
        self.make(title="NormalOne")
        page = self.client.get("/?folder=keep").content.decode()
        self.assertIn("KeptOne", page)
        self.assertNotIn("NormalOne", page)

    def test_purge_expired_trash_removes_old_only(self):
        old = self.make()
        recent = self.make()
        Artifact.objects.filter(pk=old.id).update(deleted_at=timezone.now() - timedelta(days=4))
        Artifact.objects.filter(pk=recent.id).update(deleted_at=timezone.now())
        purged = purge_expired_trash()
        self.assertEqual(purged, 1)
        self.assertFalse(Artifact.objects.filter(pk=old.id).exists())
        self.assertTrue(Artifact.objects.filter(pk=recent.id).exists())

    def test_delete_triggers_expired_purge(self):
        old = self.make()
        Artifact.objects.filter(pk=old.id).update(deleted_at=timezone.now() - timedelta(days=4))
        fresh = self.make()
        # Deleting fresh runs the purge (inline here) -> the expired one goes,
        # the freshly-trashed one stays.
        self.client.post(reverse("artifact-delete", args=[fresh.id]))
        self.assertFalse(Artifact.objects.filter(pk=old.id).exists())
        self.assertTrue(Artifact.objects.filter(pk=fresh.id, deleted_at__isnull=False).exists())

    def test_mark_all_read_across_sources(self):
        self.make(source="a")
        self.make(source="b")
        r = self.client.post(reverse("feed-mark-read"))
        self.assertEqual(r.status_code, 200)
        self.assertEqual(r.json()["marked"], 2)
        self.assertFalse(Artifact.objects.filter(owner=self.user, read_at__isnull=True).exists())

    def test_mark_all_read_scoped_to_source(self):
        a = self.make(source="hermes")
        b = self.make(source="janus")
        self.assertEqual(self.client.post(reverse("feed-mark-read"), {"source": "hermes"}).json()["marked"], 1)
        a.refresh_from_db()
        b.refresh_from_db()
        self.assertIsNotNone(a.read_at)
        self.assertIsNone(b.read_at)

    def test_mark_all_read_owner_scoped(self):
        theirs = self.make(owner=self.other)
        self.client.post(reverse("feed-mark-read"))
        theirs.refresh_from_db()
        self.assertIsNone(theirs.read_at)

    def test_mark_all_read_excludes_trashed(self):
        a = self.make()
        Artifact.objects.filter(pk=a.id).update(deleted_at=timezone.now())
        self.assertEqual(self.client.post(reverse("feed-mark-read")).json()["marked"], 0)