faea5c47-7f2b-4136-8c9b-7cbe324def9d

Committed 15 Apr 2025 03:05PM CUT coverage: 85.237% (+0.05%) from 85.192%

Build # faea5c47-7f2b-4136-8c9b-7cbe324def9d

Build Type

push

circleci

Committed by

web-flow

Commit Message

Merge pull request #5500 from mozilla/add-api-count-MPP-4012

MPP-4012 - feat(glean): log API access as Glean server event

Run Details

2470 of 3609 branches covered (68.44%)

Branch coverage included in aggregate %.

106 of 109 new or added lines in 7 files covered. (97.25%)

3 existing lines in 2 files now uncovered.

17322 of 19611 relevant lines covered (88.33%)

9.8 hits per line

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

96.3

/privaterelay/glean_interface.py

"""Relay interface to EventsServerEventLogger generated by glean_parser."""

from __future__ import annotations

from datetime import datetime
from logging import getLogger
from typing import Any, Literal, NamedTuple

from django.conf import settings
from django.contrib.auth.models import User
from django.http import HttpRequest

from ipware import get_client_ip

from emails.models import DomainAddress, RelayAddress

from .glean.server_events import GLEAN_EVENT_MOZLOG_TYPE, EventsServerEventLogger
from .types import RELAY_CHANNEL_NAME

# Enumerate the mask setting that caused an email to not be forwarded.
EmailBlockedReason = Literal[
    "block_all",  # The mask is set to block all mail
    "block_promotional",  # The mask is set to block promotional / list mail
]


def _opt_dt_to_glean(value: datetime | None) -> int:
    """Convert an optional datetime to an integer timestamp."""
    if value == datetime.min:
        return -2  # datetime was not checked
    if value is None:
        return -1  # datetime does not exist
    return int(value.timestamp())


def _opt_str_to_glean(value: str | None) -> str:
    """Convert an optional string to a (possibly empty) string."""
    return "" if value is None else value


class RequestData(NamedTuple):
    """Extract and store data from the request."""

    user_agent: str | None = None
    ip_address: str | None = None

    @classmethod
    def from_request(cls, request: HttpRequest) -> RequestData:
        user_agent = request.headers.get("user-agent", None)
        client_ip, is_routable = get_client_ip(request)
        ip_address = client_ip if (client_ip and is_routable) else None
        return cls(user_agent=user_agent, ip_address=ip_address)


class UserData(NamedTuple):
    """Extract and store data from a Relay user."""

    metrics_enabled: bool
    fxa_id: str | None = None
    n_random_masks: int = 0
    n_domain_masks: int = 0
    n_deleted_random_masks: int = 0
    n_deleted_domain_masks: int = 0
    date_joined_relay: datetime | None = None
    date_joined_premium: datetime | None = None
    premium_status: str = ""
    has_extension: bool = False
    date_got_extension: datetime | None = None

    @classmethod
    def from_user(cls, user: User) -> UserData:
        metrics_enabled = user.profile.metrics_enabled
        if not metrics_enabled:
            return cls(metrics_enabled=False)

        fxa_id = user.profile.metrics_fxa_id or None
        n_random_masks = user.relayaddress_set.count()
        n_domain_masks = user.domainaddress_set.count()
        n_deleted_random_masks = user.profile.num_deleted_relay_addresses
        n_deleted_domain_masks = user.profile.num_deleted_domain_addresses
        date_joined_relay = user.date_joined
        if user.profile.has_premium:
            if user.profile.has_phone:
                date_joined_premium = user.profile.date_subscribed_phone
            else:
                date_joined_premium = user.profile.date_subscribed
        else:
            date_joined_premium = None
        premium_status = user.profile.metrics_premium_status
        # Until more accurate date_got_extension is calculated (MPP-3765)
        # do not check for when the user got extension
        has_extension = False
        date_got_extension = datetime.min
        return cls(
            metrics_enabled=True,
            fxa_id=fxa_id,
            n_random_masks=n_random_masks,
            n_domain_masks=n_domain_masks,
            n_deleted_random_masks=n_deleted_random_masks,
            n_deleted_domain_masks=n_deleted_domain_masks,
            date_joined_relay=date_joined_relay,
            date_joined_premium=date_joined_premium,
            premium_status=premium_status,
            has_extension=has_extension,
            date_got_extension=date_got_extension,
        )


