customerTicket/tests/test_voip_calls.py

import json
from unittest.mock import Mock, patch

import pytest
import requests
from django.urls import reverse

from apps.accounts.models import User
from apps.core.models import VoipConfiguration
from apps.customers.models import ContactGroup, CustomerProfile
from apps.staff.models import OrganizationalUnit, StaffProfile
from apps.tickets.models import TicketCategory
from apps.tickets.services import create_ticket


@pytest.fixture
def voip_data(db):
    unit = OrganizationalUnit.objects.create(name="فروش تماس", slug="voip-sales")
    group = ContactGroup.objects.create(name="مخاطبان تماس", unit=unit)
    existing_customer = User.objects.create_user(
        username="existing-voip-mobile-customer",
        mobile="09384999655",
        account_type=User.AccountType.CUSTOMER,
    )
    agent = User.objects.create_superuser(
        username="voip-agent",
        password="safe-password",
        mobile=None,
        account_type=User.AccountType.STAFF,
        voip_extension="300",
        voip_mobile="09384999655",
    )
    StaffProfile.objects.create(user=agent, primary_unit=unit, display_name="کارشناس تماس")
    customer = User.objects.create_user(
        username="voip-customer",
        mobile="09139591708",
        account_type=User.AccountType.CUSTOMER,
    )
    profile = CustomerProfile.objects.create(user=customer, first_name="مخاطب", last_name="آزمایشی")
    profile.groups.add(group)
    category = TicketCategory.objects.create(unit=unit, title="تماس فروش", slug="voip-call")
    ticket = create_ticket(customer=customer, category=category, subject="پیگیری تماس", body="شرح تماس")
    VoipConfiguration.objects.create(voip_call_url="http://127.0.0.1:9000/v1/click2call")
    assert agent.voip_mobile == existing_customer.mobile
    return agent, customer, ticket


@pytest.mark.django_db
@pytest.mark.parametrize(
    "case",
    (
        {"method": "ip_phone", "agent_field": "agent_extension", "agent_value": "300", "key_prefix": "internal-"},
        {"method": "mobile", "agent_field": "agent_mobile", "agent_value": "09384999655", "key_prefix": "mobile-"},
    ),
)
def test_backend_sends_voip_call(client, voip_data, case):
    agent, customer, _ = voip_data
    method = case["method"]
    client.force_login(agent)
    response_mock = Mock(content=b"{}")
    response_mock.json.return_value = {}
    with patch("apps.integrations.voip.requests.post", return_value=response_mock) as post:
        response = client.post(
            reverse("staff_tickets:contact_call", args=[customer.public_id]),
            data=json.dumps({"method": method}),
            content_type="application/json",
        )

    assert response.status_code == 200
    call = post.call_args
    assert call.args == ("http://127.0.0.1:9000/v1/click2call",)
    assert call.kwargs["json"] == {
        "method": method,
        case["agent_field"]: case["agent_value"],
        "customer_number": "09139591708",
        "metadata": {"agent_id": str(agent.public_id), "customer_id": str(customer.public_id)},
    }
    assert call.kwargs["headers"]["Idempotency-Key"].startswith(case["key_prefix"])


@pytest.mark.django_db
def test_call_pages_hide_provider_url(client, voip_data):
    agent, customer, ticket = voip_data
    client.force_login(agent)
    for route in (
        reverse("staff_tickets:contacts"),
        reverse("staff_tickets:contact_edit", args=[customer.public_id]),
        reverse("staff_tickets:detail", args=[ticket.public_id]),
    ):
        content = client.get(route).content.decode()
        assert 'data-voip-method="ip_phone"' in content
        assert 'data-voip-method="mobile"' in content
        assert reverse("staff_tickets:contact_call", args=[customer.public_id]) in content
        assert "http://127.0.0.1:9000/v1/click2call" not in content
        assert "voip-call.js" in content


@pytest.mark.django_db
def test_call_method_availability(client, voip_data):
    agent, _, ticket = voip_data
    agent.voip_mobile = ""
    agent.save(update_fields=["voip_mobile"])
    client.force_login(agent)
    content = client.get(reverse("staff_tickets:detail", args=[ticket.public_id])).content.decode()
    assert 'data-voip-method="ip_phone"' in content
    assert 'data-voip-method="mobile"' not in content

    agent.voip_call = False
    agent.save(update_fields=["voip_call"])
    content = client.get(reverse("staff_tickets:detail", args=[ticket.public_id])).content.decode()
    assert "data-voip-call" not in content


@pytest.mark.django_db
def test_voip_error_is_minimal(client, voip_data):
    agent, customer, _ = voip_data
    client.force_login(agent)
    with patch("apps.integrations.voip.requests.post", side_effect=requests.ConnectionError("provider detail")):
        response = client.post(
            reverse("staff_tickets:contact_call", args=[customer.public_id]),
            data=json.dumps({"method": "mobile"}),
            content_type="application/json",
        )
    assert response.status_code == 502
    assert response.json() == {"success": False, "message": "برقراری تماس ناموفق بود."}

    agent.voip_call = False
    agent.save(update_fields=["voip_call"])
    response = client.post(
        reverse("staff_tickets:contact_call", args=[customer.public_id]),
        data=json.dumps({"method": "mobile"}),
        content_type="application/json",
    )
    assert response.status_code == 403


@pytest.mark.django_db
@pytest.mark.parametrize("provider_error", [requests.Timeout("timed out"), requests.HTTPError("503 unavailable")])
def test_voip_provider_failure_does_not_raise_into_staff_page(client, voip_data, provider_error):
    agent, customer, _ = voip_data
    client.force_login(agent)
    with patch("apps.integrations.voip.requests.post", side_effect=provider_error) as post:
        response = client.post(
            reverse("staff_tickets:contact_call", args=[customer.public_id]),
            data=json.dumps({"method": "mobile"}),
            content_type="application/json",
        )
    assert response.status_code == 502
    assert response.json() == {"success": False, "message": "برقراری تماس ناموفق بود."}
    assert post.call_args.kwargs["timeout"] > 0


@pytest.mark.django_db
def test_voip_invalid_provider_response_is_reported_as_failure(client, voip_data):
    agent, customer, _ = voip_data
    client.force_login(agent)
    provider_response = Mock(content=b"[]")
    provider_response.json.return_value = []
    with patch("apps.integrations.voip.requests.post", return_value=provider_response):
        response = client.post(
            reverse("staff_tickets:contact_call", args=[customer.public_id]),
            data=json.dumps({"method": "mobile"}),
            content_type="application/json",
        )
    assert response.status_code == 502
    assert response.json() == {"success": False, "message": "برقراری تماس ناموفق بود."}