• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

mozilla / fx-private-relay / 1b0c290a-0052-4fa6-9c59-ab9043d03a19

17 Mar 2025 04:52PM UTC coverage: 85.123% (-0.01%) from 85.137%
1b0c290a-0052-4fa6-9c59-ab9043d03a19

Pull #5439

circleci

vpremamozilla
changes made to baseload doc
Pull Request #5439: Update Base Load Engineer Doc

2434 of 3561 branches covered (68.35%)

Branch coverage included in aggregate %.

17048 of 19326 relevant lines covered (88.21%)

9.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.55
/phones/tests/models_tests.py
1
import random
1✔
2
from collections.abc import Iterator
1✔
3
from datetime import UTC, datetime, timedelta
1✔
4
from types import SimpleNamespace
1✔
5
from unittest.mock import Mock, call, patch
1✔
6
from uuid import uuid4
1✔
7

8
from django.conf import settings
1✔
9
from django.contrib.auth.models import User
1✔
10
from django.core.cache import cache
1✔
11
from django.core.exceptions import BadRequest, ValidationError
1✔
12
from django.test import override_settings
1✔
13

14
import pytest
1✔
15
import responses
1✔
16
from allauth.socialaccount.models import SocialAccount, SocialToken
1✔
17
from model_bakery import baker
1✔
18
from twilio.base.exceptions import TwilioRestException
1✔
19

20
from privaterelay.tests.utils import omit_markus_logs
1✔
21

22
if settings.PHONES_ENABLED:
1!
23
    from ..models import (
1✔
24
        InboundContact,
25
        RealPhone,
26
        RelayNumber,
27
        area_code_numbers,
28
        get_last_text_sender,
29
        iq_fmt,
30
        location_numbers,
31
        suggested_numbers,
32
    )
33

34

35
pytestmark = pytest.mark.skipif(
1✔
36
    not settings.PHONES_ENABLED, reason="PHONES_ENABLED is False"
37
)
38

39

40
@pytest.fixture(autouse=True)
1✔
41
def test_settings(settings):
1✔
42
    settings.TWILIO_MESSAGING_SERVICE_SID = [f"MG{uuid4().hex}"]
1✔
43
    return settings
1✔
44

45

46
@pytest.fixture
1✔
47
def twilio_number_sid():
1✔
48
    """A Twilio Incoming Number ID"""
49
    return f"PN{uuid4().hex}"
1✔
50

51

52
@pytest.fixture(autouse=True)
1✔
53
def mock_twilio_client(twilio_number_sid: str) -> Iterator[Mock]:
1✔
54
    """Mock PhonesConfig with a mock twilio client"""
55
    with patch(
1✔
56
        "phones.apps.PhonesConfig.twilio_client",
57
        spec_set=[
58
            "available_phone_numbers",
59
            "incoming_phone_numbers",
60
            "messages",
61
            "messaging",
62
        ],
63
    ) as mock_twilio_client:
64
        mock_twilio_client.available_phone_numbers = Mock(spec_set=[])
1✔
65
        mock_twilio_client.incoming_phone_numbers = Mock(spec_set=["create"])
1✔
66
        mock_twilio_client.incoming_phone_numbers.create = Mock(
1✔
67
            spec_set=[], return_value=SimpleNamespace(sid=twilio_number_sid)
68
        )
69
        mock_twilio_client.messages = Mock(spec_set=["create"])
1✔
70
        mock_twilio_client.messages.create = Mock(spec_set=[])
1✔
71
        mock_twilio_client.messaging = Mock(spec_set=["v1"])
1✔
72
        mock_twilio_client.messaging.v1 = Mock(spec_set=["services"])
1✔
73
        mock_twilio_client.messaging.v1.services = Mock(spec_set=[])
1✔
74
        yield mock_twilio_client
1✔
75

76

77
def make_phone_test_user() -> User:
1✔
78
    phone_user = baker.make(User, email="phone_user@example.com")
1✔
79
    phone_user.profile.date_subscribed = datetime.now(tz=UTC) - timedelta(days=15)
1✔
80
    phone_user.profile.save()
1✔
81
    upgrade_test_user_to_phone(phone_user)
1✔
82
    return phone_user
1✔
83

84

85
def upgrade_test_user_to_phone(user):
1✔
86
    random_sub = random.choice(settings.SUBSCRIPTIONS_WITH_PHONE)
1✔
87
    account: SocialAccount = baker.make(
1✔
88
        SocialAccount,
89
        user=user,
90
        provider="fxa",
91
        uid=str(uuid4()).replace("-", ""),
92
        extra_data={"avatar": "avatar.png", "subscriptions": [random_sub]},
93
    )
94
    baker.make(
1✔
95
        SocialToken,
96
        account=account,
97
        expires_at=datetime.now(UTC) + timedelta(1),
98
    )
99
    return user
1✔
100

101

102
@pytest.fixture(autouse=True)
1✔
103
def phone_user(db):
1✔
104
    return make_phone_test_user()
1✔
105

106

