Retain ROCS scores after temporary cleanup errors

drugex/training/scorers/rocs_cdpkit.py

@@ -399,7 +399,8 @@ def getScores(self, mols, frags=None) -> np.ndarray:
         if not unique_smiles:
             return np.zeros((len(mols), num_groups))
 
-        with tempfile.TemporaryDirectory() as tmpdir:
+        # Cleanup failures on network filesystems must not discard completed scores.
+        with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
             conf_file = self.conformer_generator.genConformers(unique_smiles, tmpdir)
             if not os.path.exists(conf_file):
                 if self.show_progress:

drugex/training/scorers/rocs_rdkit.py

@@ -414,7 +414,8 @@ def getScores(
         if not unique_smiles:
             return scores
 
-        with tempfile.TemporaryDirectory() as tmpdir:
+        # Cleanup failures on network filesystems must not discard completed scores.
+        with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
             conf_file = self.conformer_generator.genConformers(unique_smiles, tmpdir)
             if not os.path.exists(conf_file):
                 if self.show_progress:

drugex/training/scorers/runtime_tests.py

@@ -0,0 +1,70 @@
+"""Regression tests for scorer temporary-directory behavior."""
+
+import unittest
+from unittest.mock import MagicMock, patch
+
+import numpy as np
+
+from drugex.training.scorers import rocs_cdpkit, rocs_rdkit
+from drugex.training.scorers.rocs_cdpkit import CDPKitROCSScorer
+from drugex.training.scorers.rocs_rdkit import RDKitROCSScorer
+
+
+class _MissingConformerGenerator:
+    """Return a missing file after the scorer enters its temporary directory."""
+
+    def genConformers(self, smiles, output_dir):
+        return f"{output_dir}/missing.sdf"
+
+
+def _temporary_directory_mock():
+    manager = MagicMock()
+    manager.__enter__.return_value = "/tmp/drugex-scorer-test"
+    manager.__exit__.return_value = False
+    return manager
+
+
+class TemporaryDirectoryTests(unittest.TestCase):
+    """Verify both open-source backends request cleanup-error tolerance."""
+
+    def test_rdkit_ignores_cleanup_errors(self):
+        scorer = object.__new__(RDKitROCSScorer)
+        scorer.group_to_indices = [[0]]
+        scorer.show_progress = False
+        scorer.conformer_generator = _MissingConformerGenerator()
+        manager = _temporary_directory_mock()
+
+        with patch.object(
+            rocs_rdkit.tempfile,
+            "TemporaryDirectory",
+            return_value=manager,
+        ) as temporary_directory:
+            scores = scorer.getScores(["CCO"])
+
+        temporary_directory.assert_called_once_with(ignore_cleanup_errors=True)
+        np.testing.assert_array_equal(scores, np.zeros((1, 1)))
+
+    @unittest.skipUnless(
+        rocs_cdpkit.CDPL_AVAILABLE,
+        "CDPKit is not available",
+    )
+    def test_cdpkit_ignores_cleanup_errors(self):
+        scorer = object.__new__(CDPKitROCSScorer)
+        scorer.group_to_indices = [[0]]
+        scorer.show_progress = False
+        scorer.conformer_generator = _MissingConformerGenerator()
+        manager = _temporary_directory_mock()
+
+        with patch.object(
+            rocs_cdpkit.tempfile,
+            "TemporaryDirectory",
+            return_value=manager,
+        ) as temporary_directory:
+            scores = scorer.getScores(["CCO"])
+
+        temporary_directory.assert_called_once_with(ignore_cleanup_errors=True)
+        np.testing.assert_array_equal(scores, np.zeros((1, 1)))
+
+
+if __name__ == "__main__":
+    unittest.main()