35084fc5-73ae-405a-93ba-7c0998b1a2a7

Committed 15 Oct 2025 02:15PM UTC coverage: 88.889% (+0.03%) from 88.862%

Build # 35084fc5-73ae-405a-93ba-7c0998b1a2a7

Build Type

Pull #5964

circleci

Committed by

jwhitlock

Commit Message

docs: Add ticket references to TODOs

Pull Request Pull Request #5964: Remove heroku references and code paths (MPP-4425)

Run Details

2920 of 3925 branches covered (74.39%)

Branch coverage included in aggregate %.

7 of 9 new or added lines in 5 files covered. (77.78%)

1 existing line in 1 file now uncovered.

18072 of 19691 relevant lines covered (91.78%)

11.41 hits per line

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

42.11

/privaterelay/apps.py

import base64
import json
import os
from pathlib import Path
from typing import TYPE_CHECKING, Any

from django.apps import AppConfig
from django.conf import settings
from django.utils.functional import cached_property

import markus
import requests

ROOT_DIR = os.path.abspath(os.curdir)


if TYPE_CHECKING:
    from allauth.socialaccount.models import SocialApp


def configure_google_profiler() -> None:
    if (
        settings.GOOGLE_APPLICATION_CREDENTIALS == ""
        or settings.GOOGLE_CLOUD_PROFILER_CREDENTIALS_B64 == ""
    ):
        return

    # Set up Google Cloud Profiler
    service, version = get_profiler_startup_data()
    if service is None:
        return

    gcp_key_json_path = Path(settings.GOOGLE_APPLICATION_CREDENTIALS)
    if not gcp_key_json_path.exists():
        write_gcp_key_json_file(gcp_key_json_path)
    try:
        with gcp_key_json_path.open() as gcp_key_file:
            try:
                # Make sure the expected gcp_key.json file is valid json
                gcp_key_data = json.load(gcp_key_file)
                import googlecloudprofiler

                googlecloudprofiler.start(
                    service=service,
                    service_version=version,
                    project_id=gcp_key_data["project_id"],
                )
            except json.JSONDecodeError:
                print(f"error during json.load({gcp_key_json_path})")
    except Exception as exc:
        print(
            f"exception {repr(exc)}"
            " while starting google cloud profiler"
            f" with key file: {gcp_key_json_path}"
        )


def get_profiler_startup_data() -> tuple[str | None, str | None]:
    from .utils import get_version_info

    if settings.RELAY_CHANNEL not in ("dev", "stage", "prod"):
        return (None, None)

    if settings.RELAY_CHANNEL in ("dev", "stage"):
        service = f"fxprivaterelay-{settings.RELAY_CHANNEL}"
    if settings.RELAY_CHANNEL == "prod":
        service = "fxprivaterelay"

    version_info = get_version_info()
    version = version_info.get("version", "unknown")

    return service, version


def write_gcp_key_json_file(gcp_key_json_path: Path) -> None:
    """
    Create the gcp key json file from contents of GOOGLE_CLOUD_PROFILER_CREDENTIALS_B64
    """
    google_app_creds = base64.b64decode(settings.GOOGLE_CLOUD_PROFILER_CREDENTIALS_B64)
    if not google_app_creds == b"":
        with open(gcp_key_json_path, "w+") as gcp_key_file:
            gcp_key_file.write(google_app_creds.decode("utf-8"))


def configure_markus() -> None:
    backends: list[dict[str, Any]] = []
    if settings.STATSD_ENABLED and not settings.IN_PYTEST:
        backends.append(
            {
                "class": "markus.backends.datadog.DatadogMetrics",
                "options": {
                    "statsd_host": settings.STATSD_HOST,
                    "statsd_port": settings.STATSD_PORT,
                    "statsd_namespace": settings.STATSD_PREFIX,
                },
            }
        )
    if settings.STATSD_DEBUG:
        backends.append(
            {
                "class": "markus.backends.logging.LoggingMetrics",
                "options": {
                    "logger_name": "markus",
                    "leader": "METRICS",
                },
            }
        )
    markus.configure(backends=backends)


class PrivateRelayConfig(AppConfig):
    name = "privaterelay"

    def ready(self) -> None:
        configure_markus()
        configure_google_profiler()

        import privaterelay.signals  # noqa: F401 (imported but unused warning)

        try:
            del self.fxa_verifying_keys  # Clear cache
            del self.fxa_social_app  # Clear cache
        except AttributeError:
            pass

    @cached_property
    def fxa_verifying_keys(self) -> list[dict[str, Any]]:
        resp = requests.get(
            "{}/jwks".format(settings.SOCIALACCOUNT_PROVIDERS["fxa"]["OAUTH_ENDPOINT"]),
            timeout=10,
        )
        if resp.status_code == 200:
            keys: list[dict[str, Any]] = resp.json()["keys"]
            return keys
        return []

    @cached_property
    def fxa_social_app(self) -> "SocialApp":
        from allauth.socialaccount.models import SocialApp

        return SocialApp.objects.get(provider="fxa")

