Refactor builtin repositories to enum (#5976)

supervisor/store/git.py

@@ -1,5 +1,6 @@
 """Init file for Supervisor add-on Git."""
 
+from abc import ABC, abstractmethod
 import asyncio
 import functools as ft
 import logging
@@ -7,19 +8,19 @@ from pathlib import Path
 
 import git
 
-from ..const import ATTR_BRANCH, ATTR_URL, URL_HASSIO_ADDONS
+from ..const import ATTR_BRANCH, ATTR_URL
 from ..coresys import CoreSys, CoreSysAttributes
 from ..exceptions import StoreGitCloneError, StoreGitError, StoreJobError
 from ..jobs.decorator import Job, JobCondition
 from ..resolution.const import ContextType, IssueType, SuggestionType
 from ..utils import remove_folder
 from .utils import get_hash_from_repository
-from .validate import RE_REPOSITORY
+from .validate import RE_REPOSITORY, BuiltinRepository
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 
 
-class GitRepo(CoreSysAttributes):
+class GitRepo(CoreSysAttributes, ABC):
     """Manage Add-on Git repository."""
 
     builtin: bool
@@ -197,29 +198,23 @@ class GitRepo(CoreSysAttributes):
                 )
                 raise StoreGitError() from err
 
-    async def _remove(self):
+    @abstractmethod
+    async def remove(self) -> None:
         """Remove a repository."""
-        if self.lock.locked():
-            _LOGGER.warning("There is already a task in progress")
-            return
-
-        def _remove_git_dir(path: Path) -> None:
-            if not path.is_dir():
-                return
-            remove_folder(path)
-
-        async with self.lock:
-            await self.sys_run_in_executor(_remove_git_dir, self.path)
 
 
-class GitRepoHassIO(GitRepo):
-    """Supervisor add-ons repository."""
+class GitRepoBuiltin(GitRepo):
+    """Built-in add-ons repository."""
 
-    builtin: bool = False
+    builtin: bool = True
 
-    def __init__(self, coresys):
+    def __init__(self, coresys: CoreSys, repository: BuiltinRepository):
         """Initialize Git Supervisor add-on repository."""
-        super().__init__(coresys, coresys.config.path_addons_core, URL_HASSIO_ADDONS)
+        super().__init__(coresys, repository.get_path(coresys), repository.url)
+
+    async def remove(self) -> None:
+        """Raise. Cannot remove built-in repositories."""
+        raise RuntimeError("Cannot remove built-in repositories!")
 
 
 class GitRepoCustom(GitRepo):
@@ -233,7 +228,21 @@ class GitRepoCustom(GitRepo):
 
         super().__init__(coresys, path, url)
 
-    async def remove(self):
+    async def remove(self) -> None:
         """Remove a custom repository."""
+        if self.lock.locked():
+            _LOGGER.warning(
+                "Cannot remove add-on repository %s, there is already a task in progress",
+                self.url,
+            )
+            return
+
         _LOGGER.info("Removing custom add-on repository %s", self.url)
-        await self._remove()
+
+        def _remove_git_dir(path: Path) -> None:
+            if not path.is_dir():
+                return
+            remove_folder(path)
+
+        async with self.lock:
+            await self.sys_run_in_executor(_remove_git_dir, self.path)

supervisor/store/repository.py

@@ -2,7 +2,6 @@
 
 import logging
 from pathlib import Path
-from typing import cast
 
 import voluptuous as vol
 
@@ -13,9 +12,9 @@ from ..coresys import CoreSys, CoreSysAttributes
 from ..exceptions import ConfigurationFileError, StoreError
 from ..utils.common import read_json_or_yaml_file
 from .const import StoreType
-from .git import GitRepo, GitRepoCustom, GitRepoHassIO
+from .git import GitRepo, GitRepoBuiltin, GitRepoCustom
 from .utils import get_hash_from_repository
-from .validate import SCHEMA_REPOSITORY_CONFIG
+from .validate import SCHEMA_REPOSITORY_CONFIG, BuiltinRepository
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 UNKNOWN = "unknown"
@@ -34,10 +33,11 @@ class Repository(CoreSysAttributes):
             self._slug = repository
             self._type = StoreType.LOCAL
             self._latest_mtime: float | None = None
-        elif repository == StoreType.CORE:
-            self.git = GitRepoHassIO(coresys)
-            self._slug = repository
-            self._type = StoreType.CORE
+        elif repository in BuiltinRepository:
+            builtin = BuiltinRepository(repository)
+            self.git = GitRepoBuiltin(coresys, builtin)
+            self._slug = builtin.id
+            self._type = builtin.type
         else:
             self.git = GitRepoCustom(coresys, repository)
             self._slug = get_hash_from_repository(repository)
@@ -140,7 +140,7 @@ class Repository(CoreSysAttributes):
 
     async def remove(self) -> None:
         """Remove add-on repository."""
-        if not self.git or self.type == StoreType.CORE:
+        if not self.git or self.git.builtin:
             raise StoreError("Can't remove built-in repositories!", _LOGGER.error)
 
-        await cast(GitRepoCustom, self.git).remove()
+        await self.git.remove()

supervisor/store/validate.py

@@ -1,21 +1,62 @@
 """Validate add-ons options schema."""
 