class EmailMaskData(NamedTuple):
    """Extract and store data from a Relay email mask."""

    is_random_mask: bool
    has_website: bool

    @classmethod
    def from_mask(cls, mask: RelayAddress | DomainAddress) -> EmailMaskData:
        if isinstance(mask, RelayAddress):
            is_random_mask = True
            has_website = bool(mask.generated_for)
        else:
            is_random_mask = False
            has_website = False
        return EmailMaskData(is_random_mask=is_random_mask, has_website=has_website)


class RelayGleanLogger(EventsServerEventLogger):
    """Extend the generated EventsServerEventLogger for Relay usage."""

    def __init__(
        self,
        application_id: str,
        app_display_version: str,
        channel: RELAY_CHANNEL_NAME,
    ):
        if not settings.GLEAN_EVENT_MOZLOG_TYPE == GLEAN_EVENT_MOZLOG_TYPE:
            raise ValueError(
                "settings.GLEAN_EVENT_MOZLOG_TYPE must equal GLEAN_EVENT_MOZLOG_TYPE"
            )
        self._logger = getLogger(GLEAN_EVENT_MOZLOG_TYPE)
        super().__init__(
            application_id=application_id,
            app_display_version=app_display_version,
            channel=channel,
        )

    def emit_record(self, now: datetime, ping: dict[str, Any]) -> None:
        """Emit record as a log instead of a print()"""
        self._logger.info(GLEAN_EVENT_MOZLOG_TYPE, extra=ping)

    def log_email_mask_created(
        self,
        *,
        request: HttpRequest | None = None,
        mask: RelayAddress | DomainAddress,
        created_by_api: bool,
    ) -> None:
        """Log that a Relay email mask was created."""
        user_data = UserData.from_user(mask.user)
        if not user_data.metrics_enabled:
            return
        request_data = RequestData.from_request(request) if request else RequestData()
        mask_data = EmailMaskData.from_mask(mask)
        self.record_email_mask_created(
            user_agent=_opt_str_to_glean(request_data.user_agent),
            ip_address=_opt_str_to_glean(request_data.ip_address),
            fxa_id=_opt_str_to_glean(user_data.fxa_id),
            platform="",
            n_random_masks=user_data.n_random_masks,
            n_domain_masks=user_data.n_domain_masks,
            n_deleted_random_masks=user_data.n_deleted_random_masks,
            n_deleted_domain_masks=user_data.n_deleted_domain_masks,
            date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay),
            premium_status=user_data.premium_status,
            date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium),
            has_extension=user_data.has_extension,
            date_got_extension=_opt_dt_to_glean(user_data.date_got_extension),
            is_random_mask=mask_data.is_random_mask,
            created_by_api=created_by_api,
            has_website=mask_data.has_website,
        )

    def log_email_mask_label_updated(
        self,
        *,
        request: HttpRequest,
        mask: RelayAddress | DomainAddress,
    ) -> None:
        """Log that a Relay email mask's label was changed."""
        user_data = UserData.from_user(mask.user)
        if not user_data.metrics_enabled:
            return
        request_data = RequestData.from_request(request)
        mask_data = EmailMaskData.from_mask(mask)
        self.record_email_mask_label_updated(
            user_agent=_opt_str_to_glean(request_data.user_agent),
            ip_address=_opt_str_to_glean(request_data.ip_address),
            fxa_id=_opt_str_to_glean(user_data.fxa_id),
            platform="",
            n_random_masks=user_data.n_random_masks,
            n_domain_masks=user_data.n_domain_masks,
            n_deleted_random_masks=user_data.n_deleted_random_masks,
            n_deleted_domain_masks=user_data.n_deleted_domain_masks,
            date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay),
            premium_status=user_data.premium_status,
            date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium),
            has_extension=user_data.has_extension,
            date_got_extension=_opt_dt_to_glean(user_data.date_got_extension),
            is_random_mask=mask_data.is_random_mask,
        )

    def log_email_mask_deleted(
        self,
        *,
        request: HttpRequest,
        user: User,
        is_random_mask: bool,
    ) -> None:
        """Log that a Relay email mask was deleted."""
        user_data = UserData.from_user(user)
        if not user_data.metrics_enabled:
            return
        request_data = RequestData.from_request(request)
        self.record_email_mask_deleted(
            user_agent=_opt_str_to_glean(request_data.user_agent),
            ip_address=_opt_str_to_glean(request_data.ip_address),
            fxa_id=_opt_str_to_glean(user_data.fxa_id),
            platform="",
            n_random_masks=user_data.n_random_masks,
            n_domain_masks=user_data.n_domain_masks,
            n_deleted_random_masks=user_data.n_deleted_random_masks,
            n_deleted_domain_masks=user_data.n_deleted_domain_masks,
            date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay),
            premium_status=user_data.premium_status,
            date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium),
            has_extension=user_data.has_extension,
            date_got_extension=_opt_dt_to_glean(user_data.date_got_extension),
            is_random_mask=is_random_mask,
        )

    def log_email_forwarded(
        self,
        *,
        mask: RelayAddress | DomainAddress,
        is_reply: bool = False,
    ) -> None:
        """Log that an email was forwarded."""
        user_data = UserData.from_user(mask.user)
        if not user_data.metrics_enabled:
            return
        request_data = RequestData()
        mask_data = EmailMaskData.from_mask(mask)
        self.record_email_forwarded(
            user_agent=_opt_str_to_glean(request_data.user_agent),
            ip_address=_opt_str_to_glean(request_data.ip_address),
            fxa_id=_opt_str_to_glean(user_data.fxa_id),
            platform="",
            n_random_masks=user_data.n_random_masks,
            n_domain_masks=user_data.n_domain_masks,
            n_deleted_random_masks=user_data.n_deleted_random_masks,
            n_deleted_domain_masks=user_data.n_deleted_domain_masks,
            date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay),
            premium_status=user_data.premium_status,
            date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium),
            has_extension=user_data.has_extension,
            date_got_extension=_opt_dt_to_glean(user_data.date_got_extension),
            is_random_mask=mask_data.is_random_mask,
            is_reply=is_reply,
        )

    def log_email_blocked(
        self,
        *,
        mask: RelayAddress | DomainAddress,
        reason: EmailBlockedReason,
        is_reply: bool = False,
    ) -> None:
        """Log that an email was not forwarded."""
        user_data = UserData.from_user(mask.user)
        if not user_data.metrics_enabled:
            return
        request_data = RequestData()
        mask_data = EmailMaskData.from_mask(mask)
        self.record_email_blocked(
            user_agent=_opt_str_to_glean(request_data.user_agent),
            ip_address=_opt_str_to_glean(request_data.ip_address),
            fxa_id=_opt_str_to_glean(user_data.fxa_id),
            platform="",
            n_random_masks=user_data.n_random_masks,
            n_domain_masks=user_data.n_domain_masks,
            n_deleted_random_masks=user_data.n_deleted_random_masks,
            n_deleted_domain_masks=user_data.n_deleted_domain_masks,
            date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay),
            premium_status=user_data.premium_status,
            date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium),
            has_extension=user_data.has_extension,
            date_got_extension=_opt_dt_to_glean(user_data.date_got_extension),
            is_random_mask=mask_data.is_random_mask,
            is_reply=is_reply,
            reason=reason,
        )

    def log_api_accessed(self, request: HttpRequest) -> None:
        """Log that any Relay API endpoint was accessed."""
        if not request.user or not request.user.is_authenticated:
            return
        request_data = RequestData.from_request(request)
        user_data = UserData.from_user(request.user)
        self.record_api_accessed(
            user_agent=_opt_str_to_glean(request_data.user_agent),
            ip_address=_opt_str_to_glean(request_data.ip_address),
            endpoint=request.path,
            method=_opt_str_to_glean(request.method),
            fxa_id=_opt_str_to_glean(user_data.fxa_id),
        )

    def log_text_received(
        self,
        *,
        user: User,
    ) -> None:
        """Log that a text message was received."""
        user_data = UserData.from_user(user)
        if not user_data.metrics_enabled:
            return
        request_data = RequestData()
        self.record_phone_text_received(
            user_agent=_opt_str_to_glean(request_data.user_agent),
            ip_address=_opt_str_to_glean(request_data.ip_address),
            fxa_id=_opt_str_to_glean(user_data.fxa_id),
        )

    def log_call_received(
        self,
        *,
        user: User,
    ) -> None:
        """Log that a phone call was received."""
        user_data = UserData.from_user(user)
        if not user_data.metrics_enabled:
            return
        request_data = RequestData()
        self.record_phone_call_received(
            user_agent=_opt_str_to_glean(request_data.user_agent),
            ip_address=_opt_str_to_glean(request_data.ip_address),
            fxa_id=_opt_str_to_glean(user_data.fxa_id),
        )

