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

98.38

/api/tests/authentication_2024_tests.py

from datetime import datetime
from typing import NotRequired, TypedDict

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

import responses
from allauth.socialaccount.models import SocialAccount
from model_bakery import baker
from rest_framework.exceptions import APIException, AuthenticationFailed, NotFound
from rest_framework.test import APIClient

from ..authentication import (
    FXA_TOKEN_AUTH_OLD_AND_PROVEN,
    INTROSPECT_TOKEN_URL,
)
from ..authentication import (
    FxaTokenAuthentication2024 as FxaTokenAuthentication,
)
from ..authentication import (
    get_cache_key_2024 as get_cache_key,
)
from ..authentication import (
    get_fxa_uid_from_oauth_token_2024 as get_fxa_uid_from_oauth_token,
)
from ..authentication import (
    introspect_token_2024 as introspect_token,
)

MOCK_BASE = "api.authentication"

# TODO MPP-3527 - Many tests mock FxA responses. This one should specify that it is
# mocking the introspection URL. It could also be refactored to a pytest fixture, or a
# nullable.


class FxaResponse(TypedDict, total=False):
    active: bool
    sub: str
    exp: int
    error: str


class CachedFxaResponse(TypedDict):
    status_code: int
    json: NotRequired[FxaResponse | str]


def _setup_fxa_response(
    status_code: int, json: FxaResponse | str | None = None
) -> CachedFxaResponse:
    responses.add(
        responses.POST,
        INTROSPECT_TOKEN_URL,
        status=status_code,
        json=json,
    )
    if json is None:
        return {"status_code": status_code}
    return {"status_code": status_code, "json": json}


