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

mozilla / fx-private-relay / 6f94747f-109e-4815-824e-9ff978c31b89

17 Sep 2025 02:15PM UTC coverage: 88.099% (-0.03%) from 88.129%
6f94747f-109e-4815-824e-9ff978c31b89

push

circleci

web-flow
Merge pull request #5883 from mozilla/alert-fix-59

fix alert #59: remove social account logging

2920 of 3949 branches covered (73.94%)

Branch coverage included in aggregate %.

3 of 3 new or added lines in 1 file covered. (100.0%)

4 existing lines in 2 files now uncovered.

18134 of 19949 relevant lines covered (90.9%)

11.27 hits per line

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

98.7
/api/tests/privaterelay_views_tests.py
1
"""Tests for api/views/email_views.py"""
2

3
from datetime import datetime
1✔
4

5
from django.conf import LazySettings
1✔
6
from django.contrib.auth.models import User
1✔
7
from django.core.cache import cache
1✔
8
from django.test import RequestFactory, TestCase
1✔
9
from django.test.client import Client
1✔
10
from django.urls import reverse
1✔
11

12
import pytest
1✔
13
import responses
1✔
14
from allauth.socialaccount.models import SocialAccount
1✔
15
from rest_framework.test import APIClient
1✔
16

17
from api.authentication import INTROSPECT_TOKEN_URL, get_cache_key
1✔
18
from api.tests.authentication_tests import _setup_fxa_response
1✔
19
from api.views.privaterelay import FXA_PROFILE_URL
1✔
20
from privaterelay.models import Profile
1✔
21

22

23
@pytest.mark.django_db
1✔
24
def test_runtime_data_response_structure(client: Client) -> None:
1✔
25
    """Test that runtime_data returns the expected structure."""
26
    path = "/api/v1/runtime_data/"
1✔
27
    response = client.get(path)
1✔
28

29
    assert response.status_code == 200
1✔
30
    data = response.json()
1✔
31

32
    # Check that all expected keys are present
33
    expected_keys = [
1✔
34
        "FXA_ORIGIN",
35
        "PERIODICAL_PREMIUM_PRODUCT_ID",
36
        "GOOGLE_ANALYTICS_ID",
37
        "GA4_MEASUREMENT_ID",
38
        "BUNDLE_PRODUCT_ID",
39
        "MEGABUNDLE_PRODUCT_ID",
40
        "PHONE_PRODUCT_ID",
41
        "PERIODICAL_PREMIUM_PLANS",
42
        "PHONE_PLANS",
43
        "BUNDLE_PLANS",
44
        "BASKET_ORIGIN",
45
        "WAFFLE_FLAGS",
46
        "WAFFLE_SWITCHES",
47
        "WAFFLE_SAMPLES",
48
        "MAX_MINUTES_TO_VERIFY_REAL_PHONE",
49
    ]
50

51
    for key in expected_keys:
1✔
52
        assert key in data, f"Expected key {key} missing from response"
1✔
53

54
    # Check that plans data is present
55
    assert isinstance(data["PERIODICAL_PREMIUM_PLANS"], dict)
1✔
56
    assert isinstance(data["PHONE_PLANS"], dict)
1✔
57
    assert isinstance(data["BUNDLE_PLANS"], dict)
1✔
58

59

60
@pytest.mark.django_db
1✔
61
@pytest.mark.parametrize("use_subplat3", [True, False])
1✔
62
def test_runtime_data_uses_correct_plan_mapping(
1✔
63
    client: Client, settings: LazySettings, use_subplat3: bool
64
) -> None:
65
    """
66
    Test that runtime_data uses the correct plan mapping based on
67
    USE_SUBPLAT3 setting.
68
    """
69
    settings.USE_SUBPLAT3 = use_subplat3
1✔
70

71
    path = "/api/v1/runtime_data/"
1✔
72
    response = client.get(path)
1✔
73

74
    assert response.status_code == 200
1✔
75
    data = response.json()
1✔
76

77
    # Check that the correct plan data is returned
78
    if use_subplat3:
1✔
79
        # Check for SP3 plan data
80
        assert (
1✔
81
            "url"
82
            in data["PERIODICAL_PREMIUM_PLANS"]["plan_country_lang_mapping"]["US"]["*"][
83
                "monthly"
84
            ]
85
        )
