share ticket and ticket message
This commit is contained in:
javad committed
1 parent
39e64854b8
commit
79fe1cddbb
35 files changed
+1742
-95
No files matched your search
@@ -0,0 +1,57 @@
|
||||
from django.urls import reverse
|
||||
from playwright.sync_api import expect
|
||||
|
||||
from tests.e2e.pages.base import BasePage
|
||||
|
||||
|
||||
class StaffSharingPage(BasePage):
|
||||
def share_message(self, ticket, message, recipient, note):
|
||||
self.goto(reverse("staff_tickets:detail", args=[ticket.public_id]))
|
||||
controls = self.page.locator(f'[data-share-message="{message.public_id}"]')
|
||||
expect(controls.locator("form")).to_be_hidden()
|
||||
controls.locator("summary").first.focus()
|
||||
controls.locator("summary").first.press("Enter")
|
||||
expect(controls.locator("form")).to_be_visible()
|
||||
controls.get_by_label("کارشناس گیرنده").select_option(str(recipient.staff_profile.pk))
|
||||
controls.locator(".staff-share-note-field > summary").click()
|
||||
controls.get_by_label("توضیح برای گیرنده").fill(note)
|
||||
controls.get_by_role("button", name="اشتراک پیام و پیوست‌هایش").click()
|
||||
expect(self.page.get_by_text("فقط همین پیام و پیوست‌های آن در صندوق اشتراک‌های کارشناس قرار گرفت.")).to_be_visible()
|
||||
|
||||
def open_received_message(self, ticket):
|
||||
self.goto(reverse("staff_tickets:shares"))
|
||||
expect(self.page.locator('[data-share-unread-count]').first).to_contain_text("1")
|
||||
self.page.locator("a.staff-ticket").first.click()
|
||||
expect(self.page.get_by_role("heading", name=ticket.subject, exact=True)).to_be_visible()
|
||||
expect(self.page.locator('.received-share-attribution')).to_contain_text("کارشناس sender")
|
||||
expect(self.page.locator('[data-share-unread-count]').first).to_be_hidden()
|
||||
expect(self.page.locator('[data-ticket-composer]')).to_have_count(0)
|
||||
expect(self.page.get_by_role("button", name="عملیات تیکت", exact=True)).to_be_visible()
|
||||
if self.page.viewport_size["width"] < 768:
|
||||
self.page.get_by_role("button", name="اطلاعات مشتری", exact=True).click()
|
||||
expect(self.page.locator('#customer-info')).to_contain_text(ticket.customer.mobile)
|
||||
expect(self.page.locator('#customer-info')).to_be_visible()
|
||||
if self.page.viewport_size["width"] < 768:
|
||||
self.page.locator('#customer-info [data-close-panel]').click()
|
||||
|
||||
def expect_isolated_content(self, selected_text, forbidden_texts):
|
||||
expect(self.page.get_by_text(selected_text, exact=True)).to_be_visible()
|
||||
for text in forbidden_texts:
|
||||
expect(self.page.get_by_text(text, exact=True)).to_have_count(0)
|
||||
|
||||
def expect_opened_and_revoke(self, ticket, recipient_name):
|
||||
self.goto(reverse("staff_tickets:detail", args=[ticket.public_id]))
|
||||
controls = self.page.locator("[data-share-message]").filter(has=self.page.locator(".share-audit-row").filter(has_text=recipient_name))
|
||||
controls.locator("summary").first.click()
|
||||
row = controls.locator(".share-audit-row").filter(has_text=recipient_name)
|
||||
expect(row).to_contain_text("دیده‌شده")
|
||||
expect(row.get_by_role("button", name="لغو اشتراک پیام")).to_be_hidden()
|
||||
row.locator("summary").click()
|
||||
expect(row).to_contain_text("اولین مشاهده")
|
||||
controls.locator(".staff-inline-share-history > summary").click()
|
||||
expect(controls.locator(".share-event-row").filter(has_text="مشاهده پیام اشتراکی")).to_be_visible()
|
||||
row.get_by_role("button", name="لغو اشتراک پیام").click()
|
||||
|
||||
def expect_revoked_link_denied(self, url):
|
||||
self.goto(url)
|
||||
expect(self.page.get_by_text("403 Forbidden", exact=False)).to_be_visible()
|
||||
@@ -0,0 +1,33 @@
|
||||
import pytest
|
||||
|
||||
from tests.e2e.pages.auth import AuthenticationPage
|
||||
from tests.e2e.pages.sharing import StaffSharingPage
|
||||
from tests.test_staff_sharing import make_sharing_case
|
||||
|
||||
|
||||
pytestmark = [pytest.mark.e2e, pytest.mark.django_db(transaction=True)]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("viewport", [{"width": 1280, "height": 900}, {"width": 390, "height": 844}])
|
||||
def test_staff_can_share_one_message_track_opening_and_revoke(live_server, browser_page, viewport):
|
||||
case = make_sharing_case()
|
||||
password = "SharingTest123!"
|
||||
with browser_page() as page:
|
||||
page.set_viewport_size(viewport)
|
||||
AuthenticationPage(page, live_server.url).login_staff(case["sender"], password)
|
||||
StaffSharingPage(page, live_server.url).share_message(case["ticket"], case["selected"], case["recipient"], "بررسی رسید توسط مالی")
|
||||
with browser_page() as page:
|
||||
page.set_viewport_size(viewport)
|
||||
AuthenticationPage(page, live_server.url).login_staff(case["recipient"], password)
|
||||
sharing = StaffSharingPage(page, live_server.url)
|
||||
sharing.open_received_message(case["ticket"])
|
||||
sharing.expect_isolated_content(case["selected"].body, ["مکاتبه محرمانه قبل", "مکاتبه محرمانه بعد"])
|
||||
shared_url = page.url
|
||||
with browser_page() as page:
|
||||
page.set_viewport_size(viewport)
|
||||
AuthenticationPage(page, live_server.url).login_staff(case["sender"], password)
|
||||
StaffSharingPage(page, live_server.url).expect_opened_and_revoke(case["ticket"], "کارشناس recipient")
|
||||
with browser_page() as page:
|
||||
page.set_viewport_size(viewport)
|
||||
AuthenticationPage(page, live_server.url).login_staff(case["recipient"], password)
|
||||
StaffSharingPage(page, live_server.url).expect_revoked_link_denied(shared_url)
|
||||
@@ -0,0 +1,35 @@
|
||||
import pytest
|
||||
from django.urls import reverse
|
||||
|
||||
from apps.tickets.sharing import revoke_staff_share, share_staff_message, staff_message_context
|
||||
from tests.test_staff_sharing import make_sharing_case
|
||||
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
def test_inline_sharing_keeps_each_messages_recipients_and_history_separate(client):
|
||||
case = make_sharing_case()
|
||||
other_message = case["ticket"].ticket_messages.get(body="مکاتبه محرمانه قبل")
|
||||
selected_grant = share_staff_message(case["selected"], case["sender"], case["recipient"], "رسید منتخب")
|
||||
other_grant = share_staff_message(other_message, case["sender"], case["outsider"], "مکاتبه دیگر")
|
||||
client.force_login(case["recipient"])
|
||||
assert client.get(reverse("staff_tickets:shared_message", args=[selected_grant.public_id]), follow=True).status_code == 200
|
||||
assert client.post(reverse("staff_tickets:read_shares", args=[case["ticket"].public_id]), {"message_ids": [str(case["selected"].public_id)]}).status_code == 200
|
||||
client.force_login(case["sender"])
|
||||
response = client.get(reverse("staff_tickets:detail", args=[case["ticket"].public_id]))
|
||||
selected = next(item for item in response.context["conversation"] if item.pk == case["selected"].pk)
|
||||
before = next(item for item in response.context["conversation"] if item.pk == other_message.pk)
|
||||
assert [grant.pk for grant in selected.staff_message_shares] == [selected_grant.pk]
|
||||
assert selected.staff_message_shares[0].first_viewed_at is not None
|
||||
assert [grant.pk for grant in before.staff_message_shares] == [other_grant.pk]
|
||||
assert {event.metadata["action"] for event in selected.staff_share_events} == {"created", "opened"}
|
||||
assert all(event.metadata["message_public_id"] == str(selected.public_id) for event in selected.staff_share_events)
|
||||
assert all(event.metadata["user"] == case["recipient"].pk for event in selected.staff_share_events)
|
||||
assert revoke_staff_share(selected_grant, case["sender"])
|
||||
staff_message_context(case["ticket"], case["sender"], [case["selected"]])
|
||||
assert case["selected"].staff_message_shares[0].revoked_at is not None
|
||||
assert {event.metadata["action"] for event in case["selected"].staff_share_events} == {"created", "opened", "revoked"}
|
||||
context = staff_message_context(case["ticket"], case["customer"], [other_message])
|
||||
assert not context["can_share_message"]
|
||||
assert not hasattr(other_message, "staff_message_shares")
|
||||
@@ -0,0 +1,160 @@
|
||||
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
|
||||
@@ -0,0 +1,239 @@
|
||||
import pytest
|
||||
import re
|
||||
from django.contrib.auth.models import Permission
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.core.files.uploadedfile import SimpleUploadedFile
|
||||
from django.test import Client
|
||||
from django.urls import reverse
|
||||
from django.utils import timezone
|
||||
|
||||
from apps.accounts.models import User
|
||||
from apps.attachments.models import Attachment
|
||||
from apps.messaging.models import Message
|
||||
from apps.notifications.models import NotificationJob
|
||||
from apps.staff.models import OrganizationalUnit, StaffProfile
|
||||
from apps.tickets.models import MessageShare, TicketAccessGrant, TicketCategory, TicketEvent
|
||||
from apps.tickets.policies import TicketPolicy
|
||||
from apps.tickets.selectors import staff_ticket_rows
|
||||
from apps.tickets.services import create_ticket
|
||||
from apps.tickets.sharing import revoke_staff_share, share_staff_message, share_staff_ticket
|
||||
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
def make_sharing_case():
|
||||
sales = OrganizationalUnit.objects.create(name="فروش اشتراک", slug="sharing-sales")
|
||||
finance = OrganizationalUnit.objects.create(name="مالی اشتراک", slug="sharing-finance")
|
||||
users = {}
|
||||
for index, name in enumerate(("sender", "recipient", "outsider"), 1):
|
||||
user = User.objects.create_user(username=f"sharing-{name}", mobile=f"0912777000{index}", password="SharingTest123!", account_type="STAFF")
|
||||
StaffProfile.objects.create(user=user, display_name=f"کارشناس {name}", primary_unit=sales if name == "sender" else finance)
|
||||
users[name] = user
|
||||
users["sender"].user_permissions.add(Permission.objects.get(content_type__app_label="tickets", codename="share_ticket"))
|
||||
customer = User.objects.create_user(username="sharing-customer", mobile="09127770004", account_type="CUSTOMER")
|
||||
category = TicketCategory.objects.create(unit=sales, title="اشتراک", slug="sharing-category")
|
||||
ticket = create_ticket(customer=customer, category=category, selected_expert=users["sender"], subject="موضوع محرمانه کل تیکت", body="مکاتبه محرمانه قبل")
|
||||
selected = Message.objects.create(ticket=ticket, author=customer, body="رسید منتخب برای بررسی مالی")
|
||||
Message.objects.create(ticket=ticket, author=users["sender"], body="مکاتبه محرمانه بعد", is_internal=True)
|
||||
return {**users, "customer": customer, "ticket": ticket, "selected": selected}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sharing_case():
|
||||
return make_sharing_case()
|
||||
|
||||
|
||||
def attachment_for(message, content=b"private receipt", filename="receipt.pdf"):
|
||||
return Attachment.objects.create(message=message, file=SimpleUploadedFile(filename, content), original_name=filename,
|
||||
stored_name=filename, relative_path=filename, attachment_type="PDF", mime_type="application/pdf", extension=".pdf", file_size=len(content), uploaded_by=message.author)
|
||||
|
||||
|
||||
def test_ticket_sharing_records_first_open_updates_and_revocation(client, sharing_case):
|
||||
c = sharing_case
|
||||
url = reverse("staff_tickets:share", args=[c["ticket"].public_id])
|
||||
client.force_login(c["sender"])
|
||||
response = client.post(url, {"expert": c["recipient"].staff_profile.pk, "permission_level": "VIEW_ONLY", "note": "بررسی کل پرونده"})
|
||||
assert response.status_code == 302
|
||||
grant = TicketAccessGrant.objects.get(ticket=c["ticket"], user=c["recipient"])
|
||||
assert grant.first_viewed_at is None and grant.view_count == 0
|
||||
client.force_login(c["recipient"])
|
||||
assert client.get(reverse("staff_tickets:shares")).status_code == 302
|
||||
grant.refresh_from_db()
|
||||
assert grant.first_viewed_at is None # Listing is not evidence of opening.
|
||||
detail = reverse("staff_tickets:detail", args=[c["ticket"].public_id])
|
||||
assert client.get(detail).status_code == 200
|
||||
assert client.get(detail).status_code == 200
|
||||
grant.refresh_from_db()
|
||||
first_view = grant.first_viewed_at
|
||||
assert first_view and grant.last_viewed_at >= first_view and grant.view_count == 2
|
||||
assert c["ticket"].events.filter(event_type=TicketEvent.Type.SHARE_OPENED).count() == 1
|
||||
client.force_login(c["sender"])
|
||||
audit = client.get(detail).content.decode()
|
||||
assert "دیده‌شده" in audit and "کارشناس recipient" in audit and "بررسی کل پرونده" in audit
|
||||
client.post(url, {"expert": c["recipient"].staff_profile.pk, "permission_level": "REPLY", "note": "توضیح جدید"})
|
||||
grant.refresh_from_db()
|
||||
assert grant.permission_level == "REPLY" and grant.first_viewed_at == first_view
|
||||
assert c["ticket"].events.filter(event_type=TicketEvent.Type.SHARED).count() == 2
|
||||
client.post(reverse("staff_tickets:revoke", args=[c["ticket"].public_id, c["recipient"].public_id]))
|
||||
grant.refresh_from_db()
|
||||
assert grant.revoked_at and grant.revoked_by == c["sender"]
|
||||
assert not TicketPolicy.can_view(c["recipient"], c["ticket"])
|
||||
new = share_staff_ticket(c["ticket"], c["sender"], c["recipient"], "VIEW_ONLY")
|
||||
assert new.pk != grant.pk and new.first_viewed_at is None
|
||||
|
||||
|
||||
def test_one_message_share_never_grants_ticket_or_neighbor_access(client, sharing_case):
|
||||
c = sharing_case
|
||||
client.force_login(c["customer"])
|
||||
customer_url = reverse("tickets:detail", args=[c["ticket"].public_id])
|
||||
def stable_html(response):
|
||||
# Django masks CSRF tokens with fresh randomness on every render.
|
||||
return re.sub(r'(?<=["\x27])[A-Za-z0-9]{64}(?=["\x27])', 'CSRF_TOKEN', response.content.decode())
|
||||
before = stable_html(client.get(customer_url))
|
||||
jobs_before = NotificationJob.objects.count()
|
||||
grant = share_staff_message(c["selected"], c["sender"], c["recipient"], "بررسی رسید")
|
||||
assert not TicketAccessGrant.objects.filter(ticket=c["ticket"], user=c["recipient"]).exists()
|
||||
assert not TicketPolicy.can_view(c["recipient"], c["ticket"])
|
||||
assert not staff_ticket_rows(c["recipient"]).filter(pk=c["ticket"].pk).exists()
|
||||
assert NotificationJob.objects.count() == jobs_before
|
||||
assert stable_html(client.get(customer_url)) == before
|
||||
client.force_login(c["recipient"])
|
||||
assert client.get(reverse("staff_tickets:shares_count")).json()["unread_shares"] == 1
|
||||
inbox = client.get(reverse("staff_tickets:list"), {"tab": "shared"}).content.decode()
|
||||
assert c["ticket"].subject in inbox
|
||||
assert grant.first_viewed_at is None
|
||||
assert client.get(reverse("staff_tickets:shared_message", args=[grant.public_id])).status_code == 302
|
||||
detail = reverse("staff_tickets:detail", args=[c["ticket"].public_id])
|
||||
page = client.get(detail)
|
||||
body = page.content.decode()
|
||||
assert c["selected"].body in body and "بررسی رسید" in body
|
||||
assert "مکاتبه محرمانه قبل" not in body and "مکاتبه محرمانه بعد" not in body
|
||||
assert c["ticket"].subject in body and c["customer"].mobile in body
|
||||
assert 'id="ticket-actions"' in body and 'id="customer-info"' in body
|
||||
assert reverse("messages:revisions", args=[c["selected"].public_id]) not in body
|
||||
read_url = reverse("staff_tickets:read_shares", args=[c["ticket"].public_id])
|
||||
grant.refresh_from_db()
|
||||
assert grant.first_viewed_at is None
|
||||
assert client.post(read_url, {"message_ids": [str(c["selected"].public_id)]}).json()["unread_shares"] == 0
|
||||
Message.objects.create(ticket=c["ticket"], author=c["sender"], body="مکاتبه آینده محرمانه")
|
||||
assert "مکاتبه آینده محرمانه" not in client.get(detail).content.decode()
|
||||
grant.refresh_from_db()
|
||||
assert grant.first_viewed_at and grant.view_count == 1
|
||||
assert client.get(reverse("staff_tickets:shares_count")).json()["unread_shares"] == 0
|
||||
assert c["ticket"].events.filter(event_type=TicketEvent.Type.MESSAGE_SHARE_OPENED).count() == 1
|
||||
for route in ("detail", "messages", "poll"):
|
||||
assert client.get(reverse(f"staff_tickets:{route}", args=[c["ticket"].public_id])).status_code == 200
|
||||
assert client.post(reverse("staff_tickets:reply", args=[c["ticket"].public_id]), {"body": "unauthorized reply"}).status_code in (302, 403)
|
||||
assert not c["ticket"].ticket_messages.filter(body="unauthorized reply").exists()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("download", [False, True])
|
||||
def test_shared_attachment_is_scoped_to_message_and_revoked_immediately(client, sharing_case, settings, tmp_path, download):
|
||||
settings.MEDIA_ROOT = tmp_path
|
||||
settings.USE_NGINX_MEDIA = False
|
||||
c = sharing_case
|
||||
chosen = attachment_for(c["selected"])
|
||||
other = attachment_for(c["ticket"].ticket_messages.first(), b"other secret", "other.pdf")
|
||||
grant = share_staff_message(c["selected"], c["sender"], c["recipient"])
|
||||
route = "shared_message_attachment_download" if download else "shared_message_attachment"
|
||||
url = reverse(f"staff_tickets:{route}", args=[grant.public_id, chosen.public_id])
|
||||
client.force_login(c["recipient"])
|
||||
response = client.get(url)
|
||||
assert response.status_code == 200 and b"".join(response.streaming_content) == b"private receipt"
|
||||
assert "no-store" in response["Cache-Control"]
|
||||
assert client.get(reverse(f"staff_tickets:{route}", args=[grant.public_id, other.public_id])).status_code == 404
|
||||
assert client.get(reverse("attachments:view", args=[chosen.public_id])).status_code == 403
|
||||
client.force_login(c["outsider"])
|
||||
assert client.get(url).status_code == 403
|
||||
revoke_staff_share(grant, c["sender"])
|
||||
client.force_login(c["recipient"])
|
||||
assert client.get(url).status_code == 403
|
||||
assert client.get(reverse("staff_tickets:shared_message", args=[grant.public_id])).status_code == 403
|
||||
|
||||
|
||||
def test_deleted_message_and_inactive_recipient_cannot_open_share(client, sharing_case):
|
||||
c = sharing_case
|
||||
grant = share_staff_message(c["selected"], c["sender"], c["recipient"])
|
||||
url = reverse("staff_tickets:shared_message", args=[grant.public_id])
|
||||
client.force_login(c["recipient"])
|
||||
StaffProfile.objects.filter(user=c["recipient"]).update(is_active=False)
|
||||
assert client.get(url).status_code == 403
|
||||
StaffProfile.objects.filter(user=c["recipient"]).update(is_active=True)
|
||||
Message.objects.filter(pk=c["selected"].pk).update(deleted_at=timezone.now())
|
||||
assert client.get(url).status_code == 403
|
||||
assert "بررسی رسید" not in client.get(reverse("staff_tickets:shares")).content.decode()
|
||||
with pytest.raises(ValidationError):
|
||||
share_staff_message(c["selected"], c["sender"], c["recipient"])
|
||||
|
||||
|
||||
def test_share_operations_validate_recipients_authority_method_and_csrf(client, sharing_case):
|
||||
c = sharing_case
|
||||
url = reverse("staff_tickets:share_message", args=[c["ticket"].public_id, c["selected"].public_id])
|
||||
client.force_login(c["sender"])
|
||||
assert client.get(url).status_code == 405
|
||||
for target in (c["sender"], c["customer"]):
|
||||
value = getattr(target, "staff_profile", None)
|
||||
client.post(url, {"expert": value.pk if value else 999999, "note": "bad"})
|
||||
assert MessageShare.objects.count() == 0
|
||||
StaffProfile.objects.filter(user=c["recipient"]).update(is_active=False)
|
||||
client.post(url, {"expert": c["recipient"].staff_profile.pk})
|
||||
assert MessageShare.objects.count() == 0
|
||||
StaffProfile.objects.filter(user=c["recipient"]).update(is_active=True)
|
||||
csrf_client = Client(enforce_csrf_checks=True)
|
||||
csrf_client.force_login(c["sender"])
|
||||
assert csrf_client.post(url, {"expert": c["recipient"].staff_profile.pk}).status_code == 403
|
||||
client.force_login(c["outsider"])
|
||||
assert client.post(url, {"expert": c["recipient"].staff_profile.pk}).status_code == 403
|
||||
client.force_login(c["customer"])
|
||||
assert client.post(url, {"expert": c["recipient"].staff_profile.pk}).status_code == 403
|
||||
assert MessageShare.objects.count() == 0
|
||||
|
||||
|
||||
def test_duplicate_share_is_idempotent_and_regrant_preserves_history(client, sharing_case):
|
||||
c = sharing_case
|
||||
first = share_staff_message(c["selected"], c["sender"], c["recipient"], "رسید")
|
||||
duplicate = share_staff_message(c["selected"], c["sender"], c["recipient"], "رسید")
|
||||
assert duplicate.pk == first.pk
|
||||
assert c["ticket"].events.filter(event_type=TicketEvent.Type.MESSAGE_SHARED).count() == 1
|
||||
assert revoke_staff_share(first, c["sender"])
|
||||
assert not revoke_staff_share(first, c["sender"])
|
||||
new = share_staff_message(c["selected"], c["sender"], c["recipient"], "ارسال مجدد")
|
||||
assert new.pk != first.pk and new.first_viewed_at is None
|
||||
first.refresh_from_db()
|
||||
assert first.revoked_at and first.note == "رسید"
|
||||
client.force_login(c["sender"])
|
||||
body = client.get(reverse("staff_tickets:detail", args=[c["ticket"].public_id])).content.decode()
|
||||
assert "لغوشده" in body and "ارسال مجدد" in body and "هنوز دیده نشده" in body
|
||||
|
||||
|
||||
def test_staff_message_sharing_controls_survive_polling_and_pagination(client, sharing_case):
|
||||
c = sharing_case
|
||||
for index in range(53):
|
||||
Message.objects.create(ticket=c["ticket"], author=c["customer"], body=f"پیام شماره {index}")
|
||||
client.force_login(c["sender"])
|
||||
detail = client.get(reverse("staff_tickets:detail", args=[c["ticket"].public_id]))
|
||||
assert "اشتراک فقط این پیام" in detail.content.decode()
|
||||
before = detail.context["older_cursor"]
|
||||
older = client.get(reverse("staff_tickets:messages", args=[c["ticket"].public_id]), {"before": before})
|
||||
assert "اشتراک فقط این پیام" in older.content.decode()
|
||||
poll = client.get(reverse("staff_tickets:poll", args=[c["ticket"].public_id])).json()
|
||||
assert "اشتراک فقط این پیام" in poll["html"]
|
||||
client.force_login(c["customer"])
|
||||
customer_poll = client.get(reverse("tickets:poll", args=[c["ticket"].public_id])).json()
|
||||
assert "اشتراک فقط این پیام" not in customer_poll["html"]
|
||||
assert "staff-message-share" not in customer_poll["html"]
|
||||
|
||||
|
||||
def test_message_share_cannot_be_used_under_another_ticket_or_by_another_user(client, sharing_case):
|
||||
c = sharing_case
|
||||
other = create_ticket(customer=c["customer"], category=c["ticket"].category, selected_expert=c["sender"], subject="تیکت دیگر", body="دیگر")
|
||||
client.force_login(c["sender"])
|
||||
assert client.post(reverse("staff_tickets:share_message", args=[other.public_id, c["selected"].public_id]), {"expert": c["recipient"].staff_profile.pk}).status_code == 404
|
||||
grant = share_staff_message(c["selected"], c["sender"], c["recipient"])
|
||||
assert client.post(reverse("staff_tickets:revoke_message_share", args=[other.public_id, grant.public_id])).status_code == 404
|
||||
for user in (c["sender"], c["outsider"], c["customer"]):
|
||||
client.force_login(user)
|
||||
assert client.get(reverse("staff_tickets:shared_message", args=[grant.public_id])).status_code == 403
|
||||
client.logout()
|
||||
assert client.get(reverse("staff_tickets:shared_message", args=[grant.public_id])).status_code == 302
|
||||
Reference in new issue
Block a user