107
@pytest.fixture
1✔
108
def django_cache():
1✔
109
    """Return a cleared Django cache as a fixture."""
110
    cache.clear()
1✔
111
    yield cache
1✔
112
    cache.clear()
1✔
113

114

115
def test_realphone_pending_objects_includes_new(phone_user):
1✔
116
    number = "+12223334444"
1✔
117
    real_phone = RealPhone.objects.create(
1✔
118
        user=phone_user,
119
        number=number,
120
        verification_sent_date=datetime.now(UTC),
121
    )
122
    assert RealPhone.pending_objects.exists_for_number(number)
1✔
123
    recent_phone = RealPhone.recent_objects.get_for_user_number_and_verification_code(
1✔
124
        phone_user, number, real_phone.verification_code
125
    )
126
    assert recent_phone.id == real_phone.id
1✔
127

128

129
def test_realphone_pending_objects_excludes_old(phone_user):
1✔
130
    number = "+12223334444"
1✔
131
    real_phone = RealPhone.objects.create(
1✔
132
        user=phone_user,
133
        number=number,
134
        verification_sent_date=(
135
            datetime.now(UTC)
136
            - timedelta(0, 60 * settings.MAX_MINUTES_TO_VERIFY_REAL_PHONE + 1)
137
        ),
138
    )
139
    assert not RealPhone.pending_objects.exists_for_number(number)
1✔
140
    with pytest.raises(RealPhone.DoesNotExist):
1✔
141
        RealPhone.recent_objects.get_for_user_number_and_verification_code(
1✔
142
            phone_user, number, real_phone.verification_code
143
        )
144

145

146
def test_create_realphone_creates_twilio_message(phone_user, mock_twilio_client):
1✔
147
    number = "+12223334444"
1✔
148
    RealPhone.objects.create(user=phone_user, verified=True, number=number)
1✔
149
    mock_twilio_client.messages.create.assert_called_once()
1✔
150
    call_kwargs = mock_twilio_client.messages.create.call_args.kwargs
1✔
151
    assert call_kwargs["to"] == number
1✔
152
    assert "verification code" in call_kwargs["body"]
1✔
153

154

155
@override_settings(IQ_FOR_VERIFICATION=True)
1✔
156
@responses.activate
1✔
157
@pytest.mark.skipif(not settings.IQ_ENABLED, reason="IQ_ENABLED is false")
1✔
158
def test_create_realphone_creates_iq_message(phone_user):
1✔
159
    number = "+12223334444"
×
160
    iq_number = iq_fmt(number)
×
161
    resp = responses.add(
×
162
        responses.POST,
163
        settings.IQ_PUBLISH_MESSAGE_URL,
164
        status=200,
165
        match=[
166
            responses.matchers.json_params_matcher(
167
                {
168
                    "to": [iq_number],
169
                    "from": settings.IQ_MAIN_NUMBER,
170
                },
171
                strict_match=False,
172
            )
173
        ],
174
    )
175

176
    RealPhone.objects.create(user=phone_user, verified=True, number=number)
×
177

178
    assert resp.call_count == 1
×
179

180

181
def test_create_second_realphone_for_user_raises_exception(
1✔
182
    phone_user, mock_twilio_client
183
):
184
    RealPhone.objects.create(user=phone_user, verified=True, number="+12223334444")
1✔
185
    mock_twilio_client.messages.create.assert_called_once()
1✔
186
    mock_twilio_client.reset_mock()
1✔
187

188
    with pytest.raises(BadRequest):
1✔
189
        RealPhone.objects.create(user=phone_user, number="+12223335555")
1✔
190
    mock_twilio_client.messages.assert_not_called()
1✔
191

192

193
def test_create_realphone_deletes_expired_unverified_records(
1✔
194
    phone_user, mock_twilio_client
195
):
196
    # create an expired unverified record
197
    number = "+12223334444"
1✔
198
    RealPhone.objects.create(
1✔
199
        user=phone_user,
200
        number=number,
201
        verified=False,
202
        verification_sent_date=(
203
            datetime.now(UTC)
204
            - timedelta(0, 60 * settings.MAX_MINUTES_TO_VERIFY_REAL_PHONE + 1)
205
        ),
206
    )
207
    expired_verification_records = RealPhone.expired_objects.filter(number=number)
1✔
208
    assert len(expired_verification_records) >= 1
1✔
209
    mock_twilio_client.messages.create.assert_called_once()
1✔
210

211
    # now try to create the new record
212
    RealPhone.objects.create(user=baker.make(User), number=number)
1✔
213
    expired_verification_records = RealPhone.expired_objects.filter(number=number)
1✔
214
    assert len(expired_verification_records) == 0
1✔
215
    mock_twilio_client.messages.create.assert_called()
1✔
216

217

218
def test_mark_realphone_verified_sets_verified_and_date(phone_user):
1✔
219
    real_phone = RealPhone.objects.create(user=phone_user, verified=False)
1✔
220
    real_phone.mark_verified()
1✔
221
    assert real_phone.verified
1✔
222
    assert real_phone.verified_date
1✔
223

224

