Adding tests

parsl/tests/test_htex/test_resource_spec_validation.py

@@ -0,0 +1,40 @@
+import queue
+
+import mock
+import pytest
+
+from parsl.executors import HighThroughputExecutor
+from parsl.executors.high_throughput.mpi_prefix_composer import (
+    InvalidResourceSpecification,
+)
+
+
+def double(x):
+    return x * 2
+
+
+@pytest.mark.local
+def test_submit_calls_validate():
+
+    htex = HighThroughputExecutor()
+    htex.outgoing_q = mock.Mock(spec=queue.Queue)
+    htex.validate_resource_spec = mock.Mock(spec=htex.validate_resource_spec)
+
+    res_spec = {}
+    htex.submit(double, res_spec, (5,), {})
+    htex.validate_resource_spec.assert_called()
+
+
+@pytest.mark.local
+def test_resource_spec_validation():
+    htex = HighThroughputExecutor()
+    ret_val = htex.validate_resource_spec({})
+    assert ret_val is None
+
+
+@pytest.mark.local
+def test_resource_spec_validation_bad_keys():
+    htex = HighThroughputExecutor()
+
+    with pytest.raises(InvalidResourceSpecification):
+        htex.validate_resource_spec({"num_nodes": 2})

parsl/tests/test_mpi_apps/test_mpi_mode_enabled.py

@@ -6,26 +6,34 @@
 import pytest
 
 import parsl
-from parsl import bash_app, python_app
+from parsl import Config, bash_app, python_app
+from parsl.executors import MPIExecutor
 from parsl.executors.high_throughput.mpi_prefix_composer import (
     MissingResourceSpecification,
 )
-from parsl.tests.configs.htex_local import fresh_config
+from parsl.launchers import SimpleLauncher
+from parsl.providers import LocalProvider
 
 EXECUTOR_LABEL = "MPI_TEST"
 
 
 def local_setup():
-    config = fresh_config()
-    config.executors[0].label = EXECUTOR_LABEL
-    config.executors[0].max_workers_per_node = 2
-    config.executors[0].enable_mpi_mode = True
-    config.executors[0].mpi_launcher = "mpiexec"
 
     cwd = os.path.abspath(os.path.dirname(__file__))
     pbs_nodefile = os.path.join(cwd, "mocks", "pbs_nodefile")
 
-    config.executors[0].provider.worker_init = f"export PBS_NODEFILE={pbs_nodefile}"
+    config = Config(
+        executors=[
+            MPIExecutor(
+                label=EXECUTOR_LABEL,
+                max_workers_per_block=2,
+                mpi_launcher="mpiexec",
+                provider=LocalProvider(
+                    worker_init=f"export PBS_NODEFILE={pbs_nodefile}",
+                    launcher=SimpleLauncher()
+                )
+            )
+        ])
 
     parsl.load(config)
 

parsl/tests/test_mpi_apps/test_resource_spec.py

@@ -1,14 +1,18 @@
 import contextlib
 import logging
 import os
+import queue
 import typing
 import unittest
 from typing import Dict
 
+import mock
 import pytest
 
 import parsl
 from parsl.app.app import python_app
+from parsl.executors.high_throughput.executor import HighThroughputExecutor
+from parsl.executors.high_throughput.mpi_executor import MPIExecutor
 from parsl.executors.high_throughput.mpi_prefix_composer import (
     InvalidResourceSpecification,
     MissingResourceSpecification,
@@ -20,6 +24,8 @@
     get_slurm_hosts_list,
     identify_scheduler,
 )
+from parsl.launchers import SimpleLauncher
+from parsl.providers import LocalProvider
 from parsl.tests.configs.htex_local import fresh_config
 
 EXECUTOR_LABEL = "MPI_TEST"
@@ -48,23 +54,6 @@ def get_env_vars(parsl_resource_specification: Dict = {}) -> Dict:
     return parsl_vars
 
 
-@pytest.mark.local
-def test_resource_spec_env_vars():
-    resource_spec = {
-        "num_nodes": 4,
-        "ranks_per_node": 2,
-    }
-
-    assert double(5).result() == 10
-
-    future = get_env_vars(parsl_resource_specification=resource_spec)
-
-    result = future.result()
-    assert isinstance(result, Dict)
-    assert result["PARSL_NUM_NODES"] == str(resource_spec["num_nodes"])
-    assert result["PARSL_RANKS_PER_NODE"] == str(resource_spec["ranks_per_node"])
-
-
 @pytest.mark.local
 @unittest.mock.patch("subprocess.check_output", return_value=b"c203-031\nc203-032\n")
 def test_slurm_mocked_mpi_fetch(subprocess_check):
@@ -122,22 +111,43 @@ def test_top_level():
 
 @pytest.mark.local
 @pytest.mark.parametrize(
-    "resource_spec, is_mpi_enabled, exception",
+    "resource_spec, exception",
     (
-        ({"num_nodes": 2, "ranks_per_node": 1}, False, None),
-        ({"launcher_options": "--debug_foo"}, False, None),
-        ({"num_nodes": 2, "BAD_OPT": 1}, False, InvalidResourceSpecification),
-        ({}, False, None),
-        ({"num_nodes": 2, "ranks_per_node": 1}, True, None),
-        ({"launcher_options": "--debug_foo"}, True, None),
-        ({"num_nodes": 2, "BAD_OPT": 1}, True, InvalidResourceSpecification),
-        ({}, True, MissingResourceSpecification),
+
+        ({"num_nodes": 2, "ranks_per_node": 1}, None),
+        ({"launcher_options": "--debug_foo"}, None),
+        ({"num_nodes": 2, "BAD_OPT": 1}, InvalidResourceSpecification),
+        ({}, MissingResourceSpecification),
     )
 )
-def test_resource_spec(resource_spec: Dict, is_mpi_enabled: bool, exception):
+def test_mpi_resource_spec(resource_spec: Dict, exception):
+    """Test validation of resource_specification in MPIExecutor"""
+
+    mpi_ex = MPIExecutor(provider=LocalProvider(launcher=SimpleLauncher()))
+    mpi_ex.outgoing_q = mock.Mock(spec=queue.Queue)
+
     if exception:
         with pytest.raises(exception):
-            validate_resource_spec(resource_spec, is_mpi_enabled)
+            mpi_ex.validate_resource_spec(resource_spec)
     else:
-        result = validate_resource_spec(resource_spec, is_mpi_enabled)
+        result = mpi_ex.validate_resource_spec(resource_spec)
         assert result is None
+
+
+@pytest.mark.local
+@pytest.mark.parametrize(
+    "resource_spec",
+    (
+        {"num_nodes": 2, "ranks_per_node": 1},
+        {"launcher_options": "--debug_foo"},
+        {"BAD_OPT": 1},
+    )
+)
+def test_mpi_resource_spec_passed_to_htex(resource_spec: dict):
+    """HTEX should reject every resource_spec"""
+
+    htex = HighThroughputExecutor()
+    htex.outgoing_q = mock.Mock(spec=queue.Queue)
+
+    with pytest.raises(InvalidResourceSpecification):
+        htex.validate_resource_spec(resource_spec)