84353759-c057-4f20-b282-724c34504dc9

Committed 26 Nov 2025 04:22PM UTC coverage: 89.192% (+0.4%) from 88.772%

Build # 84353759-c057-4f20-b282-724c34504dc9

Build Type

Pull #6049

circleci

Committed by

jwhitlock

Commit Message

Update TermsAcceptedUserViewTest for new errors

Pull Request Pull Request #6049: fix(relay): Create alternate bearer token auth for FxA (MPP-3505)

Run Details

3016 of 4041 branches covered (74.63%)

Branch coverage included in aggregate %.

1334 of 1349 new or added lines in 9 files covered. (98.89%)

6 existing lines in 2 files now uncovered.

19067 of 20718 relevant lines covered (92.03%)

11.09 hits per line

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

97.86

/api/tests/terms_accepted_user_2024_tests.py

"""Tests for terms_accepted_user_2024 in api/views/privaterelay_views.py"""

from datetime import datetime

from django.core.cache import cache
from django.test import RequestFactory, TestCase, override_settings

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

from api.authentication import (
    FXA_TOKEN_AUTH_OLD_AND_PROVEN,
    INTROSPECT_TOKEN_URL,
)
from api.authentication import (
    get_cache_key_2024 as get_cache_key,
)
from api.tests.authentication_2024_tests import _setup_fxa_response
from api.views.privaterelay import FXA_PROFILE_URL
from privaterelay.models import Profile


