6f94747f-109e-4815-824e-9ff978c31b89

Committed 17 Sep 2025 02:15PM UTC coverage: 88.099% (-0.03%) from 88.129%

Build # 6f94747f-109e-4815-824e-9ff978c31b89

Build Type

push

circleci

Committed by

web-flow

Commit Message

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

fix alert #59: remove social account logging

Run Details

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

"""Tests for api/views/email_views.py"""

from datetime import datetime

from django.conf import LazySettings
from django.contrib.auth.models import User
from django.core.cache import cache
from django.test import RequestFactory, TestCase
from django.test.client import Client
from django.urls import reverse

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

from api.authentication import INTROSPECT_TOKEN_URL, get_cache_key
from api.tests.authentication_tests import _setup_fxa_response
from api.views.privaterelay import FXA_PROFILE_URL
from privaterelay.models import Profile


@pytest.mark.django_db
def test_runtime_data_response_structure(client: Client) -> None:
    """Test that runtime_data returns the expected structure."""
    path = "/api/v1/runtime_data/"
    response = client.get(path)

    assert response.status_code == 200
    data = response.json()

    # Check that all expected keys are present
    expected_keys = [
        "FXA_ORIGIN",
        "PERIODICAL_PREMIUM_PRODUCT_ID",
        "GOOGLE_ANALYTICS_ID",
        "GA4_MEASUREMENT_ID",
        "BUNDLE_PRODUCT_ID",
        "MEGABUNDLE_PRODUCT_ID",
        "PHONE_PRODUCT_ID",
        "PERIODICAL_PREMIUM_PLANS",
        "PHONE_PLANS",
        "BUNDLE_PLANS",
        "BASKET_ORIGIN",
        "WAFFLE_FLAGS",
        "WAFFLE_SWITCHES",
        "WAFFLE_SAMPLES",
        "MAX_MINUTES_TO_VERIFY_REAL_PHONE",
    ]

    for key in expected_keys:
        assert key in data, f"Expected key {key} missing from response"

    # Check that plans data is present
    assert isinstance(data["PERIODICAL_PREMIUM_PLANS"], dict)
    assert isinstance(data["PHONE_PLANS"], dict)
    assert isinstance(data["BUNDLE_PLANS"], dict)


@pytest.mark.django_db
@pytest.mark.parametrize("use_subplat3", [True, False])
def test_runtime_data_uses_correct_plan_mapping(
    client: Client, settings: LazySettings, use_subplat3: bool
) -> None:
    """
    Test that runtime_data uses the correct plan mapping based on
    USE_SUBPLAT3 setting.
    """
    settings.USE_SUBPLAT3 = use_subplat3

    path = "/api/v1/runtime_data/"
    response = client.get(path)

    assert response.status_code == 200
    data = response.json()

    # Check that the correct plan data is returned
    if use_subplat3:
        # Check for SP3 plan data
        assert (
            "url"
            in data["PERIODICAL_PREMIUM_PLANS"]["plan_country_lang_mapping"]["US"]["*"][
                "monthly"
            ]
        )
        assert (
            "url"
            in data["PHONE_PLANS"]["plan_country_lang_mapping"]["US"]["*"]["monthly"]
        )
        assert (
            "url"
            in data["BUNDLE_PLANS"]["plan_country_lang_mapping"]["US"]["*"]["yearly"]
        )
    else:
        # Check for regular plan data
        assert (
            "id"
            in data["PERIODICAL_PREMIUM_PLANS"]["plan_country_lang_mapping"]["US"]["*"][
                "monthly"
            ]
        )
        assert (
            "id"
            in data["PHONE_PLANS"]["plan_country_lang_mapping"]["US"]["*"]["monthly"]
        )
        assert (
            "id"
            in data["BUNDLE_PLANS"]["plan_country_lang_mapping"]["US"]["*"]["yearly"]
        )