@override_settings(FXA_TOKEN_AUTH_VERSION=FXA_TOKEN_AUTH_OLD_AND_PROVEN)
class AuthenticationMiscellaneous(TestCase):
    def setUp(self):
        self.auth = FxaTokenAuthentication
        self.factory = RequestFactory()
        self.path = "/api/v1/relayaddresses"
        self.fxa_verify_path = INTROSPECT_TOKEN_URL
        self.uid = "relay-user-fxa-uid"

    def tearDown(self):
        cache.clear()

    @responses.activate
    def test_introspect_token_catches_JSONDecodeError_raises_AuthenticationFailed(self):
        _setup_fxa_response(200)
        invalid_token = "invalid-123"

        try:
            introspect_token(invalid_token)
        except AuthenticationFailed as e:
            assert str(e.detail) == "JSONDecodeError from FXA introspect response"
            assert responses.assert_call_count(self.fxa_verify_path, 1) is True
            return
        self.fail("Should have raised AuthenticationFailed")

    @responses.activate
    def test_introspect_token_returns_fxa_introspect_response(self):
        now_time = int(datetime.now().timestamp())
        # Note: FXA iat and exp are timestamps in *milliseconds*
        exp_time = (now_time + 60 * 60) * 1000
        json_data: FxaResponse = {
            "active": True,
            "sub": self.uid,
            "exp": exp_time,
        }
        status_code = 200
        expected_fxa_resp_data = {"status_code": status_code, "json": json_data}
        _setup_fxa_response(status_code, json_data)
        valid_token = "valid-123"
        cache_key = get_cache_key(valid_token)

        assert cache.get(cache_key) is None

        fxa_resp_data = introspect_token(valid_token)
        assert responses.assert_call_count(self.fxa_verify_path, 1) is True
        assert fxa_resp_data == expected_fxa_resp_data

    @responses.activate
    def test_get_fxa_uid_from_oauth_token_returns_cached_response(self):
        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
        fxa_response = _setup_fxa_response(
            200, {"active": True, "sub": self.uid, "exp": exp_time}
        )
        cache_key = get_cache_key(user_token)

        assert cache.get(cache_key) is None

        # get FxA uid for the first time
        fxa_uid = get_fxa_uid_from_oauth_token(user_token)
        assert fxa_uid == self.uid
        assert responses.assert_call_count(self.fxa_verify_path, 1) is True
        assert cache.get(cache_key) == fxa_response

        # now check that the 2nd call did NOT make another fxa request
        fxa_uid = get_fxa_uid_from_oauth_token(user_token)
        assert fxa_uid == self.uid
        assert responses.assert_call_count(self.fxa_verify_path, 1) is True

    @responses.activate
    def test_get_fxa_uid_from_oauth_token_status_code_None_uses_cached_response_returns_error_response(  # noqa: E501
        self,
    ) -> None:
        _setup_fxa_response(200)
        invalid_token = "invalid-123"
        cache_key = get_cache_key(invalid_token)

        assert cache.get(cache_key) is None

        # get fxa response with no status code for the first time
        try:
            get_fxa_uid_from_oauth_token(invalid_token)
        except AuthenticationFailed as e:
            assert str(e.detail) == "JSONDecodeError from FXA introspect response"
            assert responses.assert_call_count(self.fxa_verify_path, 1) is True
        assert cache.get(cache_key) == {"status_code": None, "json": {}}

        # now check that the 2nd call did NOT make another fxa request
        try:
            get_fxa_uid_from_oauth_token(invalid_token)
        except APIException as e:
            assert str(e.detail) == "Previous FXA call failed, wait to retry."
            assert responses.assert_call_count(self.fxa_verify_path, 1) is True
            return
        self.fail("Should have raised APIException")

    @responses.activate
    def test_get_fxa_uid_from_oauth_token_status_code_not_200_uses_cached_response_returns_error_response(  # noqa: E501
        self,
    ) -> None:
        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(
            401, {"active": False, "sub": self.uid, "exp": exp_time}
        )
        invalid_token = "invalid-123"
        cache_key = get_cache_key(invalid_token)

        assert cache.get(cache_key) is None

        # get fxa response with none 200 response for the first time
        try:
            get_fxa_uid_from_oauth_token(invalid_token)
        except APIException as e:
            assert str(e.detail) == "Did not receive a 200 response from FXA."
            assert responses.assert_call_count(self.fxa_verify_path, 1) is True
        assert cache.get(cache_key) == fxa_response

        # now check that the 2nd call did NOT make another fxa request
        try:
            get_fxa_uid_from_oauth_token(invalid_token)
        except APIException as e:
            assert str(e.detail) == "Did not receive a 200 response from FXA."
            assert responses.assert_call_count(self.fxa_verify_path, 1) is True
            return
        self.fail("Should have raised APIException")

    @responses.activate
    def test_get_fxa_uid_from_oauth_token_not_active_uses_cached_response_returns_error_response(  # noqa: E501
        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
        fxa_response = _setup_fxa_response(
            200, {"active": False, "sub": self.uid, "exp": old_exp_time}
        )
        invalid_token = "invalid-123"
        cache_key = get_cache_key(invalid_token)

        assert cache.get(cache_key) is None

        # get fxa response with token inactive for the first time
        try:
            get_fxa_uid_from_oauth_token(invalid_token)
        except AuthenticationFailed as e:
            assert str(e.detail) == "FXA returned active: False for token."
            assert responses.assert_call_count(self.fxa_verify_path, 1) is True
        assert cache.get(cache_key) == fxa_response

        # now check that the 2nd call did NOT make another fxa request
        try:
            get_fxa_uid_from_oauth_token(invalid_token)
        except AuthenticationFailed as e:
            assert str(e.detail) == "FXA returned active: False for token."
            assert responses.assert_call_count(self.fxa_verify_path, 1) is True
            return
        self.fail("Should have raised AuthenticationFailed")

    @responses.activate
    def test_get_fxa_uid_from_oauth_token_returns_fxa_response_with_no_fxa_uid(self):
        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
        fxa_response = _setup_fxa_response(200, {"active": True, "exp": exp_time})
        cache_key = get_cache_key(user_token)

        assert cache.get(cache_key) is None

        # get fxa response with no fxa uid for the first time
        try:
            get_fxa_uid_from_oauth_token(user_token)
        except NotFound as e:
            assert str(e.detail) == "FXA did not return an FXA UID."
            assert responses.assert_call_count(self.fxa_verify_path, 1) is True
        assert cache.get(cache_key) == fxa_response

        # now check that the 2nd call did NOT make another fxa request
        try:
            get_fxa_uid_from_oauth_token(user_token)
        except NotFound as e:
            assert str(e.detail) == "FXA did not return an FXA UID."
            assert responses.assert_call_count(self.fxa_verify_path, 1) is True
            return
        self.fail("Should have raised AuthenticationFailed")


@override_settings(FXA_TOKEN_AUTH_VERSION=FXA_TOKEN_AUTH_OLD_AND_PROVEN)
class FxaTokenAuthenticationTest(TestCase):
    def setUp(self) -> None:
        self.auth = FxaTokenAuthentication()
        self.factory = RequestFactory()
        self.path = "/api/v1/relayaddresses/"
        self.fxa_verify_path = INTROSPECT_TOKEN_URL
        self.uid = "relay-user-fxa-uid"

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

    def test_no_authorization_header_returns_none(self) -> None:
        get_addresses_req = self.factory.get(self.path)
        assert self.auth.authenticate(get_addresses_req) is None

    def test_no_bearer_in_authorization_returns_none(self) -> None:
        headers = {"Authorization": "unexpected 123"}
        get_addresses_req = self.factory.get(self.path, headers=headers)
        assert self.auth.authenticate(get_addresses_req) is None

    def test_no_token_returns_400(self) -> None:
        client = APIClient()
        client.credentials(HTTP_AUTHORIZATION="Bearer ")
        response = client.get("/api/v1/relayaddresses/")
        assert response.status_code == 400
        assert response.json()["detail"] == "Missing FXA Token after 'Bearer'."

    @responses.activate
    def test_non_200_resp_from_fxa_raises_error_and_caches(self) -> None:
        fxa_response = _setup_fxa_response(401, {"error": "401"})
        not_found_token = "not-found-123"
        client = APIClient()
        client.credentials(HTTP_AUTHORIZATION=f"Bearer {not_found_token}")

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

        response = client.get("/api/v1/relayaddresses/")
        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
        assert cache.get(get_cache_key(not_found_token)) == fxa_response

        # now check that the code does NOT make another fxa request
        response = client.get("/api/v1/relayaddresses/")
        assert responses.assert_call_count(self.fxa_verify_path, 1) is True

    @responses.activate
    def test_non_200_non_json_resp_from_fxa_raises_error_and_caches(self) -> None:
        fxa_response = _setup_fxa_response(503, "Bad Gateway")
        not_found_token = "fxa-gw-error"
        client = APIClient()
        client.credentials(HTTP_AUTHORIZATION=f"Bearer {not_found_token}")

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

        response = client.get("/api/v1/relayaddresses/")
        assert response.status_code == 500

        assert responses.assert_call_count(self.fxa_verify_path, 1) is True
        assert cache.get(get_cache_key(not_found_token)) == fxa_response

        # now check that the code does NOT make another fxa request
        response = client.get("/api/v1/relayaddresses/")
        assert responses.assert_call_count(self.fxa_verify_path, 1) is True

    @responses.activate
    def test_inactive_token_responds_with_401(self) -> None:
        fxa_response = _setup_fxa_response(200, {"active": False})
        inactive_token = "inactive-123"
        client = APIClient()
        client.credentials(HTTP_AUTHORIZATION=f"Bearer {inactive_token}")

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

        response = client.get("/api/v1/relayaddresses/")
        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
        assert cache.get(get_cache_key(inactive_token)) == fxa_response

        # now check that the code does NOT make another fxa request
        response = client.get("/api/v1/relayaddresses/")
        assert responses.assert_call_count(self.fxa_verify_path, 1) is True

    @responses.activate
    def test_200_resp_from_fxa_no_matching_user_raises_APIException(self) -> None:
        fxa_response = _setup_fxa_response(
            200, {"active": True, "sub": "not-a-relay-user"}
        )
        non_user_token = "non-user-123"
        client = APIClient()
        client.credentials(HTTP_AUTHORIZATION=f"Bearer {non_user_token}")

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

        response = client.get("/api/v1/relayaddresses/")
        assert response.status_code == 403
        expected_detail = (
            "Authenticated user does not have a Relay account."
            " Have they accepted the terms?"
        )
        assert response.json()["detail"] == expected_detail
        assert cache.get(get_cache_key(non_user_token)) == fxa_response

        # the code does NOT make another fxa request
        response = client.get("/api/v1/relayaddresses/")
        assert responses.assert_call_count(self.fxa_verify_path, 1) is True

    @responses.activate
    def test_200_resp_from_fxa_inactive_Relay_user_raises_APIException(self) -> None:
        sa: SocialAccount = baker.make(SocialAccount, uid=self.uid, provider="fxa")
        sa.user.is_active = False
        sa.user.save()
        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, "sub": self.uid, "exp": exp_time})
        inactive_user_token = "inactive-user-123"
        client = APIClient()
        client.credentials(HTTP_AUTHORIZATION=f"Bearer {inactive_user_token}")

        response = client.get("/api/v1/relayaddresses/")
        assert response.status_code == 403
        expected_detail = (
            "Authenticated user does not have an active Relay account."
            " Have they been deactivated?"
        )
        assert response.json()["detail"] == expected_detail

    @responses.activate
    def test_200_resp_from_fxa_for_user_returns_user_and_caches(self) -> None:
        sa: SocialAccount = baker.make(SocialAccount, uid=self.uid, provider="fxa")
        user_token = "user-123"
        client = APIClient()
        client.credentials(HTTP_AUTHORIZATION=f"Bearer {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}
        )

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

        # check the endpoint status code
        response = client.get("/api/v1/relayaddresses/")
        assert response.status_code == 200
        assert responses.assert_call_count(self.fxa_verify_path, 1) is True
        assert cache.get(get_cache_key(user_token)) == fxa_response

        # check the function returns the right user
        headers = {"Authorization": f"Bearer {user_token}"}
        get_addresses_req = self.factory.get(self.path, headers=headers)
        auth_return = self.auth.authenticate(get_addresses_req)
        assert auth_return == (sa.user, user_token)

        # now check that the 2nd call did NOT make another fxa request
        assert responses.assert_call_count(self.fxa_verify_path, 1) is True
        assert cache.get(get_cache_key(user_token)) == fxa_response

    @responses.activate
    def test_write_requests_make_calls_to_fxa(self) -> None:
        sa: SocialAccount = baker.make(SocialAccount, uid=self.uid, provider="fxa")
        user_token = "user-123"
        client = APIClient()
        client.credentials(HTTP_AUTHORIZATION=f"Bearer {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}
        )

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

        # check the endpoint status code
        response = client.get("/api/v1/relayaddresses/")
        assert response.status_code == 200
        assert responses.assert_call_count(self.fxa_verify_path, 1) is True
        assert cache.get(get_cache_key(user_token)) == fxa_response

        # check the function returns the right user
        headers = {"Authorization": f"Bearer {user_token}"}
        get_addresses_req = self.factory.get(self.path, headers=headers)
        auth_return = self.auth.authenticate(get_addresses_req)
        assert auth_return == (sa.user, user_token)

        # now check that the 2nd GET request did NOT make another fxa request
        assert responses.assert_call_count(self.fxa_verify_path, 1) is True
        assert cache.get(get_cache_key(user_token)) == fxa_response

        headers = {"Authorization": f"Bearer {user_token}"}

        # send POST to /api/v1/relayaddresses and check that cache is used - i.e.,
        # FXA is *NOT* called
        post_addresses_req = self.factory.post(self.path, headers=headers)
        auth_return = self.auth.authenticate(post_addresses_req)
        assert responses.assert_call_count(self.fxa_verify_path, 1) is True

        # send POST to another API endpoint and check that cache is NOT used
        post_webcompat = self.factory.post(
            "/api/v1/report_webcompat_issue", headers=headers
        )
        auth_return = self.auth.authenticate(post_webcompat)
        assert responses.assert_call_count(self.fxa_verify_path, 2) is True

        # send other write requests and check that FXA *IS* called
        put_addresses_req = self.factory.put(self.path, headers=headers)
        auth_return = self.auth.authenticate(put_addresses_req)
        assert responses.assert_call_count(self.fxa_verify_path, 3) is True

        delete_addresses_req = self.factory.delete(self.path, headers=headers)
        auth_return = self.auth.authenticate(delete_addresses_req)
        assert responses.assert_call_count(self.fxa_verify_path, 4) is True

