e4e29e9a-abb7-44b4-b62b-1f2e81ba3cba

Committed 25 Aug 2025 05:10PM UTC coverage: 88.1% (+1.8%) from 86.295%

Build # e4e29e9a-abb7-44b4-b62b-1f2e81ba3cba

Build Type

Pull #5761

circleci

Committed by

vpremamozilla

Commit Message

MPP-4153 - test(mask-management): increase frontend test coverage for Mask Management pages and components

Pull Request Pull Request #5761: MPP-4293 - increase frontend test coverage for Mask Management pages and components (part 2)

Run Details

2912 of 3943 branches covered (73.85%)

Branch coverage included in aggregate %.

18136 of 19948 relevant lines covered (90.92%)

11.27 hits per line

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

92.48

/api/views/emails.py

"""API views for emails"""

from logging import getLogger
from typing import Generic, TypeVar

from django.apps import apps
from django.conf import settings
from django.contrib.auth.models import User
from django.core.exceptions import ObjectDoesNotExist
from django.db.models.query import QuerySet
from django.template.loader import render_to_string

import django_ftl
from django_filters import rest_framework as filters
from drf_spectacular.utils import OpenApiResponse, extend_schema
from rest_framework.decorators import api_view, permission_classes, throttle_classes
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.serializers import BaseSerializer
from rest_framework.status import (
    HTTP_201_CREATED,
    HTTP_400_BAD_REQUEST,
    HTTP_404_NOT_FOUND,
)
from rest_framework.throttling import UserRateThrottle
from rest_framework.viewsets import ModelViewSet
from waffle import flag_is_active

from emails.apps import EmailsConfig
from emails.models import DomainAddress, RelayAddress
from emails.utils import generate_from_header, ses_message_props
from emails.views import _get_address, wrap_html_email
from privaterelay.ftl_bundles import main as ftl_bundle
from privaterelay.utils import glean_logger

from ..permissions import IsOwner
from ..serializers.emails import (
    DomainAddressSerializer,
    FirstForwardedEmailSerializer,
    RelayAddressSerializer,
)
from . import SaveToRequestUser

logger = getLogger("events")


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",
        ]


_Address = TypeVar("_Address", RelayAddress, DomainAddress)


class AddressViewSet(Generic[_Address], SaveToRequestUser, ModelViewSet):
    def perform_create(self, serializer: BaseSerializer[_Address]) -> None:
        super().perform_create(serializer)
        if not serializer.instance:
            raise ValueError("serializer.instance must be truthy value.")
        glean_logger().log_email_mask_created(
            request=self.request,
            mask=serializer.instance,
            created_by_api=True,
        )

    def perform_update(self, serializer: BaseSerializer[_Address]) -> None:
        if not serializer.instance:
            raise ValueError("serializer.instance must be truthy value.")
        old_description = serializer.instance.description
        super().perform_update(serializer)
        new_description = serializer.instance.description
        if old_description != new_description:
            glean_logger().log_email_mask_label_updated(
                request=self.request, mask=serializer.instance
            )

    def perform_destroy(self, instance: _Address) -> None:
        user = instance.user
        is_random_mask = isinstance(instance, RelayAddress)
        super().perform_destroy(instance)
        glean_logger().log_email_mask_deleted(
            request=self.request,
            user=user,
            is_random_mask=is_random_mask,
        )


@extend_schema(tags=["emails"])
class RelayAddressViewSet(AddressViewSet[RelayAddress]):
    """An email address with a random name provided by Relay."""

    serializer_class = RelayAddressSerializer
    permission_classes = [IsAuthenticated, IsOwner]
    filterset_class = RelayAddressFilter

    def get_queryset(self) -> QuerySet[RelayAddress]:
        if isinstance(self.request.user, User):
            return RelayAddress.objects.filter(user=self.request.user)
        return RelayAddress.objects.none()


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",
        ]


@extend_schema(tags=["emails"])
class DomainAddressViewSet(AddressViewSet[DomainAddress]):
    """An email address with subdomain chosen by a Relay user."""

    serializer_class = DomainAddressSerializer
    permission_classes = [IsAuthenticated, IsOwner]
    filterset_class = DomainAddressFilter

    def get_queryset(self) -> QuerySet[DomainAddress]:
        if isinstance(self.request.user, User):
            return DomainAddress.objects.filter(
                user=self.request.user
            ).prefetch_related("user", "user__profile")
        return DomainAddress.objects.none()


