Only load translations for an integration once per test session

homeassistant/helpers/translation.py

@@ -5,6 +5,7 @@
 import asyncio
 from collections.abc import Iterable, Mapping
 from contextlib import suppress
+from dataclasses import dataclass
 import logging
 import pathlib
 import string
@@ -140,30 +141,42 @@ async def _async_get_component_strings(
     return translations_by_language
 
 
+@dataclass(slots=True)
+class _TranslationsCacheData:
+    """Data for the translation cache.
+
+    This class contains data that is designed to be shared
+    between multiple instances of the translation cache so
+    we only have to load the data once.
+    """
+
+    loaded: dict[str, set[str]]
+    cache: dict[str, dict[str, dict[str, dict[str, str]]]]
+
+
 class _TranslationCache:
     """Cache for flattened translations."""
 
-    __slots__ = ("hass", "loaded", "cache", "lock")
+    __slots__ = ("hass", "cache_data", "lock")
 
     def __init__(self, hass: HomeAssistant) -> None:
         """Initialize the cache."""
         self.hass = hass
-        self.loaded: dict[str, set[str]] = {}
-        self.cache: dict[str, dict[str, dict[str, dict[str, str]]]] = {}
+        self.cache_data = _TranslationsCacheData({}, {})
         self.lock = asyncio.Lock()
 
     @callback
     def async_is_loaded(self, language: str, components: set[str]) -> bool:
         """Return if the given components are loaded for the language."""
-        return components.issubset(self.loaded.get(language, set()))
+        return components.issubset(self.cache_data.loaded.get(language, set()))
 
     async def async_load(
         self,
         language: str,
         components: set[str],
     ) -> None:
         """Load resources into the cache."""
-        loaded = self.loaded.setdefault(language, set())
+        loaded = self.cache_data.loaded.setdefault(language, set())
         if components_to_load := components - loaded:
             # Translations are never unloaded so if there are no components to load
             # we can skip the lock which reduces contention when multiple different
@@ -193,7 +206,7 @@ def get_cached(
         components: set[str],
     ) -> dict[str, str]:
         """Read resources from the cache."""
-        category_cache = self.cache.get(language, {}).get(category, {})
+        category_cache = self.cache_data.cache.get(language, {}).get(category, {})
         # If only one component was requested, return it directly
         # to avoid merging the dictionaries and keeping additional
         # copies of the same data in memory.
@@ -207,6 +220,7 @@ def get_cached(
 
     async def _async_load(self, language: str, components: set[str]) -> None:
         """Populate the cache for a given set of components."""
+        loaded = self.cache_data.loaded
         _LOGGER.debug(
             "Cache miss for %s: %s",
             language,
@@ -240,7 +254,7 @@ async def _async_load(self, language: str, components: set[str]) -> None:
                 language, components, translation_by_language_strings[language]
             )
 
-            loaded_english_components = self.loaded.setdefault(LOCALE_EN, set())
+            loaded_english_components = loaded.setdefault(LOCALE_EN, set())
             # Since we just loaded english anyway we can avoid loading
             # again if they switch back to english.
             if loaded_english_components.isdisjoint(components):
@@ -249,7 +263,7 @@ async def _async_load(self, language: str, components: set[str]) -> None:
                 )
                 loaded_english_components.update(components)
 
-        self.loaded[language].update(components)
+        loaded[language].update(components)
 
     def _validate_placeholders(
         self,
@@ -304,7 +318,7 @@ def _build_category_cache(
     ) -> None:
         """Extract resources into the cache."""
         resource: dict[str, Any] | str
-        cached = self.cache.setdefault(language, {})
+        cached = self.cache_data.cache.setdefault(language, {})
         categories = {
             category
             for component in translation_strings.values()

tests/conftest.py

@@ -1165,6 +1165,31 @@ def mock_get_source_ip() -> Generator[patch, None, None]:
         patcher.stop()
 
 
+@pytest.fixture(autouse=True, scope="session")
+def translations_once() -> Generator[patch, None, None]:
+    """Only load translations once per session."""
+    from homeassistant.helpers.translation import _TranslationsCacheData
+
+    cache = _TranslationsCacheData({}, {})
+    patcher = patch(
+        "homeassistant.helpers.translation._TranslationsCacheData",
+        return_value=cache,
+    )
+    patcher.start()
+    try:
+        yield patcher
+    finally:
+        patcher.stop()
+
+
+@pytest.fixture
+def disable_translations_once(translations_once):
+    """Override loading translations once."""
+    translations_once.stop()
+    yield
+    translations_once.start()
+
+
 @pytest.fixture
 def mock_zeroconf() -> Generator[None, None, None]:
     """Mock zeroconf."""

tests/helpers/test_entity_platform.py

@@ -213,6 +213,7 @@ async def test_update_state_adds_entities_with_update_before_add_false(
     assert not ent.update.called
 
 
+@pytest.mark.usefixtures("disable_translations_once")
 async def test_set_scan_interval_via_platform(hass: HomeAssistant) -> None:
     """Test the setting of the scan interval via platform."""
 
@@ -260,6 +261,7 @@ def create_entity(number: int) -> MockEntity:
     await component.async_add_entities(create_entity(i) for i in range(2))
 
 
+@pytest.mark.usefixtures("disable_translations_once")
 async def test_platform_warn_slow_setup(hass: HomeAssistant) -> None:
     """Warn we log when platform setup takes a long time."""
     platform = MockPlatform()

tests/helpers/test_translation.py

@@ -16,6 +16,11 @@
 from homeassistant.setup import async_setup_component
 
 
+@pytest.fixture(autouse=True)
+def _disable_translations_once(disable_translations_once):
+    """Override loading translations once."""
+
+
 @pytest.fixture
 def mock_config_flows():
     """Mock the config flows."""