Support copying and cloning to temporary directories

foamlib/_cases/_async.py

@@ -2,7 +2,9 @@
 import multiprocessing
 import os
 import sys
+import tempfile
 from contextlib import asynccontextmanager
+from pathlib import Path
 from typing import (
     Callable,
     Optional,
@@ -14,10 +16,16 @@
 else:
     from typing import AsyncGenerator, Collection, Sequence
 
+if sys.version_info >= (3, 11):
+    from typing import Self
+else:
+    from typing_extensions import Self
+
 import aioshutil
 
 from ._recipes import _FoamCaseRecipes
 from ._subprocess import run_async
+from ._util import awaitableasynccontextmanager
 
 
 class AsyncFoamCase(_FoamCaseRecipes):
@@ -166,24 +174,60 @@ async def restore_0_dir(self) -> None:
         for name, args, kwargs in self._restore_0_dir_cmds():
             await getattr(self, name)(*args, **kwargs)
 
-    async def copy(self, dst: Union["os.PathLike[str]", str]) -> "AsyncFoamCase":
+    @awaitableasynccontextmanager
+    @asynccontextmanager
+    async def copy(
+        self, dst: Optional[Union["os.PathLike[str]", str]] = None
+    ) -> "AsyncGenerator[Self]":
         """
         Make a copy of this case.
 
-        :param dst: The destination path.
+        Use as an async context manager to automatically delete the copy when done.
+
+        :param dst: The destination path. If None, copy to a temporary directory.
         """
+        if dst is None:
+            dst = Path(tempfile.mkdtemp(), self.name)
+            tmp = True
+        else:
+            tmp = False
+
         for name, args, kwargs in self._copy_cmds(dst):
             await getattr(self, name)(*args, **kwargs)
 
-        return AsyncFoamCase(dst)
+        yield type(self)(dst)
 
-    async def clone(self, dst: Union["os.PathLike[str]", str]) -> "AsyncFoamCase":
+        if tmp:
+            assert isinstance(dst, Path)
+            await self._rmtree(dst.parent)
+        else:
+            await self._rmtree(dst)
+
+    @awaitableasynccontextmanager
+    @asynccontextmanager
+    async def clone(
+        self, dst: Optional[Union["os.PathLike[str]", str]] = None
+    ) -> "AsyncGenerator[Self]":
         """
         Clone this case (make a clean copy).
 
-        :param dst: The destination path.
+        Use as an async context manager to automatically delete the clone when done.
+
+        :param dst: The destination path. If None, clone to a temporary directory.
         """
+        if dst is None:
+            dst = Path(tempfile.mkdtemp(), self.name)
+            tmp = True
+        else:
+            tmp = False
+
         for name, args, kwargs in self._clone_cmds(dst):
             await getattr(self, name)(*args, **kwargs)
 
-        return AsyncFoamCase(dst)
+        yield type(self)(dst)
+
+        if tmp:
+            assert isinstance(dst, Path)
+            await self._rmtree(dst.parent)
+        else:
+            await self._rmtree(dst)

foamlib/_cases/_sync.py

@@ -1,9 +1,13 @@
 import os
 import shutil
 import sys
+import tempfile
+from pathlib import Path
+from types import TracebackType
 from typing import (
     Callable,
     Optional,
+    Type,
     Union,
 )
 
@@ -12,6 +16,11 @@
 else:
     from typing import Collection, Sequence
 
+if sys.version_info >= (3, 11):
+    from typing import Self
+else:
+    from typing_extensions import Self
+
 from ._recipes import _FoamCaseRecipes
 from ._subprocess import run_sync
 
@@ -45,6 +54,17 @@ def _copytree(
     ) -> None:
         shutil.copytree(src, dest, symlinks=symlinks, ignore=ignore)
 
+    def __enter__(self) -> "FoamCase":
+        return self
+
+    def __exit__(
+        self,
+        exc_type: Optional[Type[BaseException]],
+        exc_val: Optional[BaseException],
+        exc_tb: Optional[TracebackType],
+    ) -> None:
+        self._rmtree(self.path)
+
     def clean(
         self,
         *,
@@ -130,24 +150,34 @@ def restore_0_dir(self) -> None:
         for name, args, kwargs in self._restore_0_dir_cmds():
             getattr(self, name)(*args, **kwargs)
 
-    def copy(self, dst: Union["os.PathLike[str]", str]) -> "FoamCase":
+    def copy(self, dst: Optional[Union["os.PathLike[str]", str]] = None) -> "Self":
         """
         Make a copy of this case.
 
-        :param dst: The destination path.
+        Use as a context manager to automatically delete the copy when done.
+
+        :param dst: The destination path. If None, copy to a temporary directory.
         """
+        if dst is None:
+            dst = Path(tempfile.mkdtemp(), self.name)
+
         for name, args, kwargs in self._copy_cmds(dst):
             getattr(self, name)(*args, **kwargs)
 
-        return FoamCase(dst)
+        return type(self)(dst)
 
-    def clone(self, dst: Union["os.PathLike[str]", str]) -> "FoamCase":
+    def clone(self, dst: Optional[Union["os.PathLike[str]", str]] = None) -> "Self":
         """
         Clone this case (make a clean copy).
 
-        :param dst: The destination path.
+        Use as a context manager to automatically delete the clone when done.
+
+        :param dst: The destination path. If None, clone to a temporary directory.
         """
+        if dst is None:
+            dst = Path(tempfile.mkdtemp(), self.name)
+
         for name, args, kwargs in self._clone_cmds(dst):
             getattr(self, name)(*args, **kwargs)
 
-        return FoamCase(dst)
+        return type(self)(dst)

foamlib/_cases/_util.py

@@ -0,0 +1,27 @@
+from types import TracebackType
+from typing import Any, AsyncContextManager, Callable, Optional, Type
+
+
+class _AwaitableAsyncContextManager:
+    def __init__(self, cm: "AsyncContextManager[Any]"):
+        self._cm = cm
+
+    def __await__(self) -> Any:
+        return self._cm.__aenter__().__await__()
+
+    async def __aenter__(self) -> Any:
+        return await self._cm.__aenter__()
+
+    async def __aexit__(
+        self,
+        exc_type: Optional[Type[BaseException]],
+        exc_val: Optional[BaseException],
+        exc_tb: Optional[TracebackType],
+    ) -> Any:
+        return await self._cm.__aexit__(exc_type, exc_val, exc_tb)
+
+
+def awaitableasynccontextmanager(
+    cm: Callable[..., "AsyncContextManager[Any]"],
+) -> Callable[..., _AwaitableAsyncContextManager]:
+    return lambda *args, **kwargs: _AwaitableAsyncContextManager(cm(*args, **kwargs))

tests/test_cases/test_flange.py

@@ -1,19 +1,26 @@
 import os
+import sys
 from pathlib import Path
 
+if sys.version_info >= (3, 9):
+    from collections.abc import Generator
+else:
+    from typing import Generator
+
 import pytest
 from foamlib import CalledProcessError, FoamCase
 
 
 @pytest.fixture
-def flange(tmp_path: Path) -> FoamCase:
+def flange() -> "Generator[FoamCase]":
     tutorials_path = Path(os.environ["FOAM_TUTORIALS"])
     path = tutorials_path / "basic" / "laplacianFoam" / "flange"
     of11_path = tutorials_path / "legacy" / "basic" / "laplacianFoam" / "flange"
 
     case = FoamCase(path if path.exists() else of11_path)
 
-    return case.clone(tmp_path / case.name)
+    with case.clone() as clone:
+        yield clone
 
 
 @pytest.mark.parametrize("parallel", [True, False])

tests/test_cases/test_flange_async.py

@@ -1,20 +1,27 @@
 import os
+import sys
 from pathlib import Path
 
+if sys.version_info >= (3, 9):
+    from collections.abc import AsyncGenerator
+else:
+    from typing import AsyncGenerator
+
 import pytest
 import pytest_asyncio
 from foamlib import AsyncFoamCase, CalledProcessError
 
 
 @pytest_asyncio.fixture
-async def flange(tmp_path: Path) -> AsyncFoamCase:
+async def flange() -> "AsyncGenerator[AsyncFoamCase]":
     tutorials_path = Path(os.environ["FOAM_TUTORIALS"])
     path = tutorials_path / "basic" / "laplacianFoam" / "flange"
     of11_path = tutorials_path / "legacy" / "basic" / "laplacianFoam" / "flange"
 
     case = AsyncFoamCase(path if path.exists() else of11_path)
 
-    return await case.clone(tmp_path / case.name)
+    async with case.clone() as clone:
+        yield clone
 
 
 @pytest.mark.asyncio

tests/test_cases/test_pitz.py

@@ -1,20 +1,27 @@
 import os
+import sys
 from pathlib import Path
 from typing import Sequence
 
+if sys.version_info >= (3, 9):
+    from collections.abc import Generator
+else:
+    from typing import Generator
+
 import pytest
 from foamlib import FoamCase
 
 
 @pytest.fixture
-def pitz(tmp_path: Path) -> FoamCase:
+def pitz() -> "Generator[FoamCase]":
     tutorials_path = Path(os.environ["FOAM_TUTORIALS"])
     path = tutorials_path / "incompressible" / "simpleFoam" / "pitzDaily"
     of11_path = tutorials_path / "incompressibleFluid" / "pitzDaily"
 
     case = FoamCase(path if path.exists() else of11_path)
 
-    return case.clone(tmp_path / case.name)
+    with case.clone() as clone:
+        yield clone
 
 
 def test_run(pitz: FoamCase) -> None:

tests/test_cases/test_pitz_async.py

@@ -1,21 +1,28 @@
 import os
+import sys
 from pathlib import Path
 from typing import Sequence
 
+if sys.version_info >= (3, 9):
+    from collections.abc import AsyncGenerator
+else:
+    from typing import AsyncGenerator
+
 import pytest
 import pytest_asyncio
 from foamlib import AsyncFoamCase
 
 
 @pytest_asyncio.fixture
-async def pitz(tmp_path: Path) -> AsyncFoamCase:
+async def pitz() -> "AsyncGenerator[AsyncFoamCase]":
     tutorials_path = Path(os.environ["FOAM_TUTORIALS"])
     path = tutorials_path / "incompressible" / "simpleFoam" / "pitzDaily"
     of11_path = tutorials_path / "incompressibleFluid" / "pitzDaily"
 
     case = AsyncFoamCase(path if path.exists() else of11_path)
 
-    return await case.clone(tmp_path / case.name)
+    async with case.clone() as clone:
+        yield clone
 
 
 @pytest.mark.asyncio

tests/test_files/test_files.py

@@ -3,9 +3,9 @@
 from pathlib import Path
 
 if sys.version_info >= (3, 9):
-    from collections.abc import Sequence
+    from collections.abc import Generator, Sequence
 else:
-    from typing import Sequence
+    from typing import Generator, Sequence
 
 import numpy as np
 import pytest
@@ -93,14 +93,15 @@ def test_new_field(tmp_path: Path) -> None:
 
 
 @pytest.fixture
-def pitz(tmp_path: Path) -> FoamCase:
+def pitz() -> "Generator[FoamCase]":
     tutorials_path = Path(os.environ["FOAM_TUTORIALS"])
     path = tutorials_path / "incompressible" / "simpleFoam" / "pitzDaily"
     of11_path = tutorials_path / "incompressibleFluid" / "pitzDaily"
 
     case = FoamCase(path if path.exists() else of11_path)
 
-    return case.clone(tmp_path / case.name)
+    with case.clone() as clone:
+        yield clone
 
 
 def test_dimensions(pitz: FoamCase) -> None: