tests: no manual os.chdir; use util.working_directory context manager

Author: rupertnash

tests/test_examples.py

@@ -34,6 +34,7 @@
     get_windows_safe_factory,
     needs_docker,
     windows_needs_docker,
+    working_directory,
 )
 
 sys.argv = [""]
@@ -1068,16 +1069,16 @@ def test_no_js_console(factor: str) -> None:
 def test_cid_file_dir(tmp_path: Path, factor: str) -> None:
     """Test --cidfile-dir option works."""
     test_file = "cache_test_workflow.cwl"
-    cwd = Path.cwd()
-    os.chdir(tmp_path)
-    commands = factor.split()
-    commands.extend(["--cidfile-dir", str(tmp_path), get_data("tests/wf/" + test_file)])
-    error_code, stdout, stderr = get_main_output(commands)
-    assert "completed success" in stderr
-    assert error_code == 0
-    cidfiles_count = sum(1 for _ in tmp_path.glob("**/*"))
-    assert cidfiles_count == 2
-    os.chdir(cwd)
+    with working_directory(tmp_path):
+        commands = factor.split()
+        commands.extend(
+            ["--cidfile-dir", str(tmp_path), get_data("tests/wf/" + test_file)]
+        )
+        error_code, stdout, stderr = get_main_output(commands)
+        assert "completed success" in stderr
+        assert error_code == 0
+        cidfiles_count = sum(1 for _ in tmp_path.glob("**/*"))
+        assert cidfiles_count == 2
 
 
 @needs_docker
@@ -1121,22 +1122,20 @@ def test_cid_file_non_existing_dir(tmp_path: Path, factor: str) -> None:
 def test_cid_file_w_prefix(tmp_path: Path, factor: str) -> None:
     """Test that --cidfile-prefix works."""
     test_file = "cache_test_workflow.cwl"
-    cwd = Path.cwd()
-    os.chdir(tmp_path)
-    try:
-        commands = factor.split()
-        commands.extend(
-            [
-                "--record-container-id",
-                "--cidfile-prefix=pytestcid",
-                get_data("tests/wf/" + test_file),
-            ]
-        )
-        error_code, stdout, stderr = get_main_output(commands)
-    finally:
-        listing = tmp_path.iterdir()
-        os.chdir(cwd)
-        cidfiles_count = sum(1 for _ in tmp_path.glob("**/pytestcid*"))
+    with working_directory(tmp_path):
+        try:
+            commands = factor.split()
+            commands.extend(
+                [
+                    "--record-container-id",
+                    "--cidfile-prefix=pytestcid",
+                    get_data("tests/wf/" + test_file),
+                ]
+            )
+            error_code, stdout, stderr = get_main_output(commands)
+        finally:
+            listing = tmp_path.iterdir()
+            cidfiles_count = sum(1 for _ in tmp_path.glob("**/pytestcid*"))
     assert "completed success" in stderr
     assert error_code == 0
     assert cidfiles_count == 2, "{}/n{}".format(list(listing), stderr)
@@ -1397,20 +1396,20 @@ def test_v1_0_arg_empty_prefix_separate_false() -> None:
 def test_scatter_output_filenames(tmp_path: Path) -> None:
     """If a scatter step produces identically named output then confirm that the final output is renamed correctly."""
     cwd = Path.cwd()
-    os.chdir(tmp_path)
-    rtc = RuntimeContext()
-    rtc.outdir = str(cwd)
-    factory = cwltool.factory.Factory(runtime_context=rtc)
-    output_names = ["output.txt", "output.txt_2", "output.txt_3"]
-    scatter_workflow = factory.make(get_data("tests/scatter_numbers.cwl"))
-    result = scatter_workflow(range=3)
-    assert isinstance(result, dict)
-    assert "output" in result
-
-    locations = sorted([element["location"] for element in result["output"]])
+    with working_directory(tmp_path):
+        rtc = RuntimeContext()
+        rtc.outdir = str(cwd)
+        factory = cwltool.factory.Factory(runtime_context=rtc)
+        output_names = ["output.txt", "output.txt_2", "output.txt_3"]
+        scatter_workflow = factory.make(get_data("tests/scatter_numbers.cwl"))
+        result = scatter_workflow(range=3)
+        assert isinstance(result, dict)
+        assert "output" in result
+
+        locations = sorted([element["location"] for element in result["output"]])
 