86
        assert (
1✔
87
            "url"
88
            in data["PHONE_PLANS"]["plan_country_lang_mapping"]["US"]["*"]["monthly"]
89
        )
90
        assert (
1✔
91
            "url"
92
            in data["BUNDLE_PLANS"]["plan_country_lang_mapping"]["US"]["*"]["yearly"]
93
        )
94
    else:
95
        # Check for regular plan data
96
        assert (
1✔
97
            "id"
98
            in data["PERIODICAL_PREMIUM_PLANS"]["plan_country_lang_mapping"]["US"]["*"][
99
                "monthly"
100
            ]
101
        )
102
        assert (
1✔
103
            "id"
104
            in data["PHONE_PLANS"]["plan_country_lang_mapping"]["US"]["*"]["monthly"]
105
        )
106
        assert (
1✔
107
            "id"
108
            in data["BUNDLE_PLANS"]["plan_country_lang_mapping"]["US"]["*"]["yearly"]
109
        )
110

111

112
def test_patch_premium_user_subdomain_cannot_be_changed(
1✔
113
    premium_user: User, prem_api_client: Client
114
) -> None:
115
    """A premium user should not be able to edit their subdomain."""
116
    premium_profile = premium_user.profile
1✔
117
    original_subdomain = "helloworld"
1✔
118
    premium_profile.subdomain = original_subdomain
1✔
119
    premium_profile.save()
1✔
120

121
    new_subdomain = "helloworldd"
1✔
122
    response = prem_api_client.patch(
1✔
123
        reverse(viewname="profiles-detail", args=[premium_profile.id]),
124
        data={"subdomain": new_subdomain},
125
        format="json",
126
    )
127

128
    ret_data = response.json()
1✔
129
    premium_profile.refresh_from_db()
1✔
130

131
    assert ret_data.get("subdomain", [""])[0] == "This field is read only"
1✔
132
    assert premium_profile.subdomain == original_subdomain
1✔
133
    assert response.status_code == 400
1✔
134

135

136
def test_patch_profile_fields_are_read_only_by_default(
1✔
137
    premium_user: User, prem_api_client: Client
138
) -> None:
139
    """
140
    A field in the Profile model should be read only by default, and return a 400
141
    response code (see StrictReadOnlyFieldsMixin in api/serializers/__init__.py), if it
142
    is not mentioned in the ProfileSerializer class fields.
143

144
    Two fields were tested, num_address_deleted, and sent_welcome_email to see if the
145
    behavior matches what is described here:
146
    https://www.django-rest-framework.org/api-guide/serializers/#specifying-read-only-fields
147
    """
148
    premium_profile = premium_user.profile
1✔
149
    expected_num_address_deleted = premium_profile.num_address_deleted
1✔
150
    expected_sent_welcome_email = premium_profile.sent_welcome_email
1✔
151

152
    response = prem_api_client.patch(
1✔
153
        reverse(viewname="profiles-detail", args=[premium_profile.id]),
154
        data={
155
            "num_address_deleted": 5,
156
            "sent_welcome_email": True,
157
        },
158
        format="json",
159
    )
160

161
    ret_data = response.json()
1✔
162
    premium_profile.refresh_from_db()
1✔
163

164
    assert premium_profile.num_address_deleted == expected_num_address_deleted
1✔
165
    assert premium_profile.sent_welcome_email == expected_sent_welcome_email
1✔
166
    assert ret_data.get("sent_welcome_email", [""])[0] == "This field is read only"
1✔
167
    assert ret_data.get("num_address_deleted", [""])[0] == "This field is read only"
1✔
168
    assert response.status_code == 400
1✔
169

170

171
def test_profile_non_read_only_fields_update_correctly(
1✔
172
    premium_user: User, prem_api_client: Client
173
) -> None:
174
    """
175
    A field that is not read only should update correctly on a patch request.
176

177
    "Not read only" meaning that it was defined in the serializers fields, but not
178
    read_only_fields.
179
    """
180
    premium_profile = premium_user.profile
1✔
181
    old_onboarding_state = premium_profile.onboarding_state
1✔
182
    old_email_tracker_remove_value = premium_profile.remove_level_one_email_trackers
1✔
183

