Amirhossein Khalili
ff99130938
Split backend tests into unit and integration suites, add isolated fixtures and coverage enforcement, and update CI and test documentation. Cover current authentication, content, notification, payment, media, and admin flows; fix behaviors revealed by the new tests. Validate both suites and the per-module gate against local PostgreSQL.
112 lines
6.9 KiB
Python
112 lines
6.9 KiB
Python
import pytest
|
|
import json
|
|
from types import SimpleNamespace
|
|
from django.contrib import admin
|
|
from django.test import RequestFactory
|
|
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
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.integration
|
|
class TestManualNotificationTests:
|
|
|
|
def test_form_requires_title_and_plain_digest_but_allows_optional_body(self):
|
|
assert ManualNotificationForm({'title': 'Title', 'digest': 'Intro'}).is_valid()
|
|
assert not ManualNotificationForm({'title': '', 'digest': 'Intro'}).is_valid()
|
|
|
|
def test_task_sends_only_to_selected_active_users(self, mocker):
|
|
notify_user = mocker.patch('apps.notifications.tasks.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')
|
|
assert sent == 1
|
|
notify_user.assert_called_once_with(active.pk, {'type': 'manual', 'title': 'Title', 'message': 'Digest', 'body_markdown': '# Body', 'level': 'info'})
|
|
|
|
def test_admin_action_queues_valid_composer_for_selected_users(self, mocker):
|
|
delay = mocker.patch('apps.users.admin.send_manual_notification.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)
|
|
mocker.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')
|
|
assert not user.is_email_verified
|
|
user.email = 'person@example.com'
|
|
user.save(update_fields=['email'])
|
|
user.refresh_from_db()
|
|
assert user.is_email_verified
|
|
|
|
def test_admin_changelist_uses_the_composer_before_queuing(self, mocker):
|
|
delay = mocker.patch('apps.users.admin.send_manual_notification.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)
|
|
assert preview.status_code == 200
|
|
assert 'توضیح کوتاه' in preview.content.decode()
|
|
assert not delay.called
|
|
result = self.client.post(path, {**action_data, 'confirm_send': '1', 'title': 'Title', 'digest': 'Digest', 'body_markdown': '# Body'})
|
|
assert result.status_code in (200, 302)
|
|
delay.assert_called_once_with([recipient.pk], 'Title', 'Digest', '# Body')
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _pytest_client(self, client):
|
|
self.client = client
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.integration
|
|
class TestNotificationAPIIntegrationTests:
|
|
|
|
def test_detail_requires_auth_and_returns_body_only_to_recipient(self, mocker):
|
|
get = mocker.patch('apps.notifications.api.views.RedisNotificationStore.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'
|
|
assert self.client.get(path).status_code == 401
|
|
owner_response = self.client.get(path, HTTP_AUTHORIZATION=f'Bearer {create_jwt_token(owner)}')
|
|
assert owner_response.status_code == 200
|
|
assert owner_response.json()['body_markdown'] == '# Private'
|
|
assert owner_response.json()['action_url'] == '/events/workshop'
|
|
other_response = self.client.get(path, HTTP_AUTHORIZATION=f'Bearer {create_jwt_token(other)}')
|
|
assert other_response.status_code == 404
|
|
|
|
def test_list_excludes_full_body_and_uses_read_filter(self, mocker):
|
|
list_notifications_store = mocker.patch('apps.notifications.api.views.RedisNotificationStore.list')
|
|
_count = mocker.patch('apps.notifications.api.views.RedisNotificationStore.unread_count', return_value=1)
|
|
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)}')
|
|
assert response.status_code == 200
|
|
item = response.json()['notifications'][0]
|
|
assert item['has_body']
|
|
assert 'body_markdown' not in item
|
|
assert list_notifications_store.call_args.kwargs['read_filter'] == 'unread'
|
|
|
|
def test_delete_only_removes_the_authenticated_users_notification(self, mocker):
|
|
delete = mocker.patch('apps.notifications.api.views.RedisNotificationStore.delete')
|
|
_count = mocker.patch('apps.notifications.api.views.RedisNotificationStore.unread_count', return_value=0)
|
|
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'
|
|
assert self.client.delete(path, HTTP_AUTHORIZATION=f'Bearer {create_jwt_token(other)}').status_code == 404
|
|
assert self.client.delete(path, HTTP_AUTHORIZATION=f'Bearer {create_jwt_token(owner)}').status_code == 200
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _pytest_client(self, client):
|
|
self.client = client
|