Fixing singularity unit tests.

Previous implementation used the mock module to mock io.open calls to test mlcubes. This worked for all calls to io.open. New implementation invokes subprocess.check_output that seems to be using io.open internally that did not work with existing implementation. New mocking strategy is to mock io.open call only for specific file paths (e.g., paths to mlcube configuration files).

runners/mlcube_singularity/mlcube_singularity/tests/test_singularity_runner.py

@@ -14,6 +14,7 @@
 import tempfile
 import typing as t
 import unittest
+from io import open
 from pathlib import Path
 from unittest import TestCase
 from unittest.mock import mock_open, patch
@@ -26,6 +27,7 @@
 from mlcube.errors import ExecutionError
 from mlcube.shell import Shell
 
+
 try:
  client = Client.from_env()
  client.init()
@@ -59,6 +61,33 @@
 )
 
 
+_MLCUBE_CONFIG_FILE = "/some/path/to/mlcube.yaml"
+"""We will be mocking io.open call only for this file path."""
+
+_io_open = open
+"""Original function we will be mocking in these unit tests."""
+
+
+def get_custom_mock_open(file_path_to_mock: str, read_data: str) -> t.Callable:
+ """Function to help mock `io.open` only for this specific file path.
+
+ Original implementation:
+ https://stackoverflow.com/questions/67234524/python-unittest-mock-open-specific-paths-dont-mock-others
+
+ Args:
+ file_path_to_mock: Only mock the `io.open` call for this path. For others, call original `io.open`.
+ read_data: Content for the `file_path_to_mock` file.
+
+ """
+ def mocked_open() -> t.Callable:
+ def conditional_open_fn(path, *args, **kwargs):
+ if path == file_path_to_mock:
+ return mock_open(read_data=read_data)()
+ return _io_open(path, *args, **kwargs)
+ return conditional_open_fn
+ return mocked_open
+
+
 _sync_workspace_fn: t.Optional[t.Callable] = None
 
 
@@ -102,9 +131,9 @@ def tearDownClass(cls) -> None:
 
  @unittest.skipUnless(client is not None, reason="No singularity available.")
  def test_mlcube_default_entrypoints(self):
- with patch("io.open", mock_open(read_data=_MLCUBE_DEFAULT_ENTRY_POINT)):
+ with patch("io.open", new_callable=get_custom_mock_open(_MLCUBE_CONFIG_FILE, _MLCUBE_DEFAULT_ENTRY_POINT)):
  mlcube: DictConfig = MLCubeConfig.create_mlcube_config(
- "/some/path/to/mlcube.yaml",
+ _MLCUBE_CONFIG_FILE,
  runner_config=Config.DEFAULT,
  runner_cls=SingularityRun,
  )
@@ -122,9 +151,9 @@ def test_mlcube_default_entrypoints(self):
 
  @unittest.skipUnless(client is not None, reason="No singularity available.")
  def test_mlcube_custom_entrypoints(self):
- with patch("io.open", mock_open(read_data=_MLCUBE_CUSTOM_ENTRY_POINTS)):
+ with patch("io.open", new_callable=get_custom_mock_open(_MLCUBE_CONFIG_FILE, _MLCUBE_CUSTOM_ENTRY_POINTS)):
  mlcube: DictConfig = MLCubeConfig.create_mlcube_config(
- "/some/path/to/mlcube.yaml",
+ _MLCUBE_CONFIG_FILE,
  runner_config=Config.DEFAULT,
  runner_cls=SingularityRun,
  )