pyupgrade to 3.6+ syntax

Author: mr-c

cwltest/__init__.py

@@ -37,12 +37,8 @@
 UNSUPPORTED_FEATURE = 33
 DEFAULT_TIMEOUT = 600  # 10 minutes
 
-if sys.version_info < (3, 0):
-    import subprocess32 as subprocess  # nosec
-    from pipes import quote
-else:
-    import subprocess  # nosec
-    from shlex import quote
+import subprocess  # nosec
+from shlex import quote
 
 templock = threading.Lock()
 
@@ -55,7 +51,7 @@ def prepare_test_command(
     cwd,  # type: str
     verbose=False,  # type: bool
 ):  # type: (...) -> List[str]
-    """ Turn the test into a command line. """
+    """Turn the test into a command line."""
     test_command = [tool]
     test_command.extend(args)
 
@@ -72,26 +68,26 @@ def prepare_test_command(
             outdir = tempfile.mkdtemp(prefix=os.path.abspath(os.path.curdir))
             test_command.extend(
                 [
-                    "--tmp-outdir-prefix={}".format(outdir),
-                    "--tmpdir-prefix={}".format(outdir),
+                    f"--tmp-outdir-prefix={outdir}",
+                    f"--tmpdir-prefix={outdir}",
                 ]
             )
         else:
             outdir = tempfile.mkdtemp()
-    test_command.extend(["--outdir={}".format(outdir)])
+    test_command.extend([f"--outdir={outdir}"])
     if not verbose:
         test_command.extend(["--quiet"])
 
     cwd = schema_salad.ref_resolver.file_uri(cwd)
     toolpath = test["tool"]
     if toolpath.startswith(cwd):
-        toolpath = toolpath[len(cwd)+1:]
+        toolpath = toolpath[len(cwd) + 1 :]
     test_command.extend([os.path.normcase(toolpath)])
 
     jobpath = test.get("job")
     if jobpath:
         if jobpath.startswith(cwd):
-            jobpath = jobpath[len(cwd)+1:]
+            jobpath = jobpath[len(cwd) + 1 :]
         test_command.append(os.path.normcase(jobpath))
     return test_command
 
@@ -108,7 +104,7 @@ def run_test(
     global templock
 
     out = {}  # type: Dict[str,Any]
-    outdir = outstr = outerr = u""
+    outdir = outstr = outerr = ""
     test_command = []  # type: List[str]
     duration = 0.0
     prefix = ""
@@ -145,7 +141,13 @@ def run_test(
 
         start_time = time.time()
         stderr = subprocess.PIPE if not args.verbose else None
-        process = subprocess.Popen(test_command, stdout=subprocess.PIPE, stderr=stderr, universal_newlines=True, cwd=cwd)  # nosec
+        process = subprocess.Popen(  # nosec
+            test_command,
+            stdout=subprocess.PIPE,
+            stderr=stderr,
+            universal_newlines=True,
+            cwd=cwd,
+        )
         outstr, outerr = process.communicate(timeout=timeout)
         return_code = process.poll()
         duration = time.time() - start_time
@@ -167,25 +169,25 @@ def run_test(
         if test.get("should_fail", False):
             return TestResult(0, outstr, outerr, duration, args.classname)
         _logger.error(
-            u"""Test %i failed: %s""",
+            """Test %i failed: %s""",
             test_number,
             " ".join([quote(tc) for tc in test_command]),
         )
         _logger.error(test.get("doc"))
         if err.returncode == UNSUPPORTED_FEATURE:
-            _logger.error(u"Does not support required feature")
+            _logger.error("Does not support required feature")
         else:
-            _logger.error(u"Returned non-zero")
+            _logger.error("Returned non-zero")
         _logger.error(outerr)
         return TestResult(1, outstr, outerr, duration, args.classname, str(err))
     except (yamlscanner.ScannerError, TypeError) as err:
         _logger.error(
-            u"""Test %i failed: %s""",
+            """Test %i failed: %s""",
             test_number,
-            u" ".join([quote(tc) for tc in test_command]),
+            " ".join([quote(tc) for tc in test_command]),
         )
         _logger.error(outstr)
-        _logger.error(u"Parse error %s", str(err))
+        _logger.error("Parse error %s", str(err))
         _logger.error(outerr)
     except KeyboardInterrupt:
         _logger.error(
@@ -204,7 +206,7 @@ def run_test(
         return TestResult(2, outstr, outerr, timeout, args.classname, "Test timed out")
     finally:
         if process is not None and process.returncode is None:
-            _logger.error(u"""Terminating lingering process""")
+            _logger.error("""Terminating lingering process""")
             process.terminate()
             for _ in range(0, 3):
                 time.sleep(1)
@@ -217,24 +219,24 @@ def run_test(
 
     if test.get("should_fail", False):
         _logger.warning(
-            u"""Test %i failed: %s""",
+            """Test %i failed: %s""",
             test_number,
-            u" ".join([quote(tc) for tc in test_command]),
+            " ".join([quote(tc) for tc in test_command]),
         )
         _logger.warning(test.get("doc"))
-        _logger.warning(u"Returned zero but it should be non-zero")
+        _logger.warning("Returned zero but it should be non-zero")
         return TestResult(1, outstr, outerr, duration, args.classname)
 
     try:
         compare(test.get("output"), out)
     except CompareFail as ex:
         _logger.warning(
-            u"""Test %i failed: %s""",
+            """Test %i failed: %s""",
             test_number,
-            u" ".join([quote(tc) for tc in test_command]),
+            " ".join([quote(tc) for tc in test_command]),
         )
         _logger.warning(test.get("doc"))
-        _logger.warning(u"Compare failure %s", ex)
+        _logger.warning("Compare failure %s", ex)
         fail_message = str(ex)
 
     if outdir:
@@ -345,13 +347,14 @@ def arg_parser():  # type: () -> argparse.ArgumentParser
 
     pkg = pkg_resources.require("cwltest")
     if pkg:
-        ver = u"%s %s" % (sys.argv[0], pkg[0].version)
+        ver = f"{sys.argv[0]} {pkg[0].version}"
     else:
-        ver = u"%s %s" % (sys.argv[0], "unknown version")
+        ver = "{} {}".format(sys.argv[0], "unknown version")
     parser.add_argument("--version", action="version", version=ver)
 
     return parser
 
+
 def expand_number_range(nr: str) -> List[int]:
     ans: List[int] = []
     for s in nr.split(","):
@@ -362,6 +365,7 @@ def expand_number_range(nr: str) -> List[int]:
             ans.append(int(s) - 1)
     return ans
 
+
 def main():  # type: () -> int
 
     args = arg_parser().parse_args(sys.argv[1:])
@@ -558,12 +562,12 @@ def main():  # type: () -> int
             else:
                 color = "yellow"
 
-            with open("{}/{}.json".format(args.badgedir, t), "w") as out:
+            with open(f"{args.badgedir}/{t}.json", "w") as out:
                 out.write(
                     json.dumps(
                         {
-                            "subject": "{}".format(t),
-                            "status": "{}%".format(percent),
+                            "subject": f"{t}",
+                            "status": f"{percent}%",
                             "color": color,
                         }
                     )

cwltest/tests/test_categories.py

@@ -8,11 +8,17 @@
 import defusedxml.ElementTree as ET
 import schema_salad.ref_resolver
 
+
 class TestCategories(unittest.TestCase):
     maxDiff = None
 
     def test_unsupported_with_required_tests(self):
-        args = ["--test", schema_salad.ref_resolver.file_uri(get_data("tests/test-data/required-unsupported.yml"))]
+        args = [
+            "--test",
+            schema_salad.ref_resolver.file_uri(
+                get_data("tests/test-data/required-unsupported.yml")
+            ),
+        ]
         try:
             cwd = os.getcwd()
             os.chdir(get_data("tests/test-data/"))
@@ -47,7 +53,12 @@ def test_unsupported_with_required_tests(self):
         )
 
     def test_unsupported_with_optional_tests(self):
-        args = ["--test", schema_salad.ref_resolver.file_uri(get_data("tests/test-data/optional-unsupported.yml"))]
+        args = [
+            "--test",
+            schema_salad.ref_resolver.file_uri(
+                get_data("tests/test-data/optional-unsupported.yml")
+            ),
+        ]
         error_code, stdout, stderr = run_with_mock_cwl_runner(args)
         self.assertEqual(error_code, 0)
         self.assertEqual(
@@ -58,7 +69,12 @@ def test_unsupported_with_optional_tests(self):
         )
 
     def test_error_with_optional_tests(self):
-        args = ["--test", schema_salad.ref_resolver.file_uri(get_data("tests/test-data/optional-error.yml"))]
+        args = [
+            "--test",
+            schema_salad.ref_resolver.file_uri(
+                get_data("tests/test-data/optional-error.yml")
+            ),
+        ]
         error_code, stdout, stderr = run_with_mock_cwl_runner(args)
         self.assertEqual(error_code, 1)
         self.assertIn("1 failures", stderr)
@@ -67,7 +83,9 @@ def test_category_in_junit_xml(self):
         junit_xml_report = get_data("tests/test-data/junit-report.xml")
         args = [
             "--test",
-            schema_salad.ref_resolver.file_uri(get_data("tests/test-data/optional-error.yml")),
+            schema_salad.ref_resolver.file_uri(
+                get_data("tests/test-data/optional-error.yml")
+            ),
             "--junit-xml",
             junit_xml_report,
         ]

cwltest/tests/test_prepare.py

@@ -4,10 +4,10 @@
 
 
 class TestPrepareCommand(unittest.TestCase):
-    """ Test prepare_test_command() """
+    """Test prepare_test_command()"""
 
     def test_unix_relative_path(self):
-        """ Confirm unix style to windows style path corrections. """
+        """Confirm unix style to windows style path corrections."""
         command = prepare_test_command(
             tool="cwl-runner",
             args=[],
@@ -19,7 +19,7 @@ def test_unix_relative_path(self):
                 "job": "v1.0/bwa-mem-job.json",
                 "tags": ["required"],
             },
-            cwd=os.getcwd()
+            cwd=os.getcwd(),
         )
         if os.name == "nt":
             self.assertEqual(command[3], "v1.0\\bwa-mem-tool.cwl")

cwltest/utils.py

@@ -7,7 +7,7 @@
 REQUIRED = "required"
 
 
-class TestResult(object):
+class TestResult:
 
     """Encapsulate relevant test result data."""
 
@@ -30,12 +30,12 @@ def __init__(
 
     def create_test_case(self, test):
         # type: (Dict[Text, Any]) -> junit_xml.TestCase
-        doc = test.get(u"doc", "N/A").strip()
+        doc = test.get("doc", "N/A").strip()
         if test.get("tags"):
             category = ", ".join(test["tags"])
         else:
             category = REQUIRED
-        short_name = test.get(u"short_name")
+        short_name = test.get("short_name")
         case = junit_xml.TestCase(
             doc,
             elapsed_sec=self.duration,
@@ -53,12 +53,12 @@ class CompareFail(Exception):
     @classmethod
     def format(cls, expected, actual, cause=None):
         # type: (Any, Any, Any) -> CompareFail
-        message = u"expected: %s\ngot: %s" % (
+        message = "expected: {}\ngot: {}".format(
             json.dumps(expected, indent=4, sort_keys=True),
             json.dumps(actual, indent=4, sort_keys=True),
         )
         if cause:
-            message += u"\ncaused by: %s" % cause
+            message += "\ncaused by: %s" % cause
         return cls(message)
 
 
@@ -84,7 +84,7 @@ def compare_location(expected, actual):
         raise CompareFail.format(
             expected,
             actual,
-            u"%s does not end with %s" % (actual[comp], expected[comp]),
+            f"{actual[comp]} does not end with {expected[comp]}",
         )
 
 
@@ -98,7 +98,7 @@ def compare_contents(expected, actual):
             expected,
             actual,
             json.dumps(
-                u"Output file contents do not match: actual '%s' is not equal to expected '%s'"
+                "Output file contents do not match: actual '%s' is not equal to expected '%s'"
                 % (actual_contents, expected_contents)
             ),
         )
@@ -111,7 +111,7 @@ def check_keys(keys, expected, actual):
             compare(expected.get(k), actual.get(k))
         except CompareFail as e:
             raise CompareFail.format(
-                expected, actual, u"field '%s' failed comparison: %s" % (k, str(e))
+                expected, actual, f"field '{k}' failed comparison: {str(e)}"
             )
 
 
@@ -128,11 +128,11 @@ def compare_directory(expected, actual):
     # type: (Dict[str,Any], Dict[str,Any]) -> None
     if actual.get("class") != "Directory":
         raise CompareFail.format(
-            expected, actual, u"expected object with a class 'Directory'"
+            expected, actual, "expected object with a class 'Directory'"
         )
     if "listing" not in actual:
         raise CompareFail.format(
-            expected, actual, u"'listing' is mandatory field in Directory object"
+            expected, actual, "'listing' is mandatory field in Directory object"
         )
     for i in expected["listing"]:
         found = False
@@ -147,7 +147,7 @@ def compare_directory(expected, actual):
             raise CompareFail.format(
                 expected,
                 actual,
-                u"%s not found" % json.dumps(i, indent=4, sort_keys=True),
+                "%s not found" % json.dumps(i, indent=4, sort_keys=True),
             )
     compare_file(expected, actual)
 
@@ -159,12 +159,12 @@ def compare_dict(expected, actual):
             compare(expected[c], actual.get(c))
         except CompareFail as e:
             raise CompareFail.format(
-                expected, actual, u"failed comparison for key '%s': %s" % (c, e)
+                expected, actual, f"failed comparison for key '{c}': {e}"
             )
     extra_keys = set(actual.keys()).difference(list(expected.keys()))
     for k in extra_keys:
         if actual[k] is not None:
-            raise CompareFail.format(expected, actual, u"unexpected key '%s'" % k)
+            raise CompareFail.format(expected, actual, "unexpected key '%s'" % k)
 
 
 def compare(expected, actual):  # type: (Any, Any) -> None
@@ -190,7 +190,7 @@ def compare(expected, actual):  # type: (Any, Any) -> None
                 raise CompareFail.format(expected, actual)
 
             if len(expected) != len(actual):
-                raise CompareFail.format(expected, actual, u"lengths don't match")
+                raise CompareFail.format(expected, actual, "lengths don't match")
             for c in range(0, len(expected)):
                 try:
                     compare(expected[c], actual[c])