Remove is_safe_to_remove helper (#2357)

custom_components/hacs/helpers/classes/repository.py

@@ -22,7 +22,6 @@
     get_info_md_content,
     get_repository,
 )
-from custom_components.hacs.helpers.functions.is_safe_to_remove import is_safe_to_remove
 from custom_components.hacs.helpers.functions.misc import get_repository_name
 from custom_components.hacs.helpers.functions.store import async_remove_store
 from custom_components.hacs.helpers.functions.validate_repository import (
@@ -34,6 +33,7 @@
 )
 from custom_components.hacs.share import get_hacs
 from custom_components.hacs.utils.logger import getLogger
+from custom_components.hacs.utils.path import is_safe
 from custom_components.hacs.utils.queue_manager import QueueManager
 
 
@@ -439,7 +439,7 @@ async def remove_local_directory(self):
                 local_path = self.content.path.local
 
             if os.path.exists(local_path):
-                if not is_safe_to_remove(local_path):
+                if not is_safe(self.hacs, local_path):
                     self.logger.error("%s Path %s is blocked from removal", self, local_path)
                     return False
                 self.logger.debug("%s Removing %s", self, local_path)

custom_components/hacs/helpers/methods/installation.py

@@ -73,13 +73,14 @@ async def async_install_repository(repository):
             f"{repository.content.path.local}/{repository.data.persistent_directory}"
         ):
             persistent_directory = Backup(
+                hacs,
                 f"{repository.content.path.local}/{repository.data.persistent_directory}",
                 tempfile.gettempdir() + "/hacs_persistent_directory/",
             )
             await hacs.hass.async_add_executor_job(persistent_directory.create)
 
     if repository.data.installed and not repository.content.single:
-        backup = Backup(repository.content.path.local)
+        backup = Backup(hacs, repository.content.path.local)
         await hacs.hass.async_add_executor_job(backup.create)
 
     if repository.data.zip_release and version != repository.data.default_branch:

custom_components/hacs/operational/backup.py

@@ -1,22 +1,28 @@
 """Backup."""
+from __future__ import annotations
 import os
 import shutil
 import tempfile
 from time import sleep
+from typing import TYPE_CHECKING
 
-from custom_components.hacs.helpers.functions.is_safe_to_remove import is_safe_to_remove
-from custom_components.hacs.utils.logger import getLogger
 
-BACKUP_PATH = tempfile.gettempdir() + "/hacs_backup/"
+from ..utils.path import is_safe
 
-_LOGGER = getLogger()
+if TYPE_CHECKING:
+    from ..base import HacsBase
+    from ..helpers.classes.repository import HacsRepository
+
+
+BACKUP_PATH = f"{tempfile.gettempdir()}/hacs_backup/"
 
 
 class Backup:
     """Backup."""
 
-    def __init__(self, local_path, backup_path=BACKUP_PATH):
+    def __init__(self, hacs: HacsBase, local_path: str, backup_path: str = BACKUP_PATH) -> None:
         """initialize."""
+        self.hacs = hacs
         self.local_path = local_path
         self.backup_path = backup_path
         self.backup_path_full = f"{self.backup_path}{self.local_path.split('/')[-1]}"
@@ -25,10 +31,12 @@ def create(self):
         """Create a backup in /tmp"""
         if not os.path.exists(self.local_path):
             return
-        if not is_safe_to_remove(self.local_path):
+        if not is_safe(self.hacs, self.local_path):
             return
         if os.path.exists(self.backup_path):
             shutil.rmtree(self.backup_path)
+
+            # Wait for the folder to be removed
             while os.path.exists(self.backup_path):
                 sleep(0.1)
         os.makedirs(self.backup_path, exist_ok=True)
@@ -42,7 +50,7 @@ def create(self):
                 shutil.rmtree(self.local_path)
                 while os.path.exists(self.local_path):
                     sleep(0.1)
