Adding shared_fs and staging_required tags to tests

parsl/executors/globus_compute.py

@@ -2,8 +2,6 @@
 from concurrent.futures import Future
 from typing import Any, Callable, Dict, Optional, Union
 
-import typeguard
-
 from parsl.errors import OptionalModuleMissing
 from parsl.executors.base import ParslExecutor
 from parsl.utils import RepresentationMixin
@@ -33,7 +31,6 @@ def __init__(
             **kwargs,
         ):
         """
-
         Parameters
         ----------
 
@@ -125,6 +122,7 @@ def submit(self, func: Callable, resource_specification: Dict[str, Any], *args:
         self._executor.resource_specification = resource_specification or self.resource_specification
         return self._executor.submit(func, *args, **kwargs)
 
+
     def shutdown(self, wait=True, *, cancel_futures=False):
         """Clean-up the resources associated with the Executor.
 

parsl/tests/configs/globus_compute.py

@@ -11,6 +11,7 @@ def fresh_config():
     return Config(
         executors=[
             GlobusComputeExecutor(
+                label="globus_compute",
                 endpoint_id=endpoint_id
             )
         ]

parsl/tests/conftest.py

@@ -167,6 +167,11 @@ def pytest_configure(config):
         'markers',
         'globus_compute: Marks tests that require a valid globus_compute target'
     )
+    config.addinivalue_line(
+        'markers',
+        'shared_fs: Marks tests that require a shared_fs between the workers are the test client'
+    )
+
 
 
 @pytest.fixture(autouse=True, scope='session')

parsl/tests/test_bash_apps/test_basic.py

@@ -24,6 +24,7 @@ def foo(x, y, z=10, stdout=None, label=None):
     return f"echo {x} {y} {z}"
 
 
+@pytest.mark.shared_fs
 def test_command_format_1(tmpd_cwd):
     """Testing command format for BashApps"""
 
@@ -38,6 +39,7 @@ def test_command_format_1(tmpd_cwd):
     assert so_content == "1 4 10"
 
 
+@pytest.mark.shared_fs
 def test_auto_log_filename_format(caplog):
     """Testing auto log filename format for BashApps
     """
@@ -65,7 +67,7 @@ def test_auto_log_filename_format(caplog):
     for record in caplog.records:
         assert record.levelno < logging.ERROR
 
-
+@pytest.mark.shared_fs
 def test_parallel_for(tmpd_cwd, n=3):
     """Testing a simple parallel for loop"""
     outdir = tmpd_cwd / "outputs/test_parallel"

parsl/tests/test_bash_apps/test_error_codes.py

@@ -58,6 +58,7 @@ def bad_format(stderr='std.err', stdout='std.out'):
 whitelist = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'configs', '*threads*')
 
 
+@pytest.mark.shared_fs
 def test_div_0(test_fn=div_0):
     err_code = test_matrix[test_fn]['exit_code']
     f = test_fn()
@@ -73,6 +74,7 @@ def test_div_0(test_fn=div_0):
     os.remove('std.out')
 
 
+@pytest.mark.shared_fs
 def test_bash_misuse(test_fn=bash_misuse):
     err_code = test_matrix[test_fn]['exit_code']
     f = test_fn()
@@ -87,6 +89,7 @@ def test_bash_misuse(test_fn=bash_misuse):
     os.remove('std.out')
 
 
+@pytest.mark.shared_fs
 def test_command_not_found(test_fn=command_not_found):
     err_code = test_matrix[test_fn]['exit_code']
     f = test_fn()
@@ -103,6 +106,7 @@ def test_command_not_found(test_fn=command_not_found):
     return True
 
 
+@pytest.mark.shared_fs
 def test_not_executable(test_fn=not_executable):
     err_code = test_matrix[test_fn]['exit_code']
     f = test_fn()

parsl/tests/test_bash_apps/test_kwarg_storage.py

@@ -8,6 +8,7 @@ def foo(z=2, stdout=None):
     return f"echo {z}"
 
 
+@pytest.mark.shared_fs
 def test_command_format_1(tmpd_cwd):
     """Testing command format for BashApps
     """

parsl/tests/test_bash_apps/test_memoize.py