def test_patch_premium_user_subdomain_cannot_be_changed(
    premium_user: User, prem_api_client: Client
) -> None:
    """A premium user should not be able to edit their subdomain."""
    premium_profile = premium_user.profile
    original_subdomain = "helloworld"
    premium_profile.subdomain = original_subdomain
    premium_profile.save()

    new_subdomain = "helloworldd"
    response = prem_api_client.patch(
        reverse(viewname="profiles-detail", args=[premium_profile.id]),
        data={"subdomain": new_subdomain},
        format="json",
    )

    ret_data = response.json()
    premium_profile.refresh_from_db()

    assert ret_data.get("subdomain", [""])[0] == "This field is read only"
    assert premium_profile.subdomain == original_subdomain
    assert response.status_code == 400


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

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

    response = prem_api_client.patch(
        reverse(viewname="profiles-detail", args=[premium_profile.id]),
        data={
            "num_address_deleted": 5,
            "sent_welcome_email": True,
        },
        format="json",
    )

    ret_data = response.json()
    premium_profile.refresh_from_db()

    assert premium_profile.num_address_deleted == expected_num_address_deleted
    assert premium_profile.sent_welcome_email == expected_sent_welcome_email
    assert ret_data.get("sent_welcome_email", [""])[0] == "This field is read only"
    assert ret_data.get("num_address_deleted", [""])[0] == "This field is read only"
    assert response.status_code == 400


def test_profile_non_read_only_fields_update_correctly(
    premium_user: User, prem_api_client: Client
) -> None:
    """
    A field that is not read only should update correctly on a patch request.

    "Not read only" meaning that it was defined in the serializers fields, but not
    read_only_fields.
    """
    premium_profile = premium_user.profile
    old_onboarding_state = premium_profile.onboarding_state
    old_email_tracker_remove_value = premium_profile.remove_level_one_email_trackers

    response = prem_api_client.patch(
        reverse(viewname="profiles-detail", args=[premium_profile.id]),
        data={
            "onboarding_state": 1,
            "remove_level_one_email_trackers": True,
        },
        format="json",
    )

    ret_data = response.json()
    premium_profile.refresh_from_db()

    assert (
        ret_data.get("remove_level_one_email_trackers", None)
        != old_email_tracker_remove_value
    )
    assert (
        premium_profile.remove_level_one_email_trackers
        != old_email_tracker_remove_value
    )
    assert premium_profile.remove_level_one_email_trackers is True
    assert ret_data.get("remove_level_one_email_trackers", None) is True
    assert ret_data.get("onboarding_state", None) != old_onboarding_state
    assert premium_profile.onboarding_state == 1
    assert ret_data.get("onboarding_state", None) == 1
    assert response.status_code == 200


def test_profile_patch_with_model_and_serializer_fields(
    premium_user: User, prem_api_client: Client
) -> None:
    premium_profile = premium_user.profile

    response = prem_api_client.patch(
        reverse(viewname="profiles-detail", args=[premium_profile.id]),
        data={"subdomain": "vanilla", "num_address_deleted": 5},
        format="json",
    )

    premium_profile.refresh_from_db()

    assert response.status_code == 400
    assert premium_profile.subdomain != "vanilla"
    assert premium_profile.num_address_deleted == 0


def test_profile_patch_with_non_read_only_and_read_only_fields(
    premium_user, prem_api_client
):
    """A request that includes at least one read only field will give a 400 response"""
    premium_profile = premium_user.profile
    old_onboarding_state = premium_profile.onboarding_state

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

    ret_data = response.json()
    premium_profile.refresh_from_db()

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


def test_profile_patch_fields_that_dont_exist(
    premium_user: User, prem_api_client: Client
) -> None:
    """
    A request sent with only fields that don't exist give a 200 response (this is the
    default behavior django provides, we decided to leave it as is)
    """
    response = prem_api_client.patch(
        reverse(viewname="profiles-detail", args=[premium_user.profile.id]),
        data={
            "nonsense": False,
            "blabla": "blabla",
        },
        format="json",
    )

    assert response.status_code == 200