225
def test_create_relaynumber_without_realphone_raises_error(
1✔
226
    phone_user, mock_twilio_client
227
):
228
    with pytest.raises(ValidationError) as exc_info:
1✔
229
        RelayNumber.objects.create(user=phone_user, number="+19998887777")
1✔
230
    assert exc_info.value.message == "User does not have a verified real phone."
1✔
231
    mock_twilio_client.messages.create.assert_not_called()
1✔
232
    mock_twilio_client.incoming_phone_numbers.create.assert_not_called()
1✔
233

234

235
def test_create_relaynumber_when_user_already_has_one_raises_error(
1✔
236
    phone_user, mock_twilio_client
237
):
238
    mock_messages_create = mock_twilio_client.messages.create
1✔
239
    mock_number_create = mock_twilio_client.incoming_phone_numbers.create
1✔
240

241
    real_phone = "+12223334444"
1✔
242
    RealPhone.objects.create(user=phone_user, verified=True, number=real_phone)
1✔
243
    mock_messages_create.assert_called_once()
1✔
244
    mock_messages_create.reset_mock()
1✔
245

246
    relay_number = "+19998887777"
1✔
247
    relay_number_obj = RelayNumber.objects.create(user=phone_user, number=relay_number)
1✔
248

249
    mock_number_create.assert_called_once()
1✔
250
    call_kwargs = mock_number_create.call_args.kwargs
1✔
251
    assert call_kwargs["phone_number"] == relay_number
1✔
252
    assert call_kwargs["sms_application_sid"] == settings.TWILIO_SMS_APPLICATION_SID
1✔
253
    assert call_kwargs["voice_application_sid"] == settings.TWILIO_SMS_APPLICATION_SID
1✔
254

255
    mock_messages_create.assert_called_once()
1✔
256
    call_kwargs = mock_messages_create.call_args.kwargs
1✔
257
    assert "Welcome" in call_kwargs["body"]
1✔
258
    assert call_kwargs["to"] == real_phone
1✔
259
    assert relay_number_obj.vcard_lookup_key in call_kwargs["media_url"][0]
1✔
260

261
    mock_number_create.reset_mock()
1✔
262
    mock_messages_create.reset_mock()
1✔
263
    second_relay_number = "+14445556666"
1✔
264
    with pytest.raises(ValidationError) as exc_info:
1✔
265
        RelayNumber.objects.create(user=phone_user, number=second_relay_number)
1✔
266
    assert exc_info.value.message == "User can have only one relay number."
1✔
267
    mock_number_create.assert_not_called()
1✔
268
    mock_messages_create.assert_not_called()
1✔
269

270
    # Creating RelayNumber with same number is also an error
271
    with pytest.raises(ValidationError) as exc_info:
1✔
272
        RelayNumber.objects.create(user=phone_user, number=relay_number)
1✔
273
    assert exc_info.value.message == "User can have only one relay number."
1✔
274
    mock_number_create.assert_not_called()
1✔
275
    mock_messages_create.assert_not_called()
1✔
276

277

278
def test_create_duplicate_relaynumber_raises_error(phone_user, mock_twilio_client):
1✔
279
    mock_messages_create = mock_twilio_client.messages.create
1✔
280
    mock_number_create = mock_twilio_client.incoming_phone_numbers.create
1✔
281

282
    real_phone = "+12223334444"
1✔
283
    RealPhone.objects.create(user=phone_user, verified=True, number=real_phone)
1✔
284
    mock_messages_create.assert_called_once()
1✔
285
    mock_messages_create.reset_mock()
1✔
286

287
    relay_number = "+19998887777"
1✔
288
    RelayNumber.objects.create(user=phone_user, number=relay_number)
1✔
289

290
    mock_number_create.assert_called_once()
1✔
291
    call_kwargs = mock_number_create.call_args.kwargs
1✔
292
    assert call_kwargs["phone_number"] == relay_number
1✔
293
    assert call_kwargs["sms_application_sid"] == settings.TWILIO_SMS_APPLICATION_SID
1✔
294
    assert call_kwargs["voice_application_sid"] == settings.TWILIO_SMS_APPLICATION_SID
1✔
295

296
    mock_messages_create.assert_called_once()
1✔
297
    mock_number_create.reset_mock()
1✔
298
    mock_messages_create.reset_mock()
1✔
299

300
    second_user = make_phone_test_user()
1✔
301
    second_phone = "+15553334444"
1✔
302
    RealPhone.objects.create(user=second_user, verified=True, number=second_phone)
1✔
303
    mock_messages_create.assert_called_once()
1✔
304
    mock_messages_create.reset_mock()
1✔
305

306
    with pytest.raises(ValidationError) as exc_info:
1✔
307
        RelayNumber.objects.create(user=second_user, number=relay_number)
1✔
308
    assert exc_info.value.message == "This number is already claimed."
1✔
309
    mock_number_create.assert_not_called()
1✔
310
    mock_messages_create.assert_not_called()
1✔
311

312

