0dc159f6-4e4a-4180-94a8-c657384efbeb

Committed 04 Oct 2023 08:37PM CUT coverage: 74.774% (+0.06%) from 74.712%

Build # 0dc159f6-4e4a-4180-94a8-c657384efbeb

Build Type

push

circleci

Committed by

web-flow

Commit Message

Merge pull request #3955 from mozilla/fix-MPP-3420-noreversematch

fix MPP-3420: handle and log NoReverseMatch error

Run Details

1910 of 2769 branches covered (0.0%)

Branch coverage included in aggregate %.

24 of 24 new or added lines in 2 files covered. (100.0%)

6016 of 7831 relevant lines covered (76.82%)

18.35 hits per line

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

86.84

/api/views/__init__.py

import json
import logging
from django.urls.exceptions import NoReverseMatch
import requests
from typing import Mapping, Optional

from django.conf import settings
from django.contrib.auth.models import User
from django.db import IntegrityError

from drf_spectacular.utils import OpenApiResponse, extend_schema
from rest_framework.authentication import get_authorization_header
from rest_framework.exceptions import (
    AuthenticationFailed,
    ParseError,
)
from rest_framework.response import Response
from rest_framework.views import exception_handler

from allauth.account.adapter import get_adapter as get_account_adapter
from allauth.socialaccount.models import SocialAccount
from allauth.socialaccount.helpers import complete_social_login
from allauth.socialaccount.providers.fxa.provider import FirefoxAccountsProvider
from django_filters import rest_framework as filters
from waffle import get_waffle_flag_model
from waffle.models import Switch, Sample
from rest_framework import (
    decorators,
    permissions,
    response,
    status,
    viewsets,
)
from emails.utils import incr_if_enabled

from privaterelay.plans import (
    get_bundle_country_language_mapping,
    get_premium_country_language_mapping,
    get_phone_country_language_mapping,
)
from privaterelay.utils import get_countries_info_from_request_and_mapping

from emails.models import (
    DomainAddress,
    Profile,
    RelayAddress,
)

from ..authentication import get_fxa_uid_from_oauth_token
from ..exceptions import ConflictError, RelayAPIException
from ..permissions import IsOwner, CanManageFlags
from ..serializers import (
    DomainAddressSerializer,
    ProfileSerializer,
    RelayAddressSerializer,
    UserSerializer,
    FlagSerializer,
    WebcompatIssueSerializer,
)

from privaterelay.ftl_bundles import main as ftl_bundle

logger = logging.getLogger("events")
info_logger = logging.getLogger("eventsinfo")
FXA_PROFILE_URL = (
    f"{settings.SOCIALACCOUNT_PROVIDERS['fxa']['PROFILE_ENDPOINT']}/profile"
)


class SaveToRequestUser:
    def perform_create(self, serializer):
        serializer.save(user=self.request.user)


class RelayAddressFilter(filters.FilterSet):
    used_on = filters.CharFilter(field_name="used_on", lookup_expr="icontains")

    class Meta:
        model = RelayAddress
        fields = [
            "enabled",
            "description",
            "generated_for",
            "block_list_emails",
            "used_on",
            # read-only
            "id",
            "address",
            "domain",
            "created_at",
            "last_modified_at",
            "last_used_at",
            "num_forwarded",
            "num_blocked",
            "num_spam",
        ]


class RelayAddressViewSet(SaveToRequestUser, viewsets.ModelViewSet):
    serializer_class = RelayAddressSerializer
    permission_classes = [permissions.IsAuthenticated, IsOwner]
    filterset_class = RelayAddressFilter

    def get_queryset(self):
        return RelayAddress.objects.filter(user=self.request.user)


class DomainAddressFilter(filters.FilterSet):
    used_on = filters.CharFilter(field_name="used_on", lookup_expr="icontains")

    class Meta:
        model = DomainAddress
        fields = [
            "enabled",
            "description",
            "block_list_emails",
            "used_on",
            # read-only
            "id",
            "address",
            "domain",
            "created_at",
            "last_modified_at",
            "last_used_at",
            "num_forwarded",
            "num_blocked",
            "num_spam",
        ]


