Hello World
" - expected_text = "Hello World" - email_instance = mock_email_class.return_value - - # Act - _send_html_email("Subject", html_body, "target@example.com") - - # Assert - mock_email_class.assert_called_once_with( - subject="Subject", - body=expected_text, - from_email="noreply@example.com", - to=["target@example.com"], +class AutomaticRegistrationNotificationTests(SimpleTestCase): + @mock.patch("apps.events.tasks.notify_user") + @mock.patch("apps.events.tasks.Registration.objects") + def test_confirmation_delivers_in_app_with_event_link(self, registrations, notify_user): + registration = SimpleNamespace( + user_id=7, + event_id=12, + ticket_id="ticket", + event=SimpleNamespace(title="Workshop", slug="workshop"), ) - email_instance.attach_alternative.assert_called_once_with(html_body, "text/html") - email_instance.send.assert_called_once() + registrations.select_related.return_value.get.return_value = registration + result = send_registration_confirmation_notification.run("1") + self.assertEqual(result, {"sent": True}) + self.assertEqual(notify_user.call_args.args[0], 7) + self.assertEqual(notify_user.call_args.args[1]["action_url"], "/events/workshop") + + @mock.patch("apps.events.tasks.send_critical_sms.delay") + @mock.patch("apps.events.tasks.notify_user") + @mock.patch("apps.events.tasks.Registration.objects") + def test_cancellation_delivers_in_app_and_optional_sms(self, registrations, notify_user, send_sms): + registrations.select_related.return_value.get.return_value = SimpleNamespace( + user_id=7, + event_id=12, + user=SimpleNamespace(mobile="09191110002", is_mobile_verified=True), + event=SimpleNamespace(title="Workshop", slug="workshop"), + ) + self.assertEqual(send_registration_cancellation_notification.run("1"), {"sent": True}) + self.assertEqual(notify_user.call_args.args[1]["type"], "event_cancellation") + send_sms.assert_called_once_with("09191110002", "event_cancellation", "Workshop") -class RegistrationResourceTests(SimpleTestCase): +class ManualActionRemovalTests(SimpleTestCase): + def test_event_and_registration_admin_have_no_manual_send_actions(self): + self.assertFalse(EventAdmin.__dict__.get("actions_row")) + self.assertFalse(RegistrationAdmin.__dict__.get("actions_row")) + + def test_bulk_status_actions_save_each_registration_for_automatic_notification(self): + model_admin = RegistrationAdmin(Registration, admin.site) + registration = mock.Mock() + with mock.patch.object(model_admin, "message_user"): + model_admin.confirm_registrations(SimpleNamespace(), [registration]) + self.assertEqual(registration.status, Registration.StatusChoices.CONFIRMED) + registration.save.assert_called_with(update_fields=["status"]) + registration.reset_mock() + model_admin.cancel_registrations(SimpleNamespace(), [registration]) + self.assertEqual(registration.status, Registration.StatusChoices.CANCELLED) + registration.save.assert_called_with(update_fields=["status"]) + + +class EventModelTests(TestCase): def setUp(self): - self.resource = RegistrationResource() - - def test_dehydrate_ticket_id_truncates_to_eight_characters(self): - # Arrange - ticket_id = uuid.uuid4() - record = SimpleNamespace(ticket_id=ticket_id) - expected = str(ticket_id)[:8] - - # Act - result = self.resource.dehydrate_ticket_id(record) - - # Assert - self.assertEqual(result, expected) - - def test_dehydrate_ticket_id_handles_missing_values(self): - # Arrange - record = SimpleNamespace(ticket_id=None) - - # Act - result = self.resource.dehydrate_ticket_id(record) - - # Assert - self.assertEqual(result, "") - - -class AnnouncementFormTests(SimpleTestCase): - def test_statuses_field_initializes_with_confirmed_and_attended(self): - # Arrange - form = AnnouncementForm() - - # Act - initial = form.fields["statuses"].initial - - # Assert - expected = [ - Registration.StatusChoices.CONFIRMED, - Registration.StatusChoices.ATTENDED, - ] - self.assertEqual(initial, expected) - - -class EventEmailLogFactoryMixin: - def create_user(self): - unique = uuid.uuid4().hex - return User.objects.create_user( - email=f"user_{unique}@example.com", - username=f"user_{unique[:10]}", - password="pass1234", - ) - - def create_event(self, **kwargs): now = timezone.now() - defaults = { - "title": f"Event {uuid.uuid4().hex[:6]}", - "description": "Fixture event", - "start_time": now, - "end_time": now + timedelta(hours=1), - "slug": f"event-{uuid.uuid4().hex[:6]}", - "price": 0, - } - defaults.update(kwargs) - return Event.objects.create(**defaults) - - -class EventEmailLogModelTests(EventEmailLogFactoryMixin, TestCase): - def test_claim_creates_pending_log(self): - # Arrange - event = self.create_event() - user = self.create_user() - context = "send-invite" - - # Act - log, skipped = EventEmailLog.claim( - event_id=event.id, - user_id=user.id, - kind=EventEmailLog.KIND_INVITE_NON_REGISTERED, - context=context, + self.event = Event.objects.create( + title="Workshop", description="**Important** details", start_time=now + timedelta(days=1), + end_time=now + timedelta(days=1, hours=2), capacity=1, + registration_start_date=now - timedelta(days=1), + registration_end_date=now + timedelta(hours=1), ) - - # Assert - self.assertFalse(skipped) - self.assertEqual(log.status, EventEmailLog.STATUS_PENDING) - self.assertEqual(log.context_hash, EventEmailLog._hash_context(context)) - - def test_claim_returns_existing_pending_log(self): - # Arrange - event = self.create_event() - user = self.create_user() - context = "announcement" - context_hash = EventEmailLog._hash_context(context) - existing = EventEmailLog.objects.create( - event=event, - user=user, - kind=EventEmailLog.KIND_EVENT_ANNOUNCEMENT, - context_hash=context_hash, - status=EventEmailLog.STATUS_PENDING, - ) - - # Act - log, skipped = EventEmailLog.claim( - event_id=event.id, - user_id=user.id, - kind=EventEmailLog.KIND_EVENT_ANNOUNCEMENT, - context=context, - ) - - # Assert - self.assertTrue(skipped) - self.assertEqual(log.pk, existing.pk) - self.assertEqual(log.status, EventEmailLog.STATUS_PENDING) - - def test_claim_resets_failed_record(self): - # Arrange - event = self.create_event() - user = self.create_user() - context = "retry" - context_hash = EventEmailLog._hash_context(context) - log = EventEmailLog.objects.create( - event=event, - user=user, - kind=EventEmailLog.KIND_EVENT_ANNOUNCEMENT, - context_hash=context_hash, - status=EventEmailLog.STATUS_FAILED, - error="boom", - sent_at=timezone.now(), - ) - - # Act - claimed, skipped = EventEmailLog.claim( - event_id=event.id, - user_id=user.id, - kind=EventEmailLog.KIND_EVENT_ANNOUNCEMENT, - context=context, - ) - - # Assert - self.assertFalse(skipped) - self.assertEqual(claimed.pk, log.pk) - self.assertEqual(claimed.status, EventEmailLog.STATUS_PENDING) - self.assertEqual(claimed.error, "") - self.assertIsNone(claimed.sent_at) - - def test_mark_sent_sets_sent_timestamp_and_status(self): - # Arrange - event = self.create_event() - user = self.create_user() - log = EventEmailLog.objects.create( - event=event, - user=user, - kind=EventEmailLog.KIND_EVENT_ANNOUNCEMENT, - ) - - # Act - log.mark_sent() - - # Assert - self.assertEqual(log.status, EventEmailLog.STATUS_SENT) - self.assertIsNotNone(log.sent_at) - - def test_mark_failed_clears_sent_at_and_records_error(self): - # Arrange - event = self.create_event() - user = self.create_user() - log = EventEmailLog.objects.create( - event=event, - user=user, - kind=EventEmailLog.KIND_EVENT_ANNOUNCEMENT, - sent_at=timezone.now(), - ) - - # Act - log.mark_failed("timeout") - - # Assert - self.assertEqual(log.status, EventEmailLog.STATUS_FAILED) - self.assertEqual(log.error, "timeout") - self.assertIsNone(log.sent_at) - - -class EventModelTests(EventEmailLogFactoryMixin, TestCase): - def test_description_html_renders_markdown(self): - # Arrange - event = self.create_event(description="**bold** content") - - # Act - rendered = event.description_html - - # Assert - self.assertIn("bold", rendered) - - def test_is_registration_open_follows_window(self): - # Arrange - now = timezone.now() - event = self.create_event( - registration_start_date=now - timedelta(hours=2), - registration_end_date=now + timedelta(hours=2), - ) - - # Act / Assert - with mock.patch("apps.events.models.timezone.now", return_value=now): - self.assertTrue(event.is_registration_open) - - def test_is_registration_open_closed_outside_window(self): - # Arrange - now = timezone.now() - event = self.create_event( - registration_start_date=now + timedelta(hours=1), - registration_end_date=now + timedelta(hours=2), - ) - - # Act / Assert - with mock.patch("apps.events.models.timezone.now", return_value=now): - self.assertFalse(event.is_registration_open) - - def test_current_attendees_count_filters_statuses(self): - # Arrange - event = self.create_event() - user_one = self.create_user() - user_two = self.create_user() - Registration.objects.create( - event=event, - user=user_one, - status=Registration.StatusChoices.CONFIRMED, - ) - Registration.objects.create( - event=event, - user=user_two, - status=Registration.StatusChoices.CANCELLED, - ) - Registration.objects.create( - event=event, - user=self.create_user(), - status=Registration.StatusChoices.ATTENDED, - ) - - # Act - count = event.current_attendees_count - - # Assert - self.assertEqual(count, 2) - - def test_has_available_slots_respects_capacity(self): - # Arrange - event = self.create_event(capacity=2) - for _ in range(2): - Registration.objects.create( - event=event, - user=self.create_user(), - status=Registration.StatusChoices.CONFIRMED, - ) - - # Act - available_after_full = event.has_available_slots - - # Assert - self.assertFalse(available_after_full) - - def test_has_available_slots_allows_unlimited_capacity(self): - # Arrange - event = self.create_event(capacity=None) - - # Act - available = event.has_available_slots - - # Assert - self.assertTrue(available) - - -class EventTaskBehaviorTests(EventEmailLogFactoryMixin, TestCase): - def test_event_recipients_filters_by_status_and_email(self): - # Arrange - event = self.create_event() - verified = self.create_user() - verified.is_email_verified = True - verified.save(update_fields=["is_email_verified"]) - Registration.objects.create( - event=event, - user=verified, - status=Registration.StatusChoices.CONFIRMED, - ) - unverified = self.create_user() - Registration.objects.create( - event=event, - user=unverified, - status=Registration.StatusChoices.CONFIRMED, - ) - - # Act - recipients = list(_event_recipients(event, statuses=[Registration.StatusChoices.CONFIRMED])) - - # Assert - self.assertEqual(len(recipients), 1) - self.assertEqual(recipients[0].user_id, verified.id) - - @override_settings(DEFAULT_FROM_EMAIL="noreply@example.com") - @mock.patch("apps.events.tasks.EmailMultiAlternatives") - @mock.patch("apps.events.tasks.render_to_string", return_value="ok
") - @mock.patch("apps.events.tasks.strip_tags", side_effect=lambda html: "ok") - @mock.patch("apps.events.tasks.markdown.markdown", return_value="converted") - def test_send_registration_confirmation_email_sends_message( - self, - mock_markdown, - mock_strip, - mock_render, - mock_email_class, - ): - # Arrange - registration = SimpleNamespace( - pk=1, - user=SimpleNamespace(email="user@example.com", username="user-one"), - event=SimpleNamespace( - title="Title", - registration_success_markdown="**done**", - ), - ) - manager = mock.MagicMock() - manager.select_related.return_value.get.return_value = registration - - mock_email_instance = mock_email_class.return_value - - # Act - with mock.patch("apps.events.tasks.Registration.objects", manager): - send_registration_confirmation_email.run("1") - - # Assert helpers - mock_markdown.assert_called_once_with( - "**done**", - extensions=["extra", "sane_lists", "toc"], - ) - mock_render.assert_called_once_with( - "emails/event_registration_confirmation.html", - { - "user": registration.user, - "event": registration.event, - "registration": registration, - "success_html": "converted", - }, - ) - mock_strip.assert_called_once_with("ok
") - - # Assert - mock_email_class.assert_called_once() - mock_email_instance.attach_alternative.assert_called_once() - mock_email_instance.send.assert_called_once() - - -class EventAdminTests(EventEmailLogFactoryMixin, TestCase): - def setUp(self): - self.site = AdminSite() - self.event_admin = EventAdmin(Event, self.site) - self.registration_admin = RegistrationAdmin(Registration, self.site) - self.event_admin.message_user = mock.Mock() - self.registration_admin.message_user = mock.Mock() - - def test_price_display_returns_label_for_free(self): - # Arrange - now = timezone.now() - event = Event( - title="Free Event", - description="desc", - start_time=now, - end_time=now + timedelta(hours=1), - price=None, - ) - - # Act - result = self.event_admin.price_display(event) - - # Assert - self.assertEqual(result, "رایگان") - - @mock.patch("apps.events.admin.jdate", return_value="JDATE") - def test_start_time_display_calls_jdate(self, mock_jdate): - event = self.create_event() - - result = self.event_admin.start_time_display(event) - - mock_jdate.assert_called_once_with(event.start_time) - self.assertEqual(result, "JDATE") - - @mock.patch("apps.events.admin.jdate", return_value="JDATE") - def test_end_time_display_calls_jdate(self, mock_jdate): - event = self.create_event() - - result = self.event_admin.end_time_display(event) - - mock_jdate.assert_called_once_with(event.end_time) - self.assertEqual(result, "JDATE") - - def test_capacity_display_handles_unlimited(self): - event = self.create_event(capacity=None) - - result = self.event_admin.capacity_display(event) - - self.assertEqual(result, "نامحدود") - - @mock.patch("apps.events.admin.Event.current_attendees_count", new_callable=mock.PropertyMock, return_value=7) - def test_attendees_display_returns_current_attendees(self, _mock_count): - event = self.create_event() - - result = self.event_admin.attendees_display(event) - - self.assertEqual(result, 7) - - @mock.patch("apps.events.admin.Event.is_registration_open", new_callable=mock.PropertyMock, return_value=True) - def test_is_registration_open_display_returns_bool(self, _mock_open): - event = self.create_event() - - self.assertTrue(self.event_admin.is_registration_open_display(event)) - - def test_make_draft_updates_status(self): - event = self.create_event(status=Event.StatusChoices.PUBLISHED) - queryset = Event.all_objects.filter(pk=event.pk) - - self.event_admin.make_draft(None, queryset) - - self.assertEqual(Event.objects.get(pk=event.pk).status, Event.StatusChoices.DRAFT) - - def test_make_cancelled_updates_status(self): - event = self.create_event(status=Event.StatusChoices.DRAFT) - queryset = Event.all_objects.filter(pk=event.pk) - - self.event_admin.make_cancelled(None, queryset) - - self.assertEqual(Event.objects.get(pk=event.pk).status, Event.StatusChoices.CANCELLED) - - def test_make_completed_updates_status(self): - event = self.create_event(status=Event.StatusChoices.PUBLISHED) - queryset = Event.objects.filter(pk=event.pk) - - self.event_admin.make_completed(None, queryset) - - self.assertEqual(Event.objects.get(pk=event.pk).status, Event.StatusChoices.COMPLETED) - - def test_restore_events_marks_is_deleted_false(self): - event = self.create_event() - event.delete() - queryset = Event.all_objects.filter(pk=event.pk) - - self.event_admin.restore_events(None, queryset) - - self.assertFalse(Event.all_objects.get(pk=event.pk).is_deleted) - - def test_action_send_skyroom_credentials_queues_task(self): - event = self.create_event() - - with mock.patch("apps.events.admin.queue_skyroom_credentials.delay") as mock_delay: - result = self.event_admin.action_send_skyroom_credentials(mock.Mock(), event.pk) - - mock_delay.assert_called_once_with(event.pk) - self.assertEqual(result, mock.ANY) - - def test_action_send_reminder_now_queues_task(self): - event = self.create_event() - - with mock.patch("apps.events.admin.send_event_reminder_task.delay") as mock_delay: - result = self.event_admin.action_send_reminder_now(mock.Mock(), event.pk) - - mock_delay.assert_called_once_with(event.pk) - self.assertEqual(result, mock.ANY) - - def test_action_send_announcement_dispatches_queue(self): - event = self.create_event() - data = QueryDict(mutable=True) - data.update({"subject": "Hello", "body_html": "hi
"}) - data.setlist("statuses", [Registration.StatusChoices.CONFIRMED]) - request = SimpleNamespace(method="POST", POST=data, user=self.create_user()) - - with mock.patch("apps.events.admin.queue_event_announcement") as mock_queue, \ - mock.patch("apps.events.admin.redirect", return_value="redirected") as mock_redirect: - result = self.event_admin.action_send_announcement(request, event.pk) - - mock_queue.delay.assert_called_once() - mock_redirect.assert_called_once() - self.assertEqual(result, "redirected") - - def test_action_invite_other_users_queues_task(self): - event = self.create_event() - - with mock.patch("apps.events.admin.queue_invites_to_non_registered_users.delay") as mock_delay: - result = self.event_admin.action_invite_other_users(mock.Mock(), event.pk) - - mock_delay.assert_called_once_with(event.pk) - self.assertEqual(result, mock.ANY) - - def test_make_published_updates_status(self): - event = self.create_event(status=Event.StatusChoices.DRAFT) - queryset = Event.objects.filter(pk=event.pk) - - self.event_admin.make_published(None, queryset) - - self.assertEqual(Event.objects.get(pk=event.pk).status, Event.StatusChoices.PUBLISHED) - - def test_confirm_registrations_sets_status(self): - # Arrange - event = self.create_event() - user = self.create_user() - user.is_email_verified = False - user.save(update_fields=["is_email_verified"]) - registration = Registration.objects.create( - event=event, - user=user, - status=Registration.StatusChoices.PENDING, - ) - - # Act - self.registration_admin.confirm_registrations(None, Registration.objects.filter(pk=registration.pk)) - - # Assert - self.assertEqual( - Registration.objects.get(pk=registration.pk).status, - Registration.StatusChoices.CONFIRMED, - ) - - -class RegistrationAdminTests(EventEmailLogFactoryMixin, TestCase): - def setUp(self): - self.site = AdminSite() - self.admin = RegistrationAdmin(Registration, self.site) - self.admin.message_user = mock.Mock() - - def test_cancel_registrations_sets_status(self): - registration = Registration.objects.create( - event=self.create_event(), - user=self.create_user(), - status=Registration.StatusChoices.PENDING, - ) - - self.admin.cancel_registrations(None, Registration.objects.filter(pk=registration.pk)) - - self.assertEqual( - Registration.objects.get(pk=registration.pk).status, - Registration.StatusChoices.CANCELLED, - ) - - def test_mark_attended_updates_status(self): - registration = Registration.objects.create( - event=self.create_event(), - user=self.create_user(), - status=Registration.StatusChoices.CONFIRMED, - ) - - self.admin.mark_attended(None, Registration.objects.filter(pk=registration.pk)) - - self.assertEqual( - Registration.objects.get(pk=registration.pk).status, - Registration.StatusChoices.ATTENDED, - ) - - def test_restore_registrations_calls_restore(self): - registration = Registration.objects.create( - event=self.create_event(), - user=self.create_user(), - status=Registration.StatusChoices.PENDING, - ) - registration.delete() - - with mock.patch.object(Registration, "objects", Registration.all_objects): - self.admin.restore_registrations(None, Registration.all_objects.filter(pk=registration.pk)) - - self.assertFalse(Registration.all_objects.get(pk=registration.pk).is_deleted) - - def test_action_email_selected_sends_and_redirects(self): - event = self.create_event() - registration = Registration.objects.create( - event=event, - user=self.create_user(), - status=Registration.StatusChoices.PENDING, - ) - data = QueryDict(mutable=True) - data.update({"subject": "Title", "body_html": "body
"}) - request = SimpleNamespace(method="POST", POST=data, user=self.create_user()) - - with mock.patch("apps.events.admin.render_to_string", return_value="ok
"), \ - mock.patch("apps.events.admin._send_html_email") as mock_send, \ - mock.patch("apps.events.admin.redirect", return_value="redirected") as mock_redirect: - self.admin.action_email_selected(request, registration.pk) - - mock_send.assert_called_once() - mock_redirect.assert_called_once() - - def test_action_send_skyroom_credentials_queues_task(self): - registration = Registration.objects.create( - event=self.create_event(), - user=self.create_user(), - status=Registration.StatusChoices.CONFIRMED, - ) - - with mock.patch("apps.events.admin.send_skyroom_credentials_individual_task.delay") as mock_delay: - result = self.admin.action_send_skyroom_credentials(mock.Mock(), registration.pk) - - mock_delay.assert_called_once_with(registration.pk) - self.assertEqual(result, mock.ANY) - - -class EventEmailLogAdminTests(EventEmailLogFactoryMixin, TestCase): - def setUp(self): - self.site = AdminSite() - self.admin = EventEmailLogAdmin(EventEmailLog, self.site) - self.admin.message_user = mock.Mock() - - def test_user_email_returns_dash_when_missing(self): - event = self.create_event() - user = self.create_user() - user.email = "" - log = EventEmailLog.objects.create( - event=event, - user=user, - kind=EventEmailLog.KIND_EVENT_ANNOUNCEMENT, - ) - - self.assertEqual(self.admin.user_email(log), "—") - - def test_resend_selected_emails_requeues_and_clears_error(self): - event = self.create_event() - user = self.create_user() - log = EventEmailLog.objects.create( - event=event, - user=user, - kind=EventEmailLog.KIND_INVITE_NON_REGISTERED, - status=EventEmailLog.STATUS_FAILED, - error="boom", - ) - - with mock.patch("apps.events.admin.send_invite_to_user.delay") as mock_delay: - self.admin.resend_selected_emails(mock.Mock(), EventEmailLog.objects.filter(pk=log.pk)) - - log.refresh_from_db() - self.assertEqual(log.status, EventEmailLog.STATUS_PENDING) - self.assertEqual(log.error, "") - mock_delay.assert_called_once_with(log.event_id, log.user_id) - - -class EventTasksCoverageTests(EventEmailLogFactoryMixin, TestCase): - def _dummy_registration(self): - user = self.create_user() - user.is_email_verified = True - user.save(update_fields=["is_email_verified"]) - event = self.create_event() - registration = Registration.objects.create( - event=event, - user=user, - status=Registration.StatusChoices.CONFIRMED, - ) - return registration - - def test_send_registration_cancellation_email_returns_when_email_missing(self): - registration = SimpleNamespace( - user=SimpleNamespace(email=None), - event=SimpleNamespace(title="Title"), - ) - manager = mock.MagicMock() - manager.select_related.return_value.get.return_value = registration - - with mock.patch("apps.events.tasks.Registration.objects", manager), \ - mock.patch("apps.events.tasks.EmailMultiAlternatives") as mock_email: - send_registration_cancellation_email.run("1") - - mock_email.assert_not_called() - - def test_send_registration_cancellation_email_retries_on_failure(self): - registration = SimpleNamespace( - user=SimpleNamespace(email="user@example.com"), - event=SimpleNamespace(title="Title"), - ) - manager = mock.MagicMock() - manager.select_related.return_value.get.return_value = registration - email_instance = mock.MagicMock() - email_instance.send.side_effect = RuntimeError("boom") - mock_email_class = mock.MagicMock(return_value=email_instance) - - with mock.patch("apps.events.tasks.Registration.objects", manager), \ - mock.patch("apps.events.tasks.EmailMultiAlternatives", mock_email_class), \ - mock.patch("apps.events.tasks.render_to_string", return_value="ok
"), \ - mock.patch("apps.events.tasks.strip_tags", return_value="ok"), \ - mock.patch.object(send_registration_cancellation_email, "retry", side_effect=RuntimeError("retry")) as mock_retry: - with self.assertRaises(RuntimeError): - send_registration_cancellation_email.run("1") - - mock_retry.assert_called_once() - - def test_send_skyroom_credentials_individual_task_sends_email(self): - user = SimpleNamespace(email="user@example.com") - event = SimpleNamespace( - title="E", - slug="slug", - online_link="https://example.com", - ) - registration = SimpleNamespace( - event=event, - user=user, - ticket_id="abcdefghijk", - ) - manager = mock.MagicMock() - manager.get.return_value = registration - email_instance = mock.MagicMock() - - with mock.patch("apps.events.tasks.Registration.objects", manager), \ - mock.patch("apps.events.tasks.EmailMultiAlternatives", mock.MagicMock(return_value=email_instance)), \ - mock.patch("apps.events.tasks.render_to_string", return_value="ok
"), \ - mock.patch("apps.events.tasks.strip_tags", return_value="ok"): - send_skyroom_credentials_individual_task.run(1) - - email_instance.send.assert_called_once() - - def test_send_event_reminder_task_sends_messages(self): - event = SimpleNamespace(title="Ev", slug="slug") - - class DummyRegs: - def __init__(self, ids): - self.ids = ids - def select_related(self, *args, **kwargs): - return self - def distinct(self): - return self - def values_list(self, *args, **kwargs): - return self.ids - - regs = DummyRegs([1]) - with mock.patch("apps.events.tasks.Event.objects.get", return_value=event), \ - mock.patch("apps.events.tasks._event_recipients", return_value=regs), \ - mock.patch("apps.events.tasks.group") as mock_group: - mock_job = mock.MagicMock() - mock_group.return_value = mock_job - mock_job.apply_async.return_value = mock.MagicMock(id="gid") - - result = send_event_reminder_task.run(1) - - mock_group.assert_called_once() - mock_job.apply_async.assert_called_once() - self.assertEqual(result["queued"], 1) - self.assertEqual(result["group_id"], "gid") - - def test_queue_event_announcement_builds_group(self): - event = self.create_event() - class DummyQS: - def __init__(self, ids): - self.ids = ids - def select_related(self, *args, **kwargs): - return self - def exclude(self, *args, **kwargs): - return self - def distinct(self): - return self - def values_list(self, *args, **kwargs): - return self.ids - with mock.patch("apps.events.tasks.Event.objects.get", return_value=event), \ - mock.patch("apps.events.tasks._event_recipients", return_value=DummyQS([1, 2])), \ - mock.patch("apps.events.tasks.group") as mock_group: - mock_job = mock.MagicMock() - mock_group.return_value = mock_job - result = queue_event_announcement.run(event.id, "subject", "body
") - - mock_group.assert_called_once() - mock_job.apply_async.assert_called_once() - self.assertEqual(result["queued"], 2) - - def test_send_event_announcement_to_user_marks_sent(self): - event = self.create_event() - user = self.create_user() - registration = SimpleNamespace( - user=user, - event=event, - id=1, - ) - log = mock.MagicMock(status=EventEmailLog.STATUS_PENDING) - with mock.patch("apps.events.tasks.Registration.objects.select_related") as mock_select, \ - mock.patch("apps.events.tasks.EventEmailLog.claim", return_value=(log, False)), \ - mock.patch("apps.events.tasks.render_to_string", return_value="ok
"), \ - mock.patch("apps.events.tasks.strip_tags", return_value="ok"), \ - mock.patch("apps.events.tasks.EmailMultiAlternatives", return_value=mock.MagicMock()): - mock_select.return_value.get.return_value = registration - send_event_announcement_to_user._orig_run(event.id, 1, "subject", "body
") - - log.mark_sent.assert_called_once() - - def test_send_event_announcement_to_user_returns_skip(self): - log = mock.MagicMock(status=EventEmailLog.STATUS_PENDING) - with mock.patch("apps.events.tasks.Registration.objects.select_related") as mock_select, \ - mock.patch("apps.events.tasks.EventEmailLog.claim", return_value=(log, True)): - mock_select.return_value.get.return_value = SimpleNamespace(user=SimpleNamespace(id=1), event=SimpleNamespace(slug="slug")) - result = send_event_announcement_to_user._orig_run(1, 1, "subject", "body
") - - self.assertEqual(result, {"skipped": True, "status": log.status}) - - def test_queue_invites_to_non_registered_users_uses_group(self): - event = self.create_event() - class DummyUserQS: - def __init__(self, ids): - self.ids = ids - def filter(self, *args, **kwargs): - return self - def exclude(self, *args, **kwargs): - return self - def distinct(self): - return self - def values_list(self, *args, **kwargs): - return self.ids - with mock.patch("apps.events.tasks.Event.objects.get", return_value=event), \ - mock.patch("apps.events.tasks.User.objects.all", return_value=DummyUserQS([1])), \ - mock.patch("apps.events.tasks.group") as mock_group: - mock_job = mock.MagicMock() - mock_group.return_value = mock_job - result = queue_invites_to_non_registered_users.run(event.id) - - mock_job.apply_async.assert_called_once() - self.assertEqual(result["queued"], 1) - - def test_send_invite_to_user_skips_when_claimed(self): - log = mock.MagicMock(status=EventEmailLog.STATUS_PENDING) - with mock.patch("apps.events.tasks.Event.objects.get", return_value=self.create_event()), \ - mock.patch("apps.events.tasks.User.objects.get", return_value=self.create_user()), \ - mock.patch("apps.events.tasks.EventEmailLog.claim", return_value=(log, True)): - result = send_invite_to_user._orig_run(1, 1) - - self.assertEqual(result, {"skipped": True, "status": log.status}) - - def test_send_invite_to_user_sends_email(self): - msg_instance = mock.MagicMock() - target_user = self.create_user() - with mock.patch("apps.events.tasks.Event.objects.get", return_value=self.create_event()), \ - mock.patch("apps.events.tasks.User.objects.get", return_value=target_user), \ - mock.patch("apps.events.tasks.EventEmailLog.claim", return_value=(mock.MagicMock(), False)), \ - mock.patch("apps.events.tasks.render_to_string", return_value="ok
"), \ - mock.patch("apps.events.tasks._build_email_context", return_value="ctx"), \ - mock.patch("apps.events.tasks.EmailMultiAlternatives", return_value=msg_instance): - result = send_invite_to_user._orig_run(1, 1) - - msg_instance.send.assert_called_once() - self.assertEqual(result, f"Email sent to {target_user.email}") - - def test_queue_skyroom_credentials_builds_group(self): - event = self.create_event() - class DummyRegQS: - def __init__(self, ids): - self.ids = ids - def select_related(self, *args, **kwargs): - return self - def exclude(self, *args, **kwargs): - return self - def distinct(self): - return self - def values_list(self, *args, **kwargs): - return self.ids - with mock.patch("apps.events.tasks.Event.objects.get", return_value=event), \ - mock.patch("apps.events.tasks._event_recipients", return_value=DummyRegQS([1])), \ - mock.patch("apps.events.tasks.group") as mock_group: - mock_job = mock.MagicMock() - mock_group.return_value = mock_job - result = queue_skyroom_credentials.run(event.id) - - mock_job.apply_async.assert_called_once() - self.assertEqual(result["queued"], 1) - - def test_send_skyroom_credentials_to_user_skips_when_claimed(self): - log = mock.MagicMock(status=EventEmailLog.STATUS_PENDING) - with mock.patch("apps.events.tasks.Registration.objects.select_related") as mock_select, \ - mock.patch("apps.events.tasks.EventEmailLog.claim", return_value=(log, True)): - mock_select.return_value.get.return_value = SimpleNamespace( - user=SimpleNamespace(id=1, email="user@example.com"), - event=SimpleNamespace(id=1, slug="slug", online_link="https://example.com", title="E"), - ticket_id=uuid.uuid4(), - ) - result = send_skyroom_credentials_to_user._orig_run(1, 1) - - self.assertEqual(result, {"skipped": True, "status": log.status}) - - def test_send_skyroom_credentials_to_user_sends_email(self): - msg_instance = mock.MagicMock() - with mock.patch("apps.events.tasks.Registration.objects.select_related") as mock_select, \ - mock.patch("apps.events.tasks.EventEmailLog.claim", return_value=(mock.MagicMock(), False)), \ - mock.patch("apps.events.tasks.render_to_string", return_value="ok
"), \ - mock.patch("apps.events.tasks.strip_tags", return_value="ok"), \ - mock.patch("apps.events.tasks.EmailMultiAlternatives", return_value=msg_instance): - mock_select.return_value.get.return_value = SimpleNamespace( - user=SimpleNamespace(email="user@example.com", id=1), - event=SimpleNamespace(title="Title", slug="slug", online_link="https://example.com"), - ticket_id=uuid.uuid4(), - ) - send_skyroom_credentials_to_user._orig_run(1, 1) - - msg_instance.send.assert_called_once() - - def test_send_registration_confirmation_email_skips_without_email(self): - registration = SimpleNamespace( - user=SimpleNamespace(email=""), - event=SimpleNamespace(title="Title", registration_success_markdown=""), - ) - manager = mock.MagicMock() - manager.select_related.return_value.get.return_value = registration - with mock.patch("apps.events.tasks.Registration.objects", manager), \ - mock.patch("apps.events.tasks.EmailMultiAlternatives") as mock_email: - send_registration_confirmation_email.run("1") - - mock_email.assert_not_called() - - def test_send_registration_confirmation_email_retries_on_failure(self): - registration = SimpleNamespace( - user=SimpleNamespace(email="user@example.com"), - event=SimpleNamespace(title="Title", registration_success_markdown=""), - ) - manager = mock.MagicMock() - manager.select_related.return_value.get.return_value = registration - email_instance = mock.MagicMock() - email_instance.send.side_effect = RuntimeError("boom") - mock_email_class = mock.MagicMock(return_value=email_instance) - - with mock.patch("apps.events.tasks.Registration.objects", manager), \ - mock.patch("apps.events.tasks.EmailMultiAlternatives", mock_email_class), \ - mock.patch("apps.events.tasks.render_to_string", return_value="ok
"), \ - mock.patch("apps.events.tasks.strip_tags", return_value="ok"), \ - mock.patch.object(send_registration_confirmation_email, "retry", side_effect=RuntimeError("retry")) as mock_retry: - with self.assertRaises(RuntimeError): - send_registration_confirmation_email.run("1") - - mock_retry.assert_called_once() - - def test_event_recipients_disregards_verification_flag(self): - event = self.create_event() - user = self.create_user() - user.is_email_verified = False - user.save(update_fields=["is_email_verified"]) - registration = Registration.objects.create( - event=event, - user=user, - status=Registration.StatusChoices.PENDING, - ) - - recipients = _event_recipients(event, only_verified=False) - - self.assertEqual(len(recipients), 1) - self.assertEqual(recipients[0].user_id, user.id) - - def test_send_skyroom_credentials_individual_task_retries_on_failure(self): - user = SimpleNamespace(email="user@example.com") - event = SimpleNamespace(title="Title", slug="slug", online_link="https://example.com") - registration = SimpleNamespace(user=user, event=event, ticket_id="abcdef") - manager = mock.MagicMock() - manager.get.return_value = registration - - with mock.patch("apps.events.tasks.Registration.objects", manager), \ - mock.patch("apps.events.tasks.EmailMultiAlternatives", mock.MagicMock(return_value=mock.MagicMock(send=mock.Mock(side_effect=RuntimeError("boom"))))), \ - mock.patch("apps.events.tasks.render_to_string", return_value="ok
"), \ - mock.patch("apps.events.tasks.strip_tags", return_value="ok"), \ - mock.patch.object(send_skyroom_credentials_individual_task, "retry", side_effect=RuntimeError("retry")) as mock_retry: - with self.assertRaises(RuntimeError): - send_skyroom_credentials_individual_task.run(1) - - self.assertTrue(mock_retry.called) - - def test_send_event_reminder_task_propagates_failure(self): - event = SimpleNamespace(title="Ev", slug="slug") - - class DummyRegs: - def __init__(self, ids): - self.ids = ids - def select_related(self, *args, **kwargs): - return self - def distinct(self): - return self - def values_list(self, *args, **kwargs): - return self.ids - - regs = DummyRegs([1]) - with mock.patch("apps.events.tasks.Event.objects.get", return_value=event), \ - mock.patch("apps.events.tasks._event_recipients", return_value=regs), \ - mock.patch("apps.events.tasks.group") as mock_group: - mock_job = mock.MagicMock() - mock_group.return_value = mock_job - mock_job.apply_async.side_effect = RuntimeError("boom") - - with self.assertRaises(RuntimeError): - send_event_reminder_task.run(1) - - def test_send_event_reminder_to_user_marks_sent(self): - event = self.create_event() - user = self.create_user() - registration = SimpleNamespace(user=user, event=event, id=1) - log = mock.MagicMock(status=EventEmailLog.STATUS_PENDING) - msg_instance = mock.MagicMock() - with mock.patch("apps.events.tasks.Registration.objects.select_related") as mock_select, \ - mock.patch("apps.events.tasks.EventEmailLog.claim", return_value=(log, False)), \ - mock.patch("apps.events.tasks.render_to_string", return_value="ok
"), \ - mock.patch("apps.events.tasks.strip_tags", return_value="ok"), \ - mock.patch("apps.events.tasks.EmailMultiAlternatives", return_value=msg_instance): - mock_select.return_value.get.return_value = registration - result = send_event_reminder_to_user._orig_run(event.id, 1) - - msg_instance.send.assert_called_once() - log.mark_sent.assert_called_once() - self.assertEqual(result, f"Email sent to {user.email}") - - def test_send_event_announcement_to_user_handles_soft_time_limit(self): - event = self.create_event() - user = self.create_user() - registration = SimpleNamespace(user=user, event=event, id=1) - log = mock.MagicMock(status=EventEmailLog.STATUS_PENDING) - with mock.patch("apps.events.tasks.Registration.objects.select_related") as mock_select, \ - mock.patch("apps.events.tasks.EventEmailLog.claim", return_value=(log, False)), \ - mock.patch("apps.events.tasks.render_to_string", side_effect=SoftTimeLimitExceeded("timeout")), \ - mock.patch("apps.events.tasks.strip_tags") as mock_strip: - mock_select.return_value.get.return_value = registration - with self.assertRaises(SoftTimeLimitExceeded): - send_event_announcement_to_user._orig_run(1, 1, "subject", "body
") - - log.mark_failed.assert_called_once_with("Soft time limit exceeded") - - def test_send_event_announcement_to_user_handles_failure(self): - event = self.create_event() - user = self.create_user() - registration = SimpleNamespace(user=user, event=event, id=1) - log = mock.MagicMock(status=EventEmailLog.STATUS_PENDING) - with mock.patch("apps.events.tasks.Registration.objects.select_related") as mock_select, \ - mock.patch("apps.events.tasks.EventEmailLog.claim", return_value=(log, False)), \ - mock.patch("apps.events.tasks.render_to_string", return_value="ok
"), \ - mock.patch("apps.events.tasks.strip_tags", return_value="ok"), \ - mock.patch("apps.events.tasks.EmailMultiAlternatives", return_value=mock.MagicMock(send=mock.Mock(side_effect=RuntimeError("boom")))): - mock_select.return_value.get.return_value = registration - with self.assertRaises(RuntimeError): - send_event_announcement_to_user._orig_run(1, 1, "subject", "body
") - - log.mark_failed.assert_called_once() - - def test_send_invite_to_user_handles_failure(self): - event = self.create_event() - user = self.create_user() - log = mock.MagicMock(status=EventEmailLog.STATUS_PENDING) - with mock.patch("apps.events.tasks.Event.objects.get", return_value=event), \ - mock.patch("apps.events.tasks.User.objects.get", return_value=user), \ - mock.patch("apps.events.tasks.EventEmailLog.claim", return_value=(log, False)), \ - mock.patch("apps.events.tasks.render_to_string", return_value="ok
"), \ - mock.patch("apps.events.tasks._build_email_context", return_value="ctx"), \ - mock.patch("apps.events.tasks.EmailMultiAlternatives", return_value=mock.MagicMock(send=mock.Mock(side_effect=RuntimeError("boom")))): - with self.assertRaises(RuntimeError): - send_invite_to_user._orig_run(1, 1) - - log.mark_failed.assert_called_once() - - def test_send_skyroom_credentials_to_user_handles_failure(self): - event = self.create_event() - user = self.create_user() - log = mock.MagicMock(status=EventEmailLog.STATUS_PENDING) - with mock.patch("apps.events.tasks.Registration.objects.select_related") as mock_select, \ - mock.patch("apps.events.tasks.EventEmailLog.claim", return_value=(log, False)), \ - mock.patch("apps.events.tasks.render_to_string", return_value="ok
"), \ - mock.patch("apps.events.tasks.strip_tags", return_value="ok"), \ - mock.patch("apps.events.tasks.EmailMultiAlternatives", return_value=mock.MagicMock(send=mock.Mock(side_effect=RuntimeError("boom")))): - mock_select.return_value.get.return_value = SimpleNamespace( - user=user, - event=event, - ticket_id=uuid.uuid4(), - ) - with self.assertRaises(RuntimeError): - send_skyroom_credentials_to_user._orig_run(1, 1) - - log.mark_failed.assert_called_once() - - def test_queue_invites_to_non_registered_users_respects_filters(self): - event = self.create_event() - verified = self.create_user() - verified.is_email_verified = True - verified.save(update_fields=["is_email_verified"]) - inactive = self.create_user() - inactive.is_email_verified = True - inactive.is_active = False - inactive.save(update_fields=["is_email_verified", "is_active"]) - with mock.patch("apps.events.tasks.group") as mock_group: - mock_job = mock.MagicMock() - mock_group.return_value = mock_job - result = queue_invites_to_non_registered_users.run(event.id, only_verified=True, only_active=True) - - mock_job.apply_async.assert_called_once() - self.assertEqual(result["queued"], 1) + self.user = User.objects.create_user(username="attendee", mobile="09191110003", password="password") + + def test_markdown_description_and_registration_window_remain_available(self): + self.assertIn("Important", self.event.description_html) + self.assertTrue(self.event.is_registration_open) + self.event.registration_end_date = timezone.now() - timedelta(minutes=1) + self.assertFalse(self.event.is_registration_open) + + @mock.patch("apps.events.tasks.send_registration_confirmation_notification.delay") + def test_capacity_counts_confirmed_registration(self, delay): + Registration.objects.create(event=self.event, user=self.user, status=Registration.StatusChoices.CONFIRMED) + self.assertEqual(self.event.current_attendees_count, 1) + self.assertFalse(self.event.has_available_slots) + delay.assert_called_once() + + def test_unlimited_capacity_allows_registration(self): + self.event.capacity = None + self.assertTrue(self.event.has_available_slots) diff --git a/apps/notifications/api/schemas.py b/apps/notifications/api/schemas.py index 282a63b..e7baf3b 100644 --- a/apps/notifications/api/schemas.py +++ b/apps/notifications/api/schemas.py @@ -9,10 +9,10 @@ class NotificationSchema(Schema): type: str title: str message: str + has_body: bool = False level: str created_at: datetime | str is_seen: bool - delete_on_seen: bool action_url: str | None = None entity_type: str | None = None entity_id: int | str | None = None @@ -25,6 +25,10 @@ class NotificationListSchema(Schema): notifications: list[NotificationSchema] +class NotificationDetailSchema(NotificationSchema): + body_markdown: str | None = None + + class NotificationSeenIn(Schema): id: str diff --git a/apps/notifications/api/views.py b/apps/notifications/api/views.py index 99230b7..33b2158 100644 --- a/apps/notifications/api/views.py +++ b/apps/notifications/api/views.py @@ -9,6 +9,7 @@ from django.views import View from ninja import Router from apps.notifications.api.schemas import ( + NotificationDetailSchema, NotificationDeleteResponseSchema, NotificationListSchema, NotificationMarkAllReadResponseSchema, @@ -54,7 +55,9 @@ def _validate_stream_token(token: str | None) -> str: @notifications_router.get("/", response=NotificationListSchema, auth=jwt_auth) -def list_notifications(request, limit: int | None = None, offset: int = 0, type: str | None = None): +def list_notifications(request, limit: int | None = None, offset: int = 0, type: str | None = None, status: str = "all"): + if status not in {"all", "unread", "read"}: + return JsonResponse({"error": "Invalid status filter."}, status=400) user_id = str(request.auth.id) safe_limit = min(_safe_int(limit, settings.NOTIFICATION_DEFAULT_PAGE_SIZE), settings.NOTIFICATION_MAX_PAGE_SIZE) notifications, total_count = RedisNotificationStore.list( @@ -62,11 +65,12 @@ def list_notifications(request, limit: int | None = None, offset: int = 0, type: limit=safe_limit, offset=_safe_int(offset, 0), type_filter=type, + read_filter=status, ) return { "count": total_count, "unread_count": RedisNotificationStore.unread_count(user_id), - "notifications": notifications, + "notifications": [RedisNotificationStore.summary(item) for item in notifications], } @@ -78,18 +82,6 @@ def mark_notification_seen(request, data: NotificationSeenIn): return 200, {"marked_read": True, **payload} -@notifications_router.delete("/{notif_id}", response={200: NotificationDeleteResponseSchema, 404: ErrorSchema}, auth=jwt_auth) -def delete_notification(request, notif_id: str): - deleted = RedisNotificationStore.delete(str(request.auth.id), notif_id) - if not deleted: - return 404, {"error": "Notification not found."} - return 200, { - "deleted": True, - "notification_id": notif_id, - "unread_count": RedisNotificationStore.unread_count(str(request.auth.id)), - } - - @notifications_router.post("/mark-all-read", response=NotificationMarkAllReadResponseSchema, auth=jwt_auth) def mark_all_notifications_read(request, type: str | None = None): updated = RedisNotificationStore.mark_all_seen(str(request.auth.id), type_filter=type) @@ -106,6 +98,26 @@ def get_notification_stream_token(request): } +@notifications_router.get("/{notif_id}", response={200: NotificationDetailSchema, 404: ErrorSchema}, auth=jwt_auth) +def get_notification(request, notif_id: str): + notification = RedisNotificationStore.get(str(request.auth.id), notif_id) + if notification is None: + return 404, {"error": "Notification not found."} + return 200, {**notification, "has_body": bool(notification.get("body_markdown"))} + + +@notifications_router.delete("/{notif_id}", response={200: NotificationDeleteResponseSchema, 404: ErrorSchema}, auth=jwt_auth) +def delete_notification(request, notif_id: str): + deleted = RedisNotificationStore.delete(str(request.auth.id), notif_id) + if not deleted: + return 404, {"error": "Notification not found."} + return 200, { + "deleted": True, + "notification_id": notif_id, + "unread_count": RedisNotificationStore.unread_count(str(request.auth.id)), + } + + class NotificationStreamView(View): def _build_stream(self, user_id: str) -> Iterator[str]: pubsub = RedisNotificationStore.get_pubsub() @@ -118,7 +130,7 @@ class NotificationStreamView(View): ) unread_count = RedisNotificationStore.unread_count(user_id) yield f"retry: {settings.NOTIFICATION_SSE_RETRY_MS}\n\n" - yield _format_sse_event("connected", {"notifications": initial_notifications, "unread_count": unread_count}) + yield _format_sse_event("connected", {"notifications": [RedisNotificationStore.summary(item) for item in initial_notifications], "unread_count": unread_count}) yield _format_sse_event("unread_count", {"unread_count": unread_count}) pubsub.subscribe(channel) diff --git a/apps/notifications/services.py b/apps/notifications/services.py index aba2ca3..01fa743 100644 --- a/apps/notifications/services.py +++ b/apps/notifications/services.py @@ -50,16 +50,22 @@ class RedisNotificationStore: "type": payload.get("type") or "notification", "title": payload.get("title") or "", "message": payload.get("message") or "", + "body_markdown": payload.get("body_markdown") or None, "level": payload.get("level") or "info", "created_at": _isoformat_datetime(payload.get("created_at")), "is_seen": bool(payload.get("is_seen", False)), - "delete_on_seen": bool(payload.get("delete_on_seen", False)), "action_url": payload.get("action_url"), "entity_type": payload.get("entity_type"), "entity_id": payload.get("entity_id"), "meta": payload.get("meta") or {}, } + @staticmethod + def summary(data: dict) -> dict: + result = {**data, "has_body": bool(data.get("body_markdown"))} + result.pop("body_markdown", None) + return result + @classmethod def _publish_event(cls, user_id: str, event: str, data: dict) -> None: if not settings.NOTIFICATIONS_ENABLED: @@ -96,7 +102,7 @@ class RedisNotificationStore: pipe.execute() unread_count = cls.unread_count(user_id) - cls._publish_event(user_id, "notification", {"notification": data, "unread_count": unread_count}) + cls._publish_event(user_id, "notification", {"notification": cls.summary(data), "unread_count": unread_count}) cls._publish_event(user_id, "unread_count", {"unread_count": unread_count}) return data @@ -108,6 +114,7 @@ class RedisNotificationStore: limit: int | None = None, offset: int = 0, type_filter: str | None = None, + read_filter: str = "all", paginate: bool = True, ) -> tuple[list[dict], int]: ids_key = cls._ids_key(user_id) @@ -134,6 +141,10 @@ class RedisNotificationStore: continue if type_filter and data.get("type") != type_filter: continue + if read_filter == "unread" and data["is_seen"]: + continue + if read_filter == "read" and not data["is_seen"]: + continue items.append(data) if cleanup_ids: @@ -182,17 +193,6 @@ class RedisNotificationStore: data = cls.get(user_id, notif_id) if not data: return None - if data.get("delete_on_seen"): - deleted = cls.delete(user_id, notif_id) - if deleted: - return { - "notification_id": notif_id, - "deleted": True, - "notification": None, - "unread_count": cls.unread_count(user_id), - } - return None - if not data.get("is_seen"): data["is_seen"] = True redis_client.hset( @@ -204,7 +204,7 @@ class RedisNotificationStore: payload = { "notification_id": notif_id, "deleted": False, - "notification": data, + "notification": cls.summary(data), "unread_count": unread_count, } cls._publish_event(user_id, "notification_seen", payload) @@ -228,16 +228,12 @@ class RedisNotificationStore: continue if type_filter and data.get("type") != type_filter: continue - if data.get("delete_on_seen"): - pipe.zrem(cls._ids_key(user_id), notif_id) - pipe.hdel(cls._data_key(user_id), notif_id) - else: - data["is_seen"] = True - pipe.hset( - cls._data_key(user_id), - notif_id, - json.dumps(data, ensure_ascii=False, default=str), - ) + data["is_seen"] = True + pipe.hset( + cls._data_key(user_id), + notif_id, + json.dumps(data, ensure_ascii=False, default=str), + ) updated += 1 if updated: pipe.execute() diff --git a/apps/notifications/tasks.py b/apps/notifications/tasks.py index 06ee7a8..12f0a7a 100644 --- a/apps/notifications/tasks.py +++ b/apps/notifications/tasks.py @@ -1,8 +1,9 @@ import logging from celery import shared_task +from django.contrib.auth import get_user_model -from apps.notifications.services import RedisNotificationStore +from apps.notifications.services import RedisNotificationStore, notify_user logger = logging.getLogger(__name__) @@ -12,3 +13,19 @@ def cleanup_notification_retention(): removed = RedisNotificationStore.cleanup_expired() logger.info("Cleaned up %s expired notifications", removed) return removed + + +@shared_task +def send_manual_notification(user_ids: list[int], title: str, digest: str, body_markdown: str | None = None): + users = get_user_model().objects.filter(pk__in=user_ids, is_active=True, is_deleted=False) + sent = 0 + for user_id in users.values_list("pk", flat=True).iterator(): + notify_user(user_id, { + "type": "manual", + "title": title, + "message": digest, + "body_markdown": body_markdown, + "level": "info", + }) + sent += 1 + return sent diff --git a/apps/notifications/tests/__init__.py b/apps/notifications/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/apps/notifications/tests/test_notifications.py b/apps/notifications/tests/test_notifications.py new file mode 100644 index 0000000..409f8e1 --- /dev/null +++ b/apps/notifications/tests/test_notifications.py @@ -0,0 +1,170 @@ +import json +from types import SimpleNamespace +from unittest import mock + +from django.contrib import admin +from django.test import RequestFactory, SimpleTestCase, TestCase + +from apps.notifications.api.views import get_notification, list_notifications +from apps.notifications.services import RedisNotificationStore +from apps.notifications.tasks import send_manual_notification +from apps.users.admin import ManualNotificationForm, UserAdmin +from apps.users.models import User +from core.authentication import create_jwt_token + + +class NotificationStoreTests(SimpleTestCase): + def test_summary_hides_full_body_and_advertises_it(self): + notification = RedisNotificationStore._normalize_notification({ + "id": "one", "title": "Title", "message": "Digest", "body_markdown": "# Full body", + }) + summary = RedisNotificationStore.summary(notification) + self.assertTrue(summary["has_body"]) + self.assertNotIn("body_markdown", summary) + self.assertEqual(notification["body_markdown"], "# Full body") + + @mock.patch("apps.notifications.services.redis_client") + def test_read_filter_applies_before_pagination(self, redis_client): + redis_client.zrevrange.return_value = ["one", "two", "three"] + redis_client.pipeline.return_value.execute.return_value = [ + json.dumps({"id": "one", "is_seen": False}), + json.dumps({"id": "two", "is_seen": True}), + json.dumps({"id": "three", "is_seen": False}), + ] + unread, count = RedisNotificationStore.list("user", limit=1, offset=1, read_filter="unread") + self.assertEqual(count, 2) + self.assertEqual([item["id"] for item in unread], ["three"]) + read, count = RedisNotificationStore.list("user", read_filter="read") + self.assertEqual(count, 1) + self.assertEqual(read[0]["id"], "two") + + @mock.patch("apps.notifications.api.views.RedisNotificationStore") + def test_detail_is_scoped_to_authenticated_recipient(self, store): + store.get.side_effect = lambda user_id, _: {"id": "one", "body_markdown": "# Secret"} if user_id == "1" else None + owner = SimpleNamespace(auth=SimpleNamespace(id=1)) + other = SimpleNamespace(auth=SimpleNamespace(id=2)) + self.assertEqual(get_notification(owner, "one")[0], 200) + self.assertEqual(get_notification(other, "one")[0], 404) + store.get.assert_any_call("2", "one") + + @mock.patch("apps.notifications.api.views.RedisNotificationStore") + def test_list_returns_summaries_and_rejects_unknown_status(self, store): + request = SimpleNamespace(auth=SimpleNamespace(id=1)) + store.list.return_value = ([{"id": "one", "body_markdown": "# Secret"}], 1) + store.summary.side_effect = RedisNotificationStore.summary + store.unread_count.return_value = 1 + result = list_notifications(request, status="unread") + self.assertNotIn("body_markdown", result["notifications"][0]) + self.assertEqual(store.list.call_args.kwargs["read_filter"], "unread") + self.assertEqual(list_notifications(request, status="invalid").status_code, 400) + + @mock.patch.object(RedisNotificationStore, "_publish_event") + @mock.patch.object(RedisNotificationStore, "unread_count", return_value=0) + @mock.patch.object(RedisNotificationStore, "get") + @mock.patch("apps.notifications.services.redis_client") + def test_marking_old_ephemeral_notification_read_keeps_it_visible(self, redis_client, get, _count, _publish): + get.return_value = {"id": "one", "is_seen": False, "delete_on_seen": True, "body_markdown": "# Body"} + result = RedisNotificationStore.mark_seen("user", "one") + self.assertFalse(result["deleted"]) + self.assertTrue(result["notification"]["is_seen"]) + redis_client.hset.assert_called_once() + redis_client.pipeline.assert_not_called() + + +class ManualNotificationTests(TestCase): + def test_form_requires_title_and_plain_digest_but_allows_optional_body(self): + self.assertTrue(ManualNotificationForm({"title": "Title", "digest": "Intro"}).is_valid()) + self.assertFalse(ManualNotificationForm({"title": "", "digest": "Intro"}).is_valid()) + + @mock.patch("apps.notifications.tasks.notify_user") + def test_task_sends_only_to_selected_active_users(self, notify_user): + active = User.objects.create_user(username="active", mobile="09190000001", password="password") + inactive = User.objects.create_user(username="inactive", mobile="09190000002", password="password", is_active=False) + sent = send_manual_notification.run([active.pk, inactive.pk], "Title", "Digest", "# Body") + self.assertEqual(sent, 1) + notify_user.assert_called_once_with(active.pk, { + "type": "manual", "title": "Title", "message": "Digest", "body_markdown": "# Body", "level": "info", + }) + + @mock.patch("apps.users.admin.send_manual_notification.delay") + def test_admin_action_queues_valid_composer_for_selected_users(self, delay): + staff = User.objects.create_superuser(username="staff", mobile="09190000003", password="password") + recipient = User.objects.create_user(username="recipient", mobile="09190000004", password="password") + request = RequestFactory().post("/admin/users/user/", { + "confirm_send": "1", "title": "Title", "digest": "Digest", "body_markdown": "**Body**", + }) + request.user = staff + model_admin = UserAdmin(User, admin.site) + with mock.patch.object(model_admin, "message_user"): + model_admin.send_notification(request, User.objects.filter(pk=recipient.pk)) + delay.assert_called_once_with([recipient.pk], "Title", "Digest", "**Body**") + + def test_account_email_identity_is_true_when_an_email_is_supplied(self): + user = User.objects.create_user(username="emailuser", mobile="09190000005", password="password") + self.assertFalse(user.is_email_verified) + user.email = "person@example.com" + user.save(update_fields=["email"]) + user.refresh_from_db() + self.assertTrue(user.is_email_verified) + + @mock.patch("apps.users.admin.send_manual_notification.delay") + def test_admin_changelist_uses_the_composer_before_queuing(self, delay): + staff = User.objects.create_superuser(username="adminstaff", mobile="09190000008", password="password") + recipient = User.objects.create_user(username="target", mobile="09190000009", password="password") + self.client.force_login(staff) + action_data = {"action": "send_notification", "_selected_action": str(recipient.pk), "index": "0"} + path = "/admin/users/user/" + preview = self.client.post(path, action_data) + self.assertEqual(preview.status_code, 200) + self.assertContains(preview, "توضیح کوتاه") + self.assertFalse(delay.called) + result = self.client.post(path, { + **action_data, "confirm_send": "1", "title": "Title", "digest": "Digest", "body_markdown": "# Body", + }) + self.assertIn(result.status_code, (200, 302)) + delay.assert_called_once_with([recipient.pk], "Title", "Digest", "# Body") + + +class NotificationAPIIntegrationTests(TestCase): + @mock.patch("apps.notifications.api.views.RedisNotificationStore.get") + def test_detail_requires_auth_and_returns_body_only_to_recipient(self, get): + owner = User.objects.create_user(username="owner", mobile="09190000006", password="password") + other = User.objects.create_user(username="other", mobile="09190000007", password="password") + detail = RedisNotificationStore._normalize_notification({ + "id": "one", "title": "Title", "message": "Digest", "body_markdown": "# Private", + "action_url": "/events/workshop", + }) + get.side_effect = lambda user_id, _: detail if user_id == str(owner.pk) else None + path = "/api/notifications/one" + self.assertEqual(self.client.get(path).status_code, 401) + owner_response = self.client.get(path, HTTP_AUTHORIZATION=f"Bearer {create_jwt_token(owner)}") + self.assertEqual(owner_response.status_code, 200) + self.assertEqual(owner_response.json()["body_markdown"], "# Private") + self.assertEqual(owner_response.json()["action_url"], "/events/workshop") + other_response = self.client.get(path, HTTP_AUTHORIZATION=f"Bearer {create_jwt_token(other)}") + self.assertEqual(other_response.status_code, 404) + + @mock.patch("apps.notifications.api.views.RedisNotificationStore.unread_count", return_value=1) + @mock.patch("apps.notifications.api.views.RedisNotificationStore.list") + def test_list_excludes_full_body_and_uses_read_filter(self, list_notifications_store, _count): + user = User.objects.create_user(username="reader", mobile="09190000010", password="password") + full = RedisNotificationStore._normalize_notification({ + "id": "one", "title": "Title", "message": "Digest", "body_markdown": "# Private", + }) + list_notifications_store.return_value = ([full], 1) + response = self.client.get("/api/notifications/?status=unread", HTTP_AUTHORIZATION=f"Bearer {create_jwt_token(user)}") + self.assertEqual(response.status_code, 200) + item = response.json()["notifications"][0] + self.assertTrue(item["has_body"]) + self.assertNotIn("body_markdown", item) + self.assertEqual(list_notifications_store.call_args.kwargs["read_filter"], "unread") + + @mock.patch("apps.notifications.api.views.RedisNotificationStore.unread_count", return_value=0) + @mock.patch("apps.notifications.api.views.RedisNotificationStore.delete") + def test_delete_only_removes_the_authenticated_users_notification(self, delete, _count): + owner = User.objects.create_user(username="delete_owner", mobile="09190000011", password="password") + other = User.objects.create_user(username="delete_other", mobile="09190000012", password="password") + delete.side_effect = lambda user_id, _: user_id == str(owner.pk) + path = "/api/notifications/one" + self.assertEqual(self.client.delete(path, HTTP_AUTHORIZATION=f"Bearer {create_jwt_token(other)}").status_code, 404) + self.assertEqual(self.client.delete(path, HTTP_AUTHORIZATION=f"Bearer {create_jwt_token(owner)}").status_code, 200) diff --git a/apps/payments/api/views.py b/apps/payments/api/views.py index 8cc4efa..b5fe602 100644 --- a/apps/payments/api/views.py +++ b/apps/payments/api/views.py @@ -315,7 +315,7 @@ def callback(request, Authority: str | None = None, Status: str | None = None): if previous_status != Payment.OrderStatusChoices.CANCELED: _notify_payment_status( pay, - title=f"پرداخت {pay.event.title} لغو شد", + title=f"پرداخت رویداد «{pay.event.title}» لغو شد", message="پرداخت شما کامل نشد و ثبتنام نهایی انجام نگرفت.", level="warning", ) @@ -341,7 +341,7 @@ def callback(request, Authority: str | None = None, Status: str | None = None): if previous_status != Payment.OrderStatusChoices.FAILED: _notify_payment_status( pay, - title=f"پرداخت {pay.event.title} ناموفق بود", + title=f"پرداخت رویداد «{pay.event.title}» ناموفق بود", message="خطا در تأیید پرداخت رخ داد. در صورت نیاز دوباره تلاش کنید.", level="error", ) @@ -373,7 +373,7 @@ def callback(request, Authority: str | None = None, Status: str | None = None): if previous_status != Payment.OrderStatusChoices.PAID: _notify_payment_status( pay, - title=f"پرداخت {pay.event.title} تأیید شد", + title=f"پرداخت رویداد «{pay.event.title}» تأیید شد", message="پرداخت شما با موفقیت ثبت شد و ثبتنام رویداد تکمیل شده است.", level="success", ) @@ -385,7 +385,7 @@ def callback(request, Authority: str | None = None, Status: str | None = None): if previous_status != Payment.OrderStatusChoices.FAILED: _notify_payment_status( pay, - title=f"پرداخت {pay.event.title} ناموفق بود", + title=f"پرداخت رویداد «{pay.event.title}» ناموفق بود", message="تراکنش شما توسط درگاه تأیید نشد. در صورت نیاز دوباره پرداخت را انجام دهید.", level="error", ) diff --git a/apps/users/admin.py b/apps/users/admin.py index a245054..36363e1 100644 --- a/apps/users/admin.py +++ b/apps/users/admin.py @@ -1,15 +1,15 @@ from django import forms -from django.utils import timezone -from django.conf import settings from django.contrib import admin, messages from django.contrib.auth.admin import UserAdmin as BaseUserAdmin +from django.contrib.admin.helpers import ACTION_CHECKBOX_NAME +from django.template.response import TemplateResponse from import_export.admin import ImportExportModelAdmin from simplemde.widgets import SimpleMDEEditor from apps.users.models import User, University, Major from apps.users.resources import UserResource -from apps.users.tasks import send_verification_email +from apps.notifications.tasks import send_manual_notification from core.admin import SoftDeleteListFilter, BaseModelAdmin @@ -22,6 +22,12 @@ class UserAdminForm(forms.ModelForm): model = User fields = '__all__' + +class ManualNotificationForm(forms.Form): + title = forms.CharField(max_length=200, label="عنوان") + digest = forms.CharField(max_length=500, label="توضیح کوتاه", widget=forms.Textarea(attrs={"rows": 3})) + body_markdown = forms.CharField(required=False, label="متن کامل (Markdown)", widget=SimpleMDEEditor()) + @admin.register(User) class UserAdmin(BaseUserAdmin, BaseModelAdmin, ImportExportModelAdmin): form = UserAdminForm @@ -41,16 +47,10 @@ class UserAdmin(BaseUserAdmin, BaseModelAdmin, ImportExportModelAdmin): }), ('Important dates', {'fields': ('last_login', 'date_joined')}), - ('Email Verification', { - 'fields': ('is_email_verified', 'email_verification_token', 'email_verification_sent_at') - }), + ('Email identity', {'fields': ('is_email_verified',)}), ('Mobile Verification', { 'fields': ('is_mobile_verified',) }), - ('Password Reset', { - 'fields': ('password_reset_token', 'password_reset_token_expires_at'), - 'classes': ('collapse',) - }), ('Soft Delete', { 'fields': ('is_deleted', 'deleted_at'), 'classes': ('collapse',) @@ -66,50 +66,44 @@ class UserAdmin(BaseUserAdmin, BaseModelAdmin, ImportExportModelAdmin): ), ) - readonly_fields = ('email_verification_token', 'email_verification_sent_at', 'deleted_at', - 'password_reset_token', 'password_reset_token_expires_at') + readonly_fields = ('deleted_at',) - actions = BaseModelAdmin.actions + [ - 'verify_emails', - 'resend_verification_email', - ] - - @admin.action(description='Verify selected user emails') - def verify_emails(self, request, queryset): - queryset.update(is_email_verified=True) - self.message_user(request, f'Verified {queryset.count()} user emails.') + actions = BaseModelAdmin.actions + ['send_notification'] - @admin.action(description="Resend verification email") - def resend_verification_email(self, request, queryset): - qs = queryset.filter(is_email_verified=False).exclude(email__isnull=True).exclude(email="") + def get_actions(self, request): + actions = super().get_actions(request) + if not request.user.has_perm('users.send_notification'): + actions.pop('send_notification', None) + return actions - total = queryset.count() - to_send = qs.count() - skipped = total - to_send - sent = failed = 0 - - for user in qs: - try: - user.regenerate_verification_token() - user.email_verification_sent_at = timezone.now() - user.save(update_fields=["email_verification_sent_at"]) - - verification_url = f"{settings.FRONTEND_ROOT}verify-email/{user.email_verification_token}" - send_verification_email.delay(user.id, verification_url) - sent += 1 - except Exception as exc: - failed += 1 - - if sent: - self.message_user(request, f"ایمیل تأیید برای {sent} کاربر ارسال شد.", level=messages.SUCCESS) - if skipped: - self.message_user( - request, - f"{skipped} کاربر کنار گذاشته شدند (یا قبلاً تأیید شدهاند یا ایمیل ندارند).", - level=messages.WARNING, + @admin.action(description="Send in-app notification to selected users") + def send_notification(self, request, queryset): + if not request.user.has_perm('users.send_notification'): + self.message_user(request, "اجازه ارسال اعلان ندارید.", level=messages.ERROR) + return None + form = ManualNotificationForm(request.POST if 'confirm_send' in request.POST else None) + user_ids = list(queryset.filter(is_active=True, is_deleted=False).values_list('pk', flat=True)) + if not user_ids: + self.message_user(request, "کاربر فعال برای ارسال انتخاب نشده است.", level=messages.WARNING) + return None + if 'confirm_send' in request.POST and form.is_valid() and user_ids: + send_manual_notification.delay( + user_ids, + form.cleaned_data['title'], + form.cleaned_data['digest'], + form.cleaned_data['body_markdown'] or None, ) - if failed: - self.message_user(request, f"ارسال برای {failed} کاربر با خطا مواجه شد.", level=messages.ERROR) + self.message_user(request, f"ارسال اعلان برای {len(user_ids)} کاربر در صف قرار گرفت.") + return None + return TemplateResponse(request, 'admin/users/send_notification.html', { + **self.admin_site.each_context(request), + 'title': 'ارسال اعلان', + 'opts': self.model._meta, + 'form': form, + 'selected_ids': user_ids, + 'selected_count': len(user_ids), + 'action_checkbox_name': ACTION_CHECKBOX_NAME, + }) @admin.register(University) diff --git a/apps/users/api/views.py b/apps/users/api/views.py index 2f4d17a..1fb1fde 100644 --- a/apps/users/api/views.py +++ b/apps/users/api/views.py @@ -369,34 +369,6 @@ def google_oauth_claim_verify(request, data: GoogleClaimVerifySchema): return _error_response(exc) -@auth_router.get("/verify-email/{token}", response=MessageSchema) -def verify_email_legacy_guidance(request, token: str): - return { - "message": "تایید ایمیل غیرفعال شده است. برای بازیابی حساب از ورود با گوگل یا تایید شماره موبایل استفاده کنید." - } - - -@auth_router.post("/resend-verification", response=MessageSchema) -def resend_verification_legacy_guidance(request, email: str): - return { - "message": "ارسال ایمیل تایید غیرفعال شده است. برای ادامه، شماره موبایل خود را ثبت یا از ورود با گوگل استفاده کنید." - } - - -@auth_router.post("/request-password-reset", response=MessageSchema) -def request_password_reset_legacy_guidance(request): - return { - "message": "بازیابی رمز عبور با ایمیل غیرفعال شده است. از بازیابی با کد پیامکی یا ورود با گوگل استفاده کنید." - } - - -@auth_router.post("/reset-password-confirm", response=MessageSchema) -def reset_password_confirm_legacy_guidance(request): - return { - "message": "لینک بازیابی ایمیلی غیرفعال شده است. لطفاً از بازیابی رمز عبور با موبایل استفاده کنید." - } - - @auth_router.get("/profile", response=UserProfileSchema, auth=jwt_auth) def get_profile(request): return request.auth @@ -433,7 +405,7 @@ def update_profile(request, data: UserUpdateSchema): if existing.exists(): return 400, {"error": "این ایمیل قبلاً ثبت شده است."} payload["email"] = normalized_email - payload["is_email_verified"] = False if normalized_email else False + payload["is_email_verified"] = bool(normalized_email) for field, value in payload.items(): setattr(user, field, value) diff --git a/apps/users/migrations/0009_alter_user_options_and_more.py b/apps/users/migrations/0009_alter_user_options_and_more.py new file mode 100644 index 0000000..5da9390 --- /dev/null +++ b/apps/users/migrations/0009_alter_user_options_and_more.py @@ -0,0 +1,41 @@ +# Generated by Django 5.2.5 on 2026-09-26 07:57 + +from django.db import migrations + + +def align_email_identity_flags(apps, schema_editor): + users = apps.get_model('users', 'User').objects + users.filter(email__isnull=False).exclude(email='').update(is_email_verified=True) + users.filter(email__isnull=True).update(is_email_verified=False) + users.filter(email='').update(is_email_verified=False) + + +class Migration(migrations.Migration): + + dependencies = [ + ('users', '0008_alter_user_managers'), + ] + + operations = [ + migrations.RunPython(align_email_identity_flags, migrations.RunPython.noop), + migrations.AlterModelOptions( + name='user', + options={'permissions': [('send_notification', 'Can send in-app notifications')], 'verbose_name': 'User', 'verbose_name_plural': 'Users'}, + ), + migrations.RemoveField( + model_name='user', + name='email_verification_sent_at', + ), + migrations.RemoveField( + model_name='user', + name='email_verification_token', + ), + migrations.RemoveField( + model_name='user', + name='password_reset_token', + ), + migrations.RemoveField( + model_name='user', + name='password_reset_token_expires_at', + ), + ] diff --git a/apps/users/models.py b/apps/users/models.py index 7315df8..40b9307 100644 --- a/apps/users/models.py +++ b/apps/users/models.py @@ -1,10 +1,6 @@ from django.contrib.auth.models import AbstractUser, UserManager as DjangoUserManager -from django.utils import timezone from django.db import models -import uuid -from datetime import timedelta - from core.media import ( delete_image_derivatives_by_name, get_image_previous_name, @@ -80,11 +76,6 @@ class User(AbstractUser, BaseModel): ) is_email_verified = models.BooleanField(default=False) is_mobile_verified = models.BooleanField(default=False) - email_verification_token = models.UUIDField(default=uuid.uuid4, unique=True) - email_verification_sent_at = models.DateTimeField(null=True, blank=True) - - password_reset_token = models.UUIDField(null=True, blank=True, unique=True) - password_reset_token_expires_at = models.DateTimeField(null=True, blank=True) USERNAME_FIELD = 'username' REQUIRED_FIELDS = ['mobile'] @@ -95,6 +86,7 @@ class User(AbstractUser, BaseModel): db_table = 'users' verbose_name = 'User' verbose_name_plural = 'Users' + permissions = [('send_notification', 'Can send in-app notifications')] def __str__(self): identity = self.mobile or self.email or self.username @@ -124,21 +116,15 @@ class User(AbstractUser, BaseModel): is_active=True, ).exists() - def regenerate_verification_token(self): - self.email_verification_token = uuid.uuid4() - self.save(update_fields=['email_verification_token']) - - def set_password_reset_token(self): - """Generates a new password reset token and sets its expiry.""" - self.password_reset_token = uuid.uuid4() - self.password_reset_token_expires_at = timezone.now() + timedelta(hours=1) - self.save(update_fields=['password_reset_token', 'password_reset_token_expires_at']) - def save(self, *args, **kwargs): previous_image_name = get_image_previous_name(self, "profile_picture") current_image_name = self.profile_picture.name if self.profile_picture else None self.email = normalize_email_identity(self.email) self.mobile = normalize_mobile_number(self.mobile) + if kwargs.get('update_fields') is None or 'email' in kwargs['update_fields']: + self.is_email_verified = bool(self.email) + if kwargs.get('update_fields') is not None: + kwargs['update_fields'] = set(kwargs['update_fields']) | {'is_email_verified'} super().save(*args, **kwargs) diff --git a/apps/users/tasks.py b/apps/users/tasks.py index 8a8e5d8..ec304c2 100644 --- a/apps/users/tasks.py +++ b/apps/users/tasks.py @@ -70,21 +70,3 @@ def send_critical_sms(self, mobile: str, kind: str, code_or_message: str): except Exception as exc: logger.error("Failed to send SMS to %s for kind=%s: %s", mobile, kind, exc) raise self.retry(exc=exc, countdown=60) - - -@shared_task(bind=True, max_retries=1) -def send_verification_email(self, user_id, verification_url): - logger.info("Legacy verification email task skipped for user=%s url=%s", user_id, verification_url) - return {"skipped": True} - - -@shared_task(bind=True, max_retries=1) -def send_password_reset_email(self, user_id, reset_url): - logger.info("Legacy password reset email task skipped for user=%s url=%s", user_id, reset_url) - return {"skipped": True} - - -@shared_task(bind=True, max_retries=1) -def send_email_verified_success(self, user_id: int): - logger.info("Legacy verification success email task skipped for user=%s", user_id) - return {"skipped": True} diff --git a/apps/users/tests/integration/test_users.py b/apps/users/tests/integration/test_users.py index 797aca4..61d6af3 100644 --- a/apps/users/tests/integration/test_users.py +++ b/apps/users/tests/integration/test_users.py @@ -1,724 +1,85 @@ import json -import shutil -import tempfile -import uuid -from datetime import timedelta from unittest import mock -from django.conf import settings +from django.test import TestCase from django.core.files.uploadedfile import SimpleUploadedFile -from django.test import TestCase, override_settings -from django.utils import timezone -import jwt - -from apps.users.models import User, Major, University +from apps.users.models import User +from core.authentication import create_jwt_token class UsersAPIIntegrationTests(TestCase): - password = "Sup3rSecure!123" - - @classmethod - def setUpTestData(cls): - super().setUpTestData() - cls.major_cs, _ = Major.objects.get_or_create( - code="CS", defaults={"name": "Computer Science"} - ) - cls.major_gil, _ = Major.objects.get_or_create( - code="GIL_CS", defaults={"name": "Gilan Computer Science"} - ) - cls.university_ut, _ = University.objects.get_or_create( - code="UT", defaults={"name": "University of Tehran"} - ) - cls.university_gilan, _ = University.objects.get_or_create( - code="GILAN", defaults={"name": "Gilan University"} - ) - def setUp(self): - super().setUp() - patchers = [ - mock.patch("apps.users.tasks.send_verification_email.delay"), - mock.patch("apps.users.signals.send_verification_email.delay"), - mock.patch("apps.users.tasks.send_password_reset_email.delay"), - ] - ( - self.mock_send_verification_task, - self.mock_signal_verification_task, - self.mock_password_reset_task, - ) = [patcher.start() for patcher in patchers] - for patcher in patchers: - self.addCleanup(patcher.stop) - - # Helper utilities ----------------------------------------------------- - - def _numeric_student_id(self) -> str: - return str(uuid.uuid4().int)[-10:] - - def _resolve_major(self, value): - if value is None: - return None - if isinstance(value, Major): - return value - return Major.objects.filter(code=value).first() - - def _resolve_university(self, value): - if value is None: - return None - if isinstance(value, University): - return value - return University.objects.filter(code=value).first() - - def _create_user(self, **overrides) -> User: - unique = uuid.uuid4().hex[:8] - defaults = { - "username": f"user_{unique}", - "email": f"{unique}@example.com", - "student_id": self._numeric_student_id(), - "first_name": "Test", - "last_name": "User", - "year_of_study": 2, - "major": self.major_cs, - "university": self.university_ut, - } - defaults.update(overrides) - if isinstance(defaults.get("major"), str): - defaults["major"] = self._resolve_major(defaults["major"]) - if isinstance(defaults.get("university"), str): - defaults["university"] = self._resolve_university(defaults["university"]) - password = defaults.pop("password", self.password) - return User.objects.create_user(password=password, **defaults) - - def _auth_headers(self, token: str) -> dict: - return {"HTTP_AUTHORIZATION": f"Bearer {token}"} - - def _login_and_get_tokens(self, user: User, password: str | None = None) -> dict: - response = self.client.post( - "/api/auth/login", - data=json.dumps({"email": user.email, "password": password or self.password}), - content_type="application/json", + self.user = User.objects.create_user( + username="mobile_user", + mobile="09192000001", + email="person@example.com", + password="Sup3rSecure!123", + is_mobile_verified=True, ) + + def _auth(self): + return {"HTTP_AUTHORIZATION": f"Bearer {create_jwt_token(self.user)}"} + + def test_password_login_accepts_mobile_identity(self): + response = self.client.post("/api/auth/login", json.dumps({ + "identifier": self.user.mobile, + "password": "Sup3rSecure!123", + }), content_type="application/json") self.assertEqual(response.status_code, 200) - return response.json() + self.assertIn("access_token", response.json()) - def _refresh_token_value(self, user: User | None = None, **overrides) -> str: - now = timezone.now() - payload = { - "type": "refresh", - "exp": now + timedelta(minutes=5), - "iat": now, - } - if user is not None: - payload["user_id"] = user.id - payload.update(overrides) - return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM) + def test_profile_update_treats_supplied_email_as_identity(self): + response = self.client.put("/api/auth/profile", json.dumps({"email": "new@example.com"}), + content_type="application/json", **self._auth()) + self.assertEqual(response.status_code, 200) + self.user.refresh_from_db() + self.assertEqual(self.user.email, "new@example.com") + self.assertTrue(self.user.is_email_verified) - # Registration --------------------------------------------------------- + def test_profile_update_clears_email_identity_flag(self): + response = self.client.put("/api/auth/profile", json.dumps({"email": ""}), + content_type="application/json", **self._auth()) + self.assertEqual(response.status_code, 200) + self.user.refresh_from_db() + self.assertIsNone(self.user.email) + self.assertFalse(self.user.is_email_verified) - def test_register_creates_user_and_enqueues_signal(self): - # Arrange - payload = { - "username": "integration_user", - "email": "integration@example.com", - "password": "RegisterPass!9", - "student_id": "2023123456", - "first_name": "Integration", - "last_name": "Tester", - "university": self.university_ut.code, - "major": self.major_cs.code, - "year_of_study": 3, - } + def test_profile_requires_authentication_and_returns_account_identity(self): + self.assertEqual(self.client.get("/api/auth/profile").status_code, 401) + response = self.client.get("/api/auth/profile", **self._auth()) + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["mobile"], self.user.mobile) - # Act - response = self.client.post( - "/api/auth/register", data=json.dumps(payload), content_type="application/json" - ) + def test_profile_rejects_duplicate_email(self): + User.objects.create_user(username="second", mobile="09192000002", email="taken@example.com", password="password") + response = self.client.put("/api/auth/profile", json.dumps({"email": "taken@example.com"}), + content_type="application/json", **self._auth()) + self.assertEqual(response.status_code, 400) + self.user.refresh_from_db() + self.assertEqual(self.user.email, "person@example.com") - # Assert - self.assertEqual(response.status_code, 201) - self.assertTrue(User.objects.filter(email=payload["email"]).exists()) - self.assertTrue(self.mock_signal_verification_task.called) - - def test_register_rejects_short_student_id(self): - # Arrange - payload = { - "username": "short_id", - "email": "short@example.com", - "password": "RegisterPass!9", - "student_id": "123456789", # 9 digits - } - - # Act - response = self.client.post( - "/api/auth/register", data=json.dumps(payload), content_type="application/json" - ) - - # Assert + def test_profile_picture_rejects_non_image_upload(self): + image = SimpleUploadedFile("notes.txt", b"not an image", content_type="text/plain") + response = self.client.post("/api/auth/profile/picture", {"file": image}, **self._auth()) self.assertEqual(response.status_code, 400) - def test_register_rejects_duplicate_username(self): - # Arrange - existing = self._create_user(username="duplicate") - payload = { - "username": existing.username, - "email": "someone@example.com", - "password": "RegisterPass!9", - } - - # Act - response = self.client.post( - "/api/auth/register", data=json.dumps(payload), content_type="application/json" - ) - - # Assert - self.assertEqual(response.status_code, 400) - - def test_register_rejects_duplicate_email(self): - # Arrange - existing = self._create_user(email="duplicate@example.com") - payload = { - "username": "newuser", - "email": existing.email, - "password": "RegisterPass!9", - } - - # Act - response = self.client.post( - "/api/auth/register", data=json.dumps(payload), content_type="application/json" - ) - - # Assert - self.assertEqual(response.status_code, 400) - - def test_register_rejects_duplicate_student_id_in_same_university(self): - # Arrange - student_id = "2023012345" - self._create_user(student_id=student_id, university=self.university_gilan) - payload = { - "username": "dupstudent", - "email": "dupstudent@example.com", - "password": "RegisterPass!9", - "student_id": student_id, - "university": self.university_gilan.code, - } - - # Act - response = self.client.post( - "/api/auth/register", data=json.dumps(payload), content_type="application/json" - ) - - # Assert - self.assertEqual(response.status_code, 400) - - # Login & Refresh ------------------------------------------------------ - - def test_login_returns_tokens_for_verified_user(self): - # Arrange - user = self._create_user() - user.is_email_verified = True - user.save(update_fields=["is_email_verified"]) - - # Act - response = self.client.post( - "/api/auth/login", - data=json.dumps({"email": user.email, "password": self.password}), - content_type="application/json", - ) - - # Assert - self.assertEqual(response.status_code, 200) - body = response.json() - self.assertIn("access_token", body) - self.assertIn("refresh_token", body) - - def test_login_rejects_unverified_user(self): - # Arrange - user = self._create_user() - - # Act - response = self.client.post( - "/api/auth/login", - data=json.dumps({"email": user.email, "password": self.password}), - content_type="application/json", - ) - - # Assert - self.assertEqual(response.status_code, 401) - - def test_login_rejects_inactive_user(self): - # Arrange - user = self._create_user(is_email_verified=True, is_active=False) - - # Act - response = self.client.post( - "/api/auth/login", - data=json.dumps({"email": user.email, "password": self.password}), - content_type="application/json", - ) - - # Assert - self.assertEqual(response.status_code, 401) - - def test_refresh_returns_tokens(self): - # Arrange - user = self._create_user(is_email_verified=True) - tokens = self._login_and_get_tokens(user) - - # Act - response = self.client.post( - "/api/auth/refresh", - data=json.dumps({"refresh_token": tokens["refresh_token"]}), - content_type="application/json", - ) - - # Assert - self.assertEqual(response.status_code, 200) - refreshed = response.json() - self.assertIn("access_token", refreshed) - self.assertIn("refresh_token", refreshed) - - def test_refresh_rejects_non_refresh_token(self): - # Arrange - user = self._create_user(is_email_verified=True) - tokens = self._login_and_get_tokens(user) - - # Act - response = self.client.post( - "/api/auth/refresh", - data=json.dumps({"refresh_token": tokens["access_token"]}), - content_type="application/json", - ) - - # Assert - self.assertEqual(response.status_code, 401) - - def test_refresh_rejects_missing_user_id(self): - # Arrange - token = self._refresh_token_value() - - # Act - response = self.client.post( - "/api/auth/refresh", - data=json.dumps({"refresh_token": token}), - content_type="application/json", - ) - - # Assert - self.assertEqual(response.status_code, 401) - - def test_refresh_rejects_unverified_user(self): - # Arrange - user = self._create_user() - token = self._refresh_token_value(user=user) - - # Act - response = self.client.post( - "/api/auth/refresh", - data=json.dumps({"refresh_token": token}), - content_type="application/json", - ) - - # Assert - self.assertEqual(response.status_code, 401) - - def test_refresh_rejects_inactive_user(self): - # Arrange - user = self._create_user(is_email_verified=True, is_active=False) - token = self._refresh_token_value(user=user) - - # Act - response = self.client.post( - "/api/auth/refresh", - data=json.dumps({"refresh_token": token}), - content_type="application/json", - ) - - # Assert - self.assertEqual(response.status_code, 401) - - def test_refresh_rejects_expired_token(self): - # Arrange - user = self._create_user(is_email_verified=True) - token = self._refresh_token_value( - user=user, - exp=timezone.now() - timedelta(minutes=1), - ) - - # Act - response = self.client.post( - "/api/auth/refresh", - data=json.dumps({"refresh_token": token}), - content_type="application/json", - ) - - # Assert - self.assertEqual(response.status_code, 401) - - def test_refresh_rejects_invalid_token_string(self): - # Arrange - token = "not-a-valid-token" - - # Act - response = self.client.post( - "/api/auth/refresh", - data=json.dumps({"refresh_token": token}), - content_type="application/json", - ) - - # Assert - self.assertEqual(response.status_code, 401) - - # Email verification --------------------------------------------------- - - def test_verify_email_marks_user_verified(self): - # Arrange - user = self._create_user() - token = str(user.email_verification_token) - - # Act - response = self.client.get(f"/api/auth/verify-email/{token}") - - # Assert - self.assertEqual(response.status_code, 200) - user.refresh_from_db() - self.assertTrue(user.is_email_verified) - - def test_verify_email_rejects_unknown_token(self): - # Arrange - token = uuid.uuid4() - - # Act - response = self.client.get(f"/api/auth/verify-email/{token}") - - # Assert - self.assertEqual(response.status_code, 404) - - def test_resend_verification_rejects_unknown_email(self): - # Arrange - payload = {"email": "missing@example.com"} - - # Act - response = self.client.post(f"/api/auth/resend-verification?email={payload['email']}") - - # Assert - self.assertEqual(response.status_code, 404) - - # Profiles ------------------------------------------------------------- - - def test_get_profile_returns_schema_fields(self): - # Arrange - user = self._create_user(major=self.major_cs, university=self.university_gilan) - user.is_email_verified = True - user.save(update_fields=["is_email_verified"]) - tokens = self._login_and_get_tokens(user) - - # Act - response = self.client.get("/api/auth/profile", **self._auth_headers(tokens["access_token"])) - - # Assert - self.assertEqual(response.status_code, 200) - profile = response.json() - self.assertEqual(profile["major"], user.get_major_display()) - self.assertEqual(profile["university"], user.get_university_display()) - - def test_get_profile_requires_authentication(self): - # Arrange - # No token supplied. - - # Act - response = self.client.get("/api/auth/profile") - - # Assert - self.assertEqual(response.status_code, 401) - - def test_update_profile_persists_changes(self): - # Arrange - user = self._create_user(is_email_verified=True) - tokens = self._login_and_get_tokens(user) - payload = {"bio": "Updated bio", "year_of_study": 4} - - # Act - response = self.client.put( - "/api/auth/profile", - data=json.dumps(payload), - content_type="application/json", - **self._auth_headers(tokens["access_token"]), - ) - - # Assert - self.assertEqual(response.status_code, 200) - user.refresh_from_db() - self.assertEqual(user.bio, payload["bio"]) - self.assertEqual(user.year_of_study, payload["year_of_study"]) - - @override_settings(MEDIA_URL="/media/", MEDIA_ROOT=tempfile.gettempdir()) - def test_upload_profile_picture_succeeds(self): - # Arrange - user = self._create_user(is_email_verified=True) - tokens = self._login_and_get_tokens(user) - image = SimpleUploadedFile( - "avatar.png", b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR", content_type="image/png" - ) - - # Act - response = self.client.post( - "/api/auth/profile/picture", {"file": image}, **self._auth_headers(tokens["access_token"]) - ) - - # Assert - self.assertEqual(response.status_code, 200) - profile = self.client.get( - "/api/auth/profile", **self._auth_headers(tokens["access_token"]) - ).json() - self.assertIn("profile_pictures", profile["profile_picture"]) - - def test_upload_profile_picture_requires_file(self): - # Arrange - user = self._create_user(is_email_verified=True) - tokens = self._login_and_get_tokens(user) - - # Act - response = self.client.post( - "/api/auth/profile/picture", **self._auth_headers(tokens["access_token"]) - ) - - # Assert - self.assertEqual(response.status_code, 400) - - def test_upload_profile_picture_rejects_invalid_type(self): - # Arrange - user = self._create_user(is_email_verified=True) - tokens = self._login_and_get_tokens(user) - text_file = SimpleUploadedFile("doc.txt", b"text", content_type="text/plain") - - # Act - response = self.client.post( - "/api/auth/profile/picture", - {"file": text_file}, - **self._auth_headers(tokens["access_token"]), - ) - - # Assert - self.assertEqual(response.status_code, 400) - - def test_upload_profile_picture_rejects_large_files(self): - # Arrange - user = self._create_user(is_email_verified=True) - tokens = self._login_and_get_tokens(user) - large_content = b"x" * (5 * 1024 * 1024 + 1) - large_file = SimpleUploadedFile("large.png", large_content, content_type="image/png") - - # Act - response = self.client.post( - "/api/auth/profile/picture", - {"file": large_file}, - **self._auth_headers(tokens["access_token"]), - ) - - # Assert - self.assertEqual(response.status_code, 400) - - def test_delete_profile_picture_removes_file(self): - # Arrange - temp_media = tempfile.mkdtemp() - self.addCleanup(lambda: shutil.rmtree(temp_media, ignore_errors=True)) - user = self._create_user(is_email_verified=True) - tokens = self._login_and_get_tokens(user) - with override_settings(MEDIA_ROOT=temp_media, MEDIA_URL="/media/"): - image = SimpleUploadedFile("avatar.png", b"data", content_type="image/png") - self.client.post( - "/api/auth/profile/picture", - {"file": image}, - **self._auth_headers(tokens["access_token"]), - ) - - # Act - response = self.client.delete( - "/api/auth/profile/picture", **self._auth_headers(tokens["access_token"]) - ) - - # Assert - self.assertEqual(response.status_code, 200) - user.refresh_from_db() - self.assertFalse(bool(user.profile_picture)) - - # Password reset ------------------------------------------------------ - - def test_request_password_reset_enqueues_email(self): - # Arrange - user = self._create_user() - - # Act - response = self.client.post( - "/api/auth/request-password-reset", - data=json.dumps({"email": user.email}), - content_type="application/json", - ) - - # Assert - self.assertEqual(response.status_code, 200) - user.refresh_from_db() - self.assertIsNotNone(user.password_reset_token) - self.mock_password_reset_task.assert_called_once() - - def test_request_password_reset_unknown_email_returns_error(self): - # Arrange - payload = {"email": "missing@example.com"} - - # Act - response = self.client.post( - "/api/auth/request-password-reset", - data=json.dumps(payload), - content_type="application/json", - ) - - # Assert - self.assertEqual(response.status_code, 400) - - def test_reset_password_confirm_updates_credentials(self): - # Arrange - user = self._create_user() - user.set_password_reset_token() - payload = {"token": str(user.password_reset_token), "new_password": "BrandNewPass!9"} - - # Act - response = self.client.post( - "/api/auth/reset-password-confirm", - data=json.dumps(payload), - content_type="application/json", - ) - - # Assert - self.assertEqual(response.status_code, 200) - user.refresh_from_db() - self.assertIsNone(user.password_reset_token) - self.assertTrue(user.check_password(payload["new_password"])) - - def test_reset_password_confirm_rejects_expired_token(self): - # Arrange - user = self._create_user() - user.set_password_reset_token() - user.password_reset_token_expires_at = timezone.now() - timedelta(minutes=1) - user.save(update_fields=["password_reset_token_expires_at"]) - payload = {"token": str(user.password_reset_token), "new_password": "New!!!Pass"} - - # Act - response = self.client.post( - "/api/auth/reset-password-confirm", - data=json.dumps(payload), - content_type="application/json", - ) - - # Assert - self.assertEqual(response.status_code, 400) - - def test_reset_password_confirm_rejects_unknown_token(self): - # Arrange - payload = {"token": str(uuid.uuid4()), "new_password": "AnotherPass!9"} - - # Act - response = self.client.post( - "/api/auth/reset-password-confirm", - data=json.dumps(payload), - content_type="application/json", - ) - - # Assert - self.assertEqual(response.status_code, 400) - - # Admin utilities ----------------------------------------------------- - - def test_list_deleted_users_requires_privileged_user(self): - # Arrange - user = self._create_user(is_email_verified=True) - tokens = self._login_and_get_tokens(user) - - # Act - response = self.client.get( - "/api/auth/users/deleted", **self._auth_headers(tokens["access_token"]) - ) - - # Assert - self.assertEqual(response.status_code, 403) - - def test_list_deleted_users_returns_payload_for_staff(self): - # Arrange - deleted = self._create_user(is_deleted=True, deleted_at=timezone.now()) - staff = self._create_user(is_email_verified=True, is_staff=True) - tokens = self._login_and_get_tokens(staff) - - # Act - response = self.client.get( - "/api/auth/users/deleted", **self._auth_headers(tokens["access_token"]) - ) - - # Assert - self.assertEqual(response.status_code, 200) - payload = response.json() - self.assertTrue(any(item["id"] == deleted.id for item in payload)) - - def test_restore_user_requires_privileged_user(self): - # Arrange - target = self._create_user(is_deleted=True, deleted_at=timezone.now()) - user = self._create_user(is_email_verified=True) - tokens = self._login_and_get_tokens(user) - - # Act - response = self.client.post( - f"/api/auth/users/{target.id}/restore", **self._auth_headers(tokens["access_token"]) - ) - - # Assert - self.assertEqual(response.status_code, 403) - - def test_restore_user_restores_record_for_staff(self): - # Arrange - target = self._create_user(is_deleted=True, deleted_at=timezone.now()) - staff = self._create_user(is_email_verified=True, is_staff=True) - tokens = self._login_and_get_tokens(staff) - - # Act - response = self.client.post( - f"/api/auth/users/{target.id}/restore", **self._auth_headers(tokens["access_token"]) - ) - - # Assert - self.assertEqual(response.status_code, 200) - target.refresh_from_db() - self.assertFalse(target.is_deleted) - - def test_restore_user_missing_returns_error(self): - # Arrange - staff = self._create_user(is_email_verified=True, is_staff=True) - tokens = self._login_and_get_tokens(staff) - - # Act - response = self.client.post( - "/api/auth/users/999/restore", **self._auth_headers(tokens["access_token"]) - ) - - # Assert - self.assertEqual(response.status_code, 400) - - # Username checks ------------------------------------------------------ - - def test_check_username_reports_existing(self): - # Arrange - user = self._create_user() - - # Act - response = self.client.get("/api/auth/check-username", {"username": user.username}) - - # Assert + def test_check_username_reports_existing_user(self): + response = self.client.get("/api/auth/check-username", {"username": self.user.username}) self.assertEqual(response.status_code, 200) self.assertTrue(response.json()["exists"]) - def test_check_username_reports_availability(self): - # Arrange - username = "available_user" + def test_legacy_email_auth_routes_are_gone(self): + self.assertEqual(self.client.get("/api/auth/verify-email/token").status_code, 404) + self.assertEqual(self.client.post("/api/auth/resend-verification").status_code, 404) + self.assertEqual(self.client.post("/api/auth/request-password-reset").status_code, 404) - # Act - response = self.client.get("/api/auth/check-username", {"username": username}) - - # Assert + @mock.patch("apps.users.api.views.reset_password_with_otp") + def test_mobile_otp_password_reset_route_remains_available(self, reset): + response = self.client.post("/api/auth/reset-password", json.dumps({ + "mobile": self.user.mobile, + "code": "123456", + "new_password": "AnotherSecurePass!123", + }), content_type="application/json") self.assertEqual(response.status_code, 200) - self.assertFalse(response.json()["exists"]) + reset.assert_called_once() diff --git a/apps/users/tests/unit/test_users.py b/apps/users/tests/unit/test_users.py index 603a414..4d0d848 100644 --- a/apps/users/tests/unit/test_users.py +++ b/apps/users/tests/unit/test_users.py @@ -1,400 +1,30 @@ -import uuid -from datetime import timedelta -from unittest import mock +from django.test import SimpleTestCase, TestCase -from django.db.models.signals import post_save -from django.test import SimpleTestCase, TestCase, override_settings -from django.utils import timezone - -from import_export.widgets import BooleanWidget - -from apps.users.models import User, Major, University +from apps.users.models import User from apps.users.resources import UserResource -from apps.users.signals import send_verification_email_on_registration -from apps.users.tasks import ( - send_email_verified_success, - send_password_reset_email, - send_verification_email, -) -class UserFactoryMixin: - def _ensure_reference_objects(self): - if not hasattr(self, "_default_major"): - self._default_major, _ = Major.objects.get_or_create( - code="CS", - defaults={"name": "Computer Science"}, - ) - self._default_university, _ = University.objects.get_or_create( - code="UT", - defaults={"name": "University of Tehran"}, - ) - - def _resolve_major(self, value): - if value is None: - return None - if isinstance(value, Major): - return value - obj, _ = Major.objects.get_or_create(code=value, defaults={"name": value}) - return obj - - def _resolve_university(self, value): - if value is None: - return None - if isinstance(value, University): - return value - obj, _ = University.objects.get_or_create(code=value, defaults={"name": value}) - return obj - - def create_user(self, **extra_fields): - self._ensure_reference_objects() - unique = uuid.uuid4().hex - data = { - "email": f"user_{unique}@example.com", - "username": f"user_{unique[:10]}", - "first_name": "Test", - "last_name": "User", - } - password = extra_fields.pop("password", "StrongPass!123") - major = extra_fields.pop("major", self._default_major) - university = extra_fields.pop("university", self._default_university) - if isinstance(major, str): - major = self._resolve_major(major) - if isinstance(university, str): - university = self._resolve_university(university) - data.update(extra_fields) - data.setdefault("major", major) - data.setdefault("university", university) - return User.objects.create_user(password=password, **data) - - -class UserModelTests(UserFactoryMixin, TestCase): - def setUp(self): - super().setUp() - patcher = mock.patch("apps.users.signals.send_verification_email.delay") - patcher.start() - self.addCleanup(patcher.stop) - - def test_str_returns_full_name_with_email(self): - # Arrange - user = self.create_user(first_name="Ada", last_name="Lovelace") - - # Act - result = str(user) - - # Assert - expected = f"{user.get_full_name()} ({user.email})" - self.assertEqual(result, expected) - - def test_get_full_name_handles_missing_names(self): - # Arrange - user = self.create_user(first_name="Grace", last_name="") - - # Act - result = user.get_full_name() - - # Assert - self.assertEqual(result, "Grace") - - def test_regenerate_verification_token_generates_new_value(self): - # Arrange - user = self.create_user() - original_token = user.email_verification_token - - # Act - user.regenerate_verification_token() - - # Assert - self.assertNotEqual(user.email_verification_token, original_token) - - def test_set_password_reset_token_assigns_future_expiry(self): - # Arrange - user = self.create_user() - frozen = timezone.now() - - # Act - with mock.patch("apps.users.models.timezone.now", return_value=frozen): - user.set_password_reset_token() - - # Assert - self.assertIsNotNone(user.password_reset_token) - self.assertEqual( - user.password_reset_token_expires_at, - frozen + timedelta(hours=1), - ) - - def test_save_triggers_verified_task_on_state_change(self): - # Arrange - user = self.create_user() - - # Act - with mock.patch("apps.users.tasks.send_email_verified_success.delay") as mock_delay: - user.is_email_verified = True - user.save() - - # Assert - mock_delay.assert_called_once_with(user.id) - - def test_save_skips_task_when_already_verified(self): - # Arrange - user = self.create_user(is_email_verified=True) - - # Act - with mock.patch("apps.users.tasks.send_email_verified_success.delay") as mock_delay: - user.bio = "Updated bio" - user.save() - - # Assert - mock_delay.assert_not_called() - - -class UserSignalTests(TestCase): - def setUp(self): - super().setUp() - post_save.disconnect(send_verification_email_on_registration, sender=User) - self.addCleanup( - post_save.connect, - send_verification_email_on_registration, - User, - False, - ) - - @override_settings(FRONTEND_ROOT="https://frontend.example/") - @mock.patch("apps.users.signals.send_verification_email.delay") - @mock.patch("apps.users.signals.uuid.uuid4") - def test_signal_sets_username_timestamp_and_dispatches_email( - self, - mock_uuid, - mock_delay, - ): - # Arrange - fake_uuid = uuid.UUID("12345678-1234-5678-1234-567812345678") - mock_uuid.return_value = fake_uuid - fake_now = timezone.now() - user = User.objects.create( - email="new.user@example.com", - username="", - password="pass", - is_email_verified=False, - ) - - # Act - with mock.patch("apps.users.signals.timezone.now", return_value=fake_now): - send_verification_email_on_registration(User, user, created=True) - - # Assert +class UserIdentityTests(TestCase): + def test_email_is_optional_and_flags_follow_identity(self): + user = User.objects.create_user(username="mobile_only", mobile="09191110000", password="password") + self.assertIsNone(user.email) + self.assertFalse(user.is_email_verified) + user.email = " PERSON@EXAMPLE.COM " + user.save(update_fields=["email"]) user.refresh_from_db() - self.assertEqual(user.username, str(fake_uuid)[:10]) - self.assertEqual(user.email_verification_sent_at, fake_now) - expected_url = ( - f"https://frontend.example/verify-email/{user.email_verification_token}" - ) - mock_delay.assert_called_once_with(user.id, expected_url) - - @override_settings(FRONTEND_ROOT="https://frontend.example/") - @mock.patch("apps.users.signals.send_verification_email.delay") - def test_signal_preserves_existing_username(self, mock_delay): - # Arrange - fake_now = timezone.now() - user = User.objects.create( - email="existing@example.com", - username="existing_name", - password="pass", - is_email_verified=False, - ) - - # Act - with mock.patch("apps.users.signals.timezone.now", return_value=fake_now): - send_verification_email_on_registration(User, user, created=True) - - # Assert + self.assertEqual(user.email, "person@example.com") + self.assertTrue(user.is_email_verified) + user.email = None + user.save(update_fields=["email"]) user.refresh_from_db() - self.assertEqual(user.username, "existing_name") - self.assertEqual(user.email_verification_sent_at, fake_now) - mock_delay.assert_called_once() + self.assertFalse(user.is_email_verified) - @mock.patch("apps.users.signals.send_verification_email.delay") - def test_signal_skips_when_user_already_verified(self, mock_delay): - # Arrange - user = User.objects.create( - email="verified@example.com", - username="verified_user", - password="pass", - is_email_verified=True, - ) - - # Act - send_verification_email_on_registration(User, user, created=True) - - # Assert - self.assertIsNone(user.email_verification_sent_at) - mock_delay.assert_not_called() - - @mock.patch("apps.users.signals.send_verification_email.delay") - def test_signal_skips_when_email_missing(self, mock_delay): - # Arrange - user = User.objects.create( - email="", - username="no_email", - password="pass", - is_email_verified=False, - ) - - # Act - send_verification_email_on_registration(User, user, created=True) - - # Assert - self.assertIsNone(user.email_verification_sent_at) - mock_delay.assert_not_called() - - @mock.patch("apps.users.signals.send_verification_email.delay") - def test_signal_ignores_updates_to_existing_users(self, mock_delay): - # Arrange - user = User.objects.create( - email="existing-update@example.com", - username="existing_update", - password="pass", - is_email_verified=False, - ) - - # Act - send_verification_email_on_registration(User, user, created=False) - - # Assert - self.assertIsNone(user.email_verification_sent_at) - mock_delay.assert_not_called() - - -class UserTaskTests(UserFactoryMixin, TestCase): - def setUp(self): - super().setUp() - patcher = mock.patch("apps.users.signals.send_verification_email.delay") - patcher.start() - self.addCleanup(patcher.stop) - - @override_settings(DEFAULT_FROM_EMAIL="no-reply@example.com") - @mock.patch("apps.users.tasks.send_mail") - @mock.patch("apps.users.tasks.render_to_string", return_value="Hi
") - def test_send_verification_email_task_sends_expected_payload( - self, - mock_render, - mock_send_mail, - ): - # Arrange - user = self.create_user() - verification_url = "https://example.com/verify" - - # Act - result = send_verification_email.run(user.id, verification_url) - - # Assert - self.assertEqual(result, f"Verification email sent to {user.email}") - mock_render.assert_called_once_with( - "emails/verification_email.html", - {"user": user, "verification_url": verification_url}, - ) - kwargs = mock_send_mail.call_args.kwargs - self.assertEqual(kwargs["recipient_list"], [user.email]) - self.assertEqual(kwargs["from_email"], "no-reply@example.com") - self.assertEqual(kwargs["message"], "Hi") - - @override_settings(DEFAULT_FROM_EMAIL="support@example.com") - @mock.patch("apps.users.tasks.send_mail") - @mock.patch("apps.users.tasks.render_to_string", return_value="Reset
") - def test_send_password_reset_email_task_uses_reset_template( - self, - mock_render, - mock_send_mail, - ): - # Arrange - user = self.create_user() - reset_url = "https://example.com/reset" - - # Act - result = send_password_reset_email.run(user.id, reset_url) - - # Assert - self.assertEqual(result, f"Password reset email sent to {user.email}") - mock_render.assert_called_once_with( - "emails/password_reset_email.html", - {"user": user, "reset_url": reset_url}, - ) - kwargs = mock_send_mail.call_args.kwargs - self.assertEqual(kwargs["recipient_list"], [user.email]) - self.assertEqual(kwargs["from_email"], "support@example.com") - self.assertEqual(kwargs["message"], "Reset") - - @override_settings( - DEFAULT_FROM_EMAIL="success@example.com", - FRONTEND_ROOT="https://frontend.example/", - ) - @mock.patch("apps.users.tasks.send_mail") - @mock.patch("apps.users.tasks.render_to_string", return_value="Success
") - def test_send_email_verified_success_task_renders_success_template( - self, - mock_render, - mock_send_mail, - ): - # Arrange - user = self.create_user() - - # Act - result = send_email_verified_success.run(user.id) - - # Assert - self.assertEqual(result, f"verified success email sent to {user.email}") - mock_render.assert_called_once_with( - "emails/verification_success.html", - {"user": user, "home_url": "https://frontend.example/"}, - ) - kwargs = mock_send_mail.call_args.kwargs - self.assertEqual(kwargs["recipient_list"], [user.email]) - self.assertEqual(kwargs["from_email"], "success@example.com") - self.assertEqual(kwargs["message"], "Success") - - def test_send_verification_email_task_retries_on_lookup_error(self): - # Arrange - retry_patch = mock.patch.object( - send_verification_email, - "retry", - side_effect=RuntimeError("retry"), - ) - - # Act / Assert - with mock.patch( - "apps.users.tasks.User.objects.get", - side_effect=ValueError("missing"), - ), retry_patch as mock_retry: - with self.assertRaises(RuntimeError): - send_verification_email.run(999, "https://example.com/verify") - - self.assertEqual(mock_retry.call_args.kwargs.get("countdown"), 60) - self.assertIsInstance(mock_retry.call_args.kwargs.get("exc"), ValueError) + def test_display_name_prefers_mobile_when_email_absent(self): + user = User.objects.create_user(username="mobile_only", mobile="09191110001", password="password") + self.assertIn(user.mobile, str(user)) class UserResourceTests(SimpleTestCase): - def test_boolean_fields_use_boolean_widget(self): - # Arrange + def test_email_identity_flag_remains_exportable(self): resource = UserResource() - - # Act - widgets = [ - resource.fields["is_staff"].widget, - resource.fields["is_superuser"].widget, - resource.fields["is_email_verified"].widget, - ] - - # Assert - for widget in widgets: - self.assertIsInstance(widget, BooleanWidget) - - def test_field_order_matches_meta_definition(self): - # Arrange - resource = UserResource() - - # Act - field_names = tuple(resource.fields.keys()) - - # Assert - self.assertEqual(resource._meta.export_order, resource._meta.fields) - self.assertSetEqual(set(field_names), set(resource._meta.fields)) + self.assertIn("is_email_verified", resource.fields) diff --git a/config/services/celery.py b/config/services/celery.py index 79df0ac..1067abb 100644 --- a/config/services/celery.py +++ b/config/services/celery.py @@ -38,14 +38,7 @@ app.conf.beat_schedule = { 'schedule': crontab(hour=2, minute=0), 'description': 'Runs daily at 02:00 UTC to cleanup notification retention.', }, - 'process-scheduled-announcements': { - 'task': 'apps.communications.tasks.process_scheduled_announcements', - 'schedule': crontab(minute='*/15'), - 'description': 'Runs every 15 minutes to dispatch scheduled announcements.', - }, } -EMAIL_TIMEOUT_SECONDS = 10 - CELERY_TASK_SOFT_TIME_LIMIT = 20 CELERY_TASK_TIME_LIMIT = 30 diff --git a/config/settings/base.py b/config/settings/base.py index 3693655..70fd635 100644 --- a/config/settings/base.py +++ b/config/settings/base.py @@ -148,16 +148,6 @@ CSRF_TRUSTED_ORIGINS = config( CSRF_COOKIE_SECURE = config('CSRF_COOKIE_SECURE', default=True, cast=bool) SESSION_COOKIE_SECURE = config('SESSION_COOKIE_SECURE', default=True, cast=bool) -# Email Configuration -EMAIL_BACKEND = config('EMAIL_BACKEND', default='django.core.mail.backends.console.EmailBackend') -EMAIL_HOST = config('EMAIL_HOST', default='') -EMAIL_PORT = config('EMAIL_PORT', default=587, cast=int) -EMAIL_USE_SSL = config('EMAIL_USE_SSL', default=False, cast=bool) -EMAIL_USE_TLS = config('EMAIL_USE_TLS', default=True, cast=bool) -EMAIL_HOST_USER = config('EMAIL_HOST_USER', default='') -EMAIL_HOST_PASSWORD = config('EMAIL_HOST_PASSWORD', default='') -DEFAULT_FROM_EMAIL = config('DEFAULT_FROM_EMAIL', default='webmaster@localhost') - SMS_APIKEY = config('SMS_APIKEY', default='') SMS_AUTH_OTP_TEMPLATE_ID = config('SMS_AUTH_OTP_TEMPLATE_ID', default='') SMS_EVENT_CANCELLATION_TEMPLATE_ID = config('SMS_EVENT_CANCELLATION_TEMPLATE_ID', default='') @@ -236,7 +226,6 @@ os.makedirs(BASE_DIR / 'logs', exist_ok=True) BACKEND_ROOT = config('DJANGO_HOST', default='http://localhost:8000/') FRONTEND_ROOT = config('FRONTEND_ROOT', default='http://localhost:3000/') -FRONTEND_PASSWORD_RESET_PAGE = config('FRONTEND_PASSWORD_RESET_PAGE', default='http://localhost:3000/api/auth/reset-password-confirm/') FRONTEND_CALLBACK_URL = config('FRONTEND_CALLBACK_URL', default='http://localhost:3000/payments/result') GOOGLE_OAUTH_CLIENT_ID = config('GOOGLE_OAUTH_CLIENT_ID', default='') GOOGLE_OAUTH_CLIENT_SECRET = config('GOOGLE_OAUTH_CLIENT_SECRET', default='') diff --git a/config/settings/development.py b/config/settings/development.py index c06d3d5..70fd8e9 100644 --- a/config/settings/development.py +++ b/config/settings/development.py @@ -29,9 +29,6 @@ SECURE_CONTENT_TYPE_NOSNIFF = False SECURE_BROWSER_XSS_FILTER = False X_FRAME_OPTIONS = "SAMEORIGIN" -# Email backend for development -EMAIL_BACKEND = 'django.core.mail.backends.console.EmailBackend' - # Disable caching in development CACHES = { 'default': { diff --git a/config/settings/test.py b/config/settings/test.py index 11e528e..aa527fc 100644 --- a/config/settings/test.py +++ b/config/settings/test.py @@ -22,8 +22,6 @@ PASSWORD_HASHERS = [ "django.contrib.auth.hashers.MD5PasswordHasher", ] -EMAIL_BACKEND = "django.core.mail.backends.locmem.EmailBackend" - CACHES = { "default": { "BACKEND": "django.core.cache.backends.locmem.LocMemCache", diff --git a/templates/admin/users/send_notification.html b/templates/admin/users/send_notification.html new file mode 100644 index 0000000..a90743b --- /dev/null +++ b/templates/admin/users/send_notification.html @@ -0,0 +1,17 @@ +{% extends "admin/base_site.html" %} +{% block content %} +خبرنامه انجمن علوم کامپیوتر
-| - - |
-
|
-
لغو ثبتنام
-ثبتنام شما در {{ event.title }} لغو شد.
- - {% if event.start_time %} -زمان رویداد: {{ event.start_time|date:"Y-m-d H:i" }}
-اگر این تغییر را شما انجام ندادهاید، لطفاً با پشتیبانی تماس بگیرید.
-| - - | -
| - - |
انجمن علوم کامپیوتر
-سلام،
- -از اشتراک شما در خبرنامه انجمن علوم کامپیوتر متشکریم. برای تکمیل فرآیند اشتراک، لطفاً آدرس ایمیل خود را با کلیک روی دکمه زیر تأیید کنید:
- -اگر دکمه کار نمیکند، میتوانید این لینک را کپی کرده و در مرورگر خود باز کنید:
-{{ confirmation_url }}
- -پس از تأیید، شما دریافت خواهید کرد:
-توجه: این لینک تأیید به دلایل امنیتی ظرف ۲۴ ساعت منقضی خواهد شد.
- - -درخواست بازنشانی رمز عبور
-ما درخواستی برای بازنشانی رمز عبور حساب کاربری شما دریافت کردیم. اگر شما این درخواست را دادهاید، روی دکمه زیر کلیک کنید تا رمز عبور خود را بازنشانی کنید:
- - - -اگر دکمه کار نمیکند، میتوانید این لینک را کپی کرده و در مرورگر خود قرار دهید:
-{{ reset_url }}
- -این لینک بازنشانی رمز عبور به دلایل امنیتی پس از ۱ ساعت منقضی خواهد شد.
- -اگر شما درخواست بازنشانی رمز عبور ندادهاید، لطفاً این ایمیل را نادیده بگیرید. رمز عبور شما بدون تغییر باقی خواهد ماند.
- -با احترام،
انجمن علمی مهندسی کامپیوتر دانشکده فنی و مهندسی شرق گیلان
| - - |
| - - | -
به جمع ما خوش آمدید!
-- آدرس ایمیل شما با موفقیت تأیید شد و حساب کاربریتان فعال است. - از این پس میتوانید بدون محدودیت از امکانات سامانه استفاده کنید. -
- -- اگر شما این اقدام را انجام ندادهاید، لطفاً این ایمیل را نادیده بگیرید. -
- -با احترام
انجمن علمی مهندسی کامپیوتر دانشکده فنی و مهندسی شرق گیلان