1	"""Relay interface to EventsServerEventLogger generated by glean_parser."""
2
3	from __future__ import annotations	1✔
4
5	from datetime import datetime	1✔
6	from logging import getLogger	1✔
7	from typing import Any, Literal, NamedTuple	1✔
8
9	from django.conf import settings	1✔
10	from django.contrib.auth.models import User	1✔
11	from django.http import HttpRequest	1✔
12
13	from ipware import get_client_ip	1✔
14
15	from emails.models import DomainAddress, RelayAddress	1✔
16
17	from .glean.server_events import GLEAN_EVENT_MOZLOG_TYPE, EventsServerEventLogger	1✔
18	from .types import RELAY_CHANNEL_NAME	1✔
19
20	# Enumerate the mask setting that caused an email to not be forwarded.
21	EmailBlockedReason = Literal[	1✔
22	"block_all", # The mask is set to block all mail
23	"block_promotional", # The mask is set to block promotional / list mail
24	]
25
26
27	def _opt_dt_to_glean(value: datetime \| None) -> int:	1✔
28	"""Convert an optional datetime to an integer timestamp."""
29	if value == datetime.min:	1✔
30	return -2 # datetime was not checked	1✔
31	if value is None:	1✔
32	return -1 # datetime does not exist	1✔
33	return int(value.timestamp())	1✔
34
35
36	def _opt_str_to_glean(value: str \| None) -> str:	1✔
37	"""Convert an optional string to a (possibly empty) string."""
38	return "" if value is None else value	1✔
39
40
41	class RequestData(NamedTuple):	1✔
42	"""Extract and store data from the request."""
43
44	user_agent: str \| None = None	1✔
45	ip_address: str \| None = None	1✔
46
47	@classmethod	1✔
48	def from_request(cls, request: HttpRequest) -> RequestData:	1✔
49	user_agent = request.headers.get("user-agent", None)	1✔
50	client_ip, is_routable = get_client_ip(request)	1✔
51	ip_address = client_ip if (client_ip and is_routable) else None	1✔
52	return cls(user_agent=user_agent, ip_address=ip_address)	1✔
53
54
55	class UserData(NamedTuple):	1✔
56	"""Extract and store data from a Relay user."""
57
58	metrics_enabled: bool	1✔
59	fxa_id: str \| None = None	1✔
60	n_random_masks: int = 0	1✔
61	n_domain_masks: int = 0	1✔
62	n_deleted_random_masks: int = 0	1✔
63	n_deleted_domain_masks: int = 0	1✔
64	date_joined_relay: datetime \| None = None	1✔
65	date_joined_premium: datetime \| None = None	1✔
66	premium_status: str = ""	1✔
67	has_extension: bool = False	1✔
68	date_got_extension: datetime \| None = None	1✔
69
70	@classmethod	1✔
71	def from_user(cls, user: User) -> UserData:	1✔
72	metrics_enabled = user.profile.metrics_enabled	1✔
73	if not metrics_enabled:	1✔
74	return cls(metrics_enabled=False)	1✔
75
76	fxa_id = user.profile.metrics_fxa_id or None	1✔
77	n_random_masks = user.relayaddress_set.count()	1✔
78	n_domain_masks = user.domainaddress_set.count()	1✔
79	n_deleted_random_masks = user.profile.num_deleted_relay_addresses	1✔
80	n_deleted_domain_masks = user.profile.num_deleted_domain_addresses	1✔
81	date_joined_relay = user.date_joined	1✔
82	if user.profile.has_premium:	1✔
83	if user.profile.has_phone:	1✔
84	date_joined_premium = user.profile.date_subscribed_phone	1✔
85	else:
86	date_joined_premium = user.profile.date_subscribed	1✔
87	else:
88	date_joined_premium = None	1✔
89	premium_status = user.profile.metrics_premium_status	1✔
90	# Until more accurate date_got_extension is calculated (MPP-3765)
91	# do not check for when the user got extension
92	has_extension = False	1✔
93	date_got_extension = datetime.min	1✔
94	return cls(	1✔
95	metrics_enabled=True,
96	fxa_id=fxa_id,
97	n_random_masks=n_random_masks,
98	n_domain_masks=n_domain_masks,
99	n_deleted_random_masks=n_deleted_random_masks,
100	n_deleted_domain_masks=n_deleted_domain_masks,
101	date_joined_relay=date_joined_relay,
102	date_joined_premium=date_joined_premium,
103	premium_status=premium_status,
104	has_extension=has_extension,
105	date_got_extension=date_got_extension,
106	)
107
108
109	class EmailMaskData(NamedTuple):	1✔
110	"""Extract and store data from a Relay email mask."""
111
112	is_random_mask: bool	1✔
113	has_website: bool	1✔
114
115	@classmethod	1✔
116	def from_mask(cls, mask: RelayAddress \| DomainAddress) -> EmailMaskData:	1✔
117	if isinstance(mask, RelayAddress):	1✔
118	is_random_mask = True	1✔
119	has_website = bool(mask.generated_for)	1✔
120	else:
121	is_random_mask = False	1✔
122	has_website = False	1✔
123	return EmailMaskData(is_random_mask=is_random_mask, has_website=has_website)	1✔
124
125
126	class RelayGleanLogger(EventsServerEventLogger):	1✔
127	"""Extend the generated EventsServerEventLogger for Relay usage."""
128
129	def __init__(	1✔
130	self,
131	application_id: str,
132	app_display_version: str,
133	channel: RELAY_CHANNEL_NAME,
134	):
135	if not settings.GLEAN_EVENT_MOZLOG_TYPE == GLEAN_EVENT_MOZLOG_TYPE:	1!
136	raise ValueError(	×
137	"settings.GLEAN_EVENT_MOZLOG_TYPE must equal GLEAN_EVENT_MOZLOG_TYPE"
138	)
139	self._logger = getLogger(GLEAN_EVENT_MOZLOG_TYPE)	1✔
140	super().__init__(	1✔
141	application_id=application_id,
142	app_display_version=app_display_version,
143	channel=channel,
144	)
145
146	def emit_record(self, now: datetime, ping: dict[str, Any]) -> None:	1✔
147	"""Emit record as a log instead of a print()"""
148	self._logger.info(GLEAN_EVENT_MOZLOG_TYPE, extra=ping)	1✔
149
150	def log_email_mask_created(	1✔
151	self,
152	*,
153	request: HttpRequest \| None = None,
154	mask: RelayAddress \| DomainAddress,
155	created_by_api: bool,
156	) -> None:
157	"""Log that a Relay email mask was created."""
158	user_data = UserData.from_user(mask.user)	1✔
159	if not user_data.metrics_enabled:	1✔
160	return	1✔
161	request_data = RequestData.from_request(request) if request else RequestData()	1✔
162	mask_data = EmailMaskData.from_mask(mask)	1✔
163	self.record_email_mask_created(	1✔
164	user_agent=_opt_str_to_glean(request_data.user_agent),
165	ip_address=_opt_str_to_glean(request_data.ip_address),
166	fxa_id=_opt_str_to_glean(user_data.fxa_id),
167	platform="",
168	n_random_masks=user_data.n_random_masks,
169	n_domain_masks=user_data.n_domain_masks,
170	n_deleted_random_masks=user_data.n_deleted_random_masks,
171	n_deleted_domain_masks=user_data.n_deleted_domain_masks,
172	date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay),
173	premium_status=user_data.premium_status,
174	date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium),
175	has_extension=user_data.has_extension,
176	date_got_extension=_opt_dt_to_glean(user_data.date_got_extension),
177	is_random_mask=mask_data.is_random_mask,
178	created_by_api=created_by_api,
179	has_website=mask_data.has_website,
180	)
181
182	def log_email_mask_label_updated(	1✔
183	self,
184	*,
185	request: HttpRequest,
186	mask: RelayAddress \| DomainAddress,
187	) -> None:
188	"""Log that a Relay email mask's label was changed."""
189	user_data = UserData.from_user(mask.user)	1✔
190	if not user_data.metrics_enabled:	1✔
191	return	1✔
192	request_data = RequestData.from_request(request)	1✔
193	mask_data = EmailMaskData.from_mask(mask)	1✔
194	self.record_email_mask_label_updated(	1✔
195	user_agent=_opt_str_to_glean(request_data.user_agent),
196	ip_address=_opt_str_to_glean(request_data.ip_address),
197	fxa_id=_opt_str_to_glean(user_data.fxa_id),
198	platform="",
199	n_random_masks=user_data.n_random_masks,
200	n_domain_masks=user_data.n_domain_masks,
201	n_deleted_random_masks=user_data.n_deleted_random_masks,
202	n_deleted_domain_masks=user_data.n_deleted_domain_masks,
203	date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay),
204	premium_status=user_data.premium_status,
205	date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium),
206	has_extension=user_data.has_extension,
207	date_got_extension=_opt_dt_to_glean(user_data.date_got_extension),
208	is_random_mask=mask_data.is_random_mask,
209	)
210
211	def log_email_mask_deleted(	1✔
212	self,
213	*,
214	request: HttpRequest,
215	user: User,
216	is_random_mask: bool,
217	) -> None:
218	"""Log that a Relay email mask was deleted."""
219	user_data = UserData.from_user(user)	1✔
220	if not user_data.metrics_enabled:	1✔
221	return	1✔
222	request_data = RequestData.from_request(request)	1✔
223	self.record_email_mask_deleted(	1✔
224	user_agent=_opt_str_to_glean(request_data.user_agent),
225	ip_address=_opt_str_to_glean(request_data.ip_address),
226	fxa_id=_opt_str_to_glean(user_data.fxa_id),
227	platform="",
228	n_random_masks=user_data.n_random_masks,
229	n_domain_masks=user_data.n_domain_masks,
230	n_deleted_random_masks=user_data.n_deleted_random_masks,
231	n_deleted_domain_masks=user_data.n_deleted_domain_masks,
232	date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay),
233	premium_status=user_data.premium_status,
234	date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium),
235	has_extension=user_data.has_extension,
236	date_got_extension=_opt_dt_to_glean(user_data.date_got_extension),
237	is_random_mask=is_random_mask,
238	)
239
240	def log_email_forwarded(	1✔
241	self,
242	*,
243	mask: RelayAddress \| DomainAddress,
244	is_reply: bool = False,
245	) -> None:
246	"""Log that an email was forwarded."""
247	user_data = UserData.from_user(mask.user)	1✔
248	if not user_data.metrics_enabled:	1✔
249	return	1✔
250	request_data = RequestData()	1✔
251	mask_data = EmailMaskData.from_mask(mask)	1✔
252	self.record_email_forwarded(	1✔
253	user_agent=_opt_str_to_glean(request_data.user_agent),
254	ip_address=_opt_str_to_glean(request_data.ip_address),
255	fxa_id=_opt_str_to_glean(user_data.fxa_id),
256	platform="",
257	n_random_masks=user_data.n_random_masks,
258	n_domain_masks=user_data.n_domain_masks,
259	n_deleted_random_masks=user_data.n_deleted_random_masks,
260	n_deleted_domain_masks=user_data.n_deleted_domain_masks,
261	date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay),
262	premium_status=user_data.premium_status,
263	date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium),
264	has_extension=user_data.has_extension,
265	date_got_extension=_opt_dt_to_glean(user_data.date_got_extension),
266	is_random_mask=mask_data.is_random_mask,
267	is_reply=is_reply,
268	)
269
270	def log_email_blocked(	1✔
271	self,
272	*,
273	mask: RelayAddress \| DomainAddress,
274	reason: EmailBlockedReason,
275	is_reply: bool = False,
276	) -> None:
277	"""Log that an email was not forwarded."""
278	user_data = UserData.from_user(mask.user)	1✔
279	if not user_data.metrics_enabled:	1✔
280	return	1✔
281	request_data = RequestData()	1✔
282	mask_data = EmailMaskData.from_mask(mask)	1✔
283	self.record_email_blocked(	1✔
284	user_agent=_opt_str_to_glean(request_data.user_agent),
285	ip_address=_opt_str_to_glean(request_data.ip_address),
286	fxa_id=_opt_str_to_glean(user_data.fxa_id),
287	platform="",
288	n_random_masks=user_data.n_random_masks,
289	n_domain_masks=user_data.n_domain_masks,
290	n_deleted_random_masks=user_data.n_deleted_random_masks,
291	n_deleted_domain_masks=user_data.n_deleted_domain_masks,
292	date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay),
293	premium_status=user_data.premium_status,
294	date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium),
295	has_extension=user_data.has_extension,
296	date_got_extension=_opt_dt_to_glean(user_data.date_got_extension),
297	is_random_mask=mask_data.is_random_mask,
298	is_reply=is_reply,
299	reason=reason,
300	)
301
302	def log_api_accessed(self, request: HttpRequest) -> None:	1✔
303	"""Log that any Relay API endpoint was accessed."""
304	if not request.user or not request.user.is_authenticated:	1✔
305	return	1✔
306	request_data = RequestData.from_request(request)	1✔
307	user_data = UserData.from_user(request.user)	1✔
308	self.record_api_accessed(	1✔
309	user_agent=_opt_str_to_glean(request_data.user_agent),
310	ip_address=_opt_str_to_glean(request_data.ip_address),
311	endpoint=request.path,
312	method=_opt_str_to_glean(request.method),
313	fxa_id=_opt_str_to_glean(user_data.fxa_id),
314	)
315
316	def log_text_received(	1✔
317	self,
318	*,
319	user: User,
320	) -> None:
321	"""Log that a text message was received."""
322	user_data = UserData.from_user(user)	1✔
323	if not user_data.metrics_enabled:	1!
NEW 324	return	×
325	request_data = RequestData()	1✔
326	self.record_phone_text_received(	1✔
327	user_agent=_opt_str_to_glean(request_data.user_agent),
328	ip_address=_opt_str_to_glean(request_data.ip_address),
329	fxa_id=_opt_str_to_glean(user_data.fxa_id),
330	)
331
332	def log_call_received(	1✔
333	self,
334	*,
335	user: User,
336	) -> None:
337	"""Log that a phone call was received."""
338	user_data = UserData.from_user(user)	1✔
339	if not user_data.metrics_enabled:	1!
NEW 340	return	×
341	request_data = RequestData()	1✔
342	self.record_phone_call_received(	1✔
343	user_agent=_opt_str_to_glean(request_data.user_agent),
344	ip_address=_opt_str_to_glean(request_data.ip_address),
345	fxa_id=_opt_str_to_glean(user_data.fxa_id),
346	)

mozilla / fx-private-relay / faea5c47-7f2b-4136-8c9b-7cbe324def9d

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