Add minimal pytest plugin to verify that generated files are cleaned …

…up by tests (#1488)

* Add pytest plugin to verify that test modules clean up any generated file

* Format with Black and remove unused import

* Add docstring to make Pylint happy

* Format with Black

* Switch to per-test file checks and Python 3.9+ type hinting

* Apply fixtures to offending tests

* Try resolving next batch of failures

* Try addressing two failures

* Switch to using module-level fixture for tmp_path workdir

idaes/conftest.py

@@ -16,10 +16,12 @@
 
 import importlib.abc
 import importlib.machinery
+import os
+import subprocess
 import sys
-from typing import Dict
-from typing import Iterable
-from typing import List
+from collections.abc import Iterable
+from pathlib import Path
+from typing import Union
 
 import pytest
 
@@ -195,7 +197,7 @@ class ImportorskipFinder(importlib.abc.MetaPathFinder):
     a custom Loader to be used for registered modules.
     """
 
-    def __init__(self, registry: Dict[ModuleName, List[ModuleName]]):
+    def __init__(self, registry: dict[ModuleName, list[ModuleName]]):
         self._registry = registry
 
     def find_spec(self, *args, **kwargs):
@@ -217,7 +219,7 @@ class Importorskipper:
     instead of having to call importorskip() in each test module for each possibly missing module.
     """
 
-    def __init__(self, registry: Dict[ModuleName, List[ModuleName]]):
+    def __init__(self, registry: dict[ModuleName, list[ModuleName]]):
         self._registry = dict(registry)
         self._finder = ImportorskipFinder(self._registry)
 
@@ -227,7 +229,7 @@ def pytest_configure(self):
     def pytest_sessionfinish(self):
         sys.meta_path.remove(self._finder)
 
-    def pytest_report_collectionfinish(self) -> List[str]:
+    def pytest_report_collectionfinish(self) -> list[str]:
         preamble = [
             "The following modules are registered in the importorskipper plugin",
             " and will cause tests to be skipped if any of the registered modules is not found: ",
@@ -240,6 +242,126 @@ def pytest_report_collectionfinish(self) -> List[str]:
         return lines
 
 
+@pytest.fixture(scope="function")
+def run_in_tmp_path(tmp_path: Path):
+    prev_workdir = os.getcwd()
+    try:
+        os.chdir(tmp_path)
+        yield
+    finally:
+        os.chdir(prev_workdir)
+
+
+@pytest.fixture(scope="class")
+def run_class_in_tmp_path(
+    request: pytest.FixtureRequest, tmp_path_factory: pytest.TempPathFactory
+):
+    prev_workdir = os.getcwd()
+
+    modname = request.module.__name__
+    clsname = request.__class__.__name__
+    dirname = f"{modname}-{clsname}"
+
+    tmp_path = tmp_path_factory.mktemp(dirname, numbered=True)
+    try:
+        os.chdir(tmp_path)
+        yield
+    finally:
+        os.chdir(prev_workdir)
+
+
+@pytest.fixture(scope="module")
+def run_module_in_tmp_path(
+    request: pytest.FixtureRequest, tmp_path_factory: pytest.TempPathFactory
+):
+    prev_workdir = os.getcwd()
+
+    modname = request.module.__name__
+
+    tmp_path = tmp_path_factory.mktemp(modname, numbered=True)
+    try:
+        os.chdir(tmp_path)
+        yield
+    finally:
+        os.chdir(prev_workdir)
+
+
+def _get_repo_root_dir() -> Union[Path, None]:
+    try:
+        text = subprocess.check_output(
+            ["git", "rev-parse", "--show-toplevel"],
+            text=True,
+        ).strip()
+    except subprocess.CalledProcessError:
+        # no Git directory found
+        return None
+    return Path(text).resolve()
+
+
+class VerifyCleanup:
+    """
+    A pytest plugin to verify that files generated by tests are cleaned up properly.
+    """
+
+    def __init__(self, repo_root_dir: Path):
+        self._repo_root_dir = Path(repo_root_dir).resolve()
+        self._added_by_test = {}
+
+    def _get_files(self) -> list[str]:
+        args = [
+            "git",
+            "-C",
+            self._repo_root_dir,
+            "ls-files",
+            "--others",
+            "--exclude-standard",
+        ]
+        try:
+            text = subprocess.check_output(
+                args,
+                text=True,
+            ).strip()
+        except subprocess.CalledProcessError as e:
+            text = str(e)
+        return text.splitlines()
+
+    def pytest_report_collectionfinish(self) -> list[str]:
+        return [
+            f"Looking for untracked files being created by tests for repo: {self._repo_root_dir}"
+        ]
+
+    @pytest.hookimpl(hookwrapper=True)
+    def pytest_runtest_protocol(self, item):
+        before = set(self._get_files())
+        yield
+        after = set(self._get_files())
+        added = after - before
+        if added:
+            key = item.nodeid
+            self._added_by_test[key] = sorted(added)
+
+    @pytest.hookimpl(trylast=True)
+    def pytest_terminal_summary(self, terminalreporter):
+        to_report = dict(self._added_by_test)
+        if not to_report:
+            return
+
+        tr = terminalreporter
+        tr.section("Files added (and not cleaned up) by tests")
+        for item, files in to_report.items():
+            tr.write_line(item)
+            for file in files:
+                tr.write_line(f"\t{file}")
+        if to_report:
+            tr.write_line(f"{len(to_report)} test(s) did not clean up after themselves")
+            tr.write_line("The exit status of the test session will be set to failed")
+
+    @pytest.hookimpl(trylast=True)
+    def pytest_sessionfinish(self, session, exitstatus):
+        if self._added_by_test:
+            session.exitstatus = pytest.ExitCode.TESTS_FAILED
+
+
 def pytest_addhooks(pluginmanager: pytest.PytestPluginManager):
     skipper_plugin = Importorskipper(
         {
@@ -248,6 +370,9 @@ def pytest_addhooks(pluginmanager: pytest.PytestPluginManager):
     )
 
     pluginmanager.register(skipper_plugin, name="importorskipper")
+    repo_root_dir = _get_repo_root_dir()
+    if repo_root_dir is not None:
+        pluginmanager.register(VerifyCleanup(repo_root_dir))
 
 
 ####

idaes/core/solvers/tests/test_solvers.py

@@ -105,6 +105,7 @@ def test_ipopt_idaes_solve():
     assert pytest.approx(x) == pyo.value(m.x)
 
 
+@pytest.mark.usefixtures("run_in_tmp_path")
 @pytest.mark.unit
 @pytest.mark.skipif(
     not pyo.SolverFactory("ipopt_l1").available(False), reason="solver not available"

idaes/core/surrogate/pysmo/tests/test_kriging.py

@@ -22,6 +22,7 @@
 import pytest
 
 
+@pytest.mark.usefixtures("run_class_in_tmp_path")
 class TestKrigingModel:
     y = np.array(
         [

idaes/core/surrogate/pysmo/tests/test_polynomial_regression.py

@@ -153,6 +153,7 @@ def test_data_unscaling_06(self, array_type):
             FeatureScaling.data_unscaling(output_1, min_array, max_array)
 
 
+@pytest.mark.usefixtures("run_class_in_tmp_path")
 class TestPolynomialRegression:
     y = np.array(
         [

idaes/core/surrogate/pysmo/tests/test_radial_basis_function.py

@@ -152,6 +152,7 @@ def test_data_unscaling_minmax_06(self, array_type):
             FeatureScaling.data_unscaling_minmax(output_1, min_array, max_array)
 
 
+@pytest.mark.usefixtures("run_class_in_tmp_path")
 class TestRadialBasisFunction:
     y = np.array(
         [

idaes/core/surrogate/tests/test_pysmo_surrogate.py

@@ -46,7 +46,7 @@
 from idaes.core.surrogate.metrics import compute_fit_metrics
 
 
-dirpath = Path(__file__).parent.resolve()
+pytestmark = pytest.mark.usefixtures("run_module_in_tmp_path")
 
 
 # String representation of json output for testing

idaes/models_extra/power_generation/flowsheets/subcritical_power_plant/tests/test_subcritical_flowsheets.py

@@ -174,6 +174,7 @@ def test_dynamic_steam_cycle():
     assert degrees_of_freedom(m) == 0
 
 
+@pytest.mark.usefixtures("run_in_tmp_path")
 @pytest.mark.skipif(not helmholtz_available(), reason="General Helmholtz not available")
 @pytest.mark.component
 def test_subcritical_recirculation_system():

idaes/models_extra/power_generation/flowsheets/test/test_subcritical_boiler.py

@@ -35,6 +35,9 @@
 solver = get_solver()
 
 
+pytestmark = pytest.mark.usefixtures("run_module_in_tmp_path")
+
+
 @pytest.fixture(scope="module")
 def model():
     return main()

idaes/tests/test_docs.py

@@ -109,6 +109,7 @@ def _have_sphinx():
     return have_sphinx
 
 
+@pytest.mark.usefixtures("run_in_tmp_path")
 @pytest.mark.component
 def test_doctests(docs_path):
     if _have_sphinx():