-            _LOGGER.debug(
+            self.hacs.log.debug(
                 "Backup for %s, created in %s",
                 self.local_path,
                 self.backup_path_full,
@@ -65,30 +73,33 @@ def restore(self):
                 while os.path.exists(self.local_path):
                     sleep(0.1)
             shutil.copytree(self.backup_path_full, self.local_path)
-        _LOGGER.debug("Restored %s, from backup %s", self.local_path, self.backup_path_full)
+        self.hacs.log.debug("Restored %s, from backup %s", self.local_path, self.backup_path_full)
 
     def cleanup(self):
         """Cleanup backup files."""
         if os.path.exists(self.backup_path):
             shutil.rmtree(self.backup_path)
+
+            # Wait for the folder to be removed
             while os.path.exists(self.backup_path):
                 sleep(0.1)
-            _LOGGER.debug("Backup dir %s cleared", self.backup_path)
+            self.hacs.log.debug("Backup dir %s cleared", self.backup_path)
 
 
 class BackupNetDaemon:
     """BackupNetDaemon."""
 
-    def __init__(self, repository):
+    def __init__(self, hacs: HacsBase, repository: HacsRepository) -> None:
         """Initialize."""
+        self.hacs = hacs
         self.repository = repository
         self.backup_path = (
             tempfile.gettempdir() + "/hacs_persistent_netdaemon/" + repository.data.name
         )
 
     def create(self):
         """Create a backup in /tmp"""
-        if not is_safe_to_remove(self.repository.content.path.local):
+        if not is_safe(self.hacs, self.repository.content.path.local):
             return
         if os.path.exists(self.backup_path):
             shutil.rmtree(self.backup_path)
@@ -117,4 +128,4 @@ def cleanup(self):
             shutil.rmtree(self.backup_path)
             while os.path.exists(self.backup_path):
                 sleep(0.1)
-            _LOGGER.debug("Backup dir %s cleared", self.backup_path)
+            self.hacs.log.debug("Backup dir %s cleared", self.backup_path)

custom_components/hacs/utils/path.py

@@ -10,12 +10,11 @@
 
 def is_safe(hacs: Hacs, path: str | Path) -> bool:
     """Helper to check if path is safe to remove."""
-    paths = [
-        Path(f"{hacs.core.config_path}/{hacs.configuration.appdaemon_path}"),
-        Path(f"{hacs.core.config_path}/{hacs.configuration.netdaemon_path}"),
-        Path(f"{hacs.core.config_path}/{hacs.configuration.plugin_path}"),
-        Path(f"{hacs.core.config_path}/{hacs.configuration.python_script_path}"),
-        Path(f"{hacs.core.config_path}/{hacs.configuration.theme_path}"),
-        Path(f"{hacs.core.config_path}/custom_components/"),
-    ]
-    return Path(path) not in paths
+    return Path(path).as_posix() not in (
+        Path(f"{hacs.core.config_path}/{hacs.configuration.appdaemon_path}").as_posix(),
+        Path(f"{hacs.core.config_path}/{hacs.configuration.netdaemon_path}").as_posix(),
+        Path(f"{hacs.core.config_path}/{hacs.configuration.plugin_path}").as_posix(),
+        Path(f"{hacs.core.config_path}/{hacs.configuration.python_script_path}").as_posix(),
+        Path(f"{hacs.core.config_path}/{hacs.configuration.theme_path}").as_posix(),
+        Path(f"{hacs.core.config_path}/custom_components/").as_posix(),
+    )

tests/hacsbase/test_backup.py

@@ -5,11 +5,11 @@
 from custom_components.hacs.operational.backup import Backup, BackupNetDaemon
 
 
-def test_file(tmpdir):
+def test_file(hacs, tmpdir):
     with open(f"{tmpdir.dirname}/dummy_file", "w") as dummy:
         dummy.write("")
 
-    backup = Backup(f"{tmpdir.dirname}/dummy_file")
+    backup = Backup(hacs, f"{tmpdir.dirname}/dummy_file")
     backup.create()
 
     assert not os.path.exists(backup.local_path)
@@ -22,10 +22,10 @@ def test_file(tmpdir):
     assert not os.path.exists(backup.backup_path_full)
 
 
-def test_directory(tmpdir):
+def test_directory(hacs, tmpdir):
     os.makedirs(f"{tmpdir.dirname}/dummy_directory", exist_ok=True)
 
-    backup = Backup(f"{tmpdir.dirname}/dummy_directory")
+    backup = Backup(hacs, f"{tmpdir.dirname}/dummy_directory")
     backup.create()
 
     assert not os.path.exists(backup.local_path)
@@ -38,17 +38,17 @@ def test_directory(tmpdir):
     assert not os.path.exists(backup.backup_path_full)
 
 
-def test_muilti(tmpdir):
-    backup = Backup(f"{tmpdir.dirname}/dummy_directory")
+def test_muilti(hacs, tmpdir):
+    backup = Backup(hacs, f"{tmpdir.dirname}/dummy_directory")
     backup.create()
     backup.create()
 
 
-def test_netdaemon_backup(repository_netdaemon):
+def test_netdaemon_backup(hacs, repository_netdaemon):
     repository = repository_netdaemon
     repository.content.path.local = repository.localpath
     os.makedirs(repository.content.path.local, exist_ok=True)
-    backup = BackupNetDaemon(repository)
+    backup = BackupNetDaemon(hacs, repository)
     backup.cleanup()
     with open(f"{repository.content.path.local}/dummy_file.yaml", "w") as dummy:
         dummy.write("test: test")

tests/utils/test_path.py

@@ -0,0 +1,9 @@
+from custom_components.hacs.base import HacsBase
+from custom_components.hacs.utils import path
+
+
+def test_is_safe(hacs: HacsBase) -> None:
+    assert path.is_safe(hacs, "/test")
+    assert not path.is_safe(hacs, f"{hacs.core.config_path}/{hacs.configuration.theme_path}/")
+    assert not path.is_safe(hacs, f"{hacs.core.config_path}/custom_components/")
+    assert not path.is_safe(hacs, f"{hacs.core.config_path}/custom_components")