240 lines
14 KiB
Python
240 lines
14 KiB
Python
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
|