class DomainAddressViewSet(SaveToRequestUser, viewsets.ModelViewSet):
    serializer_class = DomainAddressSerializer
    permission_classes = [permissions.IsAuthenticated, IsOwner]
    filterset_class = DomainAddressFilter

    def get_queryset(self):
        return DomainAddress.objects.filter(user=self.request.user)

    def perform_create(self, serializer):
        try:
            serializer.save(user=self.request.user)
        except IntegrityError:
            domain_address = DomainAddress.objects.filter(
                user=self.request.user, address=serializer.validated_data.get("address")
            ).first()
            raise ConflictError(
                {"id": domain_address.id, "full_address": domain_address.full_address}
            )


class ProfileViewSet(viewsets.ModelViewSet):
    serializer_class = ProfileSerializer
    permission_classes = [permissions.IsAuthenticated, IsOwner]
    http_method_names = ["get", "post", "head", "put", "patch"]

    def get_queryset(self):
        return Profile.objects.filter(user=self.request.user)


class UserViewSet(viewsets.ModelViewSet):
    serializer_class = UserSerializer
    permission_classes = [permissions.IsAuthenticated, IsOwner]
    http_method_names = ["get", "head"]

    def get_queryset(self):
        return User.objects.filter(id=self.request.user.id)


@extend_schema(
    responses={
        201: OpenApiResponse(description="Created; returned when user is created."),
        202: OpenApiResponse(
            description="Accepted; returned when user already exists."
        ),
        400: OpenApiResponse(
            description="Bad request; returned when request is missing Authorization: Bearer header or token value."
        ),
        401: OpenApiResponse(
            description="Unauthorized; returned when the FXA token is invalid or expired."
        ),
    },
)
@decorators.api_view(["POST"])
@decorators.permission_classes([permissions.AllowAny])
@decorators.authentication_classes([])
def terms_accepted_user(request):
    """
    Create a Relay user from an FXA token.

    See [API Auth doc][api-auth-doc] for details.

    [api-auth-doc]: https://github.com/mozilla/fx-private-relay/blob/main/docs/api_auth.md#firefox-oauth-token-authentication-and-accept-terms-of-service
    """
    # Setting authentication_classes to empty due to
    # authentication still happening despite permissions being set to allowany
    # https://forum.djangoproject.com/t/solved-allowany-override-does-not-work-on-apiview/9754
    authorization = get_authorization_header(request).decode()
    if not authorization or not authorization.startswith("Bearer "):
        raise ParseError("Missing Bearer header.")

    token = authorization.split(" ")[1]
    if token == "":
        raise ParseError("Missing FXA Token after 'Bearer'.")

    try:
        fxa_uid = get_fxa_uid_from_oauth_token(token, use_cache=False)
    except AuthenticationFailed as e:
        # AuthenticationFailed exception returns 403 instead of 401 because we are not
        # using the proper config that comes with the authentication_classes
        # Read more: https://www.django-rest-framework.org/api-guide/authentication/#custom-authentication
        return response.Response(
            data={"detail": e.detail.title()}, status=e.status_code
        )
    status_code = 201

    try:
        sa = SocialAccount.objects.get(uid=fxa_uid, provider="fxa")
        status_code = 202
    except SocialAccount.DoesNotExist:
        # User does not exist, create a new Relay user
        fxa_profile_resp = requests.get(
            FXA_PROFILE_URL, headers={"Authorization": f"Bearer {token}"}
        )

        # this is not exactly the request object that FirefoxAccountsProvider expects, but
        # it has all of the necssary attributes to initiatlize the Provider
        provider = FirefoxAccountsProvider(request)
        # This may not save the new user that was created
        # https://github.com/pennersr/django-allauth/blob/77368a84903d32283f07a260819893ec15df78fb/allauth/socialaccount/providers/base/provider.py#L44
        social_login = provider.sociallogin_from_response(
            request, json.loads(fxa_profile_resp.content)
        )
        # Complete social login is called by callback
        # (see https://github.com/pennersr/django-allauth/blob/77368a84903d32283f07a260819893ec15df78fb/allauth/socialaccount/providers/oauth/views.py#L118)
        # which is what we are mimicking to
        # create new SocialAccount, User, and Profile for the new Relay user from Firefox
        # Since this is a Resource Provider/Server flow and are NOT a Relying Party (RP) of FXA
        # No social token information is stored (no Social Token object created).
        try:
            complete_social_login(request, social_login)
            # complete_social_login writes ['account_verified_email', 'user_created', '_auth_user_id', '_auth_user_backend', '_auth_user_hash']
            # on request.session which sets the cookie because complete_social_login does the "login"
            # The user did not actually log in, logout to clear the session
            if request.user.is_authenticated:
                get_account_adapter(request).logout(request)
        except NoReverseMatch as e:
            # TODO: use this logging to fix the underlying issue
            # https://mozilla-hub.atlassian.net/browse/MPP-3473
            if "socialaccount_signup" in e.args[0]:
                logger.error(
                    "socialaccount_signup_error",
                    extra={
                        "exception": str(e),
                        "fxa_uid": fxa_uid,
                        "social_login_state": social_login.state,
                    },
                )
                return response.Response(status=500)
            raise e
        sa = SocialAccount.objects.get(uid=fxa_uid, provider="fxa")
        # Indicate profile was created from the resource flow
        profile = sa.user.profile
        profile.created_by = "firefox_resource"
        profile.save()
    info_logger.info(
        "terms_accepted_user",
        extra={"social_account": sa.uid, "status_code": status_code},
    )
    return response.Response(status=status_code)