184
    response = prem_api_client.patch(
1✔
185
        reverse(viewname="profiles-detail", args=[premium_profile.id]),
186
        data={
187
            "onboarding_state": 1,
188
            "remove_level_one_email_trackers": True,
189
        },
190
        format="json",
191
    )
192

193
    ret_data = response.json()
1✔
194
    premium_profile.refresh_from_db()
1✔
195

196
    assert (
1✔
197
        ret_data.get("remove_level_one_email_trackers", None)
198
        != old_email_tracker_remove_value
199
    )
200
    assert (
1✔
201
        premium_profile.remove_level_one_email_trackers
202
        != old_email_tracker_remove_value
203
    )
204
    assert premium_profile.remove_level_one_email_trackers is True
1✔
205
    assert ret_data.get("remove_level_one_email_trackers", None) is True
1✔
206
    assert ret_data.get("onboarding_state", None) != old_onboarding_state
1✔
207
    assert premium_profile.onboarding_state == 1
1✔
208
    assert ret_data.get("onboarding_state", None) == 1
1✔
209
    assert response.status_code == 200
1✔
210

211

212
def test_profile_patch_with_model_and_serializer_fields(
1✔
213
    premium_user: User, prem_api_client: Client
214
) -> None:
215
    premium_profile = premium_user.profile
1✔
216

217
    response = prem_api_client.patch(
1✔
218
        reverse(viewname="profiles-detail", args=[premium_profile.id]),
219
        data={"subdomain": "vanilla", "num_address_deleted": 5},
220
        format="json",
221
    )
222

223
    premium_profile.refresh_from_db()
1✔
224

225
    assert response.status_code == 400
1✔
226
    assert premium_profile.subdomain != "vanilla"
1✔
227
    assert premium_profile.num_address_deleted == 0
1✔
228

229

230
def test_profile_patch_with_non_read_only_and_read_only_fields(
1✔
231
    premium_user, prem_api_client
232
):
233
    """A request that includes at least one read only field will give a 400 response"""
234
    premium_profile = premium_user.profile
1✔
235
    old_onboarding_state = premium_profile.onboarding_state
1✔
236

237
    response = prem_api_client.patch(
1✔
238
        reverse(viewname="profiles-detail", args=[premium_profile.id]),
239
        data={"onboarding_state": 1, "subdomain": "vanilla"},
240
        format="json",
241
    )
242

243
    ret_data = response.json()
1✔
244
    premium_profile.refresh_from_db()
1✔
245

246
    assert premium_profile.onboarding_state == old_onboarding_state
1✔
247
    assert ret_data.get("subdomain", [""])[0] == "This field is read only"
1✔
248
    assert response.status_code == 400
1✔
249

250

251
def test_profile_patch_fields_that_dont_exist(
1✔
252
    premium_user: User, prem_api_client: Client
253
) -> None:
254
    """
255
    A request sent with only fields that don't exist give a 200 response (this is the
256
    default behavior django provides, we decided to leave it as is)
257
    """
258
    response = prem_api_client.patch(
1✔
259
        reverse(viewname="profiles-detail", args=[premium_user.profile.id]),
260
        data={
261
            "nonsense": False,
262
            "blabla": "blabla",
263
        },
264
        format="json",
265
    )
266

267
    assert response.status_code == 200
1✔
268

269

270
@pytest.mark.usefixtures("fxa_social_app")
1✔
271
class TermsAcceptedUserViewTest(TestCase):
1✔
272
    def setUp(self) -> None:
1✔
273
        self.factory = RequestFactory()
1✔
274
        self.path = "/api/v1/terms-accepted-user/"
1✔
275
        self.fxa_verify_path = INTROSPECT_TOKEN_URL
1✔
276
        self.uid = "relay-user-fxa-uid"
1✔
277

278
    def _setup_client(self, token: str) -> None:
1✔
279
        self.client = APIClient()
1✔
280
        self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
1✔
281

282
    def tearDown(self) -> None:
1✔
283
        cache.clear()
1✔
284

285
    @responses.activate
1✔
286
    def test_201_new_user_created_and_202_user_exists(self) -> None:
1✔
287
        email = "user@email.com"
1✔
288
        user_token = "user-123"
1✔
289
        self._setup_client(user_token)
1✔
290
        now_time = int(datetime.now().timestamp())
1✔
291
        # Note: FXA iat and exp are timestamps in *milliseconds*
