166 lines
7.8 KiB
Python
166 lines
7.8 KiB
Python
"""Notice deletion regressions; SMS providers are always mocked."""
|
|
from urllib.parse import urlsplit
|
|
|
|
import pytest
|
|
from django.core.exceptions import PermissionDenied, ValidationError
|
|
from django.core.management import call_command
|
|
from django.test import Client
|
|
from django.urls import reverse
|
|
|
|
from apps.payments import policies, services
|
|
from apps.payments.access import issue_payment_link
|
|
from apps.payments.models import PaymentAccessConfiguration, PaymentAuditEvent, PaymentNotice, PaymentReceipt
|
|
from apps.notifications.models import NotificationJob
|
|
from tests.test_payments import case, data, submit, upload
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
@pytest.mark.parametrize('actor_key,role,superuser', [
|
|
('creator', None, False), ('outsider', 'CEO', False),
|
|
('outsider', 'GENERAL_ADMIN', False), ('outsider', None, True),
|
|
])
|
|
def test_only_creator_ceo_and_admin_can_delete(case, client, actor_key, role, superuser):
|
|
actor = case[actor_key]
|
|
if role:
|
|
actor.staff_profile.role = role
|
|
actor.staff_profile.save()
|
|
actor.is_superuser = superuser
|
|
actor.save()
|
|
client.force_login(actor)
|
|
url = reverse('staff_payments:notice_delete', args=[case['notice'].public_id])
|
|
assert client.post(url, {'confirm': 'delete'}).status_code == 302
|
|
case['notice'].refresh_from_db()
|
|
assert case['notice'].deleted_at is not None
|
|
assert PaymentAuditEvent.objects.filter(action='NOTICE_DELETED', actor=actor).count() == 1
|
|
assert not policies.visible(PaymentNotice.objects.all(), case['customer']).exists()
|
|
|
|
|
|
def test_other_staff_and_customer_cannot_delete(case, client):
|
|
for actor in [case['outsider'], case['customer']]:
|
|
client.force_login(actor)
|
|
url = reverse('staff_payments:notice_delete', args=[case['notice'].public_id])
|
|
assert client.post(url, {'confirm': 'delete'}).status_code == 403
|
|
with pytest.raises(PermissionDenied):
|
|
services.delete_notice(actor, case['notice'])
|
|
case['notice'].refresh_from_db()
|
|
assert case['notice'].deleted_at is None
|
|
|
|
|
|
@pytest.mark.parametrize('status', PaymentReceipt.Status.values)
|
|
@pytest.mark.django_db(transaction=True)
|
|
def test_any_linked_receipt_blocks_deletion_even_if_hidden(case, client, status):
|
|
receipt = submit(case)
|
|
PaymentReceipt.objects.filter(pk=receipt.pk).update(status=status)
|
|
configuration = PaymentAccessConfiguration.objects.first() or PaymentAccessConfiguration()
|
|
configuration.open_view = False
|
|
configuration.save()
|
|
assert not policies.visible(PaymentReceipt.objects.filter(pk=receipt.pk), case['creator']).exists()
|
|
assert not policies.can_delete_notice(case['creator'], case['notice'])
|
|
with pytest.raises(ValidationError, match='فیش ثبت شده'):
|
|
services.delete_notice(case['creator'], case['notice'])
|
|
client.force_login(case['creator'])
|
|
response = client.post(reverse('staff_payments:notice_delete', args=[case['notice'].public_id]),
|
|
{'confirm': 'delete'}, follow=True)
|
|
assert 'حذف آن امکانپذیر نیست' in response.content.decode()
|
|
assert not list(response.context['receipts'])
|
|
case['notice'].refresh_from_db()
|
|
assert case['notice'].deleted_at is None
|
|
assert PaymentReceipt.objects.filter(pk=receipt.pk).exists()
|
|
|
|
|
|
def test_confirmation_post_and_csrf_are_required(case, client):
|
|
client.force_login(case['creator'])
|
|
url = reverse('staff_payments:notice_delete', args=[case['notice'].public_id])
|
|
assert client.get(url).status_code == 405
|
|
assert client.post(url).status_code == 302
|
|
csrf_client = Client(enforce_csrf_checks=True)
|
|
csrf_client.force_login(case['creator'])
|
|
assert csrf_client.post(url, {'confirm': 'delete'}).status_code == 403
|
|
case['notice'].refresh_from_db()
|
|
assert case['notice'].deleted_at is None
|
|
|
|
|
|
@pytest.mark.django_db(transaction=True)
|
|
def test_deleted_notice_disappears_and_links_and_stale_forms_stop_working(case, client):
|
|
_, url = issue_payment_link(notice=case['notice'])
|
|
assert client.get(urlsplit(url).path, follow=True).status_code == 200
|
|
# Retain a valid form built before deletion to exercise the service race check.
|
|
from apps.payments.forms import ReceiptForm
|
|
form = ReceiptForm(data(case), {'receipt_image': upload()}, customer=case['customer'])
|
|
assert form.is_valid(), form.errors
|
|
services.delete_notice(case['creator'], case['notice'])
|
|
assert client.get(urlsplit(url).path).status_code == 410
|
|
assert client.get(reverse('payments:receipt_new')).status_code == 410
|
|
with pytest.raises(ValidationError, match='لغو شده'):
|
|
services.submit_receipt(case['customer'], case['customer'], form.cleaned_data)
|
|
client.force_login(case['customer'])
|
|
assert client.get(reverse('payments:notice_detail', args=[case['notice'].public_id])).status_code == 404
|
|
client.force_login(case['creator'])
|
|
response = client.get(reverse('staff_payments:desk'), {'tab': 'notices'})
|
|
assert case['notice'].number not in response.content.decode()
|
|
assert client.get(reverse('staff_payments:notice_detail', args=[case['notice'].public_id])).status_code == 404
|
|
|
|
|
|
def test_independent_receipt_does_not_block_notice_deletion(case):
|
|
submit(case, linked=False)
|
|
services.delete_notice(case['creator'], case['notice'])
|
|
case['notice'].refresh_from_db()
|
|
assert case['notice'].deleted_at
|
|
|
|
|
|
def test_deletion_cancels_queued_sms_and_worker_skips_stale_jobs(case, monkeypatch):
|
|
services.delete_notice(case['creator'], case['notice'])
|
|
job = NotificationJob.objects.get(event_type='PAYMENT_NOTICE_CREATED')
|
|
assert job.status == 'CANCELLED'
|
|
# Simulate a stale lease recovery or retry of a job for an already deleted notice.
|
|
NotificationJob.objects.filter(pk=job.pk).update(status='RETRYING')
|
|
from apps.notifications.management.commands import process_notifications
|
|
def forbidden_provider():
|
|
pytest.fail('Deleted notices must never initialize an SMS provider')
|
|
monkeypatch.setattr(process_notifications, 'get_sms_provider', forbidden_provider)
|
|
call_command('process_notifications', once=True, job_id=job.pk, channel='SMS')
|
|
job.refresh_from_db()
|
|
assert job.status == 'CANCELLED'
|
|
|
|
|
|
def test_role_revocation_is_rechecked_and_repeated_delete_is_idempotent(case):
|
|
actor = case['outsider']
|
|
actor.staff_profile.role = 'CEO'
|
|
actor.staff_profile.save()
|
|
assert policies.can_delete_notice(actor, case['notice'])
|
|
type(actor.staff_profile).objects.filter(pk=actor.staff_profile.pk).update(role='SALES_EXPERT')
|
|
with pytest.raises(PermissionDenied):
|
|
services.delete_notice(actor, case['notice'])
|
|
services.delete_notice(case['creator'], case['notice'])
|
|
services.delete_notice(case['creator'], case['notice'])
|
|
assert PaymentAuditEvent.objects.filter(action='NOTICE_DELETED').count() == 1
|
|
|
|
|
|
@pytest.mark.parametrize('actor_key,role', [('creator', None), ('outsider', 'CEO'), ('outsider', 'GENERAL_ADMIN')])
|
|
def test_delete_actors_can_reach_notice_when_general_view_is_closed(case, client, actor_key, role):
|
|
configuration = PaymentAccessConfiguration.objects.first() or PaymentAccessConfiguration()
|
|
configuration.open_view = False
|
|
configuration.save()
|
|
actor = case[actor_key]
|
|
if role:
|
|
actor.staff_profile.role = role
|
|
actor.staff_profile.save()
|
|
client.force_login(actor)
|
|
response = client.get(reverse('staff_payments:notice_detail', args=[case['notice'].public_id]))
|
|
assert response.status_code == 200
|
|
assert response.context['can_delete_notice']
|
|
|
|
|
|
@pytest.mark.parametrize('profile_inactive', [True, False])
|
|
def test_inactive_creator_cannot_delete(case, profile_inactive):
|
|
actor = case['creator']
|
|
if profile_inactive:
|
|
actor.staff_profile.is_active = False
|
|
actor.staff_profile.save()
|
|
else:
|
|
actor.is_active = False
|
|
actor.save()
|
|
with pytest.raises(PermissionDenied):
|
|
services.delete_notice(actor, case['notice'])
|