b06e8ab5-baf6-4a94-ad8f-b818825c1988

Committed 21 Jun 2024 01:53PM UTC coverage: 85.369% (+0.01%) from 85.357%

Build # b06e8ab5-baf6-4a94-ad8f-b818825c1988

Build Type

push

circleci

Committed by

web-flow

Commit Message

Merge pull request #4801 from mozilla/api-client-mpp-3827

MPP-3827: Switch from `RequestsClient` to `APIClient`, remove transaction tests

Run Details

3982 of 5113 branches covered (77.88%)

Branch coverage included in aggregate %.

11 of 27 new or added lines in 2 files covered. (40.74%)

1 existing line in 1 file now uncovered.

15739 of 17988 relevant lines covered (87.5%)

10.32 hits per line

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

36.04

/api/tests/iq_views_tests.py

from unittest.mock import Mock, patch

from django.conf import settings

import pytest
import responses
from allauth.socialaccount.models import SocialAccount
from rest_framework.test import APIClient
from twilio.rest import Client

if settings.PHONES_ENABLED:
    from api.views.phones import compute_iq_mac
    from phones.models import InboundContact, iq_fmt

from api.tests.phones_views_tests import _make_real_phone, _make_relay_number
from phones.tests.models_tests import make_phone_test_user

pytestmark = pytest.mark.skipif(
    not settings.PHONES_ENABLED, reason="PHONES_ENABLED is False"
)
pytestmark = pytest.mark.skipif(not settings.IQ_ENABLED, reason="IQ_ENABLED is False")

INBOUND_SMS_PATH = "/api/v1/inbound_sms_iq/"


@pytest.fixture()
def phone_user(db):
    return make_phone_test_user()


@pytest.fixture(autouse=True)
def mocked_twilio_client():
    """
    Mock PhonesConfig with a mock twilio client
    """
    with patch(
        "phones.apps.PhonesConfig.twilio_client", spec_set=Client
    ) as mock_twilio_client:
        mock_fetch = Mock(
            return_value=Mock(
                country_code="US", phone_number="+12223334444", carrier="verizon"
            )
        )
        mock_twilio_client.lookups.v1.phone_numbers = Mock(
            return_value=Mock(fetch=mock_fetch)
        )
        yield mock_twilio_client


def _make_real_phone_with_mock_iq(phone_user, **kwargs):
    responses.add(responses.POST, settings.IQ_PUBLISH_MESSAGE_URL, status=200)
    real_phone = _make_real_phone(phone_user, **kwargs)
    responses.reset()
    return real_phone


@pytest.mark.django_db
def test_iq_endpoint_missing_verificationtoken_header():
    client = APIClient()
    response = client.post(INBOUND_SMS_PATH)
    assert response.status_code == 401
    response_body = response.json()
    assert "missing Verificationtoken header" in response_body["detail"]


@pytest.mark.django_db
def test_iq_endpoint_missing_messageid_header():
    client = APIClient()
    response = client.post(INBOUND_SMS_PATH, headers={"Verificationtoken": "valid"})

    assert response.status_code == 401
    response_body = response.json()
    assert "missing MessageId header" in response_body["detail"]


@pytest.mark.django_db
def test_iq_endpoint_invalid_hash():
    message_id = "9a09df23-01f3-4e0f-adbc-2a783878a574"
    client = APIClient()
    response = client.post(
        INBOUND_SMS_PATH,
        headers={"Verificationtoken": "invalid value", "MessageId": message_id},
    )

    assert response.status_code == 401
    response_body = response.json()
    assert "verficiationToken != computed sha256" in response_body["detail"]


@pytest.mark.django_db
def test_iq_endpoint_valid_hash_no_auth_failed_status():
    message_id = "9a09df23-01f3-4e0f-adbc-2a783878a574"
    token = compute_iq_mac(message_id)
    client = APIClient()
    response = client.post(
        INBOUND_SMS_PATH,
        headers={"Verificationtoken": token, "MessageId": message_id},
    )

    assert response.status_code == 400