@override_settings(
    FXA_TOKEN_AUTH_VERSION=FXA_TOKEN_AUTH_OLD_AND_PROVEN
)  # noqa: S106 # Possible hardcoded password
@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 response
        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 terms_accepted_user_2024 in api/views/privaterelay_views.py"""
2
3	from datetime import datetime	1✔
4
5	from django.core.cache import cache	1✔
6	from django.test import RequestFactory, TestCase, override_settings	1✔
7
8	import pytest	1✔
9	import responses	1✔
10	from allauth.socialaccount.models import SocialAccount	1✔
11	from rest_framework.test import APIClient	1✔
12
13	from api.authentication import (	1✔
14	FXA_TOKEN_AUTH_OLD_AND_PROVEN,
15	INTROSPECT_TOKEN_URL,
16	)
17	from api.authentication import (	1✔
18	get_cache_key_2024 as get_cache_key,
19	)
20	from api.tests.authentication_2024_tests import _setup_fxa_response	1✔
21	from api.views.privaterelay import FXA_PROFILE_URL	1✔
22	from privaterelay.models import Profile	1✔
23
24
25	@override_settings(	1✔
26	FXA_TOKEN_AUTH_VERSION=FXA_TOKEN_AUTH_OLD_AND_PROVEN
27	) # noqa: S106 # Possible hardcoded password
28	@pytest.mark.usefixtures("fxa_social_app")	1✔
29	class TermsAcceptedUserViewTest(TestCase):	1✔
30	def setUp(self) -> None:	1✔
31	self.factory = RequestFactory()	1✔
32	self.path = "/api/v1/terms-accepted-user/"	1✔
33	self.fxa_verify_path = INTROSPECT_TOKEN_URL	1✔
34	self.uid = "relay-user-fxa-uid"	1✔
35
36	def _setup_client(self, token: str) -> None:	1✔
37	self.client = APIClient()	1✔
38	self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")	1✔
39
40	def tearDown(self) -> None:	1✔
41	cache.clear()	1✔
42
43	@responses.activate	1✔
44	def test_201_new_user_created_and_202_user_exists(self) -> None:	1✔
45	email = "user@email.com"	1✔
46	user_token = "user-123"	1✔
47	self._setup_client(user_token)	1✔
48	now_time = int(datetime.now().timestamp())	1✔
49	# Note: FXA iat and exp are timestamps in milliseconds
50	exp_time = (now_time + 60 * 60) * 1000	1✔
51	fxa_response = _setup_fxa_response(	1✔
52	200, {"active": True, "sub": self.uid, "exp": exp_time}
53	)
54	# setup fxa profile response
55	profile_json = {	1✔
56	"email": email,
57	"amrValues": ["pwd", "email"],
58	"twoFactorAuthentication": False,
59	"metricsEnabled": True,
60	"uid": self.uid,
61	"avatar": "https://profile.stage.mozaws.net/v1/avatar/t",
62	"avatarDefault": False,
63	}
64	responses.add(	1✔
65	responses.GET,
66	FXA_PROFILE_URL,
67	status=200,
68	json=profile_json,
69	)
70	cache_key = get_cache_key(user_token)	1✔
71
72	# get fxa response with 201 response for new user and profile created
73	response = self.client.post(self.path)	1✔
74	assert response.status_code == 201	1✔
75	assert hasattr(response, "data")	1✔
76	assert response.data is None	1✔
77	# ensure no session cookie was set
78	assert len(response.cookies.keys()) == 1	1✔
79	assert "csrftoken" in response.cookies	1✔
80	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
81	assert responses.assert_call_count(FXA_PROFILE_URL, 1) is True	1✔
82	assert cache.get(cache_key) == fxa_response	1✔
83	assert SocialAccount.objects.filter(user__email=email).count() == 1	1✔
84	assert Profile.objects.filter(user__email=email).count() == 1	1✔
85	assert Profile.objects.get(user__email=email).created_by == "firefox_resource"	1✔
86
87	# now check that the 2nd call returns 202
88	response = self.client.post(self.path)	1✔
89	assert response.status_code == 202	1✔
90	assert hasattr(response, "data")	1✔
91	assert response.data is None	1✔
92	assert responses.assert_call_count(self.fxa_verify_path, 2) is True	1✔
93	assert responses.assert_call_count(FXA_PROFILE_URL, 1) is True	1✔
94
95	@responses.activate	1✔
96	def test_failed_profile_fetch_for_new_user_returns_500(self) -> None:	1✔
97	user_token = "user-123"	1✔
98	self._setup_client(user_token)	1✔
99	now_time = int(datetime.now().timestamp())	1✔
100	exp_time = (now_time + 60 * 60) * 1000	1✔
101	_setup_fxa_response(200, {"active": True, "sub": self.uid, "exp": exp_time})	1✔
102	# FxA profile server is down
103	responses.add(responses.GET, FXA_PROFILE_URL, status=502, body="")	1✔
104	response = self.client.post(self.path)	1✔
105
106	assert response.status_code == 500	1✔
107	assert response.json()["detail"] == (	1✔
108	"Did not receive a 200 response for account profile."
109	)
110
111	def test_no_authorization_header_returns_400(self) -> None:	1✔
112	client = APIClient()	1✔
113	response = client.post(self.path)	1✔
114
115	assert response.status_code == 400	1✔
116	assert response.json()["detail"] == "Missing Bearer header."	1✔
117
118	def test_no_token_returns_400(self) -> None:	1✔
119	client = APIClient()	1✔
120	client.credentials(HTTP_AUTHORIZATION="Bearer ")	1✔
121	response = client.post(self.path)	1✔
122
123	assert response.status_code == 400	1✔
124	assert response.json()["detail"] == "Missing FXA Token after 'Bearer'."	1✔
125
126	@responses.activate	1✔
127	def test_invalid_bearer_token_error_from_fxa_returns_500_and_cache_returns_500(	1✔
128	self,
129	) -> None:
130	_setup_fxa_response(401, {"error": "401"})	1✔
131	not_found_token = "not-found-123"	1✔
132	self._setup_client(not_found_token)	1✔
133
134	assert cache.get(get_cache_key(not_found_token)) is None	1✔
135
136	response = self.client.post(self.path)	1✔
137	assert response.status_code == 500	1✔
138	assert response.json()["detail"] == "Did not receive a 200 response from FXA."	1✔
139	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
140
141	@responses.activate	1✔
142	def test_jsondecodeerror_returns_401_and_cache_returns_500(	1✔
143	self,
144	) -> None:
145	_setup_fxa_response(200)	1✔
146	invalid_token = "invalid-123"	1✔
147	cache_key = get_cache_key(invalid_token)	1✔
148	self._setup_client(invalid_token)	1✔
149
150	assert cache.get(cache_key) is None	1✔
151
152	# get fxa response with no status code for the first time
153	response = self.client.post(self.path)	1✔
154	assert response.status_code == 401	1✔
155	assert (	1✔
156	response.json()["detail"] == "Jsondecodeerror From Fxa Introspect Response"
157	)
158	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
159
160	@responses.activate	1✔
161	def test_non_200_response_from_fxa_returns_500_and_cache_returns_500(	1✔
162	self,
163	) -> None:
164	now_time = int(datetime.now().timestamp())	1✔
165	# Note: FXA iat and exp are timestamps in milliseconds
166	exp_time = (now_time + 60 * 60) * 1000	1✔
167	_setup_fxa_response(401, {"active": False, "sub": self.uid, "exp": exp_time})	1✔
168	invalid_token = "invalid-123"	1✔
169	cache_key = get_cache_key(invalid_token)	1✔
170	self._setup_client(invalid_token)	1✔
171
172	assert cache.get(cache_key) is None	1✔
173
174	# get fxa response with non-200 response for the first time
175	response = self.client.post(self.path)	1✔
176	assert response.status_code == 500	1✔
177	assert response.json()["detail"] == "Did not receive a 200 response from FXA."	1✔
178	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
179
180	@responses.activate	1✔
181	def test_inactive_fxa_oauth_token_returns_401_and_cache_returns_401(	1✔
182	self,
183	) -> None:
184	now_time = int(datetime.now().timestamp())	1✔
185	# Note: FXA iat and exp are timestamps in milliseconds
186	old_exp_time = (now_time - 60 * 60) * 1000	1✔
187	_setup_fxa_response(	1✔
188	200, {"active": False, "sub": self.uid, "exp": old_exp_time}
189	)
190	invalid_token = "invalid-123"	1✔
191	cache_key = get_cache_key(invalid_token)	1✔
192	self._setup_client(invalid_token)	1✔
193
194	assert cache.get(cache_key) is None	1✔
195
196	# get fxa response with token inactive for the first time
197	response = self.client.post(self.path)	1✔
198	assert response.status_code == 401	1✔
199	assert response.json()["detail"] == "Fxa Returned Active: False For Token."	1✔
200	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
201
202	@responses.activate	1✔
203	def test_fxa_responds_with_no_fxa_uid_returns_404_and_cache_returns_404(	1✔
204	self,
205	) -> None:
206	user_token = "user-123"	1✔
207	now_time = int(datetime.now().timestamp())	1✔
208	# Note: FXA iat and exp are timestamps in milliseconds
209	exp_time = (now_time + 60 * 60) * 1000	1✔
210	_setup_fxa_response(200, {"active": True, "exp": exp_time})	1✔
211	cache_key = get_cache_key(user_token)	1✔
212	self._setup_client(user_token)	1✔
213
214	assert cache.get(cache_key) is None	1✔
215
216	# get fxa response with no fxa uid for the first time
217	response = self.client.post(self.path)	1✔
218	assert response.status_code == 404	1✔
219	assert response.json()["detail"] == "FXA did not return an FXA UID."	1✔
220	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
221
222
223	def _setup_client(token: str) -> APIClient:	1✔
NEW 224	client = APIClient()	×
NEW 225	client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")	×
NEW 226	return client	×

mozilla / fx-private-relay / 84353759-c057-4f20-b282-724c34504dc9

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