161 lines
9.8 KiB
Python
161 lines
9.8 KiB
Python
import pytest
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
from apps.messaging.models import Message
|
|
from apps.staff.models import StaffProfile
|
|
from apps.tickets.models import MessageShare, TicketAccessGrant
|
|
from apps.tickets.policies import TicketPolicy
|
|
from apps.tickets.services import change_priority, change_status
|
|
from apps.tickets.sharing import share_staff_message, share_staff_ticket, revoke_staff_share
|
|
from tests.test_staff_sharing import make_sharing_case
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
def test_shared_queue_groups_messages_and_searches_normal_ticket_metadata(client):
|
|
c = make_sharing_case()
|
|
second = Message.objects.create(ticket=c['ticket'], author=c['customer'], body='رسید دوم')
|
|
grants = [share_staff_message(m, c['sender'], c['recipient']) for m in (c['selected'], second)]
|
|
client.force_login(c['recipient'])
|
|
listing = reverse('staff_tickets:list')
|
|
response = client.get(listing, {'tab': 'shared'})
|
|
assert len(response.context['tickets']) == 1
|
|
assert c['ticket'].subject in response.content.decode()
|
|
assert len(client.get(listing, {'tab': 'shared', 'q': c['ticket'].subject}).context['tickets']) == 1
|
|
assert len(client.get(listing, {'tab': 'shared', 'q': second.body}).context['tickets']) == 1
|
|
assert len(client.get(listing, {'tab': 'shared', 'status': c['ticket'].status, 'category': c['ticket'].category_id}).context['tickets']) == 1
|
|
assert len(client.get(listing, {'tab': 'shared', 'q': c['customer'].mobile}).context['tickets']) == 1
|
|
assert client.get(listing, {'tab': 'shared', 'q': 'مکاتبه محرمانه قبل'}).context['tickets'] == []
|
|
assert client.get(reverse('staff_tickets:shares_count')).json()['unread_shares'] == 2
|
|
read = reverse('staff_tickets:read_shares', args=[c['ticket'].public_id])
|
|
assert client.post(read, {'message_ids': ['bad-uuid', str(c['ticket'].ticket_messages.first().public_id)]}).json()['unread_shares'] == 2
|
|
assert client.post(read, {'message_ids': [str(second.public_id)]}).json()['unread_shares'] == 1
|
|
grants[0].refresh_from_db()
|
|
assert grants[0].first_viewed_at is None
|
|
revoke_staff_share(grants[0], c['sender'])
|
|
assert client.get(reverse('staff_tickets:shares_count')).json()['unread_shares'] == 0
|
|
assert client.get(listing, {'tab': 'shared', 'unread': 1}).context['tickets'] == []
|
|
|
|
|
|
def test_limited_poll_tracks_revocation_without_leaking_hidden_activity(client):
|
|
c = make_sharing_case()
|
|
grant = share_staff_message(c['selected'], c['sender'], c['recipient'])
|
|
client.force_login(c['recipient'])
|
|
url = reverse('staff_tickets:poll', args=[c['ticket'].public_id])
|
|
baseline = client.get(url).json()
|
|
assert baseline['count'] == 1 and c['selected'].body in baseline['html']
|
|
assert 'مکاتبه محرمانه' not in baseline['html']
|
|
Message.objects.create(ticket=c['ticket'], author=c['sender'], body='hidden future')
|
|
same = client.get(url, {'version': baseline['version'], 'cursor': baseline['cursor']}).json()
|
|
assert same['changed'] is False
|
|
second = Message.objects.create(ticket=c['ticket'], author=c['customer'], body='second shared')
|
|
share_staff_message(second, c['sender'], c['recipient'])
|
|
revoke_staff_share(grant, c['sender'])
|
|
changed = client.get(url, {'version': baseline['version'], 'cursor': baseline['cursor']}).json()
|
|
assert changed['replace'] and changed['count'] == 1
|
|
assert second.body in changed['html'] and c['selected'].body not in changed['html']
|
|
older = client.get(reverse('staff_tickets:messages', args=[c['ticket'].public_id]), {'before': c['selected'].public_id})
|
|
assert 'hidden future' not in older.content.decode()
|
|
Message.objects.filter(pk=second.pk).update(deleted_at=timezone.now())
|
|
assert client.get(url).status_code == 403
|
|
assert client.get(reverse('staff_tickets:detail', args=[c['ticket'].public_id])).status_code == 403
|
|
|
|
|
|
@pytest.mark.parametrize('role', StaffProfile.Role.values)
|
|
def test_every_active_staff_role_can_operate_and_share_accessible_tickets(client, role):
|
|
c = make_sharing_case()
|
|
sender = c['sender']
|
|
sender.user_permissions.clear()
|
|
StaffProfile.objects.filter(user=sender).update(role=role)
|
|
sender.refresh_from_db()
|
|
assert TicketPolicy.can_share(sender, c['ticket'])
|
|
assert TicketPolicy.can_transfer(sender, c['ticket'])
|
|
change_status(c['ticket'], sender, 'OPEN')
|
|
change_priority(c['ticket'], sender, 'URGENT')
|
|
share_staff_message(c['selected'], sender, c['recipient'])
|
|
share_staff_ticket(c['ticket'], sender, c['outsider'], 'VIEW_ONLY')
|
|
client.force_login(sender)
|
|
body = client.get(reverse('staff_tickets:detail', args=[c['ticket'].public_id])).content.decode()
|
|
assert 'ثبت وضعیت' in body and 'ثبت اولویت' in body and 'انتقال تیکت' in body
|
|
assert 'اشتراک با کارشناس' in body
|
|
assert not TicketPolicy.can_operate(c['recipient'], c['ticket'])
|
|
assert TicketPolicy.can_operate(c['outsider'], c['ticket'])
|
|
assert not TicketPolicy.can_operate(c['customer'], c['ticket'])
|
|
StaffProfile.objects.filter(user=sender).update(is_active=False)
|
|
sender.refresh_from_db()
|
|
assert not TicketPolicy.can_share(sender, c['ticket'])
|
|
|
|
|
|
def test_old_shared_messages_stay_unread_until_displayed_and_poll_keeps_pagination(client):
|
|
c = make_sharing_case()
|
|
share_staff_message(c['selected'], c['sender'], c['recipient'])
|
|
for index in range(51):
|
|
message = Message.objects.create(ticket=c['ticket'], author=c['customer'], body=f'رسید مجاز {index}')
|
|
share_staff_message(message, c['sender'], c['recipient'])
|
|
client.force_login(c['recipient'])
|
|
response = client.get(reverse('staff_tickets:detail', args=[c['ticket'].public_id]))
|
|
assert len(response.context['conversation']) == 50 and response.context['has_older']
|
|
assert c['selected'].body not in response.content.decode()
|
|
read = reverse('staff_tickets:read_shares', args=[c['ticket'].public_id])
|
|
last = response.context['conversation'][-1]
|
|
assert client.post(read, {'message_ids': [str(last.public_id)]}).json()['unread_shares'] == 51
|
|
old = client.get(reverse('staff_tickets:messages', args=[c['ticket'].public_id]), {'before': response.context['older_cursor']})
|
|
assert c['selected'].body in old.content.decode() and 'مکاتبه محرمانه' not in old.content.decode()
|
|
poll = client.get(reverse('staff_tickets:poll', args=[c['ticket'].public_id])).json()
|
|
assert poll['staff_limited'] and 'data-load-older' in poll['html']
|
|
|
|
|
|
def test_full_and_message_shares_have_one_row_and_full_access_preserves_conversation(client):
|
|
c = make_sharing_case()
|
|
share_staff_message(c['selected'], c['sender'], c['recipient'])
|
|
share_staff_ticket(c['ticket'], c['sender'], c['recipient'], 'VIEW_ONLY')
|
|
client.force_login(c['recipient'])
|
|
queue = client.get(reverse('staff_tickets:list'), {'tab': 'shared'})
|
|
assert len(queue.context['tickets']) == 1
|
|
assert c['ticket'].subject in queue.content.decode()
|
|
body = client.get(reverse('staff_tickets:detail', args=[c['ticket'].public_id])).content.decode()
|
|
assert 'مکاتبه محرمانه قبل' in body and 'مکاتبه محرمانه بعد' in body
|
|
assert client.get(reverse('staff_tickets:shares_count')).json()['unread_shares'] == 1
|
|
response = client.post(reverse('staff_tickets:transfer', args=[c['ticket'].public_id]), {'expert': c['outsider'].staff_profile.pk})
|
|
assert response.status_code == 302
|
|
c['ticket'].refresh_from_db()
|
|
assert c['ticket'].owner == c['outsider']
|
|
|
|
|
|
def test_global_staff_access_does_not_duplicate_or_restrict_a_message_shared_ticket(client):
|
|
from django.contrib.auth.models import Permission
|
|
c = make_sharing_case()
|
|
share_staff_message(c['selected'], c['sender'], c['recipient'])
|
|
c['recipient'].user_permissions.add(Permission.objects.get(content_type__app_label='tickets', codename='view_all_tickets'))
|
|
client.force_login(c['recipient'])
|
|
queue = client.get(reverse('staff_tickets:list'), {'tab': 'shared'})
|
|
assert len(queue.context['tickets']) == 1
|
|
assert c['ticket'].subject in queue.content.decode()
|
|
assert not getattr(queue.context['tickets'][0], 'limited_share', False)
|
|
|
|
|
|
def test_message_recipient_has_normal_ticket_details_with_only_shared_message_content(client):
|
|
from apps.customers.models import CustomerProfile
|
|
from apps.tickets.models import Ticket
|
|
c = make_sharing_case()
|
|
CustomerProfile.objects.create(user=c['customer'], first_name='مشتری', last_name='اشتراک')
|
|
Ticket.objects.filter(pk=c['ticket'].pk).update(province='اصفهان', city='گلپایگان', priority='URGENT')
|
|
c['ticket'].refresh_from_db()
|
|
share_staff_message(c['selected'], c['sender'], c['recipient'])
|
|
client.force_login(c['recipient'])
|
|
listing = client.get(reverse('staff_tickets:list'), {'tab': 'shared'})
|
|
html = listing.content.decode()
|
|
assert c['ticket'].subject in html and 'مشتری اشتراک' in html
|
|
assert 'گلپایگان، اصفهان' in html and 'فوری' in html
|
|
detail = client.get(reverse('staff_tickets:detail', args=[c['ticket'].public_id]))
|
|
html = detail.content.decode()
|
|
for value in (c['ticket'].subject, c['customer'].mobile, 'مشتری اشتراک', 'گلپایگان، اصفهان', 'فوری', 'سابقه اخیر', 'id="customer-info"', 'id="ticket-actions"', 'data-ticket-status'):
|
|
assert value in html
|
|
assert 'limited-shared-conversation' not in html
|
|
assert c['selected'].body in html
|
|
assert 'مکاتبه محرمانه قبل' not in html and 'مکاتبه محرمانه بعد' not in html
|
|
assert not TicketAccessGrant.objects.filter(ticket=c['ticket'], user=c['recipient']).exists()
|
|
assert not TicketPolicy.can_view(c['recipient'], c['ticket'])
|
|
poll = client.get(reverse('staff_tickets:poll', args=[c['ticket'].public_id])).json()
|
|
assert poll['status'] == c['ticket'].get_status_display() and poll['count'] == 1
|