class FirstForwardedEmailRateThrottle(UserRateThrottle):
    rate = settings.FIRST_EMAIL_RATE_LIMIT


@permission_classes([IsAuthenticated])
@extend_schema(
    tags=["emails"],
    request=FirstForwardedEmailSerializer,
    responses={
        201: OpenApiResponse(description="Email sent to user."),
        400: OpenApiResponse(description="Invalid mask."),
        401: OpenApiResponse(description="Authentication required."),
        403: OpenApiResponse(description="Flag 'free_user_onboarding' is required."),
        404: OpenApiResponse(description="Unable to find the mask."),
    },
)
@api_view(["POST"])
@throttle_classes([FirstForwardedEmailRateThrottle])
def first_forwarded_email(request):
    """
    Requires `free_user_onboarding` flag to be active for the user.

    Send the `first_forwarded_email.html` email to the user via a mask.
    See [/emails/first_forwarded_email](/emails/first_forwarded_email).

    Note: `mask` value must be a `RelayAddress` that belongs to the authenticated user.
    A `DomainAddress` will not work.
    """
    if not flag_is_active(request, "free_user_onboarding"):
        # Return Permission Denied error
        return Response(
            {"detail": "Requires free_user_onboarding waffle flag."}, status=403
        )

    serializer = FirstForwardedEmailSerializer(data=request.data)
    if not serializer.is_valid():
        return Response(serializer.errors, status=HTTP_400_BAD_REQUEST)

    mask = str(serializer.data.get("mask"))
    user = request.user
    try:
        address = _get_address(mask)
        RelayAddress.objects.get(user=user, address=address)
    except ObjectDoesNotExist:
        return Response(f"{mask} does not exist for user.", status=HTTP_404_NOT_FOUND)
    profile = user.profile
    app_config = apps.get_app_config("emails")
    if not isinstance(app_config, EmailsConfig):
        raise TypeError("app_config must be type EmailsConfig")
    ses_client = app_config.ses_client
    if not ses_client:
        raise ValueError("ses_client must be truthy value.")
    if not settings.RELAY_FROM_ADDRESS:
        raise ValueError("settings.RELAY_FROM_ADDRESS must have a value.")
    with django_ftl.override(profile.language):
        translated_subject = ftl_bundle.format("forwarded-email-hero-header")
    first_forwarded_email_html = render_to_string(
        "emails/first_forwarded_email.html",
        {
            "SITE_ORIGIN": settings.SITE_ORIGIN,
        },
    )
    from_address = generate_from_header(settings.RELAY_FROM_ADDRESS, mask)
    wrapped_email = wrap_html_email(
        first_forwarded_email_html,
        profile.language,
        profile.has_premium,
        from_address,
        0,
    )
    ses_client.send_email(
        Destination={
            "ToAddresses": [user.email],
        },
        Source=from_address,
        Message={
            "Subject": ses_message_props(translated_subject),
            "Body": {
                "Html": ses_message_props(wrapped_email),
            },
        },
    )
    logger.info(f"Sent first_forwarded_email to user ID: {user.id}")
    return Response(status=HTTP_201_CREATED)