@@ -9,9 +9,7 @@ def fail_on_presence(outputs=()):
     return 'if [ -f {0} ] ; then exit 1 ; else touch {0}; fi'.format(outputs[0])
 
 
-# This test is an oddity that requires a shared-FS and simply
-# won't work if there's a staging provider.
-# @pytest.mark.sharedFS_required
+@pytest.mark.shared_fs
 def test_bash_memoization(tmpd_cwd, n=2):
     """Testing bash memoization
     """
@@ -29,9 +27,7 @@ def fail_on_presence_kw(outputs=(), foo=None):
     return 'if [ -f {0} ] ; then exit 1 ; else touch {0}; fi'.format(outputs[0])
 
 
-# This test is an oddity that requires a shared-FS and simply
-# won't work if there's a staging provider.
-# @pytest.mark.sharedFS_required
+@pytest.mark.shared_fs
 def test_bash_memoization_keywords(tmpd_cwd, n=2):
     """Testing bash memoization
     """

parsl/tests/test_bash_apps/test_memoize_ignore_args.py

@@ -1,5 +1,7 @@
 import os
 
+import pytest
+
 import parsl
 from parsl.app.app import bash_app
 
@@ -21,6 +23,7 @@ def no_checkpoint_stdout_app_ignore_args(stdout=None):
     return "echo X"
 
 
+@pytest.mark.shared_fs
 def test_memo_stdout(tmpd_cwd):
     path_x = tmpd_cwd / "test.memo.stdout.x"
 

parsl/tests/test_bash_apps/test_memoize_ignore_args_regr.py

@@ -29,6 +29,7 @@ def no_checkpoint_stdout_app(stdout=None):
     return "echo X"
 
 
+@pytest.mark.shared_fs
 def test_memo_stdout(tmpd_cwd):
     assert const_list_x == const_list_x_arg
 

parsl/tests/test_bash_apps/test_multiline.py

@@ -14,6 +14,7 @@ def multiline(inputs=(), outputs=(), stderr=None, stdout=None):
     """.format(inputs=inputs, outputs=outputs)
 
 
+@pytest.mark.shared_fs
 def test_multiline(tmpd_cwd):
     so, se = tmpd_cwd / "std.out", tmpd_cwd / "std.err"
     f = multiline(

parsl/tests/test_bash_apps/test_stdout.py

@@ -34,6 +34,7 @@ def echo_to_streams(msg, stderr=None, stdout=None):
 ]
 
 
+@pytest.mark.shared_fs
 @pytest.mark.parametrize('spec', speclist, ids=testids)
 def test_bad_stdout_specs(spec):
     """Testing bad stdout spec cases"""
@@ -91,6 +92,7 @@ def test_bad_stderr_file():
 
 
 @pytest.mark.executor_supports_std_stream_tuples
+@pytest.mark.shared_fs
 def test_stdout_truncate(tmpd_cwd, caplog):
     """Testing truncation of prior content of stdout"""
 
@@ -110,6 +112,7 @@ def test_stdout_truncate(tmpd_cwd, caplog):
         assert record.levelno < logging.ERROR
 
 
+@pytest.mark.shared_fs
 def test_stdout_append(tmpd_cwd, caplog):
     """Testing appending to prior content of stdout (default open() mode)"""
 

parsl/tests/test_docs/test_from_slides.py

@@ -1,5 +1,7 @@
 import os
 
+import pytest
+
 from parsl.app.app import bash_app, python_app
 from parsl.data_provider.files import File
 
@@ -15,6 +17,7 @@ def cat(inputs=[]):
         return f.readlines()
 
 
+@pytest.mark.staging_required
 def test_slides():
     """Testing code snippet from slides """
 

parsl/tests/test_docs/test_kwargs.py

@@ -1,6 +1,8 @@
 """Functions used to explain kwargs"""
 from pathlib import Path
 
+import pytest
+
 from parsl import File, python_app
 
 
@@ -19,6 +21,7 @@ def reduce_app(inputs=()):
     assert reduce_future.result() == 6
 
 
+@pytest.mark.shared_fs
 def test_outputs(tmpd_cwd):
     @python_app()
     def write_app(message, outputs=()):

parsl/tests/test_docs/test_workflow1.py

@@ -22,6 +22,7 @@ def save(message, outputs=[]):
     return 'echo {m} &> {o}'.format(m=message, o=outputs[0])
 
 
+@pytest.mark.shared_fs
 @pytest.mark.staging_required
 def test_procedural(N=2):
     """Procedural workflow example from docs on