def _prepare_valid_iq_request_client() -> APIClient:
    message_id = "9a09df23-01f3-4e0f-adbc-2a783878a574"
    token = compute_iq_mac(message_id)
    return APIClient(
        headers={"Verificationtoken": token, "MessageId": message_id},
    )


@pytest.mark.django_db
def test_iq_endpoint_missing_required_params():
    client = _prepare_valid_iq_request_client()
    resp = client.post(INBOUND_SMS_PATH, {})

    assert resp.status_code == 400
    resp_body = resp.json()

    # FIXME: why is this a list?
    assert "Request missing from, to, or text" in resp_body[0]


@pytest.mark.django_db
def test_iq_endpoint_unknown_number():
    unknown_number = "234567890"
    client = _prepare_valid_iq_request_client()
    data = {
        "from": "5556660000",
        "to": [
            unknown_number,
        ],
        "text": "test body",
    }

    resp = client.post(INBOUND_SMS_PATH, data, "json")

    assert resp.status_code == 400
    resp_body = resp.json()
    assert "Could not find relay number." in resp_body[0]


def test_iq_endpoint_disabled_number(phone_user):
    # TODO: should we return empty 200 to iQ when number is disabled?
    _make_real_phone_with_mock_iq(phone_user, verified=True)
    relay_number = _make_relay_number(phone_user, "inteliquent")
    relay_number.enabled = False
    relay_number.save()
    client = _prepare_valid_iq_request_client()
    formatted_to = iq_fmt(relay_number.number)
    data = {
        "from": "5556660000",
        "to": [
            formatted_to,
        ],
        "text": "test body",
    }

    resp = client.post(INBOUND_SMS_PATH, data, "json")
    assert resp.status_code == 200


@responses.activate
def test_iq_endpoint_success(phone_user):
    _make_real_phone_with_mock_iq(phone_user, verified=True)
    relay_number = _make_relay_number(phone_user, "inteliquent")
    pre_inbound_remaining_texts = relay_number.remaining_texts
    client = _prepare_valid_iq_request_client()
    formatted_to = iq_fmt(relay_number.number)
    data = {
        "from": "5556660000",
        "to": [
            formatted_to,
        ],
        "text": "test body",
    }

    # add response for forwarded text
    rsp = responses.add(responses.POST, settings.IQ_PUBLISH_MESSAGE_URL, status=200)
    resp = client.post(INBOUND_SMS_PATH, data, "json")

    assert resp.status_code == 200
    relay_number.refresh_from_db()
    assert relay_number.texts_forwarded == 1
    assert relay_number.remaining_texts == pre_inbound_remaining_texts - 1
    assert rsp.call_count == 1


@responses.activate
def test_reply_with_no_remaining_texts(phone_user):
    real_phone = _make_real_phone_with_mock_iq(phone_user, verified=True)
    relay_number = _make_relay_number(phone_user, "inteliquent")
    relay_number.remaining_texts = 0
    relay_number.save()
    # add a response for error message sent to user
    rsp = responses.add(responses.POST, settings.IQ_PUBLISH_MESSAGE_URL, status=200)

    client = _prepare_valid_iq_request_client()
    formatted_to = iq_fmt(relay_number.number)
    formatted_from = iq_fmt(real_phone.number)
    data = {
        "from": formatted_from,
        "to": [
            formatted_to,
        ],
        "text": "test reply",
    }
    resp = client.post(INBOUND_SMS_PATH, data, "json")

    assert resp.status_code == 400
    decoded_content = resp.content.decode()
    assert "Number is out of texts" in decoded_content
    assert rsp.call_count == 0
    relay_number.refresh_from_db()
    assert relay_number.remaining_texts == 0