1	"""API views for emails"""
2
3	from logging import getLogger	1✔
4	from typing import Generic, TypeVar	1✔
5
6	from django.apps import apps	1✔
7	from django.conf import settings	1✔
8	from django.contrib.auth.models import User	1✔
9	from django.core.exceptions import ObjectDoesNotExist	1✔
10	from django.db.models.query import QuerySet	1✔
11	from django.template.loader import render_to_string	1✔
12
13	import django_ftl	1✔
14	from django_filters import rest_framework as filters	1✔
15	from drf_spectacular.utils import OpenApiResponse, extend_schema	1✔
16	from rest_framework.decorators import api_view, permission_classes, throttle_classes	1✔
17	from rest_framework.permissions import IsAuthenticated	1✔
18	from rest_framework.response import Response	1✔
19	from rest_framework.serializers import BaseSerializer	1✔
20	from rest_framework.status import (	1✔
21	HTTP_201_CREATED,
22	HTTP_400_BAD_REQUEST,
23	HTTP_404_NOT_FOUND,
24	)
25	from rest_framework.throttling import UserRateThrottle	1✔
26	from rest_framework.viewsets import ModelViewSet	1✔
27	from waffle import flag_is_active	1✔
28
29	from emails.apps import EmailsConfig	1✔
30	from emails.models import DomainAddress, RelayAddress	1✔
31	from emails.utils import generate_from_header, ses_message_props	1✔
32	from emails.views import _get_address, wrap_html_email	1✔
33	from privaterelay.ftl_bundles import main as ftl_bundle	1✔
34	from privaterelay.utils import glean_logger	1✔
35
36	from ..permissions import IsOwner	1✔
37	from ..serializers.emails import (	1✔
38	DomainAddressSerializer,
39	FirstForwardedEmailSerializer,
40	RelayAddressSerializer,
41	)
42	from . import SaveToRequestUser	1✔
43
44	logger = getLogger("events")	1✔
45
46
47	class RelayAddressFilter(filters.FilterSet):	1✔
48	used_on = filters.CharFilter(field_name="used_on", lookup_expr="icontains")	1✔
49
50	class Meta:	1✔
51	model = RelayAddress	1✔
52	fields = [	1✔
53	"enabled",
54	"description",
55	"generated_for",
56	"block_list_emails",
57	"used_on",
58	# read-only
59	"id",
60	"address",
61	"domain",
62	"created_at",
63	"last_modified_at",
64	"last_used_at",
65	"num_forwarded",
66	"num_blocked",
67	"num_spam",
68	]
69
70
71	_Address = TypeVar("_Address", RelayAddress, DomainAddress)	1✔
72
73
74	class AddressViewSet(Generic[_Address], SaveToRequestUser, ModelViewSet):	1✔
75	def perform_create(self, serializer: BaseSerializer[_Address]) -> None:	1✔
76	super().perform_create(serializer)	1✔
77	if not serializer.instance:	1!
78	raise ValueError("serializer.instance must be truthy value.")	×
79	glean_logger().log_email_mask_created(	1✔
80	request=self.request,
81	mask=serializer.instance,
82	created_by_api=True,
83	)
84
85	def perform_update(self, serializer: BaseSerializer[_Address]) -> None:	1✔
86	if not serializer.instance:	1!
87	raise ValueError("serializer.instance must be truthy value.")	×
88	old_description = serializer.instance.description	1✔
89	super().perform_update(serializer)	1✔
90	new_description = serializer.instance.description	1✔
91	if old_description != new_description:	1✔
92	glean_logger().log_email_mask_label_updated(	1✔
93	request=self.request, mask=serializer.instance
94	)
95
96	def perform_destroy(self, instance: _Address) -> None:	1✔
97	user = instance.user	1✔
98	is_random_mask = isinstance(instance, RelayAddress)	1✔
99	super().perform_destroy(instance)	1✔
100	glean_logger().log_email_mask_deleted(	1✔
101	request=self.request,
102	user=user,
103	is_random_mask=is_random_mask,
104	)
105
106
107	@extend_schema(tags=["emails"])	1✔
108	class RelayAddressViewSet(AddressViewSet[RelayAddress]):	1✔
109	"""An email address with a random name provided by Relay."""
110
111	serializer_class = RelayAddressSerializer	1✔
112	permission_classes = [IsAuthenticated, IsOwner]	1✔
113	filterset_class = RelayAddressFilter	1✔
114
115	def get_queryset(self) -> QuerySet[RelayAddress]:	1✔
116	if isinstance(self.request.user, User):	1✔
117	return RelayAddress.objects.filter(user=self.request.user)	1✔
118	return RelayAddress.objects.none()	1✔
119
120
121	class DomainAddressFilter(filters.FilterSet):	1✔
122	used_on = filters.CharFilter(field_name="used_on", lookup_expr="icontains")	1✔
123
124	class Meta:	1✔
125	model = DomainAddress	1✔
126	fields = [	1✔
127	"enabled",
128	"description",
129	"block_list_emails",
130	"used_on",
131	# read-only
132	"id",
133	"address",
134	"domain",
135	"created_at",
136	"last_modified_at",
137	"last_used_at",
138	"num_forwarded",
139	"num_blocked",
140	"num_spam",
141	]
142
143
144	@extend_schema(tags=["emails"])	1✔
145	class DomainAddressViewSet(AddressViewSet[DomainAddress]):	1✔
146	"""An email address with subdomain chosen by a Relay user."""
147
148	serializer_class = DomainAddressSerializer	1✔
149	permission_classes = [IsAuthenticated, IsOwner]	1✔
150	filterset_class = DomainAddressFilter	1✔
151
152	def get_queryset(self) -> QuerySet[DomainAddress]:	1✔
153	if isinstance(self.request.user, User):	1✔
154	return DomainAddress.objects.filter(	1✔
155	user=self.request.user
156	).prefetch_related("user", "user__profile")
157	return DomainAddress.objects.none()	1✔
158
159
160	class FirstForwardedEmailRateThrottle(UserRateThrottle):	1✔
161	rate = settings.FIRST_EMAIL_RATE_LIMIT	1✔
162
163
164	@permission_classes([IsAuthenticated])	1✔
165	@extend_schema(	1✔
166	tags=["emails"],
167	request=FirstForwardedEmailSerializer,
168	responses={
169	201: OpenApiResponse(description="Email sent to user."),
170	400: OpenApiResponse(description="Invalid mask."),
171	401: OpenApiResponse(description="Authentication required."),
172	403: OpenApiResponse(description="Flag 'free_user_onboarding' is required."),
173	404: OpenApiResponse(description="Unable to find the mask."),
174	},
175	)
176	@api_view(["POST"])	1✔
177	@throttle_classes([FirstForwardedEmailRateThrottle])	1✔
178	def first_forwarded_email(request):	1✔
179	"""
180	Requires `free_user_onboarding` flag to be active for the user.
181
182	Send the `first_forwarded_email.html` email to the user via a mask.
183	See [/emails/first_forwarded_email](/emails/first_forwarded_email).
184
185	Note: `mask` value must be a `RelayAddress` that belongs to the authenticated user.
186	A `DomainAddress` will not work.
187	"""
188	if not flag_is_active(request, "free_user_onboarding"):	1✔
189	# Return Permission Denied error
190	return Response(	1✔
191	{"detail": "Requires free_user_onboarding waffle flag."}, status=403
192	)
193
194	serializer = FirstForwardedEmailSerializer(data=request.data)	1✔
195	if not serializer.is_valid():	1✔
196	return Response(serializer.errors, status=HTTP_400_BAD_REQUEST)	1✔
197
198	mask = str(serializer.data.get("mask"))	1✔
199	user = request.user	1✔
200	try:	1✔
201	address = _get_address(mask)	1✔
202	RelayAddress.objects.get(user=user, address=address)	1✔
203	except ObjectDoesNotExist:	1✔
204	return Response(f"{mask} does not exist for user.", status=HTTP_404_NOT_FOUND)	1✔
205	profile = user.profile	1✔
206	app_config = apps.get_app_config("emails")	1✔
207	if not isinstance(app_config, EmailsConfig):	1!
208	raise TypeError("app_config must be type EmailsConfig")	×
209	ses_client = app_config.ses_client	1✔
210	if not ses_client:	1!
211	raise ValueError("ses_client must be truthy value.")	×
212	if not settings.RELAY_FROM_ADDRESS:	1!
213	raise ValueError("settings.RELAY_FROM_ADDRESS must have a value.")	×
214	with django_ftl.override(profile.language):	1✔
215	translated_subject = ftl_bundle.format("forwarded-email-hero-header")	1✔
216	first_forwarded_email_html = render_to_string(	1✔
217	"emails/first_forwarded_email.html",
218	{
219	"SITE_ORIGIN": settings.SITE_ORIGIN,
220	},
221	)
222	from_address = generate_from_header(settings.RELAY_FROM_ADDRESS, mask)	1✔
223	wrapped_email = wrap_html_email(	1✔
224	first_forwarded_email_html,
225	profile.language,
226	profile.has_premium,
227	from_address,
228	0,
229	)
230	ses_client.send_email(	1✔
231	Destination={
232	"ToAddresses": [user.email],
233	},
234	Source=from_address,
235	Message={
236	"Subject": ses_message_props(translated_subject),
237	"Body": {
238	"Html": ses_message_props(wrapped_email),
239	},
240	},
241	)
242	logger.info(f"Sent first_forwarded_email to user ID: {user.id}")	1✔
243	return Response(status=HTTP_201_CREATED)	1✔

mozilla / fx-private-relay / e4e29e9a-abb7-44b4-b62b-1f2e81ba3cba

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