109 lines
5.4 KiB
Python
109 lines
5.4 KiB
Python
import pytest
|
|
from django.test import Client
|
|
from django.urls import reverse
|
|
|
|
from apps.customers.models import CustomerProfile
|
|
from apps.messaging.models import Message
|
|
from apps.tickets.sharing import share_staff_ticket
|
|
from tests.test_staff_sharing import make_sharing_case
|
|
from tests.e2e.pages.auth import AuthenticationPage
|
|
from tests.e2e.pages.message_reads import MessageReadsPage, MessageReadsRacePage
|
|
|
|
pytestmark = [pytest.mark.e2e, pytest.mark.django_db(transaction=True)]
|
|
PASSWORD = 'SharingTest123!'
|
|
|
|
|
|
@pytest.fixture
|
|
def reading_case(settings):
|
|
settings.PASSWORD_HASHERS = ['django.contrib.auth.hashers.MD5PasswordHasher']
|
|
case = make_sharing_case()
|
|
case['customer'].set_password(PASSWORD)
|
|
case['customer'].save(update_fields=['password'])
|
|
CustomerProfile.objects.create(user=case['customer'], first_name='مشتری', last_name='خواننده')
|
|
share_staff_ticket(case['ticket'], case['sender'], case['recipient'], 'VIEW_ONLY')
|
|
case['probes'] = [Message.objects.create(ticket=case['ticket'], author=case['sender'],
|
|
body=f'پیام آزمایشی {i}\n' + ('متن برای آزمایش محدوده مشاهده. ' * 45)) for i in range(12)]
|
|
return case
|
|
|
|
|
|
@pytest.mark.parametrize('staff,width', [(True, 1280), (False, 390)])
|
|
def test_only_on_screen_messages_receive_seen_and_scrolling_acknowledges_more(live_server, browser_page, reading_case, staff, width):
|
|
c = reading_case
|
|
namespace = 'staff_tickets' if staff else 'tickets'
|
|
detail = reverse(namespace + ':detail', args=[c['ticket'].public_id])
|
|
read = reverse(namespace + (':read_shares' if staff else ':read'), args=[c['ticket'].public_id])
|
|
reader_name = 'کارشناس recipient' if staff else 'مشتری خواننده'
|
|
oldest, latest = c['probes'][0], c['probes'][-1]
|
|
with browser_page() as page:
|
|
page.set_viewport_size({'width': width, 'height': 900})
|
|
auth = AuthenticationPage(page, live_server.url)
|
|
if staff:
|
|
auth.login_staff(c['recipient'], PASSWORD)
|
|
else:
|
|
auth.login_customer(c['customer'], PASSWORD)
|
|
MessageReadsPage(page, live_server.url).open_and_acknowledge(detail, read, latest)
|
|
with browser_page() as page:
|
|
AuthenticationPage(page, live_server.url).login_staff(c['sender'], PASSWORD)
|
|
view = MessageReadsPage(page, live_server.url)
|
|
view.goto(reverse('staff_tickets:detail', args=[c['ticket'].public_id]))
|
|
view.expect_unseen(oldest)
|
|
view.expect_seen_by(latest, reader_name)
|
|
with browser_page() as page:
|
|
auth = AuthenticationPage(page, live_server.url)
|
|
if staff:
|
|
auth.login_staff(c['recipient'], PASSWORD)
|
|
else:
|
|
auth.login_customer(c['customer'], PASSWORD)
|
|
view = MessageReadsPage(page, live_server.url)
|
|
view.open_and_acknowledge(detail, read, latest)
|
|
view.scroll_and_acknowledge(oldest, read)
|
|
with browser_page() as page:
|
|
AuthenticationPage(page, live_server.url).login_staff(c['sender'], PASSWORD)
|
|
view = MessageReadsPage(page, live_server.url)
|
|
view.goto(reverse('staff_tickets:detail', args=[c['ticket'].public_id]))
|
|
view.expect_seen_by(oldest, reader_name)
|
|
|
|
|
|
def test_background_poll_does_not_mark_seen_and_failed_acknowledgement_retries(live_server, browser_page, reading_case):
|
|
c = reading_case
|
|
detail = reverse('staff_tickets:detail', args=[c['ticket'].public_id])
|
|
read = reverse('staff_tickets:read_shares', args=[c['ticket'].public_id])
|
|
poll = reverse('staff_tickets:poll', args=[c['ticket'].public_id])
|
|
latest = c['probes'][-1]
|
|
with browser_page() as page:
|
|
view = MessageReadsPage(page, live_server.url)
|
|
view.emulate_background_tab()
|
|
AuthenticationPage(page, live_server.url).login_staff(c['recipient'], PASSWORD)
|
|
view.open_in_background(detail, poll)
|
|
# A second server render exposes any erroneous background receipts.
|
|
view.open_in_background(detail, poll)
|
|
view.expect_unseen(latest)
|
|
view.fail_first_acknowledgement(read)
|
|
view.activate_and_acknowledge(latest, read)
|
|
# The first response is deliberately 503; the successful retry must arrive.
|
|
view.wait_for_successful_acknowledgement(read)
|
|
with browser_page() as page:
|
|
AuthenticationPage(page, live_server.url).login_staff(c['sender'], PASSWORD)
|
|
view = MessageReadsPage(page, live_server.url)
|
|
view.goto(detail)
|
|
view.expect_seen_by(latest, 'کارشناس recipient')
|
|
|
|
|
|
def test_revocation_wins_over_a_poll_response_already_in_flight(live_server, browser_page, reading_case):
|
|
c = reading_case
|
|
detail = reverse('staff_tickets:detail', args=[c['ticket'].public_id])
|
|
read = reverse('staff_tickets:read_shares', args=[c['ticket'].public_id])
|
|
sender = Client()
|
|
sender.force_login(c['sender'])
|
|
sender.get(detail)
|
|
csrf = sender.cookies['csrftoken'].value
|
|
cookie = f"sessionid={sender.cookies['sessionid'].value}; csrftoken={csrf}"
|
|
with browser_page() as reader:
|
|
view = MessageReadsRacePage(reader, live_server.url)
|
|
view.hold_poll_after_receiving_its_body()
|
|
AuthenticationPage(reader, live_server.url).login_staff(c['recipient'], PASSWORD)
|
|
view.open_and_acknowledge(detail, read, c['probes'][-1])
|
|
view.revoke_ticket(c['ticket'], c['recipient'], cookie, csrf)
|
|
view.scroll_to_revoked_message(c['probes'][0], read)
|
|
view.release_poll_and_expect_no_messages()
|