purge_old_runs() added

src/purge/purge_old_runs.py

@@ -0,0 +1,35 @@
+import logging
+import os
+import shutil
+import datetime
+
+logger = logging.getLogger(__name__)
+
+def purge_old_runs(run_dir: str, max_age_hours: float = 1.0) -> None:
+    """
+    Deletes runs in the specified run_dir older than max_age_hours.
+    """
+    # raise exception if the run_dir is not an absolute path
+    if not os.path.isabs(run_dir):
+        raise ValueError(f"run_dir must be an absolute path: {run_dir}")
+    
+    now = datetime.datetime.now()
+    cutoff = now - datetime.timedelta(hours=max_age_hours)
+
+    for run_id in os.listdir(run_dir):
+        run_path = os.path.join(run_dir, run_id)
+        if not os.path.isdir(run_path):
+            continue  # Skip files
+
+        try:
+            # Get the modification time of the directory
+            mtime = datetime.datetime.fromtimestamp(os.path.getmtime(run_path))
+
+            if mtime < cutoff:
+                logger.debug(f"Deleting old run: {run_id} from {run_dir}")
+                shutil.rmtree(run_path) # Delete the directory and all its contents
+            else:
+                logger.debug(f"Skipping {run_id} in {run_dir}, last modified: {mtime}")
+
+        except Exception as e:
+            logger.error(f"Error processing {run_id} in {run_dir}: {e}")

src/purge/tests/test_purge_old_runs.py

@@ -0,0 +1,85 @@
+import unittest
+import os
+import shutil
+import time
+from src.purge.purge_old_runs import purge_old_runs
+
+class TestPurgeOldRuns(unittest.TestCase):
+    def setUp(self):
+        """
+        Set up test environment before each test.
+        """
+        # Create a temporary directory for the runs
+        self.test_run_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "test_run"))
+        if os.path.exists(self.test_run_dir):
+            shutil.rmtree(self.test_run_dir)
+        os.makedirs(self.test_run_dir, exist_ok=True)
+
+        # Create some dummy run directories with different modification times
+        self.create_dummy_run("run1", hours_old=0.5)
+        self.create_dummy_run("run2", hours_old=1.5)  # Should be purged
+        self.create_dummy_run("run3", hours_old=2) # Should be purged
+        self.create_dummy_run("run4", hours_old=0.25)
+        self.create_dummy_run("run5", hours_old=1)  # Boundary condition, might be purged
+        self.create_dummy_run("run6", hours_old=0)  # Today
+        self.create_dummy_file("not_a_run.txt") # a file that should be left alone
+
+    def tearDown(self):
+        """
+        Clean up test environment after each test.
+        """
+        # Remove the temporary run directory and its contents
+        if os.path.exists(self.test_run_dir):
+            shutil.rmtree(self.test_run_dir)
+
+    def create_dummy_run(self, run_id, hours_old):
+        """
+        Creates a dummy run directory with a specific modification time.
+        """
+        run_path = os.path.join(self.test_run_dir, run_id)
+        os.makedirs(run_path, exist_ok=True)
+
+        # Set the modification time of the directory
+        mtime = time.time() - (hours_old * 3600)  # seconds
+        os.utime(run_path, (mtime, mtime))
+
+    def create_dummy_file(self, filename):
+        """Create a dummy file in the test directory."""
+        filepath = os.path.join(self.test_run_dir, filename)
+        with open(filepath, "w") as f:
+            f.write("This is a dummy file.")
+
+    def test_purge_old_runs(self):
+        """Tests the purge_old_runs function."""
+        max_age_hours = 0.95
+        purge_old_runs(self.test_run_dir, max_age_hours=max_age_hours)  # Pass the directory
+
+        # Check which runs should have been purged
+        runs_to_keep = ["run1", "run4", "run6", "not_a_run.txt"]
+        runs_to_purge = ["run2", "run3", "run5"]
+
+        for run_id in runs_to_keep:
+            run_path = os.path.join(self.test_run_dir, run_id)
+            self.assertTrue(os.path.exists(run_path), f"Run {run_id} should not have been purged.")
+
+        for run_id in runs_to_purge:
+            run_path = os.path.join(self.test_run_dir, run_id)
+            self.assertFalse(os.path.exists(run_path), f"Run {run_id} should have been purged.")
+
+    def test_purge_no_runs(self):
+        """Test when no runs are older than the max_age_hours."""
+
+        # Set all runs to be very recent.
+        for item in os.listdir(self.test_run_dir):
+          item_path = os.path.join(self.test_run_dir, item)
+          mtime = time.time()  # Now
+          os.utime(item_path, (mtime, mtime))
+
+        max_age_hours = 1.0
+        purge_old_runs(self.test_run_dir, max_age_hours=max_age_hours)  # Pass the directory
+
+        # All runs should still exist.
+        expected_runs = ["run1", "run2", "run3", "run4", "run5", "run6", "not_a_run.txt"]
+        for run_id in expected_runs:
+            run_path = os.path.join(self.test_run_dir, run_id)
+            self.assertTrue(os.path.exists(run_path), f"Run {run_id} should not have been purged.")