@responses.activate
def test_reply_with_no_phone_capability(phone_user):
    real_phone = _make_real_phone_with_mock_iq(phone_user, verified=True)
    relay_number = _make_relay_number(phone_user, "inteliquent")
    sa = SocialAccount.objects.get(user=phone_user)
    sa.extra_data = {"avatar": "avatar.png", "subscriptions": []}
    sa.save()
    # add response that should NOT be called
    rsp = responses.add(responses.POST, settings.IQ_PUBLISH_MESSAGE_URL, status=200)

    client = _prepare_valid_iq_request_client()
    formatted_to = iq_fmt(relay_number.number)
    formatted_from = iq_fmt(real_phone.number)
    data = {
        "from": formatted_from,
        "to": [
            formatted_to,
        ],
        "text": "test reply",
    }
    resp = client.post(INBOUND_SMS_PATH, data, "json")

    assert resp.status_code == 400
    decoded_content = resp.content.decode()
    assert "Number owner does not have phone service" in decoded_content
    assert rsp.call_count == 0


@responses.activate
def test_reply_without_previous_sender_error(phone_user):
    real_phone = _make_real_phone_with_mock_iq(phone_user, verified=True)
    relay_number = _make_relay_number(phone_user, "inteliquent")
    client = _prepare_valid_iq_request_client()
    formatted_to = iq_fmt(relay_number.number)
    formatted_from = iq_fmt(real_phone.number)

    # add a response for sending error back to user
    error_msg = "You can only reply to phone numbers that have sent you a text message."
    rsp = responses.add(
        responses.POST,
        settings.IQ_PUBLISH_MESSAGE_URL,
        status=200,
        match=[
            responses.matchers.json_params_matcher(
                {
                    "from": formatted_to,
                    "to": [formatted_from],
                    "text": f"Message failed to send. {error_msg}",
                }
            )
        ],
    )

    data = {
        "from": formatted_from,
        "to": [
            formatted_to,
        ],
        "text": "test reply",
    }
    resp = client.post(INBOUND_SMS_PATH, data, "json")

    assert resp.status_code == 400
    decoded_content = resp.content.decode()
    assert error_msg in decoded_content
    assert rsp.call_count == 1


@responses.activate
def test_reply_with_previous_sender_works(phone_user):
    real_phone = _make_real_phone_with_mock_iq(phone_user, verified=True)
    relay_number = _make_relay_number(phone_user, "inteliquent")
    inbound_contact = InboundContact.objects.create(
        relay_number=relay_number, inbound_number="+15556660000"
    )
    client = _prepare_valid_iq_request_client()
    formatted_contact = iq_fmt(inbound_contact.inbound_number)
    formatted_relay = iq_fmt(relay_number.number)
    formatted_real = iq_fmt(real_phone.number)

    # add a response for sending error back to user
    rsp = responses.add(
        responses.POST,
        settings.IQ_PUBLISH_MESSAGE_URL,
        status=200,
        match=[
            responses.matchers.json_params_matcher(
                {
                    "from": formatted_relay,
                    "to": [formatted_contact],
                    "text": "test reply",
                }
            )
        ],
    )

    data = {
        "from": formatted_real,
        "to": [
            formatted_relay,
        ],
        "text": "test reply",
    }
    resp = client.post(INBOUND_SMS_PATH, data, "json")

    assert resp.status_code == 200
    assert rsp.call_count == 1