313
@pytest.fixture
1✔
314
def real_phone_us(phone_user, mock_twilio_client):
1✔
315
    """Create a US-based RealPhone for phone_user, with a reset twilio_client."""
316
    real_phone = RealPhone.objects.create(
1✔
317
        user=phone_user,
318
        number="+12223334444",
319
        verified=True,
320
        verification_sent_date=datetime.now(UTC),
321
    )
322
    mock_twilio_client.messages.create.assert_called_once()
1✔
323
    mock_twilio_client.messages.create.reset_mock()
1✔
324
    return real_phone
1✔
325

326

327
def test_create_relaynumber_creates_twilio_incoming_number_and_sends_welcome(
1✔
328
    phone_user, real_phone_us, mock_twilio_client, settings, twilio_number_sid
329
):
330
    """A successful relay phone creation sends a welcome message."""
331
    relay_number = "+19998887777"
1✔
332
    relay_number_obj = RelayNumber.objects.create(user=phone_user, number=relay_number)
1✔
333

334
    mock_twilio_client.incoming_phone_numbers.create.assert_called_once_with(
1✔
335
        phone_number=relay_number,
336
        sms_application_sid=settings.TWILIO_SMS_APPLICATION_SID,
337
        voice_application_sid=settings.TWILIO_SMS_APPLICATION_SID,
338
    )
339
    mock_services = mock_twilio_client.messaging.v1.services
1✔
340
    mock_services.assert_called_once_with(settings.TWILIO_MESSAGING_SERVICE_SID[0])
1✔
341
    mock_services.return_value.phone_numbers.create.assert_called_once_with(
1✔
342
        phone_number_sid=twilio_number_sid
343
    )
344

345
    mock_messages_create = mock_twilio_client.messages.create
1✔
346
    mock_messages_create.assert_called_once()
1✔
347
    call_kwargs = mock_messages_create.call_args.kwargs
1✔
348
    assert "Welcome" in call_kwargs["body"]
1✔
349
    assert call_kwargs["to"] == real_phone_us.number
1✔
350
    assert relay_number_obj.vcard_lookup_key in call_kwargs["media_url"][0]
1✔
351

352

353
def test_create_relaynumber_with_two_real_numbers(
1✔
354
    phone_user, mock_twilio_client, settings, twilio_number_sid
355
):
356
    """A user with a second unverified RealPhone is OK."""
357
    RealPhone.objects.create(user=phone_user, number="+12223334444", verified=False)
1✔
358
    phone2 = RealPhone.objects.create(
1✔
359
        user=phone_user, number="+12223335555", verified=False
360
    )
361
    phone2.mark_verified()
1✔
362
    mock_twilio_client.reset_mock()
1✔
363

364
    relay_number = "+19998887777"
1✔
365
    relay_number_obj = RelayNumber.objects.create(user=phone_user, number=relay_number)
1✔
366

367
    mock_twilio_client.incoming_phone_numbers.create.assert_called_once_with(
1✔
368
        phone_number=relay_number,
369
        sms_application_sid=settings.TWILIO_SMS_APPLICATION_SID,
370
        voice_application_sid=settings.TWILIO_SMS_APPLICATION_SID,
371
    )
372
    mock_services = mock_twilio_client.messaging.v1.services
1✔
373
    mock_services.assert_called_once_with(settings.TWILIO_MESSAGING_SERVICE_SID[0])
1✔
374
    mock_services.return_value.phone_numbers.create.assert_called_once_with(
1✔
375
        phone_number_sid=twilio_number_sid
376
    )
377

378
    mock_messages_create = mock_twilio_client.messages.create
1✔
379
    mock_messages_create.assert_called_once()
1✔
380
    call_kwargs = mock_messages_create.call_args.kwargs
1✔
381
    assert "Welcome" in call_kwargs["body"]
1✔
382
    assert call_kwargs["to"] == phone2.number
1✔
383
    assert relay_number_obj.vcard_lookup_key in call_kwargs["media_url"][0]
1✔
384

385

386
def test_create_relaynumber_already_registered_with_service(
1✔
387
    phone_user, real_phone_us, mock_twilio_client, caplog, settings, twilio_number_sid
388
):
389
    """
390
    It is OK if the relay phone is already registered with a messaging service.
391

392
    This is not likely in production, since relay phone acquisition and registration
393
    is a single step, but can happen when manually moving relay phones between users.
394
    """
395
    twilio_service_sid = settings.TWILIO_MESSAGING_SERVICE_SID[0]
1✔
396

397
    # Twilio responds that the phone number is already registered
398
    mock_services = mock_twilio_client.messaging.v1.services
1✔
399
    mock_messaging_number_create = mock_services.return_value.phone_numbers.create
1✔
400
    mock_messaging_number_create.side_effect = TwilioRestException(
1✔
401
        uri=f"/Services/{twilio_service_sid}/PhoneNumbers",
402
        msg=(
403
            "Unable to create record:"
404
            " Phone Number or Short Code is already in the Messaging Service."
405
        ),
406
        method="POST",
407
        status=409,
408
        code=21710,
409
    )
410

411
    # Does not raise exception