@pytest.mark.usefixtures("fxa_social_app")
class TermsAcceptedUserViewTest(TestCase):
    def setUp(self) -> None:
        self.factory = RequestFactory()
        self.path = "/api/v1/terms-accepted-user/"
        self.fxa_verify_path = INTROSPECT_TOKEN_URL
        self.uid = "relay-user-fxa-uid"

    def _setup_client(self, token: str) -> None:
        self.client = APIClient()
        self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")

    def tearDown(self) -> None:
        cache.clear()

    @responses.activate
    def test_201_new_user_created_and_202_user_exists(self) -> None:
        email = "user@email.com"
        user_token = "user-123"
        self._setup_client(user_token)
        now_time = int(datetime.now().timestamp())
        # Note: FXA iat and exp are timestamps in *milliseconds*
        exp_time = (now_time + 60 * 60) * 1000
        fxa_response = _setup_fxa_response(
            200, {"active": True, "sub": self.uid, "exp": exp_time}
        )
        # setup fxa profile reponse
        profile_json = {
            "email": email,
            "amrValues": ["pwd", "email"],
            "twoFactorAuthentication": False,
            "metricsEnabled": True,
            "uid": self.uid,
            "avatar": "https://profile.stage.mozaws.net/v1/avatar/t",
            "avatarDefault": False,
        }
        responses.add(
            responses.GET,
            FXA_PROFILE_URL,
            status=200,
            json=profile_json,
        )
        cache_key = get_cache_key(user_token)

        # get fxa response with 201 response for new user and profile created
        response = self.client.post(self.path)
        assert response.status_code == 201
        assert hasattr(response, "data")
        assert response.data is None
        # ensure no session cookie was set
        assert len(response.cookies.keys()) == 1
        assert "csrftoken" in response.cookies
        assert responses.assert_call_count(self.fxa_verify_path, 1) is True
        assert responses.assert_call_count(FXA_PROFILE_URL, 1) is True
        assert cache.get(cache_key) == fxa_response
        assert SocialAccount.objects.filter(user__email=email).count() == 1
        assert Profile.objects.filter(user__email=email).count() == 1
        assert Profile.objects.get(user__email=email).created_by == "firefox_resource"

        # now check that the 2nd call returns 202
        response = self.client.post(self.path)
        assert response.status_code == 202
        assert hasattr(response, "data")
        assert response.data is None
        assert responses.assert_call_count(self.fxa_verify_path, 2) is True
        assert responses.assert_call_count(FXA_PROFILE_URL, 1) is True

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

        assert response.status_code == 500
        assert response.json()["detail"] == (
            "Did not receive a 200 response for account profile."
        )

    def test_no_authorization_header_returns_400(self) -> None:
        client = APIClient()
        response = client.post(self.path)

        assert response.status_code == 400
        assert response.json()["detail"] == "Missing Bearer header."

    def test_no_token_returns_400(self) -> None:
        client = APIClient()
        client.credentials(HTTP_AUTHORIZATION="Bearer ")
        response = client.post(self.path)

        assert response.status_code == 400
        assert response.json()["detail"] == "Missing FXA Token after 'Bearer'."

    @responses.activate
    def test_invalid_bearer_token_error_from_fxa_returns_500_and_cache_returns_500(
        self,
    ) -> None:
        _setup_fxa_response(401, {"error": "401"})
        not_found_token = "not-found-123"
        self._setup_client(not_found_token)

        assert cache.get(get_cache_key(not_found_token)) is None

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

    @responses.activate
    def test_jsondecodeerror_returns_401_and_cache_returns_500(
        self,
    ) -> None:
        _setup_fxa_response(200)
        invalid_token = "invalid-123"
        cache_key = get_cache_key(invalid_token)
        self._setup_client(invalid_token)

        assert cache.get(cache_key) is None

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

    @responses.activate
    def test_non_200_response_from_fxa_returns_500_and_cache_returns_500(
        self,
    ) -> None:
        now_time = int(datetime.now().timestamp())
        # Note: FXA iat and exp are timestamps in *milliseconds*
        exp_time = (now_time + 60 * 60) * 1000
        _setup_fxa_response(401, {"active": False, "sub": self.uid, "exp": exp_time})
        invalid_token = "invalid-123"
        cache_key = get_cache_key(invalid_token)
        self._setup_client(invalid_token)

        assert cache.get(cache_key) is None

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

    @responses.activate
    def test_inactive_fxa_oauth_token_returns_401_and_cache_returns_401(
        self,
    ) -> None:
        now_time = int(datetime.now().timestamp())
        # Note: FXA iat and exp are timestamps in *milliseconds*
        old_exp_time = (now_time - 60 * 60) * 1000
        _setup_fxa_response(
            200, {"active": False, "sub": self.uid, "exp": old_exp_time}
        )
        invalid_token = "invalid-123"
        cache_key = get_cache_key(invalid_token)
        self._setup_client(invalid_token)

        assert cache.get(cache_key) is None

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

    @responses.activate
    def test_fxa_responds_with_no_fxa_uid_returns_404_and_cache_returns_404(
        self,
    ) -> None:
        user_token = "user-123"
        now_time = int(datetime.now().timestamp())
        # Note: FXA iat and exp are timestamps in *milliseconds*
        exp_time = (now_time + 60 * 60) * 1000
        _setup_fxa_response(200, {"active": True, "exp": exp_time})
        cache_key = get_cache_key(user_token)
        self._setup_client(user_token)

        assert cache.get(cache_key) is None

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


def _setup_client(token: str) -> APIClient:
    client = APIClient()
    client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
    return client

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	×

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

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