149 lines
5.3 KiB
Python
149 lines
5.3 KiB
Python
from datetime import date
|
|
from unittest import mock
|
|
|
|
from django.test import Client, SimpleTestCase, TestCase
|
|
|
|
from apps.asterisk_connector.ami import AmiClient, AmiError, parse_ami_packet
|
|
from apps.asterisk_connector.services import (
|
|
TelephonyError,
|
|
channel_for_employee,
|
|
originate_between,
|
|
sanitize_exten,
|
|
)
|
|
from apps.employees.models import Employee
|
|
from core.models import ConnectorConfig
|
|
from core.test_utils import login_2fa
|
|
|
|
|
|
def make_employee(**overrides):
|
|
n = Employee.objects.count() + 1
|
|
defaults = {
|
|
'username': f'ami.user{n}',
|
|
'email': f'ami.user{n}@company.local',
|
|
'first_name': 'Тест',
|
|
'last_name': 'Ами',
|
|
'hire_date': date(2026, 1, 1),
|
|
'user_location_type': 'pharmacy',
|
|
'mattermost_sync_enabled': False,
|
|
'ad_sync_enabled': False,
|
|
'is_staff': False,
|
|
}
|
|
defaults.update(overrides)
|
|
return Employee.objects.create(**defaults)
|
|
|
|
|
|
class AmiProtocolTests(SimpleTestCase):
|
|
def test_parse_packet(self):
|
|
packet = parse_ami_packet('Response: Success\r\nMessage: Authentication accepted\r\n')
|
|
self.assertEqual(packet['Response'], 'Success')
|
|
|
|
def test_send_login_and_originate(self):
|
|
packets = [
|
|
b'Asterisk Call Manager/2.10\r\n',
|
|
b'Response: Success\r\nMessage: Authentication accepted\r\n\r\n',
|
|
b'Response: Success\r\nMessage: Originate successfully queued\r\n\r\n',
|
|
b'Response: Goodbye\r\n\r\n',
|
|
]
|
|
|
|
class FakeSock:
|
|
def __init__(self):
|
|
self.sent = []
|
|
self._packets = list(packets)
|
|
|
|
def recv(self, _n):
|
|
return self._packets.pop(0) if self._packets else b''
|
|
|
|
def sendall(self, data):
|
|
self.sent.append(data)
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
sock = FakeSock()
|
|
with mock.patch('apps.asterisk_connector.ami.socket.create_connection', return_value=sock):
|
|
with AmiClient('127.0.0.1', 5038, 'acore', 'secret') as client:
|
|
reply = client.originate(
|
|
channel='PJSIP/101',
|
|
exten='202',
|
|
context='from-internal',
|
|
caller_id='Test <101>',
|
|
)
|
|
self.assertEqual(reply['Response'], 'Success')
|
|
self.assertTrue(any(b'Action: Login' in chunk for chunk in sock.sent))
|
|
self.assertTrue(any(b'Channel: PJSIP/101' in chunk for chunk in sock.sent))
|
|
|
|
|
|
class ChannelMappingTests(TestCase):
|
|
def test_sip_uses_pjsip_template(self):
|
|
emp = make_employee(internal_number='101', internal_line_type='sip')
|
|
self.assertEqual(channel_for_employee(emp), 'PJSIP/101')
|
|
|
|
def test_analog_uses_dahdi(self):
|
|
emp = make_employee(internal_number='12', internal_line_type='analog')
|
|
self.assertEqual(channel_for_employee(emp), 'DAHDI/12')
|
|
|
|
def test_custom_template(self):
|
|
emp = make_employee(internal_number='55', internal_line_type='sip')
|
|
self.assertEqual(
|
|
channel_for_employee(emp, {'sip_channel': 'SIP/{exten}'}),
|
|
'SIP/55',
|
|
)
|
|
|
|
def test_bad_exten(self):
|
|
with self.assertRaises(TelephonyError):
|
|
sanitize_exten('ab')
|
|
|
|
|
|
class OriginateViewTests(TestCase):
|
|
def setUp(self):
|
|
self.caller = make_employee(
|
|
internal_number='101', internal_line_type='sip', username='ami.caller',
|
|
email='ami.caller@company.local',
|
|
)
|
|
self.callee = make_employee(
|
|
internal_number='202', internal_line_type='analog', username='ami.callee',
|
|
email='ami.callee@company.local',
|
|
)
|
|
ConnectorConfig.objects.create(
|
|
connector_type='asterisk',
|
|
is_enabled=True,
|
|
host='pbx.local',
|
|
port=5038,
|
|
username='acore',
|
|
password='secret',
|
|
extra_config={'originate_context': 'from-internal'},
|
|
)
|
|
self.client = Client()
|
|
login_2fa(self.client, self.caller)
|
|
|
|
def test_originate_posts_to_ami(self):
|
|
fake = mock.MagicMock()
|
|
fake.__enter__.return_value = fake
|
|
fake.__exit__.return_value = False
|
|
fake.originate.return_value = {'Response': 'Success'}
|
|
with mock.patch(
|
|
'apps.asterisk_connector.services._client_from_config',
|
|
return_value=fake,
|
|
):
|
|
response = self.client.post('/telephony/call/', {'employee_id': str(self.callee.pk)})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn('HX-Retarget', response)
|
|
fake.originate.assert_called_once()
|
|
kwargs = fake.originate.call_args.kwargs
|
|
self.assertEqual(kwargs['channel'], 'PJSIP/101')
|
|
self.assertEqual(kwargs['exten'], '202')
|
|
|
|
def test_disabled_without_internal(self):
|
|
bare = make_employee(username='ami.bare', email='ami.bare@company.local')
|
|
login_2fa(self.client, bare)
|
|
response = self.client.post('/telephony/call/', {'employee_id': str(self.callee.pk)})
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_login_failure_surfaces_message(self):
|
|
with mock.patch(
|
|
'apps.asterisk_connector.services._client_from_config',
|
|
side_effect=AmiError('Authentication failed'),
|
|
):
|
|
with self.assertRaises(TelephonyError):
|
|
originate_between(self.caller, self.callee)
|