1	from unittest.mock import Mock, patch	1✔
2
3	from django.conf import settings	1✔
4
5	import pytest	1✔
6	import responses	1✔
7	from allauth.socialaccount.models import SocialAccount	1✔
8	from rest_framework.test import APIClient	1✔
9	from twilio.rest import Client	1✔
10
11	if settings.PHONES_ENABLED:	1!
12	from api.views.phones import compute_iq_mac	1✔
13	from phones.models import InboundContact, iq_fmt	1✔
14
15	from api.tests.phones_views_tests import _make_real_phone, _make_relay_number	1✔
16	from phones.tests.models_tests import make_phone_test_user	1✔
17
18	pytestmark = pytest.mark.skipif(	1✔
19	not settings.PHONES_ENABLED, reason="PHONES_ENABLED is False"
20	)
21	pytestmark = pytest.mark.skipif(not settings.IQ_ENABLED, reason="IQ_ENABLED is False")	1✔
22
23	INBOUND_SMS_PATH = "/api/v1/inbound_sms_iq/"	1✔
24
25
26	@pytest.fixture()	1✔
27	def phone_user(db):	1✔
NEW 28	return make_phone_test_user()	×
29
30
31	@pytest.fixture(autouse=True)	1✔
32	def mocked_twilio_client():	1✔
33	"""
34	Mock PhonesConfig with a mock twilio client
35	"""
36	with patch(	×
37	"phones.apps.PhonesConfig.twilio_client", spec_set=Client
38	) as mock_twilio_client:
39	mock_fetch = Mock(	×
40	return_value=Mock(
41	country_code="US", phone_number="+12223334444", carrier="verizon"
42	)
43	)
44	mock_twilio_client.lookups.v1.phone_numbers = Mock(	×
45	return_value=Mock(fetch=mock_fetch)
46	)
47	yield mock_twilio_client	×
48
49
50	def _make_real_phone_with_mock_iq(phone_user, **kwargs):	1✔
51	responses.add(responses.POST, settings.IQ_PUBLISH_MESSAGE_URL, status=200)	×
52	real_phone = _make_real_phone(phone_user, **kwargs)	×
53	responses.reset()	×
54	return real_phone	×
55
56
57	@pytest.mark.django_db	1✔
58	def test_iq_endpoint_missing_verificationtoken_header():	1✔
NEW 59	client = APIClient()	×
60	response = client.post(INBOUND_SMS_PATH)	×
61	assert response.status_code == 401	×
62	response_body = response.json()	×
63	assert "missing Verificationtoken header" in response_body["detail"]	×
64
65
66	@pytest.mark.django_db	1✔
67	def test_iq_endpoint_missing_messageid_header():	1✔
NEW 68	client = APIClient()	×
NEW 69	response = client.post(INBOUND_SMS_PATH, headers={"Verificationtoken": "valid"})	×
70
71	assert response.status_code == 401	×
72	response_body = response.json()	×
73	assert "missing MessageId header" in response_body["detail"]	×
74
75
76	@pytest.mark.django_db	1✔
77	def test_iq_endpoint_invalid_hash():	1✔
78	message_id = "9a09df23-01f3-4e0f-adbc-2a783878a574"	×
NEW 79	client = APIClient()	×
NEW 80	response = client.post(	×
81	INBOUND_SMS_PATH,
82	headers={"Verificationtoken": "invalid value", "MessageId": message_id},
83	)
84
85	assert response.status_code == 401	×
86	response_body = response.json()	×
87	assert "verficiationToken != computed sha256" in response_body["detail"]	×
88
89
90	@pytest.mark.django_db	1✔
91	def test_iq_endpoint_valid_hash_no_auth_failed_status():	1✔
92	message_id = "9a09df23-01f3-4e0f-adbc-2a783878a574"	×
93	token = compute_iq_mac(message_id)	×
NEW 94	client = APIClient()	×
NEW 95	response = client.post(	×
96	INBOUND_SMS_PATH,
97	headers={"Verificationtoken": token, "MessageId": message_id},
98	)
99
100	assert response.status_code == 400	×
101
102
103	def _prepare_valid_iq_request_client() -> APIClient:	1✔
104	message_id = "9a09df23-01f3-4e0f-adbc-2a783878a574"	×
105	token = compute_iq_mac(message_id)	×
NEW 106	return APIClient(	×
107	headers={"Verificationtoken": token, "MessageId": message_id},
108	)
109
110
111	@pytest.mark.django_db	1✔
112	def test_iq_endpoint_missing_required_params():	1✔
113	client = _prepare_valid_iq_request_client()	×
UNCOV 114	resp = client.post(INBOUND_SMS_PATH, {})	×
115
116	assert resp.status_code == 400	×
117	resp_body = resp.json()	×
118
119	# FIXME: why is this a list?
120	assert "Request missing from, to, or text" in resp_body[0]	×
121
122
123	@pytest.mark.django_db	1✔
124	def test_iq_endpoint_unknown_number():	1✔
125	unknown_number = "234567890"	×
126	client = _prepare_valid_iq_request_client()	×
127	data = {	×
128	"from": "5556660000",
129	"to": [
130	unknown_number,
131	],
132	"text": "test body",
133	}
134
NEW 135	resp = client.post(INBOUND_SMS_PATH, data, "json")	×
136
137	assert resp.status_code == 400	×
138	resp_body = resp.json()	×
139	assert "Could not find relay number." in resp_body[0]	×
140
141
142	def test_iq_endpoint_disabled_number(phone_user):	1✔
143	# TODO: should we return empty 200 to iQ when number is disabled?
144	_make_real_phone_with_mock_iq(phone_user, verified=True)	×
145	relay_number = _make_relay_number(phone_user, "inteliquent")	×
146	relay_number.enabled = False	×
147	relay_number.save()	×
148	client = _prepare_valid_iq_request_client()	×
149	formatted_to = iq_fmt(relay_number.number)	×
150	data = {	×
151	"from": "5556660000",
152	"to": [
153	formatted_to,
154	],
155	"text": "test body",
156	}
157
NEW 158	resp = client.post(INBOUND_SMS_PATH, data, "json")	×
159	assert resp.status_code == 200	×
160
161
162	@responses.activate	1✔
163	def test_iq_endpoint_success(phone_user):	1✔
164	_make_real_phone_with_mock_iq(phone_user, verified=True)	×
165	relay_number = _make_relay_number(phone_user, "inteliquent")	×
166	pre_inbound_remaining_texts = relay_number.remaining_texts	×
167	client = _prepare_valid_iq_request_client()	×
168	formatted_to = iq_fmt(relay_number.number)	×
169	data = {	×
170	"from": "5556660000",
171	"to": [
172	formatted_to,
173	],
174	"text": "test body",
175	}
176
177	# add response for forwarded text
178	rsp = responses.add(responses.POST, settings.IQ_PUBLISH_MESSAGE_URL, status=200)	×
NEW 179	resp = client.post(INBOUND_SMS_PATH, data, "json")	×
180
181	assert resp.status_code == 200	×
182	relay_number.refresh_from_db()	×
183	assert relay_number.texts_forwarded == 1	×
184	assert relay_number.remaining_texts == pre_inbound_remaining_texts - 1	×
185	assert rsp.call_count == 1	×
186
187
188	@responses.activate	1✔
189	def test_reply_with_no_remaining_texts(phone_user):	1✔
190	real_phone = _make_real_phone_with_mock_iq(phone_user, verified=True)	×
191	relay_number = _make_relay_number(phone_user, "inteliquent")	×
192	relay_number.remaining_texts = 0	×
193	relay_number.save()	×
194	# add a response for error message sent to user
195	rsp = responses.add(responses.POST, settings.IQ_PUBLISH_MESSAGE_URL, status=200)	×
196
197	client = _prepare_valid_iq_request_client()	×
198	formatted_to = iq_fmt(relay_number.number)	×
199	formatted_from = iq_fmt(real_phone.number)	×
200	data = {	×
201	"from": formatted_from,
202	"to": [
203	formatted_to,
204	],
205	"text": "test reply",
206	}
NEW 207	resp = client.post(INBOUND_SMS_PATH, data, "json")	×
208
209	assert resp.status_code == 400	×
210	decoded_content = resp.content.decode()	×
211	assert "Number is out of texts" in decoded_content	×
212	assert rsp.call_count == 0	×
213	relay_number.refresh_from_db()	×
214	assert relay_number.remaining_texts == 0	×
215
216
217	@responses.activate	1✔
218	def test_reply_with_no_phone_capability(phone_user):	1✔
219	real_phone = _make_real_phone_with_mock_iq(phone_user, verified=True)	×
220	relay_number = _make_relay_number(phone_user, "inteliquent")	×
221	sa = SocialAccount.objects.get(user=phone_user)	×
222	sa.extra_data = {"avatar": "avatar.png", "subscriptions": []}	×
223	sa.save()	×
224	# add response that should NOT be called
225	rsp = responses.add(responses.POST, settings.IQ_PUBLISH_MESSAGE_URL, status=200)	×
226
227	client = _prepare_valid_iq_request_client()	×
228	formatted_to = iq_fmt(relay_number.number)	×
229	formatted_from = iq_fmt(real_phone.number)	×
230	data = {	×
231	"from": formatted_from,
232	"to": [
233	formatted_to,
234	],
235	"text": "test reply",
236	}
NEW 237	resp = client.post(INBOUND_SMS_PATH, data, "json")	×
238
239	assert resp.status_code == 400	×
240	decoded_content = resp.content.decode()	×
241	assert "Number owner does not have phone service" in decoded_content	×
242	assert rsp.call_count == 0	×
243
244
245	@responses.activate	1✔
246	def test_reply_without_previous_sender_error(phone_user):	1✔
247	real_phone = _make_real_phone_with_mock_iq(phone_user, verified=True)	×
248	relay_number = _make_relay_number(phone_user, "inteliquent")	×
249	client = _prepare_valid_iq_request_client()	×
250	formatted_to = iq_fmt(relay_number.number)	×
251	formatted_from = iq_fmt(real_phone.number)	×
252
253	# add a response for sending error back to user
254	error_msg = "You can only reply to phone numbers that have sent you a text message."	×
255	rsp = responses.add(	×
256	responses.POST,
257	settings.IQ_PUBLISH_MESSAGE_URL,
258	status=200,
259	match=[
260	responses.matchers.json_params_matcher(
261	{
262	"from": formatted_to,
263	"to": [formatted_from],
264	"text": f"Message failed to send. {error_msg}",
265	}
266	)
267	],
268	)
269
270	data = {	×
271	"from": formatted_from,
272	"to": [
273	formatted_to,
274	],
275	"text": "test reply",
276	}
NEW 277	resp = client.post(INBOUND_SMS_PATH, data, "json")	×
278
279	assert resp.status_code == 400	×
280	decoded_content = resp.content.decode()	×
281	assert error_msg in decoded_content	×
282	assert rsp.call_count == 1	×
283
284
285	@responses.activate	1✔
286	def test_reply_with_previous_sender_works(phone_user):	1✔
287	real_phone = _make_real_phone_with_mock_iq(phone_user, verified=True)	×
288	relay_number = _make_relay_number(phone_user, "inteliquent")	×
289	inbound_contact = InboundContact.objects.create(	×
290	relay_number=relay_number, inbound_number="+15556660000"
291	)
292	client = _prepare_valid_iq_request_client()	×
293	formatted_contact = iq_fmt(inbound_contact.inbound_number)	×
294	formatted_relay = iq_fmt(relay_number.number)	×
295	formatted_real = iq_fmt(real_phone.number)	×
296
297	# add a response for sending error back to user
298	rsp = responses.add(	×
299	responses.POST,
300	settings.IQ_PUBLISH_MESSAGE_URL,
301	status=200,
302	match=[
303	responses.matchers.json_params_matcher(
304	{
305	"from": formatted_relay,
306	"to": [formatted_contact],
307	"text": "test reply",
308	}
309	)
310	],
311	)
312
313	data = {	×
314	"from": formatted_real,
315	"to": [
316	formatted_relay,
317	],
318	"text": "test reply",
319	}
NEW 320	resp = client.post(INBOUND_SMS_PATH, data, "json")	×
321
322	assert resp.status_code == 200	×
323	assert rsp.call_count == 1	×

mozilla / fx-private-relay / b06e8ab5-baf6-4a94-ad8f-b818825c1988

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

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