292
        exp_time = (now_time + 60 * 60) * 1000
1✔
293
        fxa_response = _setup_fxa_response(
1✔
294
            200, {"active": True, "sub": self.uid, "exp": exp_time}
295
        )
296
        # setup fxa profile reponse
297
        profile_json = {
1✔
298
            "email": email,
299
            "amrValues": ["pwd", "email"],
300
            "twoFactorAuthentication": False,
301
            "metricsEnabled": True,
302
            "uid": self.uid,
303
            "avatar": "https://profile.stage.mozaws.net/v1/avatar/t",
304
            "avatarDefault": False,
305
        }
306
        responses.add(
1✔
307
            responses.GET,
308
            FXA_PROFILE_URL,
309
            status=200,
310
            json=profile_json,
311
        )
312
        cache_key = get_cache_key(user_token)
1✔
313

314
        # get fxa response with 201 response for new user and profile created
315
        response = self.client.post(self.path)
1✔
316
        assert response.status_code == 201
1✔
317
        assert hasattr(response, "data")
1✔
318
        assert response.data is None
1✔
319
        # ensure no session cookie was set
320
        assert len(response.cookies.keys()) == 1
1✔
321
        assert "csrftoken" in response.cookies
1✔
322
        assert responses.assert_call_count(self.fxa_verify_path, 1) is True
1✔
323
        assert responses.assert_call_count(FXA_PROFILE_URL, 1) is True
1✔
324
        assert cache.get(cache_key) == fxa_response
1✔
325
        assert SocialAccount.objects.filter(user__email=email).count() == 1
1✔
326
        assert Profile.objects.filter(user__email=email).count() == 1
1✔
327
        assert Profile.objects.get(user__email=email).created_by == "firefox_resource"
1✔
328

329
        # now check that the 2nd call returns 202
330
        response = self.client.post(self.path)
1✔
331
        assert response.status_code == 202
1✔
332
        assert hasattr(response, "data")
1✔
333
        assert response.data is None
1✔
334
        assert responses.assert_call_count(self.fxa_verify_path, 2) is True
1✔
335
        assert responses.assert_call_count(FXA_PROFILE_URL, 1) is True
1✔
336

337
    @responses.activate
1✔
338
    def test_failed_profile_fetch_for_new_user_returns_500(self) -> None:
1✔
339
        user_token = "user-123"
1✔
340
        self._setup_client(user_token)
1✔
341
        now_time = int(datetime.now().timestamp())
1✔
342
        exp_time = (now_time + 60 * 60) * 1000
1✔
343
        _setup_fxa_response(200, {"active": True, "sub": self.uid, "exp": exp_time})
1✔
344
        # FxA profile server is down
345
        responses.add(responses.GET, FXA_PROFILE_URL, status=502, body="")
1✔
346
        response = self.client.post(self.path)
1✔
347

348
        assert response.status_code == 500
1✔
349
        assert response.json()["detail"] == (
1✔
350
            "Did not receive a 200 response for account profile."
351
        )
352

353
    def test_no_authorization_header_returns_400(self) -> None:
1✔
354
        client = APIClient()
1✔
355
        response = client.post(self.path)
1✔
356

357
        assert response.status_code == 400
1✔
358
        assert response.json()["detail"] == "Missing Bearer header."
1✔
359

360
    def test_no_token_returns_400(self) -> None:
1✔
361
        client = APIClient()
1✔
362
        client.credentials(HTTP_AUTHORIZATION="Bearer ")
1✔
363
        response = client.post(self.path)
1✔
364

365
        assert response.status_code == 400
1✔
366
        assert response.json()["detail"] == "Missing FXA Token after 'Bearer'."
1✔
367

368
    @responses.activate
1✔
369
    def test_invalid_bearer_token_error_from_fxa_returns_500_and_cache_returns_500(
1✔
370
        self,
371
    ) -> None:
372
        _setup_fxa_response(401, {"error": "401"})
1✔
373
        not_found_token = "not-found-123"
1✔
374
        self._setup_client(not_found_token)
1✔
375

376
        assert cache.get(get_cache_key(not_found_token)) is None
1✔
377

378
        response = self.client.post(self.path)
1✔
379
        assert response.status_code == 500
1✔
380
        assert response.json()["detail"] == "Did not receive a 200 response from FXA."