-    assert (
-        locations[0].endswith("output.txt")
-        and locations[1].endswith("output.txt_2")
-        and locations[2].endswith("output.txt_3")
-    ), f"Locations {locations} do not end with {output_names}"
+        assert (
+            locations[0].endswith("output.txt")
+            and locations[1].endswith("output.txt_2")
+            and locations[2].endswith("output.txt_3")
+        ), f"Locations {locations} do not end with {output_names}"

tests/test_singularity.py

@@ -21,25 +21,25 @@
 def test_singularity_pullfolder(tmp_path: Path) -> None:
     workdir = tmp_path / "working_dir_new"
     workdir.mkdir()
-    os.chdir(workdir)
-    pullfolder = tmp_path / "pullfolder"
-    pullfolder.mkdir()
-    env = os.environ.copy()
-    env["SINGULARITY_PULLFOLDER"] = str(pullfolder)
-    result_code, stdout, stderr = get_main_output(
-        [
-            "--singularity",
-            get_data("tests/sing_pullfolder_test.cwl"),
-            "--message",
-            "hello",
-        ],
-        env=env,
-    )
-    print(stdout)
-    print(stderr)
-    assert result_code == 0
-    image = pullfolder / "debian.img"
-    assert image.exists()
+    with working_directory(workdir):
+        pullfolder = tmp_path / "pullfolder"
+        pullfolder.mkdir()
+        env = os.environ.copy()
+        env["SINGULARITY_PULLFOLDER"] = str(pullfolder)
+        result_code, stdout, stderr = get_main_output(
+            [
+                "--singularity",
+                get_data("tests/sing_pullfolder_test.cwl"),
+                "--message",
+                "hello",
+            ],
+            env=env,
+        )
+        print(stdout)
+        print(stderr)
+        assert result_code == 0
+        image = pullfolder / "debian.img"
+        assert image.exists()
 
 
 @needs_singularity
@@ -97,32 +97,37 @@ def test_singularity_incorrect_image_pull() -> None:
 def test_singularity_local(tmp_path: Path) -> None:
     workdir = tmp_path / "working_dir"
     workdir.mkdir()
-    os.chdir(workdir)
-    result_code, stdout, stderr = get_main_output(
-        [
-            "--singularity",
-            get_data("tests/sing_pullfolder_test.cwl"),
-            "--message",
-            "hello",
-        ]
-    )
-    assert result_code == 0
+    with working_directory(workdir):
+        result_code, stdout, stderr = get_main_output(
+            [
+                "--singularity",
+                get_data("tests/sing_pullfolder_test.cwl"),
+                "--message",
+                "hello",
+            ]
+        )
+        assert result_code == 0
 
 
 @needs_singularity_2_6
 def test_singularity_docker_image_id_in_tool(tmp_path: Path) -> None:
     workdir = tmp_path / "working_dir"
     workdir.mkdir()
-    os.chdir(workdir)
-    result_code, stdout, stderr = get_main_output(
-        [
-            "--singularity",
-            get_data("tests/sing_pullfolder_test.cwl"),
-            "--message",
-            "hello",
-        ]
-    )
-    result_code1, stdout, stderr = get_main_output(
-        ["--singularity", get_data("tests/debian_image_id.cwl"), "--message", "hello"]
-    )
-    assert result_code1 == 0
+    with working_directory(workdir):
+        result_code, stdout, stderr = get_main_output(
+            [
+                "--singularity",
+                get_data("tests/sing_pullfolder_test.cwl"),
+                "--message",
+                "hello",
+            ]
+        )
+        result_code1, stdout, stderr = get_main_output(
+            [
+                "--singularity",
+                get_data("tests/debian_image_id.cwl"),
+                "--message",
+                "hello",
+            ]
+        )
+        assert result_code1 == 0

tests/test_udocker.py

@@ -8,7 +8,7 @@
 import pytest
 from _pytest.tmpdir import TempPathFactory
 
-from .util import get_data, get_main_output
+from .util import get_data, get_main_output, working_directory
 
 LINUX = sys.platform in ("linux", "linux2")
 
@@ -19,59 +19,57 @@ def udocker(tmp_path_factory: TempPathFactory) -> str:
     test_cwd = os.getcwd()
     test_environ = copy.copy(os.environ)
     docker_install_dir = str(tmp_path_factory.mktemp("udocker"))