1	import base64	1✔
2	import json	1✔
3	import os	1✔
4	from pathlib import Path	1✔
5	from typing import TYPE_CHECKING, Any	1✔
6
7	from django.apps import AppConfig	1✔
8	from django.conf import settings	1✔
9	from django.utils.functional import cached_property	1✔
10
11	import markus	1✔
12	import requests	1✔
13
14	ROOT_DIR = os.path.abspath(os.curdir)	1✔
15
16
17	if TYPE_CHECKING:
18	from allauth.socialaccount.models import SocialApp
19
20
21	def configure_google_profiler() -> None:	1✔
22	if (	1!
23	settings.GOOGLE_APPLICATION_CREDENTIALS == ""
24	or settings.GOOGLE_CLOUD_PROFILER_CREDENTIALS_B64 == ""
25	):
26	return	1✔
27
28	# Set up Google Cloud Profiler
29	service, version = get_profiler_startup_data()	×
30	if service is None:	×
31	return	×
32
33	gcp_key_json_path = Path(settings.GOOGLE_APPLICATION_CREDENTIALS)	×
34	if not gcp_key_json_path.exists():	×
35	write_gcp_key_json_file(gcp_key_json_path)	×
36	try:	×
37	with gcp_key_json_path.open() as gcp_key_file:	×
38	try:	×
39	# Make sure the expected gcp_key.json file is valid json
40	gcp_key_data = json.load(gcp_key_file)	×
41	import googlecloudprofiler	×
42
43	googlecloudprofiler.start(	×
44	service=service,
45	service_version=version,
46	project_id=gcp_key_data["project_id"],
47	)
48	except json.JSONDecodeError:	×
49	print(f"error during json.load({gcp_key_json_path})")	×
50	except Exception as exc:	×
51	print(	×
52	f"exception {repr(exc)}"
53	" while starting google cloud profiler"
54	f" with key file: {gcp_key_json_path}"
55	)
56
57
58	def get_profiler_startup_data() -> tuple[str \| None, str \| None]:	1✔
59	from .utils import get_version_info	×
60
NEW 61	if settings.RELAY_CHANNEL not in ("dev", "stage", "prod"):	×
62	return (None, None)	×
63
NEW 64	if settings.RELAY_CHANNEL in ("dev", "stage"):	×
65	service = f"fxprivaterelay-{settings.RELAY_CHANNEL}"	×
66	if settings.RELAY_CHANNEL == "prod":	×
67	service = "fxprivaterelay"	×
68
69	version_info = get_version_info()	×
70	version = version_info.get("version", "unknown")	×
71
72	return service, version	×
73
74
75	def write_gcp_key_json_file(gcp_key_json_path: Path) -> None:	1✔
76	"""
77	Create the gcp key json file from contents of GOOGLE_CLOUD_PROFILER_CREDENTIALS_B64
78	"""
79	google_app_creds = base64.b64decode(settings.GOOGLE_CLOUD_PROFILER_CREDENTIALS_B64)	×
80	if not google_app_creds == b"":	×
81	with open(gcp_key_json_path, "w+") as gcp_key_file:	×
82	gcp_key_file.write(google_app_creds.decode("utf-8"))	×
83
84
85	def configure_markus() -> None:	1✔
86	backends: list[dict[str, Any]] = []	1✔
87	if settings.STATSD_ENABLED and not settings.IN_PYTEST:	1!
88	backends.append(	×
89	{
90	"class": "markus.backends.datadog.DatadogMetrics",
91	"options": {
92	"statsd_host": settings.STATSD_HOST,
93	"statsd_port": settings.STATSD_PORT,
94	"statsd_namespace": settings.STATSD_PREFIX,
95	},
96	}
97	)
98	if settings.STATSD_DEBUG:	1!
99	backends.append(	×
100	{
101	"class": "markus.backends.logging.LoggingMetrics",
102	"options": {
103	"logger_name": "markus",
104	"leader": "METRICS",
105	},
106	}
107	)
108	markus.configure(backends=backends)	1✔
109
110
111	class PrivateRelayConfig(AppConfig):	1✔
112	name = "privaterelay"	1✔
113
114	def ready(self) -> None:	1✔
115	configure_markus()	1✔
116	configure_google_profiler()	1✔
117
118	import privaterelay.signals # noqa: F401 (imported but unused warning)	1✔
119
120	try:	1✔
121	del self.fxa_verifying_keys # Clear cache	1✔
122	del self.fxa_social_app # Clear cache	×
123	except AttributeError:	1✔
124	pass	1✔
125
126	@cached_property	1✔
127	def fxa_verifying_keys(self) -> list[dict[str, Any]]:	1✔
128	resp = requests.get(	×
129	"{}/jwks".format(settings.SOCIALACCOUNT_PROVIDERS["fxa"]["OAUTH_ENDPOINT"]),
130	timeout=10,
131	)
132	if resp.status_code == 200:	×
133	keys: list[dict[str, Any]] = resp.json()["keys"]	×
134	return keys	×
135	return []	×
136
137	@cached_property	1✔
138	def fxa_social_app(self) -> "SocialApp":	1✔
139	from allauth.socialaccount.models import SocialApp	1✔
140
141	return SocialApp.objects.get(provider="fxa")	1✔

mozilla / fx-private-relay / 35084fc5-73ae-405a-93ba-7c0998b1a2a7

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