diff --git a/pkgs/clan-cli/clan_lib/flake/flake.py b/pkgs/clan-cli/clan_lib/flake/flake.py index 8be0ff0a0..9229da0a6 100644 --- a/pkgs/clan-cli/clan_lib/flake/flake.py +++ b/pkgs/clan-cli/clan_lib/flake/flake.py @@ -9,6 +9,7 @@ from tempfile import NamedTemporaryFile from typing import Any from clan_lib.errors import ClanError +from clan_lib.nix import nix_config log = logging.getLogger(__name__) @@ -319,9 +320,7 @@ class FlakeCacheEntry: # strings need to be checked if they are store paths # if they are, we store them as a dict with the outPath key # this is to mirror nix behavior, where the outPath of an attrset is used if no further key is specified - elif isinstance(value, str) and value.startswith( - os.environ.get("NIX_STORE_DIR", "/nix/store") - ): + elif isinstance(value, str) and self._is_store_path(value): assert selectors == [] self.value = {"outPath": FlakeCacheEntry(value)} @@ -335,14 +334,68 @@ class FlakeCacheEntry: msg = f"Cannot insert {value} into cache, already have {self.value}" raise TypeError(msg) + def _is_store_path(self, value: str) -> bool: + """Check if a string is a nix store path.""" + # A store path is any path that has "store" as one of its parent directories + # and contains a hash-prefixed name after it + path_parts = Path(value).parts + + try: + store_idx = path_parts.index("store") + except ValueError: + return False + + # Check if there's at least one more component after "store" + if store_idx + 1 < len(path_parts): + # Check if the component after store looks like a nix store item + # (starts with a hash) + store_item = path_parts[store_idx + 1] + # Basic check: nix store items typically start with a hash + return len(store_item) > 32 and "-" in store_item + + return False + + def _normalize_store_path(self, store_path: str) -> Path | None: + """ + Normalize a store path to use the current NIX_STORE_DIR. + Returns None if the path is not a valid store path. + """ + # Extract the store item (hash-name) from the path + path_parts = Path(store_path).parts + + # Find the index of "store" in the path + try: + store_idx = path_parts.index("store") + except ValueError: + return None + + if store_idx + 1 < len(path_parts): + store_item = path_parts[store_idx + 1] + + # Get the current store path + # Check if we're using a test store first + test_store = os.environ.get("CLAN_TEST_STORE") + if test_store: + # In test mode, the store is at CLAN_TEST_STORE/nix/store + current_store = str(Path(test_store) / "nix" / "store") + else: + # Otherwise use nix config + config = nix_config() + current_store = config.get("store", "/nix/store") + + return Path(current_store) / store_item if current_store else None + + return None + def is_cached(self, selectors: list[Selector]) -> bool: selector: Selector # for store paths we have to check if they still exist, otherwise they have to be rebuild and are thus not cached - if isinstance(self.value, str) and self.value.startswith( - os.environ.get("NIX_STORE_DIR", "/nix/store") - ): - return Path(self.value).exists() + if isinstance(self.value, str) and self._is_store_path(self.value): + normalized_path = self._normalize_store_path(self.value) + if normalized_path: + return normalized_path.exists() + return False # if self.value is not dict but we request more selectors, we assume we are cached and an error will be thrown in the select function if isinstance(self.value, str | float | int | None): diff --git a/pkgs/clan-cli/clan_lib/tests/test_flake_cache_chroot.py b/pkgs/clan-cli/clan_lib/tests/test_flake_cache_chroot.py new file mode 100644 index 000000000..6ee561e69 --- /dev/null +++ b/pkgs/clan-cli/clan_lib/tests/test_flake_cache_chroot.py @@ -0,0 +1,108 @@ +"""Test flake cache with chroot stores.""" + +import os +import tempfile +from pathlib import Path + +import pytest + +from clan_lib.flake.flake import FlakeCache, FlakeCacheEntry + + +def test_flake_cache_with_chroot_store( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Test that flake cache works correctly with chroot stores.""" + # Create a mock store path + normal_store = "/nix/store" + chroot_store = str(tmp_path / "nix" / "store") + + # Create the chroot store directory + Path(chroot_store).mkdir(parents=True) + + # Create a fake derivation in the chroot store with proper nix store format + fake_drv = "abcd1234abcd1234abcd1234abcd1234-test-package" + fake_store_path = f"{chroot_store}/{fake_drv}" + Path(fake_store_path).mkdir(parents=True) + + # Test 1: Cache entry with normal store path + cache = FlakeCache() + + # Insert a store path that doesn't exist in the normal store + non_existent_path = f"{normal_store}/{fake_drv}" + cache.insert({"test": {"package": non_existent_path}}, "") + + # Without chroot store, this should be uncached (path doesn't exist) + assert not cache.is_cached("test.package") + + # Test 2: Set CLAN_TEST_STORE to chroot store parent + # CLAN_TEST_STORE should point to the parent of nix/store + monkeypatch.setenv("CLAN_TEST_STORE", str(tmp_path)) + + # Create a new cache with the chroot store + cache2 = FlakeCache() + + # Insert the same path but now it should use chroot store + cache2.insert({"test": {"package": fake_store_path}}, "") + + # This should be cached because the path exists in chroot store + assert cache2.is_cached("test.package") + + # Test 3: Cache persistence with chroot store + cache_file = tmp_path / "cache.json" + cache2.save_to_file(cache_file) + + # Load cache in a new instance + cache3 = FlakeCache() + cache3.load_from_file(cache_file) + + # Should still be cached with chroot store + assert cache3.is_cached("test.package") + + # Test 4: Cache validation fails when chroot store changes + monkeypatch.setenv("CLAN_TEST_STORE", "/different") + + # Same cache should now be invalid + assert not cache3.is_cached("test.package") + + +def test_flake_cache_entry_store_path_validation() -> None: + """Test that FlakeCacheEntry correctly validates store paths.""" + # Test with default store + entry = FlakeCacheEntry() + + # Insert a non-existent store path with proper format + fake_path = "/nix/store/abcd1234abcd1234abcd1234abcd1234-fake-package" + entry.insert(fake_path, []) + + # Should not be cached because path doesn't exist + assert not entry.is_cached([]) + + # Test with environment variable + with tempfile.TemporaryDirectory() as tmpdir: + # Create nix/store structure + store_dir = Path(tmpdir) / "nix" / "store" + store_dir.mkdir(parents=True) + + # Create a fake store path with proper format + fake_drv = "test1234test1234test1234test1234-package" + fake_path_obj = store_dir / fake_drv + fake_path_obj.mkdir() + fake_path = str(fake_path_obj) + + # Set CLAN_TEST_STORE to parent of store dir + old_env = os.environ.get("CLAN_TEST_STORE") + try: + os.environ["CLAN_TEST_STORE"] = str(tmpdir) + + entry2 = FlakeCacheEntry() + entry2.insert(str(fake_path), []) + + # Should be cached because path exists + assert entry2.is_cached([]) + + finally: + if old_env is None: + os.environ.pop("CLAN_TEST_STORE", None) + else: + os.environ["CLAN_TEST_STORE"] = old_env