parsl/tests/test_python_apps/test_outputs.py

@@ -16,6 +16,7 @@ def double(x, outputs=[]):
 whitelist = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'configs', '*threads*')
 
 
+@pytest.mark.shared_fs
 def test_launch_apps(tmpd_cwd, n=2):
     outdir = tmpd_cwd / "outputs"
     outdir.mkdir()

parsl/tests/test_regression/test_226.py

@@ -53,6 +53,7 @@ def test_get_dataframe():
     assert res.equals(data), 'Unexpected dataframe'
 
 
+@pytest.mark.shared_fs
 def test_bash_default_arg():
     if os.path.exists('std.out'):
         os.remove('std.out')

parsl/tests/test_staging/test_docs_1.py

@@ -12,6 +12,7 @@ def convert(inputs=[], outputs=[]):
 
 
 @pytest.mark.cleannet
+@pytest.mark.shared_fs
 def test():
     # create an remote Parsl file
     inp = File('ftp://ftp.iana.org/pub/mirror/rirstats/arin/ARIN-STATS-FORMAT-CHANGE.txt')

parsl/tests/test_staging/test_output_chain_filenames.py

@@ -1,5 +1,7 @@
 from concurrent.futures import Future
 
+import pytest
+
 from parsl import File
 from parsl.app.app import bash_app
 
@@ -14,6 +16,7 @@ def app2(inputs=(), outputs=(), stdout=None, stderr=None, mock=False):
     return f"echo '{inputs[0]}' > {outputs[0]}"
 
 
+@pytest.mark.shared_fs
 def test_behavior(tmpd_cwd):
     expected_path = str(tmpd_cwd / "simple-out.txt")
     app1_future = app1(

parsl/tests/test_staging/test_staging_ftp.py

@@ -15,6 +15,7 @@ def sort_strings(inputs=[], outputs=[]):
 
 
 @pytest.mark.cleannet
+@pytest.mark.staging_required
 def test_staging_ftp():
     """Test staging for an ftp file
 

parsl/tests/test_staging/test_staging_https.py

@@ -48,6 +48,7 @@ def sort_strings_additional_executor(inputs=(), outputs=()):
 
 
 @pytest.mark.cleannet
+@pytest.mark.staging_required
 def test_staging_https_cleannet(tmpd_cwd):
     unsorted_file = File(_unsorted_url)
     sorted_file = File(tmpd_cwd / 'sorted.txt')
@@ -68,6 +69,7 @@ def test_staging_https_local(tmpd_cwd):
 
 
 @pytest.mark.cleannet
+@pytest.mark.staging_required
 def test_staging_https_kwargs(tmpd_cwd):
     unsorted_file = File(_unsorted_url)
     sorted_file = File(tmpd_cwd / 'sorted.txt')
@@ -78,6 +80,7 @@ def test_staging_https_kwargs(tmpd_cwd):
 
 
 @pytest.mark.cleannet
+@pytest.mark.staging_required
 def test_staging_https_args(tmpd_cwd):
     unsorted_file = File(_unsorted_url)
     sorted_file = File(tmpd_cwd / 'sorted.txt')

parsl/tests/test_staging/test_staging_stdout.py

@@ -15,6 +15,7 @@ def output_to_stds(*, stdout=parsl.AUTO_LOGNAME, stderr=parsl.AUTO_LOGNAME):
     return "echo hello ; echo goodbye >&2"
 
 
+@pytest.mark.shared_fs
 def test_stdout_staging_file(tmpd_cwd, caplog):
     basename = str(tmpd_cwd) + "/stdout.txt"
     stdout_file = File("file://" + basename)
@@ -30,6 +31,7 @@ def test_stdout_staging_file(tmpd_cwd, caplog):
         assert record.levelno < logging.ERROR
 
 
+@pytest.mark.shared_fs
 def test_stdout_stderr_staging_zip(tmpd_cwd, caplog):
     zipfile_name = str(tmpd_cwd) + "/staging.zip"
     stdout_relative_path = "somewhere/test-out.txt"