1✔
381
        assert responses.assert_call_count(self.fxa_verify_path, 1) is True
1✔
382

383
    @responses.activate
1✔
384
    def test_jsondecodeerror_returns_401_and_cache_returns_500(
1✔
385
        self,
386
    ) -> None:
387
        _setup_fxa_response(200)
1✔
388
        invalid_token = "invalid-123"
1✔
389
        cache_key = get_cache_key(invalid_token)
1✔
390
        self._setup_client(invalid_token)
1✔
391

392
        assert cache.get(cache_key) is None
1✔
393

394
        # get fxa response with no status code for the first time
395
        response = self.client.post(self.path)
1✔
396
        assert response.status_code == 401
1✔
397
        assert (
1✔
398
            response.json()["detail"] == "Jsondecodeerror From Fxa Introspect Response"
399
        )
400
        assert responses.assert_call_count(self.fxa_verify_path, 1) is True
1✔
401

402
    @responses.activate
1✔
403
    def test_non_200_response_from_fxa_returns_500_and_cache_returns_500(
1✔
404
        self,
405
    ) -> None:
406
        now_time = int(datetime.now().timestamp())
1✔
407
        # Note: FXA iat and exp are timestamps in *milliseconds*
408
        exp_time = (now_time + 60 * 60) * 1000
1✔
409
        _setup_fxa_response(401, {"active": False, "sub": self.uid, "exp": exp_time})
1✔
410
        invalid_token = "invalid-123"
1✔
411
        cache_key = get_cache_key(invalid_token)
1✔
412
        self._setup_client(invalid_token)
1✔
413

414
        assert cache.get(cache_key) is None
1✔
415

416
        # get fxa response with non-200 response for the first time
417
        response = self.client.post(self.path)
1✔
418
        assert response.status_code == 500
1✔
419
        assert response.json()["detail"] == "Did not receive a 200 response from FXA."
1✔
420
        assert responses.assert_call_count(self.fxa_verify_path, 1) is True
1✔
421

422
    @responses.activate
1✔
423
    def test_inactive_fxa_oauth_token_returns_401_and_cache_returns_401(
1✔
424
        self,
425
    ) -> None:
426
        now_time = int(datetime.now().timestamp())
1✔
427
        # Note: FXA iat and exp are timestamps in *milliseconds*
428
        old_exp_time = (now_time - 60 * 60) * 1000
1✔
429
        _setup_fxa_response(
1✔
430
            200, {"active": False, "sub": self.uid, "exp": old_exp_time}
431
        )
432
        invalid_token = "invalid-123"
1✔
433
        cache_key = get_cache_key(invalid_token)
1✔
434
        self._setup_client(invalid_token)
1✔
435

436
        assert cache.get(cache_key) is None
1✔
437

438
        # get fxa response with token inactive for the first time
439
        response = self.client.post(self.path)
1✔
440
        assert response.status_code == 401
1✔
441
        assert response.json()["detail"] == "Fxa Returned Active: False For Token."
1✔
442
        assert responses.assert_call_count(self.fxa_verify_path, 1) is True
1✔
443

444
    @responses.activate
1✔
445
    def test_fxa_responds_with_no_fxa_uid_returns_404_and_cache_returns_404(
1✔
446
        self,
447
    ) -> None:
448
        user_token = "user-123"
1✔
449
        now_time = int(datetime.now().timestamp())
1✔
450
        # Note: FXA iat and exp are timestamps in *milliseconds*
451
        exp_time = (now_time + 60 * 60) * 1000
1✔
452
        _setup_fxa_response(200, {"active": True, "exp": exp_time})
1✔
453
        cache_key = get_cache_key(user_token)
1✔
454
        self._setup_client(user_token)
1✔
455

456
        assert cache.get(cache_key) is None
1✔
457

458
        # get fxa response with no fxa uid for the first time
459
        response = self.client.post(self.path)
1✔
460
        assert response.status_code == 404
1✔
461
        assert response.json()["detail"] == "FXA did not return an FXA UID."
1✔
462
        assert responses.assert_call_count(self.fxa_verify_path, 1) is True
1✔
463

464

465
def _setup_client(token: str) -> APIClient:
1✔
UNCOV
466
    client = APIClient()
×
UNCOV
467
    client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
×
UNCOV
468
    return client
×
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

© 2026 Coveralls, Inc