import io import zipfile from django.contrib.auth import get_user_model from django.core.cache import cache from django.core.files.uploadedfile import SimpleUploadedFile from django.test import TestCase from django.urls import reverse from library.models import Project, ProjectContributor from .models import ( Announcement, AnnouncementComment, AnnouncementRead, Notification, ) from .notifications import notify_project_change def _zip_bytes(entries=None): buf = io.BytesIO() with zipfile.ZipFile(buf, 'w') as zf: for name, content in (entries or {}).items(): zf.writestr(name, content) return buf.getvalue() class AnnouncementsTestCase(TestCase): def setUp(self): User = get_user_model() self.staff = User.objects.create_user(username='Boss', password='pw', is_staff=True) self.normal = User.objects.create_user(username='NormalUser', password='pw') self.normal2 = User.objects.create_user(username='NormalUser2', password='pw') self.project = Project.objects.create( slug='test-pack', title='Test Pack', category='mod', owner=self.normal, ) cache.clear() def _auth(self, user=None): if user is not None: self.client.force_login(user) session = self.client.session session['authorized'] = True session.save() def _upload_version(self, name, content): self.client.post( reverse('library:api_upload_temp'), {'kind': 'version', 'file': SimpleUploadedFile(name, content, content_type='application/zip')}, HTTP_X_REQUESTED_WITH='XMLHttpRequest', ) # --- Global announcements: staff management --- def test_staff_can_create_edit_delete(self): self._auth(self.staff) resp = self.client.post( reverse('announcements:create'), {'title': 'Welcome', 'body': 'Hello **world**', 'published': 'on'}, ) announcement = Announcement.objects.get(title='Welcome') self.assertRedirects(resp, reverse('announcements:detail', args=[announcement.pk])) self.assertTrue(announcement.published) self.assertEqual(announcement.created_by, self.staff) resp = self.client.post( reverse('announcements:edit', args=[announcement.pk]), {'title': 'Welcome v2', 'body': 'Edited', 'published': 'on'}, ) announcement.refresh_from_db() self.assertEqual(announcement.title, 'Welcome v2') self.client.post(reverse('announcements:delete', args=[announcement.pk])) self.assertFalse(Announcement.objects.filter(pk=announcement.pk).exists()) def test_non_staff_cannot_manage(self): self._auth(self.normal) resp = self.client.post( reverse('announcements:create'), {'title': 'Nope', 'body': 'x', 'published': 'on'}, ) self.assertEqual(resp.status_code, 403) self.assertFalse(Announcement.objects.exists()) def test_unpublished_hidden_from_list_and_detail(self): draft = Announcement.objects.create( title='Draft', body='secret', created_by=self.staff, published=False, ) published = Announcement.objects.create( title='Live', body='ok', created_by=self.staff, published=True, ) self._auth(self.normal) resp = self.client.get(reverse('announcements:list')) self.assertContains(resp, 'Live') self.assertNotContains(resp, 'Draft') self.assertEqual(self.client.get(reverse('announcements:detail', args=[draft.pk])).status_code, 404) self.assertEqual(self.client.get(reverse('announcements:detail', args=[published.pk])).status_code, 200) # --- Read receipts --- def test_detail_marks_read_once(self): announcement = Announcement.objects.create(title='A', body='b', created_by=self.staff) self._auth(self.normal) self.assertEqual(self.client.get(reverse('announcements:detail', args=[announcement.pk])).status_code, 200) self.assertEqual(AnnouncementRead.objects.filter(announcement=announcement, user=self.normal).count(), 1) self.client.get(reverse('announcements:detail', args=[announcement.pk])) self.assertEqual(AnnouncementRead.objects.filter(announcement=announcement, user=self.normal).count(), 1) # --- Announcement comments --- def test_comment_lifecycle(self): announcement = Announcement.objects.create(title='A', body='b', created_by=self.staff) self._auth(self.normal) self.client.post(reverse('announcements:add_comment', args=[announcement.pk]), {'body': 'First!'}) comment = AnnouncementComment.objects.get(announcement=announcement) self.assertEqual(comment.user, self.normal) # Owner edits own comment. self.client.post(reverse('announcements:edit_comment', args=[announcement.pk, comment.pk]), {'body': 'Edited!'}) comment.refresh_from_db() self.assertEqual(comment.body, 'Edited!') # Another non-staff user cannot edit or delete it. self._auth(self.normal2) self.assertEqual( self.client.post(reverse('announcements:edit_comment', args=[announcement.pk, comment.pk]), {'body': 'x'}).status_code, 403, ) self.assertEqual( self.client.post(reverse('announcements:delete_comment', args=[announcement.pk, comment.pk])).status_code, 403, ) # Staff can delete any comment. self._auth(self.staff) self.client.post(reverse('announcements:delete_comment', args=[announcement.pk, comment.pk])) self.assertFalse(AnnouncementComment.objects.filter(pk=comment.pk).exists()) # --- Personal notifications --- def test_version_change_notifies_owner_and_contributors_not_actor(self): ProjectContributor.objects.create(project=self.project, user=self.normal2, added_by=self.normal) self._auth(self.normal2) self._upload_version('pack.zip', _zip_bytes()) self.client.post(reverse('library:version_upload', args=[self.project.slug]), {'version_name': '1.0.0'}) self.assertTrue(Notification.objects.filter(user=self.normal, kind='version', project=self.project).exists()) self.assertFalse(Notification.objects.filter(user=self.normal2, kind='version').exists()) def test_comment_change_notifies_owner(self): self._auth(self.normal2) self.client.post( reverse('library:add_comment', args=[self.project.slug]), {'body': 'Nice pack'}, ) self.assertTrue(Notification.objects.filter(user=self.normal, kind='comment', project=self.project).exists()) def test_retention_prunes_to_200(self): for i in range(210): notify_project_change(self.project, self.normal2, 'version', f'change {i}') self.assertEqual(Notification.objects.filter(user=self.normal).count(), 200) def test_mark_all_read(self): for i in range(3): notify_project_change(self.project, self.normal2, 'edit', f'edit {i}') self._auth(self.normal) self.client.post(reverse('announcements:mark_all_read')) self.assertFalse(Notification.objects.filter(user=self.normal, read=False).exists()) def test_project_detail_auto_marks_read(self): notify_project_change(self.project, self.normal2, 'comment', 'someone commented') self._auth(self.normal) self.client.get(reverse('library:project_detail', args=[self.project.slug])) self.assertFalse(Notification.objects.filter(user=self.normal, read=False).exists()) # --- Home --- def test_home_guests_see_global_only(self): Announcement.objects.create(title='Global', body='hi', created_by=self.staff) notify_project_change(self.project, self.normal2, 'edit', 'edited it') self._auth() resp = self.client.get(reverse('landing:home')) self.assertContains(resp, 'Global') self.assertNotContains(resp, 'For you') def test_home_logged_sees_global_and_personal(self): Announcement.objects.create(title='Global', body='hi', created_by=self.staff) notify_project_change(self.project, self.normal2, 'edit', 'edited the pack') self._auth(self.normal) resp = self.client.get(reverse('landing:home')) self.assertContains(resp, 'Global') self.assertContains(resp, 'For you') self.assertContains(resp, 'edited the pack')