@decorators.api_view()
@decorators.permission_classes([permissions.AllowAny])
def runtime_data(request):
    flags = get_waffle_flag_model().get_all()
    flag_values = [(f.name, f.is_active(request)) for f in flags]
    switches = Switch.get_all()
    switch_values = [(s.name, s.is_active()) for s in switches]
    samples = Sample.get_all()
    sample_values = [(s.name, s.is_active()) for s in samples]
    return response.Response(
        {
            "FXA_ORIGIN": settings.FXA_BASE_ORIGIN,
            "PERIODICAL_PREMIUM_PRODUCT_ID": settings.PERIODICAL_PREMIUM_PROD_ID,
            "GOOGLE_ANALYTICS_ID": settings.GOOGLE_ANALYTICS_ID,
            "BUNDLE_PRODUCT_ID": settings.BUNDLE_PROD_ID,
            "PHONE_PRODUCT_ID": settings.PHONE_PROD_ID,
            "PERIODICAL_PREMIUM_PLANS": get_countries_info_from_request_and_mapping(
                request, get_premium_country_language_mapping()
            ),
            "PHONE_PLANS": get_countries_info_from_request_and_mapping(
                request, get_phone_country_language_mapping()
            ),
            "BUNDLE_PLANS": get_countries_info_from_request_and_mapping(
                request, get_bundle_country_language_mapping()
            ),
            "BASKET_ORIGIN": settings.BASKET_ORIGIN,
            "WAFFLE_FLAGS": flag_values,
            "WAFFLE_SWITCHES": switch_values,
            "WAFFLE_SAMPLES": sample_values,
            "MAX_MINUTES_TO_VERIFY_REAL_PHONE": (
                settings.MAX_MINUTES_TO_VERIFY_REAL_PHONE
            ),
        }
    )


class FlagFilter(filters.FilterSet):
    class Meta:
        model = get_waffle_flag_model()
        fields = [
            "name",
            "everyone",
            # "users",
            # read-only
            "id",
        ]


class FlagViewSet(viewsets.ModelViewSet):
    serializer_class = FlagSerializer
    permission_classes = [permissions.IsAuthenticated, CanManageFlags]
    filterset_class = FlagFilter
    http_method_names = ["get", "post", "head", "patch"]

    def get_queryset(self):
        flags = get_waffle_flag_model().objects
        return flags


@decorators.permission_classes([permissions.IsAuthenticated])
@extend_schema(methods=["POST"], request=WebcompatIssueSerializer)
@decorators.api_view(["POST"])
def report_webcompat_issue(request):
    serializer = WebcompatIssueSerializer(data=request.data)
    if serializer.is_valid():
        info_logger.info("webcompat_issue", extra=serializer.data)
        incr_if_enabled("webcompat_issue", 1)
        for k, v in serializer.data.items():
            if v and k != "issue_on_domain":
                incr_if_enabled(f"webcompat_issue_{k}", 1)
        return response.Response(status=status.HTTP_201_CREATED)
    return response.Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)


