Removing old projects during tests

src/simstadt/__init__.py

@@ -9,10 +9,10 @@ from .results import (
     create_simstadt_results,
 )
 from .runner import run_workflow_with_citygml
+from .utils import clean_old_workflows
 from .workflows import greenwater_simulation, heatdemand_simulation, photovoltaic_simulation
 
 # TODO: Define Template class, with already defined Workflows?
-# TODO: Define clean_old_files, and use it on Testrepo and TempRepo
 
 __all__ = [
     "KPI",
@@ -23,6 +23,7 @@ __all__ = [
     "GreenWaterResults",
     "SolarPotentialResults",
     "create_simstadt_results",
+    "clean_old_workflows",
     "run_workflow_with_citygml",
     "photovoltaic_simulation",
     "heatdemand_simulation",

src/simstadt/utils.py

@@ -3,12 +3,15 @@ import logging
 import os
 import random
 import re
+import shutil
 import string
 from contextlib import contextmanager
 from pathlib import Path
 
 import numpy as np
 
+WORKFLOW_FOLDER_PATTERN = re.compile(r"(20\d{6}_\d{4})_[a-z0-9]{4}")
+
 log = logging.getLogger(__name__)
 
 
@@ -34,6 +37,19 @@ def remove_random_id(text: str) -> str:
     return re.sub(r"^(20\d{6}_\d{4}_)?[a-z0-9]{4}_", "", text)
 
 
+def clean_old_workflows(folder: Path, pattern: str = "20*.flow", hours: int = 1) -> None:
+    """Remove timestamped workflow folders older than `hours` from `folder`.
+
+    Only deletes paths whose names match the random_id timestamp pattern.
+    """
+    cutoff = (datetime.datetime.now() - datetime.timedelta(hours=hours)).strftime("%Y%m%d_%H%M_")
+    for path in sorted(folder.glob(pattern)):
+        if m := WORKFLOW_FOLDER_PATTERN.match(path.name):
+            if m.group(1) < cutoff:
+                log.info("Removing old workflow: %s", path.name)
+                shutil.rmtree(path) if path.is_dir() else path.unlink()
+
+
 @contextmanager
 def chdir(path: Path):
     """Context manager that temporarily changes the working directory."""

tests/conftest.py

@@ -8,6 +8,18 @@ TEST_REPOSITORY = Path(__file__).parent / "data" / "TestRepo"
 TEST_PROJECT_PATH = TEST_REPOSITORY / "CGSC.proj"
 
 
+@pytest.fixture(scope="session", autouse=True)
+def clean_old_test_workflows():
+    """Remove stale workflow folders from test and temp repositories before the session."""
+    from simstadt.runner import SIMSTADT_TEMP_REPO
+    from simstadt.utils import clean_old_workflows
+
+    for proj in TEST_REPOSITORY.glob("*.proj"):
+        clean_old_workflows(proj)
+    for proj in SIMSTADT_TEMP_REPO.glob("*.proj"):
+        clean_old_workflows(proj)
+
+
 @pytest.fixture(scope="session")
 def test_repository() -> Path:
     return TEST_REPOSITORY