-    os.chdir(docker_install_dir)
-
-    url = "https://raw.githubusercontent.com/jorge-lip/udocker-builds/master/tarballs/udocker-1.1.4.tar.gz"
-    install_cmds = [
-        ["curl", url, "-o", "./udocker-tarball.tgz"],
-        ["tar", "xzvf", "udocker-tarball.tgz", "udocker"],
-        [
-            "bash",
-            "-c",
-            "UDOCKER_TARBALL={}/udocker-tarball.tgz ./udocker install".format(
-                docker_install_dir
-            ),
-        ],
-    ]
-
-    test_environ["UDOCKER_DIR"] = os.path.join(docker_install_dir, ".udocker")
-    test_environ["HOME"] = docker_install_dir
-
-    results = []
-    for _ in range(3):
-        results = [subprocess.call(cmds, env=test_environ) for cmds in install_cmds]
-        if sum(results) == 0:
-            break
-        subprocess.call(["rm", "./udocker"])
-
-    assert sum(results) == 0
-
-    udocker_path = os.path.join(docker_install_dir, "udocker")
-    os.chdir(test_cwd)
+    with working_directory(docker_install_dir):
+
+        url = "https://raw.githubusercontent.com/jorge-lip/udocker-builds/master/tarballs/udocker-1.1.4.tar.gz"
+        install_cmds = [
+            ["curl", url, "-o", "./udocker-tarball.tgz"],
+            ["tar", "xzvf", "udocker-tarball.tgz", "udocker"],
+            [
+                "bash",
+                "-c",
+                "UDOCKER_TARBALL={}/udocker-tarball.tgz ./udocker install".format(
+                    docker_install_dir
+                ),
+            ],
+        ]
+
+        test_environ["UDOCKER_DIR"] = os.path.join(docker_install_dir, ".udocker")
+        test_environ["HOME"] = docker_install_dir
+
+        results = []
+        for _ in range(3):
+            results = [subprocess.call(cmds, env=test_environ) for cmds in install_cmds]
+            if sum(results) == 0:
+                break
+            subprocess.call(["rm", "./udocker"])
+
+        assert sum(results) == 0
+
+        udocker_path = os.path.join(docker_install_dir, "udocker")
+
     return udocker_path
 
 
 @pytest.mark.skipif(not LINUX, reason="LINUX only")
 def test_udocker_usage_should_not_write_cid_file(udocker: str, tmp_path: Path) -> None:
     """Confirm that no cidfile is made when udocker is used."""
-    cwd = Path.cwd()
-    os.chdir(tmp_path)
-
-    test_file = "tests/wf/wc-tool.cwl"
-    job_file = "tests/wf/wc-job.json"
-    error_code, stdout, stderr = get_main_output(
-        [
-            "--debug",
-            "--default-container",
-            "debian",
-            "--user-space-docker-cmd=" + udocker,
-            get_data(test_file),
-            get_data(job_file),
-        ]
-    )
 
-    cidfiles_count = sum(1 for _ in tmp_path.glob("*.cid"))
-    os.chdir(cwd)
+    with working_directory(tmp_path):
+        test_file = "tests/wf/wc-tool.cwl"
+        job_file = "tests/wf/wc-job.json"
+        error_code, stdout, stderr = get_main_output(
+            [
+                "--debug",
+                "--default-container",
+                "debian",
+                "--user-space-docker-cmd=" + udocker,
+                get_data(test_file),
+                get_data(job_file),
+            ]
+        )
+
+        cidfiles_count = sum(1 for _ in tmp_path.glob("*.cid"))
 
     assert "completed success" in stderr, stderr
     assert cidfiles_count == 0
@@ -83,19 +81,17 @@ def test_udocker_usage_should_not_write_cid_file(udocker: str, tmp_path: Path) -
 )
 def test_udocker_should_display_memory_usage(udocker: str, tmp_path: Path) -> None:
     """Confirm that memory ussage is logged even with udocker."""
-    cwd = Path.cwd()
-    os.chdir(tmp_path)
-    error_code, stdout, stderr = get_main_output(
-        [
-            "--enable-ext",
-            "--default-container=debian",
-            "--user-space-docker-cmd=" + udocker,
-            get_data("tests/wf/timelimit.cwl"),
-            "--sleep_time",
-            "10",
-        ]
-    )
-    os.chdir(cwd)
+    with working_directory(tmp_path):
+        error_code, stdout, stderr = get_main_output(
+            [
+                "--enable-ext",
+                "--default-container=debian",
+                "--user-space-docker-cmd=" + udocker,
+                get_data("tests/wf/timelimit.cwl"),
+                "--sleep_time",
+                "10",
+            ]
+        )
 
     assert "completed success" in stderr, stderr
     assert "Max memory" in stderr, stderr