def relay_exception_handler(exc: Exception, context: Mapping) -> Optional[Response]:
    """
    Add error information to response data.

    When the error is a RelayAPIException, these additional fields may be present and
    the information will be translated if an Accept-Language header is added to the request:

    error_code - A string identifying the error, for client-side translation
    error_context - Additional data needed for client-side translation
    """

    response = exception_handler(exc, context)

    if response and isinstance(exc, RelayAPIException):
        error_codes = exc.get_codes()
        error_context = exc.error_context()
        if isinstance(error_codes, str):
            response.data["error_code"] = error_codes

            # Build Fluent error ID
            ftl_id_sub = "api-error-"
            ftl_id_error = error_codes.replace("_", "-")
            ftl_id = ftl_id_sub + ftl_id_error

            # Replace default message with Fluent string
            response.data["detail"] = ftl_bundle.format(ftl_id, error_context)

        if error_context:
            response.data["error_context"] = error_context

        response.data["error_code"] = error_codes

    return response

1	import json	1✔
2	import logging	1✔
3	from django.urls.exceptions import NoReverseMatch	1✔
4	import requests	1✔
5	from typing import Mapping, Optional	1✔
6
7	from django.conf import settings	1✔
8	from django.contrib.auth.models import User	1✔
9	from django.db import IntegrityError	1✔
10
11	from drf_spectacular.utils import OpenApiResponse, extend_schema	1✔
12	from rest_framework.authentication import get_authorization_header	1✔
13	from rest_framework.exceptions import (	1✔
14	AuthenticationFailed,
15	ParseError,
16	)
17	from rest_framework.response import Response	1✔
18	from rest_framework.views import exception_handler	1✔
19
20	from allauth.account.adapter import get_adapter as get_account_adapter	1✔
21	from allauth.socialaccount.models import SocialAccount	1✔
22	from allauth.socialaccount.helpers import complete_social_login	1✔
23	from allauth.socialaccount.providers.fxa.provider import FirefoxAccountsProvider	1✔
24	from django_filters import rest_framework as filters	1✔
25	from waffle import get_waffle_flag_model	1✔
26	from waffle.models import Switch, Sample	1✔
27	from rest_framework import (	1✔
28	decorators,
29	permissions,
30	response,
31	status,
32	viewsets,
33	)
34	from emails.utils import incr_if_enabled	1✔
35
36	from privaterelay.plans import (	1✔
37	get_bundle_country_language_mapping,
38	get_premium_country_language_mapping,
39	get_phone_country_language_mapping,
40	)
41	from privaterelay.utils import get_countries_info_from_request_and_mapping	1✔
42
43	from emails.models import (	1✔
44	DomainAddress,
45	Profile,
46	RelayAddress,
47	)
48
49	from ..authentication import get_fxa_uid_from_oauth_token	1✔
50	from ..exceptions import ConflictError, RelayAPIException	1✔
51	from ..permissions import IsOwner, CanManageFlags	1✔
52	from ..serializers import (	1✔
53	DomainAddressSerializer,
54	ProfileSerializer,
55	RelayAddressSerializer,
56	UserSerializer,
57	FlagSerializer,
58	WebcompatIssueSerializer,
59	)
60
61	from privaterelay.ftl_bundles import main as ftl_bundle	1✔
62
63	logger = logging.getLogger("events")	1✔
64	info_logger = logging.getLogger("eventsinfo")	1✔
65	FXA_PROFILE_URL = (	1✔
66	f"{settings.SOCIALACCOUNT_PROVIDERS['fxa']['PROFILE_ENDPOINT']}/profile"
67	)
68
69
70	class SaveToRequestUser:	1✔
71	def perform_create(self, serializer):	1✔
72	serializer.save(user=self.request.user)	1✔
73
74
75	class RelayAddressFilter(filters.FilterSet):	1✔
76	used_on = filters.CharFilter(field_name="used_on", lookup_expr="icontains")	1✔
77
78	class Meta:	1✔
79	model = RelayAddress	1✔
80	fields = [	1✔
81	"enabled",
82	"description",
83	"generated_for",
84	"block_list_emails",
85	"used_on",
86	# read-only
87	"id",
88	"address",
89	"domain",
90	"created_at",
91	"last_modified_at",
92	"last_used_at",
93	"num_forwarded",
94	"num_blocked",
95	"num_spam",
96	]
97
98
99	class RelayAddressViewSet(SaveToRequestUser, viewsets.ModelViewSet):	1✔
100	serializer_class = RelayAddressSerializer	1✔
101	permission_classes = [permissions.IsAuthenticated, IsOwner]	1✔
102	filterset_class = RelayAddressFilter	1✔
103
104	def get_queryset(self):	1✔
105	return RelayAddress.objects.filter(user=self.request.user)	1✔
106
107
108	class DomainAddressFilter(filters.FilterSet):	1✔
109	used_on = filters.CharFilter(field_name="used_on", lookup_expr="icontains")	1✔
110
111	class Meta:	1✔
112	model = DomainAddress	1✔
113	fields = [	1✔
114	"enabled",
115	"description",
116	"block_list_emails",
117	"used_on",
118	# read-only
119	"id",
120	"address",
121	"domain",
122	"created_at",
123	"last_modified_at",
124	"last_used_at",
125	"num_forwarded",
126	"num_blocked",
127	"num_spam",
128	]
129
130
131	class DomainAddressViewSet(SaveToRequestUser, viewsets.ModelViewSet):	1✔
132	serializer_class = DomainAddressSerializer	1✔
133	permission_classes = [permissions.IsAuthenticated, IsOwner]	1✔
134	filterset_class = DomainAddressFilter	1✔
135
136	def get_queryset(self):	1✔
137	return DomainAddress.objects.filter(user=self.request.user)	×
138
139	def perform_create(self, serializer):	1✔
140	try:	1✔
141	serializer.save(user=self.request.user)	1✔
142	except IntegrityError:	1✔
143	domain_address = DomainAddress.objects.filter(	×
144	user=self.request.user, address=serializer.validated_data.get("address")
145	).first()
146	raise ConflictError(	×
147	{"id": domain_address.id, "full_address": domain_address.full_address}
148	)
149
150
151	class ProfileViewSet(viewsets.ModelViewSet):	1✔
152	serializer_class = ProfileSerializer	1✔
153	permission_classes = [permissions.IsAuthenticated, IsOwner]	1✔
154	http_method_names = ["get", "post", "head", "put", "patch"]	1✔
155
156	def get_queryset(self):	1✔
157	return Profile.objects.filter(user=self.request.user)	1✔
158
159
160	class UserViewSet(viewsets.ModelViewSet):	1✔
161	serializer_class = UserSerializer	1✔
162	permission_classes = [permissions.IsAuthenticated, IsOwner]	1✔
163	http_method_names = ["get", "head"]	1✔
164
165	def get_queryset(self):	1✔
166	return User.objects.filter(id=self.request.user.id)	×
167
168
169	@extend_schema(	1✔
170	responses={
171	201: OpenApiResponse(description="Created; returned when user is created."),
172	202: OpenApiResponse(
173	description="Accepted; returned when user already exists."
174	),
175	400: OpenApiResponse(
176	description="Bad request; returned when request is missing Authorization: Bearer header or token value."
177	),
178	401: OpenApiResponse(
179	description="Unauthorized; returned when the FXA token is invalid or expired."
180	),
181	},
182	)
183	@decorators.api_view(["POST"])	1✔
184	@decorators.permission_classes([permissions.AllowAny])	1✔
185	@decorators.authentication_classes([])	1✔
186	def terms_accepted_user(request):	1✔
187	"""
188	Create a Relay user from an FXA token.
189
190	See [API Auth doc][api-auth-doc] for details.
191
192	[api-auth-doc]: https://github.com/mozilla/fx-private-relay/blob/main/docs/api_auth.md#firefox-oauth-token-authentication-and-accept-terms-of-service
193	"""
194	# Setting authentication_classes to empty due to
195	# authentication still happening despite permissions being set to allowany
196	# https://forum.djangoproject.com/t/solved-allowany-override-does-not-work-on-apiview/9754
197	authorization = get_authorization_header(request).decode()	1✔
198	if not authorization or not authorization.startswith("Bearer "):	1✔
199	raise ParseError("Missing Bearer header.")	1✔
200
201	token = authorization.split(" ")[1]	1✔
202	if token == "":	1✔
203	raise ParseError("Missing FXA Token after 'Bearer'.")	1✔
204
205	try:	1✔
206	fxa_uid = get_fxa_uid_from_oauth_token(token, use_cache=False)	1✔
207	except AuthenticationFailed as e:	1✔
208	# AuthenticationFailed exception returns 403 instead of 401 because we are not
209	# using the proper config that comes with the authentication_classes
210	# Read more: https://www.django-rest-framework.org/api-guide/authentication/#custom-authentication
211	return response.Response(	1✔
212	data={"detail": e.detail.title()}, status=e.status_code
213	)
214	status_code = 201	1✔
215
216	try:	1✔
217	sa = SocialAccount.objects.get(uid=fxa_uid, provider="fxa")	1✔
218	status_code = 202	1✔
219	except SocialAccount.DoesNotExist:	1✔
220	# User does not exist, create a new Relay user
221	fxa_profile_resp = requests.get(	1✔
222	FXA_PROFILE_URL, headers={"Authorization": f"Bearer {token}"}
223	)
224
225	# this is not exactly the request object that FirefoxAccountsProvider expects, but
226	# it has all of the necssary attributes to initiatlize the Provider
227	provider = FirefoxAccountsProvider(request)	1✔
228	# This may not save the new user that was created
229	# https://github.com/pennersr/django-allauth/blob/77368a84903d32283f07a260819893ec15df78fb/allauth/socialaccount/providers/base/provider.py#L44
230	social_login = provider.sociallogin_from_response(	1✔
231	request, json.loads(fxa_profile_resp.content)
232	)
233	# Complete social login is called by callback
234	# (see https://github.com/pennersr/django-allauth/blob/77368a84903d32283f07a260819893ec15df78fb/allauth/socialaccount/providers/oauth/views.py#L118)
235	# which is what we are mimicking to
236	# create new SocialAccount, User, and Profile for the new Relay user from Firefox
237	# Since this is a Resource Provider/Server flow and are NOT a Relying Party (RP) of FXA
238	# No social token information is stored (no Social Token object created).
239	try:	1✔
240	complete_social_login(request, social_login)	1✔
241	# complete_social_login writes ['account_verified_email', 'user_created', '_auth_user_id', '_auth_user_backend', '_auth_user_hash']
242	# on request.session which sets the cookie because complete_social_login does the "login"
243	# The user did not actually log in, logout to clear the session
244	if request.user.is_authenticated:	1!
245	get_account_adapter(request).logout(request)	1✔
246	except NoReverseMatch as e:	1✔
247	# TODO: use this logging to fix the underlying issue
248	# https://mozilla-hub.atlassian.net/browse/MPP-3473
249	if "socialaccount_signup" in e.args[0]:	1!
250	logger.error(	1✔
251	"socialaccount_signup_error",
252	extra={
253	"exception": str(e),
254	"fxa_uid": fxa_uid,
255	"social_login_state": social_login.state,
256	},
257	)
258	return response.Response(status=500)	1✔
259	raise e	×
260	sa = SocialAccount.objects.get(uid=fxa_uid, provider="fxa")	1✔
261	# Indicate profile was created from the resource flow
262	profile = sa.user.profile	1✔
263	profile.created_by = "firefox_resource"	1✔
264	profile.save()	1✔
265	info_logger.info(	1✔
266	"terms_accepted_user",
267	extra={"social_account": sa.uid, "status_code": status_code},
268	)
269	return response.Response(status=status_code)	1✔
270
271
272	@decorators.api_view()	1✔
273	@decorators.permission_classes([permissions.AllowAny])	1✔
274	def runtime_data(request):	1✔
275	flags = get_waffle_flag_model().get_all()	1✔
276	flag_values = [(f.name, f.is_active(request)) for f in flags]	1✔
277	switches = Switch.get_all()	1✔
278	switch_values = [(s.name, s.is_active()) for s in switches]	1✔
279	samples = Sample.get_all()	1✔
280	sample_values = [(s.name, s.is_active()) for s in samples]	1✔
281	return response.Response(	1✔
282	{
283	"FXA_ORIGIN": settings.FXA_BASE_ORIGIN,
284	"PERIODICAL_PREMIUM_PRODUCT_ID": settings.PERIODICAL_PREMIUM_PROD_ID,
285	"GOOGLE_ANALYTICS_ID": settings.GOOGLE_ANALYTICS_ID,
286	"BUNDLE_PRODUCT_ID": settings.BUNDLE_PROD_ID,
287	"PHONE_PRODUCT_ID": settings.PHONE_PROD_ID,
288	"PERIODICAL_PREMIUM_PLANS": get_countries_info_from_request_and_mapping(
289	request, get_premium_country_language_mapping()
290	),
291	"PHONE_PLANS": get_countries_info_from_request_and_mapping(
292	request, get_phone_country_language_mapping()
293	),
294	"BUNDLE_PLANS": get_countries_info_from_request_and_mapping(
295	request, get_bundle_country_language_mapping()
296	),
297	"BASKET_ORIGIN": settings.BASKET_ORIGIN,
298	"WAFFLE_FLAGS": flag_values,
299	"WAFFLE_SWITCHES": switch_values,
300	"WAFFLE_SAMPLES": sample_values,
301	"MAX_MINUTES_TO_VERIFY_REAL_PHONE": (
302	settings.MAX_MINUTES_TO_VERIFY_REAL_PHONE
303	),
304	}
305	)
306
307
308	class FlagFilter(filters.FilterSet):	1✔
309	class Meta:	1✔
310	model = get_waffle_flag_model()	1✔
311	fields = [	1✔
312	"name",
313	"everyone",
314	# "users",
315	# read-only
316	"id",
317	]
318
319
320	class FlagViewSet(viewsets.ModelViewSet):	1✔
321	serializer_class = FlagSerializer	1✔
322	permission_classes = [permissions.IsAuthenticated, CanManageFlags]	1✔
323	filterset_class = FlagFilter	1✔
324	http_method_names = ["get", "post", "head", "patch"]	1✔
325
326	def get_queryset(self):	1✔
327	flags = get_waffle_flag_model().objects	×
328	return flags	×
329
330
331	@decorators.permission_classes([permissions.IsAuthenticated])	1✔
332	@extend_schema(methods=["POST"], request=WebcompatIssueSerializer)	1✔
333	@decorators.api_view(["POST"])	1✔
334	def report_webcompat_issue(request):	1✔
335	serializer = WebcompatIssueSerializer(data=request.data)	×
336	if serializer.is_valid():	×
337	info_logger.info("webcompat_issue", extra=serializer.data)	×
338	incr_if_enabled("webcompat_issue", 1)	×
339	for k, v in serializer.data.items():	×
340	if v and k != "issue_on_domain":	×
341	incr_if_enabled(f"webcompat_issue_{k}", 1)	×
342	return response.Response(status=status.HTTP_201_CREATED)	×
343	return response.Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)	×
344
345
346	def relay_exception_handler(exc: Exception, context: Mapping) -> Optional[Response]:	1✔
347	"""
348	Add error information to response data.
349
350	When the error is a RelayAPIException, these additional fields may be present and
351	the information will be translated if an Accept-Language header is added to the request:
352
353	error_code - A string identifying the error, for client-side translation
354	error_context - Additional data needed for client-side translation
355	"""
356
357	response = exception_handler(exc, context)	1✔
358
359	if response and isinstance(exc, RelayAPIException):	1✔
360	error_codes = exc.get_codes()	1✔
361	error_context = exc.error_context()	1✔
362	if isinstance(error_codes, str):	1!
363	response.data["error_code"] = error_codes	1✔
364
365	# Build Fluent error ID
366	ftl_id_sub = "api-error-"	1✔
367	ftl_id_error = error_codes.replace("_", "-")	1✔
368	ftl_id = ftl_id_sub + ftl_id_error	1✔
369
370	# Replace default message with Fluent string
371	response.data["detail"] = ftl_bundle.format(ftl_id, error_context)	1✔
372
373	if error_context:	1✔
374	response.data["error_context"] = error_context	1✔
375
376	response.data["error_code"] = error_codes	1✔
377
378	return response	1✔

mozilla / fx-private-relay / 0dc159f6-4e4a-4180-94a8-c657384efbeb

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