412
    relay_number = "+19998887777"
1✔
413
    RelayNumber.objects.create(user=phone_user, number=relay_number)
1✔
414

415
    mock_twilio_client.incoming_phone_numbers.create.assert_called_once_with(
1✔
416
        phone_number=relay_number,
417
        sms_application_sid=settings.TWILIO_SMS_APPLICATION_SID,
418
        voice_application_sid=settings.TWILIO_SMS_APPLICATION_SID,
419
    )
420
    mock_services.assert_called_once_with(twilio_service_sid)
1✔
421
    mock_messaging_number_create.assert_called_once_with(
1✔
422
        phone_number_sid=twilio_number_sid
423
    )
424
    mock_twilio_client.messages.create.assert_called_once()
1✔
425
    records = omit_markus_logs(caplog)
1✔
426
    assert len(records) == 1
1✔
427
    record = records[0]
1✔
428
    assert record.msg == "twilio_messaging_service"
1✔
429
    assert getattr(record, "code") == 21710
1✔
430

431

432
def test_create_relaynumber_fail_if_all_services_are_full(
1✔
433
    phone_user, real_phone_us, mock_twilio_client, settings, caplog, twilio_number_sid
434
):
435
    """If the Twilio Messaging Service pool is full, an exception is raised."""
436
    twilio_service_sid = settings.TWILIO_MESSAGING_SERVICE_SID[0]
1✔
437

438
    # Twilio responds that the pool is full
439
    mock_services = mock_twilio_client.messaging.v1.services
1✔
440
    mock_messaging_number_create = mock_services.return_value.phone_numbers.create
1✔
441
    mock_messaging_number_create.side_effect = TwilioRestException(
1✔
442
        uri=f"/Services/{twilio_service_sid}/PhoneNumbers",
443
        msg=("Unable to create record: Number Pool size limit reached"),
444
        method="POST",
445
        status=412,
446
        code=21714,
447
    )
448

449
    # "Pool full" exception is raised
450
    with pytest.raises(Exception) as exc_info:
1✔
451
        RelayNumber.objects.create(user=phone_user, number="+19998887777")
1✔
452
    assert (
1✔
453
        str(exc_info.value) == "All services in TWILIO_MESSAGING_SERVICE_SID are full"
454
    )
455

456
    mock_messaging_number_create.assert_called_once_with(
1✔
457
        phone_number_sid=twilio_number_sid
458
    )
459
    mock_twilio_client.messages.create.assert_not_called()
1✔
460
    records = omit_markus_logs(caplog)
1✔
461
    assert len(records) == 1
1✔
462
    record = records[0]
1✔
463
    assert record.msg == "twilio_messaging_service"
1✔
464
    assert getattr(record, "code") == 21714
1✔
465

466

467
def test_create_relaynumber_no_service(
1✔
468
    phone_user, real_phone_us, mock_twilio_client, settings, caplog
469
):
470
    """If no Twilio Messaging Service IDs are defined, registration is skipped."""
471
    settings.TWILIO_MESSAGING_SERVICE_SID = []
1✔
472

473
    RelayNumber.objects.create(user=phone_user, number="+19998887777")
1✔
474

475
    mock_services = mock_twilio_client.messaging.v1.services
1✔
476
    mock_services.return_value.phone_numbers.create.assert_not_called()
1✔
477
    mock_twilio_client.messages.create.assert_called_once()
1✔
478
    records = omit_markus_logs(caplog)
1✔
479
    assert len(records) == 1
1✔
480
    record = records[0]
1✔
481
    assert record.msg == (
1✔
482
        "Skipping Twilio Messaging Service registration, since"
483
        " TWILIO_MESSAGING_SERVICE_SID is empty."
484
    )
485

486

487
def test_create_relaynumber_fallback_to_second_service(
1✔
488
    phone_user,
489
    real_phone_us,
490
    mock_twilio_client,
491
    settings,
492
    django_cache,
493
    caplog,
494
    twilio_number_sid,
495
):
496
    """The fallback messaging pool if the first is full."""
497
    twilio_service1_sid = f"MG{uuid4().hex}"
1✔
498
    twilio_service2_sid = f"MG{uuid4().hex}"
1✔
499
    settings.TWILIO_MESSAGING_SERVICE_SID = [twilio_service1_sid, twilio_service2_sid]
1✔
500
    django_cache.set("twilio_messaging_service_closed", "")
1✔
501

502
    # Twilio responds that pool 1 is full, pool 2 is OK
503
    mock_services = mock_twilio_client.messaging.v1.services
1✔
504
    mock_messaging_number_create = mock_services.return_value.phone_numbers.create
1✔
505
    mock_messaging_number_create.side_effect = [
1✔
506
        TwilioRestException(
507
            uri=f"/Services/{twilio_service1_sid}/PhoneNumbers",
508
            msg=("Unable to create record: Number Pool size limit reached"),
509
            method="POST",
510
            status=412,
511
            code=21714,
512
        ),
513
        None,
514
    ]
515

516
    RelayNumber.objects.create(user=phone_user, number="+19998887777")
