diff --git a/supervisor/store/repository.py b/supervisor/store/repository.py index 519312691..2d888066a 100644 --- a/supervisor/store/repository.py +++ b/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.""" diff --git a/supervisor/utils/__init__.py b/supervisor/utils/__init__.py index cce78850f..339191aca 100644 --- a/supervisor/utils/__init__.py +++ b/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 = {} diff --git a/tests/addons/test_addon.py b/tests/addons/test_addon.py index 7acd244ee..8c52a25a7 100644 --- a/tests/addons/test_addon.py +++ b/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 diff --git a/tests/store/test_store_manager.py b/tests/store/test_store_manager.py index 43a6609ec..ac3fba200 100644 --- a/tests/store/test_store_manager.py +++ b/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 diff --git a/tests/utils/test_utils_init.py b/tests/utils/test_utils_init.py new file mode 100644 index 000000000..5bcee7533 --- /dev/null +++ b/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