1	from datetime import datetime	1✔
2	from typing import NotRequired, TypedDict	1✔
3
4	from django.core.cache import cache	1✔
5	from django.test import RequestFactory, TestCase, override_settings	1✔
6
7	import responses	1✔
8	from allauth.socialaccount.models import SocialAccount	1✔
9	from model_bakery import baker	1✔
10	from rest_framework.exceptions import APIException, AuthenticationFailed, NotFound	1✔
11	from rest_framework.test import APIClient	1✔
12
13	from ..authentication import (	1✔
14	FXA_TOKEN_AUTH_OLD_AND_PROVEN,
15	INTROSPECT_TOKEN_URL,
16	)
17	from ..authentication import (	1✔
18	FxaTokenAuthentication2024 as FxaTokenAuthentication,
19	)
20	from ..authentication import (	1✔
21	get_cache_key_2024 as get_cache_key,
22	)
23	from ..authentication import (	1✔
24	get_fxa_uid_from_oauth_token_2024 as get_fxa_uid_from_oauth_token,
25	)
26	from ..authentication import (	1✔
27	introspect_token_2024 as introspect_token,
28	)
29
30	MOCK_BASE = "api.authentication"	1✔
31
32	# TODO MPP-3527 - Many tests mock FxA responses. This one should specify that it is
33	# mocking the introspection URL. It could also be refactored to a pytest fixture, or a
34	# nullable.
35
36
37	class FxaResponse(TypedDict, total=False):	1✔
38	active: bool	1✔
39	sub: str	1✔
40	exp: int	1✔
41	error: str	1✔
42
43
44	class CachedFxaResponse(TypedDict):	1✔
45	status_code: int	1✔
46	json: NotRequired[FxaResponse \| str]	1✔
47
48
49	def _setup_fxa_response(	1✔
50	status_code: int, json: FxaResponse \| str \| None = None
51	) -> CachedFxaResponse:
52	responses.add(	1✔
53	responses.POST,
54	INTROSPECT_TOKEN_URL,
55	status=status_code,
56	json=json,
57	)
58	if json is None:	1✔
59	return {"status_code": status_code}	1✔
60	return {"status_code": status_code, "json": json}	1✔
61
62
63	@override_settings(FXA_TOKEN_AUTH_VERSION=FXA_TOKEN_AUTH_OLD_AND_PROVEN)	1✔
64	class AuthenticationMiscellaneous(TestCase):	1✔
65	def setUp(self):	1✔
66	self.auth = FxaTokenAuthentication	1✔
67	self.factory = RequestFactory()	1✔
68	self.path = "/api/v1/relayaddresses"	1✔
69	self.fxa_verify_path = INTROSPECT_TOKEN_URL	1✔
70	self.uid = "relay-user-fxa-uid"	1✔
71
72	def tearDown(self):	1✔
73	cache.clear()	1✔
74
75	@responses.activate	1✔
76	def test_introspect_token_catches_JSONDecodeError_raises_AuthenticationFailed(self):	1✔
77	_setup_fxa_response(200)	1✔
78	invalid_token = "invalid-123"	1✔
79
80	try:	1✔
81	introspect_token(invalid_token)	1✔
82	except AuthenticationFailed as e:	1✔
83	assert str(e.detail) == "JSONDecodeError from FXA introspect response"	1✔
84	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
85	return	1✔
NEW 86	self.fail("Should have raised AuthenticationFailed")	×
87
88	@responses.activate	1✔
89	def test_introspect_token_returns_fxa_introspect_response(self):	1✔
90	now_time = int(datetime.now().timestamp())	1✔
91	# Note: FXA iat and exp are timestamps in milliseconds
92	exp_time = (now_time + 60 * 60) * 1000	1✔
93	json_data: FxaResponse = {	1✔
94	"active": True,
95	"sub": self.uid,
96	"exp": exp_time,
97	}
98	status_code = 200	1✔
99	expected_fxa_resp_data = {"status_code": status_code, "json": json_data}	1✔
100	_setup_fxa_response(status_code, json_data)	1✔
101	valid_token = "valid-123"	1✔
102	cache_key = get_cache_key(valid_token)	1✔
103
104	assert cache.get(cache_key) is None	1✔
105
106	fxa_resp_data = introspect_token(valid_token)	1✔
107	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
108	assert fxa_resp_data == expected_fxa_resp_data	1✔
109
110	@responses.activate	1✔
111	def test_get_fxa_uid_from_oauth_token_returns_cached_response(self):	1✔
112	user_token = "user-123"	1✔
113	now_time = int(datetime.now().timestamp())	1✔
114	# Note: FXA iat and exp are timestamps in milliseconds
115	exp_time = (now_time + 60 * 60) * 1000	1✔
116	fxa_response = _setup_fxa_response(	1✔
117	200, {"active": True, "sub": self.uid, "exp": exp_time}
118	)
119	cache_key = get_cache_key(user_token)	1✔
120
121	assert cache.get(cache_key) is None	1✔
122
123	# get FxA uid for the first time
124	fxa_uid = get_fxa_uid_from_oauth_token(user_token)	1✔
125	assert fxa_uid == self.uid	1✔
126	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
127	assert cache.get(cache_key) == fxa_response	1✔
128
129	# now check that the 2nd call did NOT make another fxa request
130	fxa_uid = get_fxa_uid_from_oauth_token(user_token)	1✔
131	assert fxa_uid == self.uid	1✔
132	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
133
134	@responses.activate	1✔
135	def test_get_fxa_uid_from_oauth_token_status_code_None_uses_cached_response_returns_error_response( # noqa: E501	1✔
136	self,
137	) -> None:
138	_setup_fxa_response(200)	1✔
139	invalid_token = "invalid-123"	1✔
140	cache_key = get_cache_key(invalid_token)	1✔
141
142	assert cache.get(cache_key) is None	1✔
143
144	# get fxa response with no status code for the first time
145	try:	1✔
146	get_fxa_uid_from_oauth_token(invalid_token)	1✔
147	except AuthenticationFailed as e:	1✔
148	assert str(e.detail) == "JSONDecodeError from FXA introspect response"	1✔
149	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
150	assert cache.get(cache_key) == {"status_code": None, "json": {}}	1✔
151
152	# now check that the 2nd call did NOT make another fxa request
153	try:	1✔
154	get_fxa_uid_from_oauth_token(invalid_token)	1✔
155	except APIException as e:	1✔
156	assert str(e.detail) == "Previous FXA call failed, wait to retry."	1✔
157	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
158	return	1✔
NEW 159	self.fail("Should have raised APIException")	×
160
161	@responses.activate	1✔
162	def test_get_fxa_uid_from_oauth_token_status_code_not_200_uses_cached_response_returns_error_response( # noqa: E501	1✔
163	self,
164	) -> None:
165	now_time = int(datetime.now().timestamp())	1✔
166	# Note: FXA iat and exp are timestamps in milliseconds
167	exp_time = (now_time + 60 * 60) * 1000	1✔
168	fxa_response = _setup_fxa_response(	1✔
169	401, {"active": False, "sub": self.uid, "exp": exp_time}
170	)
171	invalid_token = "invalid-123"	1✔
172	cache_key = get_cache_key(invalid_token)	1✔
173
174	assert cache.get(cache_key) is None	1✔
175
176	# get fxa response with none 200 response for the first time
177	try:	1✔
178	get_fxa_uid_from_oauth_token(invalid_token)	1✔
179	except APIException as e:	1✔
180	assert str(e.detail) == "Did not receive a 200 response from FXA."	1✔
181	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
182	assert cache.get(cache_key) == fxa_response	1✔
183
184	# now check that the 2nd call did NOT make another fxa request
185	try:	1✔
186	get_fxa_uid_from_oauth_token(invalid_token)	1✔
187	except APIException as e:	1✔
188	assert str(e.detail) == "Did not receive a 200 response from FXA."	1✔
189	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
190	return	1✔
NEW 191	self.fail("Should have raised APIException")	×
192
193	@responses.activate	1✔
194	def test_get_fxa_uid_from_oauth_token_not_active_uses_cached_response_returns_error_response( # noqa: E501	1✔
195	self,
196	) -> None:
197	now_time = int(datetime.now().timestamp())	1✔
198	# Note: FXA iat and exp are timestamps in milliseconds
199	old_exp_time = (now_time - 60 * 60) * 1000	1✔
200	fxa_response = _setup_fxa_response(	1✔
201	200, {"active": False, "sub": self.uid, "exp": old_exp_time}
202	)
203	invalid_token = "invalid-123"	1✔
204	cache_key = get_cache_key(invalid_token)	1✔
205
206	assert cache.get(cache_key) is None	1✔
207
208	# get fxa response with token inactive for the first time
209	try:	1✔
210	get_fxa_uid_from_oauth_token(invalid_token)	1✔
211	except AuthenticationFailed as e:	1✔
212	assert str(e.detail) == "FXA returned active: False for token."	1✔
213	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
214	assert cache.get(cache_key) == fxa_response	1✔
215
216	# now check that the 2nd call did NOT make another fxa request
217	try:	1✔
218	get_fxa_uid_from_oauth_token(invalid_token)	1✔
219	except AuthenticationFailed as e:	1✔
220	assert str(e.detail) == "FXA returned active: False for token."	1✔
221	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
222	return	1✔
NEW 223	self.fail("Should have raised AuthenticationFailed")	×
224
225	@responses.activate	1✔
226	def test_get_fxa_uid_from_oauth_token_returns_fxa_response_with_no_fxa_uid(self):	1✔
227	user_token = "user-123"	1✔
228	now_time = int(datetime.now().timestamp())	1✔
229	# Note: FXA iat and exp are timestamps in milliseconds
230	exp_time = (now_time + 60 * 60) * 1000	1✔
231	fxa_response = _setup_fxa_response(200, {"active": True, "exp": exp_time})	1✔
232	cache_key = get_cache_key(user_token)	1✔
233
234	assert cache.get(cache_key) is None	1✔
235
236	# get fxa response with no fxa uid for the first time
237	try:	1✔
238	get_fxa_uid_from_oauth_token(user_token)	1✔
239	except NotFound as e:	1✔
240	assert str(e.detail) == "FXA did not return an FXA UID."	1✔
241	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
242	assert cache.get(cache_key) == fxa_response	1✔
243
244	# now check that the 2nd call did NOT make another fxa request
245	try:	1✔
246	get_fxa_uid_from_oauth_token(user_token)	1✔
247	except NotFound as e:	1✔
248	assert str(e.detail) == "FXA did not return an FXA UID."	1✔
249	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
250	return	1✔
NEW 251	self.fail("Should have raised AuthenticationFailed")	×
252
253
254	@override_settings(FXA_TOKEN_AUTH_VERSION=FXA_TOKEN_AUTH_OLD_AND_PROVEN)	1✔
255	class FxaTokenAuthenticationTest(TestCase):	1✔
256	def setUp(self) -> None:	1✔
257	self.auth = FxaTokenAuthentication()	1✔
258	self.factory = RequestFactory()	1✔
259	self.path = "/api/v1/relayaddresses/"	1✔
260	self.fxa_verify_path = INTROSPECT_TOKEN_URL	1✔
261	self.uid = "relay-user-fxa-uid"	1✔
262
263	def tearDown(self) -> None:	1✔
264	cache.clear()	1✔
265
266	def test_no_authorization_header_returns_none(self) -> None:	1✔
267	get_addresses_req = self.factory.get(self.path)	1✔
268	assert self.auth.authenticate(get_addresses_req) is None	1✔
269
270	def test_no_bearer_in_authorization_returns_none(self) -> None:	1✔
271	headers = {"Authorization": "unexpected 123"}	1✔
272	get_addresses_req = self.factory.get(self.path, headers=headers)	1✔
273	assert self.auth.authenticate(get_addresses_req) is None	1✔
274
275	def test_no_token_returns_400(self) -> None:	1✔
276	client = APIClient()	1✔
277	client.credentials(HTTP_AUTHORIZATION="Bearer ")	1✔
278	response = client.get("/api/v1/relayaddresses/")	1✔
279	assert response.status_code == 400	1✔
280	assert response.json()["detail"] == "Missing FXA Token after 'Bearer'."	1✔
281
282	@responses.activate	1✔
283	def test_non_200_resp_from_fxa_raises_error_and_caches(self) -> None:	1✔
284	fxa_response = _setup_fxa_response(401, {"error": "401"})	1✔
285	not_found_token = "not-found-123"	1✔
286	client = APIClient()	1✔
287	client.credentials(HTTP_AUTHORIZATION=f"Bearer {not_found_token}")	1✔
288
289	assert cache.get(get_cache_key(not_found_token)) is None	1✔
290
291	response = client.get("/api/v1/relayaddresses/")	1✔
292	assert response.status_code == 500	1✔
293	assert response.json()["detail"] == "Did not receive a 200 response from FXA."	1✔
294
295	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
296	assert cache.get(get_cache_key(not_found_token)) == fxa_response	1✔
297
298	# now check that the code does NOT make another fxa request
299	response = client.get("/api/v1/relayaddresses/")	1✔
300	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
301
302	@responses.activate	1✔
303	def test_non_200_non_json_resp_from_fxa_raises_error_and_caches(self) -> None:	1✔
304	fxa_response = _setup_fxa_response(503, "Bad Gateway")	1✔
305	not_found_token = "fxa-gw-error"	1✔
306	client = APIClient()	1✔
307	client.credentials(HTTP_AUTHORIZATION=f"Bearer {not_found_token}")	1✔
308
309	assert cache.get(get_cache_key(not_found_token)) is None	1✔
310
311	response = client.get("/api/v1/relayaddresses/")	1✔
312	assert response.status_code == 500	1✔
313
314	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
315	assert cache.get(get_cache_key(not_found_token)) == fxa_response	1✔
316
317	# now check that the code does NOT make another fxa request
318	response = client.get("/api/v1/relayaddresses/")	1✔
319	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
320
321	@responses.activate	1✔
322	def test_inactive_token_responds_with_401(self) -> None:	1✔
323	fxa_response = _setup_fxa_response(200, {"active": False})	1✔
324	inactive_token = "inactive-123"	1✔
325	client = APIClient()	1✔
326	client.credentials(HTTP_AUTHORIZATION=f"Bearer {inactive_token}")	1✔
327
328	assert cache.get(get_cache_key(inactive_token)) is None	1✔
329
330	response = client.get("/api/v1/relayaddresses/")	1✔
331	assert response.status_code == 401	1✔
332	assert response.json()["detail"] == "FXA returned active: False for token."	1✔
333	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
334	assert cache.get(get_cache_key(inactive_token)) == fxa_response	1✔
335
336	# now check that the code does NOT make another fxa request
337	response = client.get("/api/v1/relayaddresses/")	1✔
338	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
339
340	@responses.activate	1✔
341	def test_200_resp_from_fxa_no_matching_user_raises_APIException(self) -> None:	1✔
342	fxa_response = _setup_fxa_response(	1✔
343	200, {"active": True, "sub": "not-a-relay-user"}
344	)
345	non_user_token = "non-user-123"	1✔
346	client = APIClient()	1✔
347	client.credentials(HTTP_AUTHORIZATION=f"Bearer {non_user_token}")	1✔
348
349	assert cache.get(get_cache_key(non_user_token)) is None	1✔
350
351	response = client.get("/api/v1/relayaddresses/")	1✔
352	assert response.status_code == 403	1✔
353	expected_detail = (	1✔
354	"Authenticated user does not have a Relay account."
355	" Have they accepted the terms?"
356	)
357	assert response.json()["detail"] == expected_detail	1✔
358	assert cache.get(get_cache_key(non_user_token)) == fxa_response	1✔
359
360	# the code does NOT make another fxa request
361	response = client.get("/api/v1/relayaddresses/")	1✔
362	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
363
364	@responses.activate	1✔
365	def test_200_resp_from_fxa_inactive_Relay_user_raises_APIException(self) -> None:	1✔
366	sa: SocialAccount = baker.make(SocialAccount, uid=self.uid, provider="fxa")	1✔
367	sa.user.is_active = False	1✔
368	sa.user.save()	1✔
369	now_time = int(datetime.now().timestamp())	1✔
370	# Note: FXA iat and exp are timestamps in milliseconds
371	exp_time = (now_time + 60 * 60) * 1000	1✔
372	_setup_fxa_response(200, {"active": True, "sub": self.uid, "exp": exp_time})	1✔
373	inactive_user_token = "inactive-user-123"	1✔
374	client = APIClient()	1✔
375	client.credentials(HTTP_AUTHORIZATION=f"Bearer {inactive_user_token}")	1✔
376
377	response = client.get("/api/v1/relayaddresses/")	1✔
378	assert response.status_code == 403	1✔
379	expected_detail = (	1✔
380	"Authenticated user does not have an active Relay account."
381	" Have they been deactivated?"
382	)
383	assert response.json()["detail"] == expected_detail	1✔
384
385	@responses.activate	1✔
386	def test_200_resp_from_fxa_for_user_returns_user_and_caches(self) -> None:	1✔
387	sa: SocialAccount = baker.make(SocialAccount, uid=self.uid, provider="fxa")	1✔
388	user_token = "user-123"	1✔
389	client = APIClient()	1✔
390	client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")	1✔
391	now_time = int(datetime.now().timestamp())	1✔
392	# Note: FXA iat and exp are timestamps in milliseconds
393	exp_time = (now_time + 60 * 60) * 1000	1✔
394	fxa_response = _setup_fxa_response(	1✔
395	200, {"active": True, "sub": self.uid, "exp": exp_time}
396	)
397
398	assert cache.get(get_cache_key(user_token)) is None	1✔
399
400	# check the endpoint status code
401	response = client.get("/api/v1/relayaddresses/")	1✔
402	assert response.status_code == 200	1✔
403	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
404	assert cache.get(get_cache_key(user_token)) == fxa_response	1✔
405
406	# check the function returns the right user
407	headers = {"Authorization": f"Bearer {user_token}"}	1✔
408	get_addresses_req = self.factory.get(self.path, headers=headers)	1✔
409	auth_return = self.auth.authenticate(get_addresses_req)	1✔
410	assert auth_return == (sa.user, user_token)	1✔
411
412	# now check that the 2nd call did NOT make another fxa request
413	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
414	assert cache.get(get_cache_key(user_token)) == fxa_response	1✔
415
416	@responses.activate	1✔
417	def test_write_requests_make_calls_to_fxa(self) -> None:	1✔
418	sa: SocialAccount = baker.make(SocialAccount, uid=self.uid, provider="fxa")	1✔
419	user_token = "user-123"	1✔
420	client = APIClient()	1✔
421	client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")	1✔
422	now_time = int(datetime.now().timestamp())	1✔
423	# Note: FXA iat and exp are timestamps in milliseconds
424	exp_time = (now_time + 60 * 60) * 1000	1✔
425	fxa_response = _setup_fxa_response(	1✔
426	200, {"active": True, "sub": self.uid, "exp": exp_time}
427	)
428
429	assert cache.get(get_cache_key(user_token)) is None	1✔
430
431	# check the endpoint status code
432	response = client.get("/api/v1/relayaddresses/")	1✔
433	assert response.status_code == 200	1✔
434	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
435	assert cache.get(get_cache_key(user_token)) == fxa_response	1✔
436
437	# check the function returns the right user
438	headers = {"Authorization": f"Bearer {user_token}"}	1✔
439	get_addresses_req = self.factory.get(self.path, headers=headers)	1✔
440	auth_return = self.auth.authenticate(get_addresses_req)	1✔
441	assert auth_return == (sa.user, user_token)	1✔
442
443	# now check that the 2nd GET request did NOT make another fxa request
444	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
445	assert cache.get(get_cache_key(user_token)) == fxa_response	1✔
446
447	headers = {"Authorization": f"Bearer {user_token}"}	1✔
448
449	# send POST to /api/v1/relayaddresses and check that cache is used - i.e.,
450	# FXA is NOT called
451	post_addresses_req = self.factory.post(self.path, headers=headers)	1✔
452	auth_return = self.auth.authenticate(post_addresses_req)	1✔
453	assert responses.assert_call_count(self.fxa_verify_path, 1) is True	1✔
454
455	# send POST to another API endpoint and check that cache is NOT used
456	post_webcompat = self.factory.post(	1✔
457	"/api/v1/report_webcompat_issue", headers=headers
458	)
459	auth_return = self.auth.authenticate(post_webcompat)	1✔
460	assert responses.assert_call_count(self.fxa_verify_path, 2) is True	1✔
461
462	# send other write requests and check that FXA IS called
463	put_addresses_req = self.factory.put(self.path, headers=headers)	1✔
464	auth_return = self.auth.authenticate(put_addresses_req)	1✔
465	assert responses.assert_call_count(self.fxa_verify_path, 3) is True	1✔
466
467	delete_addresses_req = self.factory.delete(self.path, headers=headers)	1✔
468	auth_return = self.auth.authenticate(delete_addresses_req)	1✔
469	assert responses.assert_call_count(self.fxa_verify_path, 4) is True	1✔

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