1✔
517

518
    mock_services.assert_has_calls(
1✔
519
        [
520
            call(twilio_service1_sid),
521
            call().phone_numbers.create(phone_number_sid=twilio_number_sid),
522
            call(twilio_service2_sid),
523
            call().phone_numbers.create(phone_number_sid=twilio_number_sid),
524
        ]
525
    )
526
    mock_twilio_client.messages.create.assert_called_once()
1✔
527

528
    assert django_cache.get("twilio_messaging_service_closed") == twilio_service1_sid
1✔
529
    records = omit_markus_logs(caplog)
1✔
530
    assert len(records) == 1
1✔
531
    record = records[0]
1✔
532
    assert record.msg == "twilio_messaging_service"
1✔
533
    assert getattr(record, "code") == 21714
1✔
534

535

536
def test_create_relaynumber_skip_known_full_service(
1✔
537
    phone_user,
538
    real_phone_us,
539
    mock_twilio_client,
540
    settings,
541
    django_cache,
542
    caplog,
543
    twilio_number_sid,
544
):
545
    """If a pool has been marked as full, it is skipped."""
546
    twilio_service1_sid = f"MG{uuid4().hex}"
1✔
547
    twilio_service2_sid = f"MG{uuid4().hex}"
1✔
548
    settings.TWILIO_MESSAGING_SERVICE_SID = [twilio_service1_sid, twilio_service2_sid]
1✔
549
    django_cache.set("twilio_messaging_service_closed", twilio_service1_sid)
1✔
550

551
    RelayNumber.objects.create(user=phone_user, number="+19998887777")
1✔
552

553
    mock_services = mock_twilio_client.messaging.v1.services
1✔
554
    mock_services.assert_called_once_with(twilio_service2_sid)
1✔
555
    mock_services.return_value.phone_numbers.create.assert_called_once_with(
1✔
556
        phone_number_sid=twilio_number_sid
557
    )
558
    mock_twilio_client.messages.create.assert_called_once()
1✔
559
    assert django_cache.get("twilio_messaging_service_closed") == twilio_service1_sid
1✔
560
    assert len(omit_markus_logs(caplog)) == 0
1✔
561

562

563
def test_create_relaynumber_other_messaging_error_raised(
1✔
564
    phone_user,
565
    real_phone_us,
566
    mock_twilio_client,
567
    settings,
568
    caplog,
569
    django_cache,
570
    twilio_number_sid,
571
):
572
    """If adding to a pool raises a different error, it is skipped."""
573
    twilio_service_sid = settings.TWILIO_MESSAGING_SERVICE_SID[0]
1✔
574

575
    # Twilio responds that pool 1 is full, pool 2 is OK
576
    mock_services = mock_twilio_client.messaging.v1.services
1✔
577
    mock_messaging_number_create = mock_services.return_value.phone_numbers.create
1✔
578
    mock_messaging_number_create.side_effect = TwilioRestException(
1✔
579
        uri=f"/Services/{twilio_service_sid}/PhoneNumbers",
580
        msg=(
581
            "Unable to create record:"
582
            " Phone Number is associated with another Messaging Service"
583
        ),
584
        method="POST",
585
        status=409,
586
        code=21712,
587
    )
588

589
    with pytest.raises(TwilioRestException):
1✔
590
        RelayNumber.objects.create(user=phone_user, number="+19998887777")
1✔
591

592
    mock_services.assert_called_once_with(twilio_service_sid)
1✔
593
    mock_messaging_number_create.assert_called_once_with(
1✔
594
        phone_number_sid=twilio_number_sid
595
    )
596
    mock_twilio_client.messages.create.assert_not_called()
1✔
597
    assert django_cache.get("twilio_messaging_service_closed") is None
1✔
598
    assert caplog.messages == ["twilio_messaging_service"]
1✔
599
    assert caplog.records[0].code == 21712
1✔
600

601

602
@pytest.fixture
1✔
603
def real_phone_ca(phone_user, mock_twilio_client):
1✔
604
    """Create a CA-based RealPhone for phone_user, with a reset twilio_client."""
605
    real_phone = RealPhone.objects.create(
1✔
606
        user=phone_user,
607
        number="+14035551234",
608
        verified=True,
609
        verification_sent_date=datetime.now(UTC),
610
        country_code="CA",
611
    )
612
    mock_twilio_client.messages.create.assert_called_once()
1✔
613
    mock_twilio_client.messages.create.reset_mock()
1✔
614
    return real_phone
1✔
615

616

617
def test_create_relaynumber_canada(
1✔
618
    phone_user, real_phone_ca, mock_twilio_client, twilio_number_sid
619
):
620
    relay_number = "+17805551234"
1✔
621
    relay_number_obj = RelayNumber.objects.create(user=phone_user, number=relay_number)
1✔
622
    assert relay_number_obj.country_code == "CA"
1✔
623

624
    mock_number_create = mock_twilio_client.incoming_phone_numbers.create
1✔
625
    mock_number_create.assert_called_once()
1✔
626
    call_kwargs = mock_number_create.call_args.kwargs
