Check local store repository for changes (#5845)

* Check local store repository for changes

Instead of simply assume that the local store repository got changed,
use mtime to check if there have been any changes to the local store.
This mimics a similar behavior to the git repository store updates.

Before this change, we end up in the updated repo code path, which
caused a re-read of all add-ons on every store reload, even though
nothing changed at all. Store reloads are triggered by Home Assistant
Core every 5 minutes.

* Fix pytest failure

Now that we actually only reload metadata if the local store changed
we have to fake the change as well to fix the store manager tests.

* Fix path cache update test for local store repository

* Take root directory into account/add pytest

* Rename utils/__init__.py tests to test_utils_init.py

supervisor/store/repository.py

@@ -5,6 +5,8 @@ from pathlib import Path
 
 import voluptuous as vol
 
+from supervisor.utils import get_latest_mtime
+
 from ..const import ATTR_MAINTAINER, ATTR_NAME, ATTR_URL, FILE_SUFFIX_CONFIGURATION
 from ..coresys import CoreSys, CoreSysAttributes
 from ..exceptions import ConfigurationFileError, StoreError
@@ -19,10 +21,10 @@ UNKNOWN = "unknown"
 
 
 class Repository(CoreSysAttributes):
-    """Repository in Supervisor."""
+    """Add-on store repository in Supervisor."""
 
     def __init__(self, coresys: CoreSys, repository: str):
-        """Initialize repository object."""
+        """Initialize add-on store repository object."""
         self.coresys: CoreSys = coresys
         self.git: GitRepo | None = None
 
@@ -30,6 +32,7 @@ class Repository(CoreSysAttributes):
         if repository == StoreType.LOCAL:
             self._slug = repository
             self._type = StoreType.LOCAL
+            self._latest_mtime: float | None = None
         elif repository == StoreType.CORE:
             self.git = GitRepoHassIO(coresys)
             self._slug = repository
@@ -102,14 +105,37 @@ class Repository(CoreSysAttributes):
     async def load(self) -> None:
         """Load addon repository."""
         if not self.git:
+            self._latest_mtime, _ = await self.sys_run_in_executor(
+                get_latest_mtime, self.sys_config.path_addons_local
+            )
             return
         await self.git.load()
 
     async def update(self) -> bool:
-        """Update add-on repository."""
+        """Update add-on repository.
+
+        Returns True if the repository was updated.
+        """
         if not await self.sys_run_in_executor(self.validate):
             return False
-        return self.type == StoreType.LOCAL or await self.git.pull()
+
+        if self.type != StoreType.LOCAL:
+            return await self.git.pull()
+
+        # Check local modifications
+        latest_mtime, modified_path = await self.sys_run_in_executor(
+            get_latest_mtime, self.sys_config.path_addons_local
+        )
+        if self._latest_mtime != latest_mtime:
+            _LOGGER.debug(
+                "Local modifications detected in %s repository: %s",
+                self.slug,
+                modified_path,
+            )
+            self._latest_mtime = latest_mtime
+            return True
+
+        return False
 
     async def remove(self) -> None:
         """Remove add-on repository."""

supervisor/utils/__init__.py

@@ -131,6 +131,28 @@ def remove_folder_with_excludes(
             item.rename(folder / item.name)
 
 
+def get_latest_mtime(directory: Path) -> tuple[float, Path]:
+    """Get the last modification time of directories and files in a directory.
+
+    Must be run in an executor. The root directory is included too, this means
+    that often the root directory is returned as the last modified file if a
+    new file is created in it.
+    """
+    latest_mtime = directory.stat().st_mtime
+    latest_path = directory
+    for path in directory.rglob("*"):
+        try:
+            mtime = path.stat().st_mtime
+            if mtime > latest_mtime:
+                latest_mtime = mtime
+                latest_path = path
+        except FileNotFoundError:
+            # File might disappear between listing and stat. Parent
+            # directory modification date will flag such a change.
+            continue
+    return latest_mtime, latest_path
+
+
 def clean_env() -> dict[str, str]:
     """Return a clean env from system."""
     new_env = {}

tests/addons/test_addon.py

@@ -790,13 +790,17 @@ def test_auto_update_available(coresys: CoreSys, install_addon_example: Addon):
 
 async def test_paths_cache(coresys: CoreSys, install_addon_ssh: Addon):
     """Test cache for key paths that may or may not exist."""
-    with patch("supervisor.addons.addon.Path.exists", return_value=True):
-        assert not install_addon_ssh.with_logo
-        assert not install_addon_ssh.with_icon
-        assert not install_addon_ssh.with_changelog
-        assert not install_addon_ssh.with_documentation
+    assert not install_addon_ssh.with_logo
+    assert not install_addon_ssh.with_icon
+    assert not install_addon_ssh.with_changelog
+    assert not install_addon_ssh.with_documentation
 
+    with (
+        patch("supervisor.addons.addon.Path.exists", return_value=True),
+        patch("supervisor.store.repository.Repository.update", return_value=True),
+    ):
         await coresys.store.reload(coresys.store.get("local"))
+
         assert install_addon_ssh.with_logo
         assert install_addon_ssh.with_icon
         assert install_addon_ssh.with_changelog

tests/store/test_store_manager.py

@@ -250,5 +250,8 @@ async def test_addon_version_timestamp(coresys: CoreSys, install_addon_example:
 
     # If a new version is seen processing repo, reset to utc now
     install_addon_example.data_store["version"] = "1.1.0"
-    await coresys.store.reload()
-    assert timestamp < install_addon_example.latest_version_timestamp
+
+    # Signal the store repositories got updated
+    with patch("supervisor.store.repository.Repository.update", return_value=True):
+        await coresys.store.reload()
+        assert timestamp < install_addon_example.latest_version_timestamp

tests/utils/test_utils_init.py

@@ -0,0 +1,85 @@
+"""Unit tests for the utils function."""
+
+import os
+from pathlib import Path
+import tempfile
+
+from supervisor.utils import get_latest_mtime  # Adjust the import as needed
+
+
+def test_get_latest_mtime_with_files():
+    """Test the latest mtime with files in the directory."""
+    with tempfile.TemporaryDirectory() as tmpdirname:
+        tmpdir = Path(tmpdirname)
+
+        # Set an old mtime for the directory itself
+        dir_mtime_initial = 1000000000
+
+        # Create first file
+        file1 = tmpdir / "file1.txt"
+        file1.write_text("First file")
+        # After creating file1, directory mtime was modified.
+        os.utime(tmpdir, (dir_mtime_initial, dir_mtime_initial))
+
+        file1_mtime = 1000000100
+        os.utime(file1, (file1_mtime, file1_mtime))
+
+        # Reset directory mtime back to older so that file1 is correctly detected
+        os.utime(tmpdir, (dir_mtime_initial, dir_mtime_initial))
+
+        # Verify file1 is detected
+        latest_mtime1, latest_path1 = get_latest_mtime(tmpdir)
+        assert latest_path1 == file1
+        assert latest_mtime1 == file1_mtime
+
+        # Create second file
+        file2 = tmpdir / "file2.txt"
+        file2.write_text("Second file")
+
+        # Verify change is detected
+        # Often the directory itself is the latest modified file
+        # because a new file was created in it. But this is not
+        # guaranteed, and also not relevant for the test.
+        latest_mtime2, _ = get_latest_mtime(tmpdir)
+        assert latest_mtime2 > latest_mtime1
+
+
+def test_get_latest_mtime_directory_when_empty():
+    """Test the latest mtime when the directory cleared."""
+    with tempfile.TemporaryDirectory() as tmpdirname:
+        tmpdir = Path(tmpdirname)
+
+        # Set initial mtime for the directory
+        dir_mtime_initial = 1000000000
+
+        # Create a file
+        file1 = tmpdir / "file1.txt"
+        file1.write_text("Temporary file")
+        file1_mtime = 1000000100
+        # After creating file1, directory mtime was modified.
+        os.utime(tmpdir, (dir_mtime_initial, dir_mtime_initial))
+        os.utime(file1, (file1_mtime, file1_mtime))
+
+        # Verify the file is the latest
+        latest_mtime1, latest_path = get_latest_mtime(tmpdir)
+        assert latest_path == file1
+        assert latest_mtime1 == file1_mtime
+
+        # Now delete the file
+        file1.unlink()
+
+        # Now the directory itself should be the latest
+        latest_mtime2, latest_path = get_latest_mtime(tmpdir)
+        assert latest_path == tmpdir
+        assert latest_mtime2 > latest_mtime1
+
+
+def test_get_latest_mtime_empty_directory():
+    """Test the latest mtime when the directory is empty."""
+    with tempfile.TemporaryDirectory() as tmpdirname:
+        tmpdir = Path(tmpdirname)
+
+        # Directory is empty
+        latest_mtime, latest_path = get_latest_mtime(tmpdir)
+        assert latest_path == tmpdir
+        assert latest_mtime > 0