+from enum import StrEnum
+from pathlib import Path
+
 import voluptuous as vol
 
-from ..const import ATTR_MAINTAINER, ATTR_NAME, ATTR_REPOSITORIES, ATTR_URL
+from ..const import (
+    ATTR_MAINTAINER,
+    ATTR_NAME,
+    ATTR_REPOSITORIES,
+    ATTR_URL,
+    URL_HASSIO_ADDONS,
+)
+from ..coresys import CoreSys
 from ..validate import RE_REPOSITORY
 from .const import StoreType
+from .utils import get_hash_from_repository
 
 URL_COMMUNITY_ADDONS = "https://github.com/hassio-addons/repository"
 URL_ESPHOME = "https://github.com/esphome/home-assistant-addon"
 URL_MUSIC_ASSISTANT = "https://github.com/music-assistant/home-assistant-addon"
-BUILTIN_REPOSITORIES = {
-    StoreType.CORE,
-    StoreType.LOCAL,
-    URL_COMMUNITY_ADDONS,
-    URL_ESPHOME,
-    URL_MUSIC_ASSISTANT,
-}
+
+
+class BuiltinRepository(StrEnum):
+    """Built-in add-on repository."""
+
+    CORE = StoreType.CORE.value
+    LOCAL = StoreType.LOCAL.value
+    COMMUNITY_ADDONS = URL_COMMUNITY_ADDONS
+    ESPHOME = URL_ESPHOME
+    MUSIC_ASSISTANT = URL_MUSIC_ASSISTANT
+
+    def __init__(self, value: str) -> None:
+        """Initialize repository item."""
+        if value == StoreType.LOCAL:
+            self.id = value
+            self.url = ""
+            self.type = StoreType.LOCAL
+        elif value == StoreType.CORE:
+            self.id = value
+            self.url = URL_HASSIO_ADDONS
+            self.type = StoreType.CORE
+        else:
+            self.id = get_hash_from_repository(value)
+            self.url = value
+            self.type = StoreType.GIT
+
+    def get_path(self, coresys: CoreSys) -> Path:
+        """Get path to git repo for repository."""
+        if self.id == StoreType.LOCAL:
+            return coresys.config.path_addons_local
+        if self.id == StoreType.CORE:
+            return coresys.config.path_addons_core
+        return Path(coresys.config.path_addons_git, self.id)
+
+
+BUILTIN_REPOSITORIES = {r.value for r in BuiltinRepository}
+
 
 # pylint: disable=no-value-for-parameter
 SCHEMA_REPOSITORY_CONFIG = vol.Schema(

tests/conftest.py

@@ -409,7 +409,7 @@ async def coresys(
         coresys_obj.init_websession = AsyncMock()
 
     # Don't remove files/folders related to addons and stores
-    with patch("supervisor.store.git.GitRepo._remove"):
+    with patch("supervisor.store.git.GitRepoCustom.remove"):
         yield coresys_obj
 
     await coresys_obj.dbus.unload()

tests/store/test_repository_git.py

@@ -15,6 +15,13 @@ from supervisor.store.git import GitRepo
 REPO_URL = "https://github.com/awesome-developer/awesome-repo"
 
 
+class GitRepoTest(GitRepo):
+    """Implementation of GitRepo for tests that allows direct setting of path."""
+
+    async def remove(self) -> None:
+        """Not implemented."""
+
+
 @pytest.fixture(name="clone_from")
 async def fixture_clone_from():
     """Mock git clone_from."""
@@ -28,7 +35,7 @@ async def test_git_clone(
 ):
     """Test git clone."""
     fragment = f"#{branch}" if branch else ""
-    repo = GitRepo(coresys, tmp_path, f"{REPO_URL}{fragment}")
+    repo = GitRepoTest(coresys, tmp_path, f"{REPO_URL}{fragment}")
 
     await repo.clone.__wrapped__(repo)
 
@@ -56,7 +63,7 @@ async def test_git_clone_error(
     coresys: CoreSys, tmp_path: Path, clone_from: AsyncMock, git_error: Exception
 ):
     """Test git clone error."""
-    repo = GitRepo(coresys, tmp_path, REPO_URL)
+    repo = GitRepoTest(coresys, tmp_path, REPO_URL)
 
     clone_from.side_effect = git_error
     with pytest.raises(StoreGitCloneError):
@@ -68,7 +75,7 @@ async def test_git_clone_error(
 async def test_git_load(coresys: CoreSys, tmp_path: Path):
     """Test git load."""
     repo_dir = tmp_path / "repo"
-    repo = GitRepo(coresys, repo_dir, REPO_URL)
+    repo = GitRepoTest(coresys, repo_dir, REPO_URL)
     repo.clone = AsyncMock()
 
     # Test with non-existing git repo root directory
@@ -106,7 +113,7 @@ async def test_git_load(coresys: CoreSys, tmp_path: Path):
 async def test_git_load_error(coresys: CoreSys, tmp_path: Path, git_errors: Exception):
     """Test git load error."""
     coresys.hardware.disk.get_disk_free_space = lambda x: 5000
-    repo = GitRepo(coresys, tmp_path, REPO_URL)
+    repo = GitRepoTest(coresys, tmp_path, REPO_URL)
 
     # Pretend we have a repo
     (tmp_path / ".git").mkdir()