1✔
627
    assert call_kwargs["phone_number"] == relay_number
1✔
628
    assert call_kwargs["sms_application_sid"] == settings.TWILIO_SMS_APPLICATION_SID
1✔
629
    assert call_kwargs["voice_application_sid"] == settings.TWILIO_SMS_APPLICATION_SID
1✔
630

631
    # Omit Canadian numbers for US A2P 10DLC messaging service
632
    mock_twilio_client.messaging.v1.services.assert_not_called()
1✔
633

634
    # A welcome message is sent
635
    mock_messages_create = mock_twilio_client.messages.create
1✔
636
    mock_messages_create.assert_called_once()
1✔
637
    call_kwargs = mock_messages_create.call_args.kwargs
1✔
638
    assert "Welcome" in call_kwargs["body"]
1✔
639
    assert call_kwargs["to"] == real_phone_ca.number
1✔
640
    assert relay_number_obj.vcard_lookup_key in call_kwargs["media_url"][0]
1✔
641

642

643
def test_relaynumber_remaining_minutes_returns_properly_formats_remaining_seconds(
1✔
644
    phone_user, real_phone_us, mock_twilio_client
645
):
646
    relay_number = "+13045551234"
1✔
647
    relay_number_obj = RelayNumber.objects.create(user=phone_user, number=relay_number)
1✔
648

649
    # Freshly created RelayNumber should have 3000 seconds => 50 minutes
650
    assert relay_number_obj.remaining_minutes == 50
1✔
651

652
    # After receiving calls remaining_minutes property should return the rounded down
653
    # to a positive integer
654
    relay_number_obj.remaining_seconds = 522
1✔
655
    relay_number_obj.save()
1✔
656
    assert relay_number_obj.remaining_minutes == 8
1✔
657

658
    # If more call time is spent than allotted (negative remaining_seconds),
659
    # the remaining_minutes property should return zero
660
    relay_number_obj.remaining_seconds = -522
1✔
661
    relay_number_obj.save()
1✔
662
    assert relay_number_obj.remaining_minutes == 0
1✔
663

664

665
def test_suggested_numbers_bad_request_for_user_without_real_phone(
1✔
666
    phone_user, mock_twilio_client
667
):
668
    with pytest.raises(BadRequest):
1✔
669
        suggested_numbers(phone_user)
1✔
670
    mock_twilio_client.available_phone_numbers.assert_not_called()
1✔
671

672

673
def test_suggested_numbers_bad_request_for_user_who_already_has_number(
1✔
674
    phone_user, real_phone_us, mock_twilio_client
675
):
676
    RelayNumber.objects.create(user=phone_user, number="+19998887777")
1✔
677
    with pytest.raises(BadRequest):
1✔
678
        suggested_numbers(phone_user)
1✔
679
    mock_twilio_client.available_phone_numbers.assert_not_called()
1✔
680

681

682
def test_suggested_numbers(phone_user, real_phone_us, mock_twilio_client):
1✔
683
    mock_list = Mock(return_value=[Mock() for i in range(5)])
1✔
684
    mock_twilio_client.available_phone_numbers = Mock(
1✔
685
        return_value=Mock(local=Mock(list=mock_list))
686
    )
687

688
    suggested_numbers(phone_user)
1✔
689
    available_numbers_calls = mock_twilio_client.available_phone_numbers.call_args_list
1✔
690
    assert available_numbers_calls == [call("US")]
1✔
691
    assert mock_list.call_args_list == [
1✔
692
        call(contains="+1222333****", limit=10),
693
        call(contains="+122233***44", limit=10),
694
        call(contains="+12223******", limit=10),
695
        call(contains="+1***3334444", limit=10),
696
        call(contains="+1222*******", limit=10),
697
        call(limit=10),
698
    ]
699

700

701
def test_suggested_numbers_ca(phone_user, mock_twilio_client):
1✔
702
    real_phone = "+14035551234"
1✔
703
    RealPhone.objects.create(
1✔
704
        user=phone_user, verified=True, number=real_phone, country_code="CA"
705
    )
706
    mock_list = Mock(return_value=[Mock() for i in range(5)])
1✔
707
    mock_twilio_client.available_phone_numbers = Mock(
1✔
708
        return_value=Mock(local=Mock(list=mock_list))
709
    )
710

711
    suggested_numbers(phone_user)
1✔
712
    available_numbers_calls = mock_twilio_client.available_phone_numbers.call_args_list
1✔
713
    assert available_numbers_calls == [call("CA")]
1✔
714
    assert mock_list.call_args_list == [
1✔
715
        call(contains="+1403555****", limit=10),
716
        call(contains="+140355***34", limit=10),
717
        call(contains="+14035******", limit=10),
718
        call(contains="+1***5551234", limit=10),
719
        call(contains="+1403*******", limit=10),
720
        call(limit=10),
721
    ]
722

723

724
def test_location_numbers(mock_twilio_client):
1✔
725
    mock_list = Mock(return_value=[Mock() for i in range(5)])
