feat: provide method for retrieving the cache_asset path of a given asset

snakemake_interface_software_deployment_plugins/__init__.py

@@ -399,7 +399,7 @@ async def managed_cache_asset(self, asset: str) -> None:
         )
         os.close(fd)
         tmp_cache_path = Path(tmp_cache_path)
-        cache_path = self.cache_path / asset
+        cache_path = self.get_cache_asset_path(asset)
         try:
             await self.cache_asset(asset, tmp_cache_path)
             os.replace(tmp_cache_path, cache_path)
@@ -415,6 +415,13 @@ async def managed_cache_asset(self, asset: str) -> None:
     def cache_path(self) -> Path:
         return self._cache_prefix
 
+    @property
+    def get_cache_asset_path(self, asset: str) -> Path:
+        """Return the path to the cached asset.
+        This does not check whether the asset exists.
+        """
+        return self.cache_path / asset
+
     async def remove_cache(self) -> None:
         """Remove the cached environment assets."""
         for asset in await self.get_cache_assets():

snakemake_interface_software_deployment_plugins/tests.py

@@ -101,6 +101,10 @@ def test_shellcmd(self, tmp_path):
         assert cmd != decorated_cmd
         assert env.run_cmd(decorated_cmd).returncode == 0
 
+    async def _cache_assets(env):
+        for asset in await env.get_cache_assets():
+            await env.managed_cache_asset(asset)
+
     def test_deploy(self, tmp_path):
         env = self._get_env(tmp_path)
         self._deploy(env, tmp_path)
@@ -114,13 +118,7 @@ def test_cache(self, tmp_path):
 
         assert isinstance(env, CacheableEnvBase)
 
-        async def cache_assets(env):
-            for asset in await env.get_cache_assets():
-                await env.managed_cache_asset(asset)
-
-        asyncio.run(cache_assets(env))
-
-        self._deploy(env, tmp_path)
+        asyncio.run(self.cache_assets(env))
 
         assert any(env.cache_path.iterdir())
 
@@ -134,7 +132,6 @@ def test_pin(self, tmp_path):
         asyncio.run(env.pin())
         assert env.pinfile.exists()
         print("Pinfile content:", env.pinfile.read_text(), sep="\n")
-        self._deploy(env, tmp_path)
 
     def test_report_software(self, tmp_path):
         env = self._get_env(tmp_path)
@@ -257,6 +254,11 @@ def _deploy(self, env: EnvBase, tmp_path):
         if not env.is_deployable():
             pytest.skip("Environment is not deployable.")
 
+        if env.is_pinnable():
+            asyncio.run(env.pin())
+        if env.is_cacheable():
+            asyncio.run(self._cache_assets(env))
+
         assert isinstance(env, DeployableEnvBase)
         asyncio.run(env.deploy())
         assert any((tmp_path / env.spec.module().__name__ / "deployments").iterdir())