1✔
726
    mock_twilio_client.available_phone_numbers = Mock(
1✔
727
        return_value=(Mock(local=Mock(list=mock_list)))
728
    )
729

730
    location_numbers("Miami, FL")
1✔
731

732
    available_numbers_calls = mock_twilio_client.available_phone_numbers.call_args_list
1✔
733
    assert available_numbers_calls == [call("US")]
1✔
734
    assert mock_list.call_args_list == [call(in_locality="Miami, FL", limit=10)]
1✔
735

736

737
def test_area_code_numbers(mock_twilio_client):
1✔
738
    mock_list = Mock(return_value=[Mock() for i in range(5)])
1✔
739
    mock_twilio_client.available_phone_numbers = Mock(
1✔
740
        return_value=(Mock(local=Mock(list=mock_list)))
741
    )
742

743
    area_code_numbers("918")
1✔
744

745
    available_numbers_calls = mock_twilio_client.available_phone_numbers.call_args_list
1✔
746
    assert available_numbers_calls == [call("US")]
1✔
747
    assert mock_list.call_args_list == [call(area_code="918", limit=10)]
1✔
748

749

750
def test_save_store_phone_log_no_relay_number_does_nothing() -> None:
1✔
751
    user = make_phone_test_user()
1✔
752
    user.profile.store_phone_log = True
1✔
753
    user.profile.save()
1✔
754

755
    user.profile.refresh_from_db()
1✔
756
    assert user.profile.store_phone_log
1✔
757

758
    user.profile.store_phone_log = False
1✔
759
    user.profile.save()
1✔
760
    assert not user.profile.store_phone_log
1✔
761

762

763
def test_save_store_phone_log_true_doesnt_delete_data() -> None:
1✔
764
    user = make_phone_test_user()
1✔
765
    baker.make(RealPhone, user=user, verified=True)
1✔
766
    relay_number = baker.make(RelayNumber, user=user)
1✔
767
    inbound_contact = baker.make(InboundContact, relay_number=relay_number)
1✔
768
    user.profile.store_phone_log = True
1✔
769
    user.profile.save()
1✔
770

771
    inbound_contact.refresh_from_db()
1✔
772
    assert inbound_contact
1✔
773

774

775
def test_save_store_phone_log_false_deletes_data() -> None:
1✔
776
    user = make_phone_test_user()
1✔
777
    baker.make(RealPhone, user=user, verified=True)
1✔
778
    relay_number = baker.make(RelayNumber, user=user)
1✔
779
    inbound_contact = baker.make(InboundContact, relay_number=relay_number)
1✔
780
    user.profile.store_phone_log = False
1✔
781
    user.profile.save()
1✔
782

783
    with pytest.raises(InboundContact.DoesNotExist):
1✔
784
        inbound_contact.refresh_from_db()
1✔
785

786

787
def test_get_last_text_sender_returning_None():
1✔
788
    user = make_phone_test_user()
1✔
789
    baker.make(RealPhone, user=user, verified=True)
1✔
790
    relay_number = baker.make(RelayNumber, user=user)
1✔
791

792
    assert get_last_text_sender(relay_number) is None
1✔
793

794

795
def test_get_last_text_sender_returning_one():
1✔
796
    user = make_phone_test_user()
1✔
797
    baker.make(RealPhone, user=user, verified=True)
1✔
798
    relay_number = baker.make(RelayNumber, user=user)
1✔
799
    inbound_contact = baker.make(
1✔
800
        InboundContact, relay_number=relay_number, last_inbound_type="text"
801
    )
802

803
    assert get_last_text_sender(relay_number) == inbound_contact
1✔
804

805

806
def test_get_last_text_sender_lots_of_inbound_returns_one():
1✔
807
    user = make_phone_test_user()
1✔
808
    baker.make(RealPhone, user=user, verified=True)
1✔
809
    relay_number = baker.make(RelayNumber, user=user)
1✔
810
    baker.make(
1✔
811
        InboundContact,
812
        relay_number=relay_number,
813
        last_inbound_type="call",
814
        last_inbound_date=datetime.now(UTC) - timedelta(days=4),
815
    )
816
    baker.make(
1✔
817
        InboundContact,
818
        relay_number=relay_number,
819
        last_inbound_type="text",
820
        last_inbound_date=datetime.now(UTC) - timedelta(days=3),
821
    )
822
    baker.make(
1✔
823
        InboundContact,
824
        relay_number=relay_number,
825
        last_inbound_type="call",
826
        last_inbound_date=datetime.now(UTC) - timedelta(days=2),
827
    )
828
    baker.make(
1✔
829
        InboundContact,
830
        relay_number=relay_number,
831
        last_inbound_type="text",
832
        last_inbound_date=datetime.now(UTC) - timedelta(days=1),
833
    )
834
    inbound_contact = baker.make(
1✔
835
        InboundContact,
836
        relay_number=relay_number,
837
        last_inbound_type="text",
838
        last_inbound_date=datetime.now(UTC),
839
    )
840

841
    assert get_last_text_sender(relay_number) == inbound_contact
1✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc