Fix f-string

test/test_dag.py

@@ -291,7 +291,7 @@ def test_undetermined(temp_factory):
 
 
 def test_auxiliary_steps(temp_factory, clear_now_and_after):
-    graph = textwrap.dedent(('''
+    graph = textwrap.dedent('''
         [K: provides='{name}.txt']
         output: f"{name}.txt"
 
@@ -308,7 +308,7 @@ def test_auxiliary_steps(temp_factory, clear_now_and_after):
         [C_3]
         input: 'a.txt'
 
-        '''))
+        ''')
     # a.txt exists and b.txt does not exist
     temp_factory('a.txt')
     clear_now_and_after('b.txt')

test/test_docker_actions.py

@@ -50,14 +50,14 @@ def timeout_func():
             # important: KeyboardInterrupt does not interrupt time.sleep()
             # because KeyboardInterrupt is handled by Python interpreter but
             # time.sleep() calls a system function.
-            raise TimeoutException("Timed out for operation {}".format(msg))
+            raise TimeoutException(f"Timed out for operation {msg}")
         finally:
             # if the action ends in specified time, timer is canceled
             timer.cancel()
     else:
 
         def signal_handler(signum, frame):
-            raise TimeoutException("Timed out for option {}".format(msg))
+            raise TimeoutException(f"Timed out for option {msg}")
 
         signal.signal(signal.SIGALRM, signal_handler)
         signal.alarm(seconds)

test/test_execute.py

@@ -47,7 +47,7 @@ def test_command_line(clear_now_and_after):
 a =1
 """)
     result = subprocess.check_output("sos --version", stderr=subprocess.STDOUT, shell=True).decode()
-    assert result.startswith("sos {}".format(__version__))
+    assert result.startswith(f"sos {__version__}")
     assert (subprocess.call("sos", stderr=subprocess.DEVNULL, stdout=subprocess.DEVNULL, shell=True) == 0)
     assert (subprocess.call("sos -h", stderr=subprocess.DEVNULL, stdout=subprocess.DEVNULL, shell=True) == 0)
     assert (subprocess.call(
@@ -1002,7 +1002,7 @@ def test_dynamic_output(temp_factory):
             with open(ff, 'w') as h:
                 h.write('a')
         """)
-    assert env.sos_dict["test"] == ["temp/something{}.html".format(x) for x in range(4)]
+    assert env.sos_dict["test"] == [f"temp/something{x}.html" for x in range(4)]
 
 
 def test_dynamic_input(temp_factory):
@@ -1028,13 +1028,13 @@ def test_dynamic_input(temp_factory):
     wf = script.workflow()
     Base_Executor(wf).run()
     assert env.sos_dict["test"], (
-        sos_targets([os.path.join("temp", "test_{}.txt.bak".format(x)) for x in range(5)]) ==
+        sos_targets([os.path.join("temp", f"test_{x}.txt.bak") for x in range(5)]) ==
         f"Expecting {[os.path.join('temp', 'test_{}.txt.bak'.format(x)) for x in range(5)]} observed {env.sos_dict['test']}"
     )
     # this time we use th existing signature
     Base_Executor(wf).run()
     assert env.sos_dict["test"], (
-        sos_targets([os.path.join("temp", "test_{}.txt.bak".format(x)) for x in range(5)]) ==
+        sos_targets([os.path.join("temp", f"test_{x}.txt.bak") for x in range(5)]) ==
         f"Expecting {[os.path.join('temp', 'test_{}.txt.bak'.format(x)) for x in range(5)]} observed {env.sos_dict['test']}"
     )
 
@@ -1207,7 +1207,7 @@ def test_removed_intermediate_files(clear_now_and_after):
 
 def test_stopped_output():
     """test output with stopped step"""
-    for file in ["{}.txt".format(a) for a in range(10)]:
+    for file in [f"{a}.txt" for a in range(10)]:
         if file_target(file).exists():
             file_target(file).unlink()
     execute_workflow("""
@@ -1224,9 +1224,9 @@ def test_stopped_output():
         """)
     for idx in range(10):
         if idx % 2 == 0:
-            assert not file_target("{}.txt".format(idx)).target_exists()
+            assert not file_target(f"{idx}.txt").target_exists()
         else:
-            assert file_target("{}.txt".format(idx)).target_exists()
+            assert file_target(f"{idx}.txt").target_exists()
             file_target(f"{idx}.txt").unlink()
 
 

test/test_parser.py

@@ -115,17 +115,17 @@ def test_sections():
     # bad names
     for badname in ["56_1", "_a", "a_", "1x", "*", "?"]:
         with pytest.raises(ParsingError):
-            SoS_Script("[{}]".format(badname))
+            SoS_Script(f"[{badname}]")
     # bad options
     for badoption in ["ss"]:
         with pytest.raises(ParsingError):
-            SoS_Script("[0:{}]".format(badoption))
+            SoS_Script(f"[0:{badoption}]")
     # allowed names
     for name in ["a5", "a_5", "*_0", "a*1_100"]:
-        SoS_Script("[{}]".format(name))
+        SoS_Script(f"[{name}]")
     # allowed names with alias
     for name in ["a5 (p1)", "a_5 (something fun)", "*_0 (no way)", "a*1_100"]:
-        SoS_Script("[{}]".format(name))
+        SoS_Script(f"[{name}]")
     # duplicate sections
     with pytest.raises(ParsingError):
         SoS_Script("""[1]\n[1]""")
@@ -1084,7 +1084,7 @@ def test_group_by(temp_factory, clear_now_and_after):
         sos_targets("a7.txt", "a8.txt", "a9.txt"),
     ]
     # number of files should be divisible by group_by
-    temp_factory(["a{}.txt".format(x) for x in range(1, 10)])
+    temp_factory([f"a{x}.txt" for x in range(1, 10)])
     execute_workflow(
         """
         [0]
@@ -1166,7 +1166,7 @@ def test_group_by(temp_factory, clear_now_and_after):
     ]
 
     # group_by='pairlabel3'
-    temp_factory(["c{}.txt".format(x) for x in range(1, 7)])
+    temp_factory([f"c{x}.txt" for x in range(1, 7)])
 
     execute_workflow(
         """
@@ -1212,7 +1212,7 @@ def test_group_by(temp_factory, clear_now_and_after):
         ),
     ]
     # group_by='pairlabel3'
-    temp_factory(["c{}.txt".format(x) for x in range(1, 7)])
+    temp_factory([f"c{x}.txt" for x in range(1, 7)])
 
     execute_workflow(
         """
@@ -1279,7 +1279,7 @@ def grp(x):
 def test_output_group_by(temp_factory):
     """Test group_by parameter of step output"""
     # group_by = 'all'
-    temp_factory(["a{}.txt".format(x) for x in range(4)])
+    temp_factory([f"a{x}.txt" for x in range(4)])
     #
     execute_workflow(
         """

test/test_remove.py

@@ -10,11 +10,11 @@
 
 def assertExists(fdlist):
     for fd in fdlist:
-        assert os.path.exists(fd), '{} does not exist'.format(fd)
+        assert os.path.exists(fd), f'{fd} does not exist'
 
 def assertNonExists(fdlist):
     for fd in fdlist:
-        assert not os.path.exists(fd), '{} still exists'.format(fd)
+        assert not os.path.exists(fd), f'{fd} still exists'
 
 def test_setup(test_workflow):
     assertExists([

test/test_signature.py

@@ -462,7 +462,7 @@ def test_loop_wise_signature(clear_now_and_after):
     assert ts1 == os.path.getmtime('myfile_11.txt')
     #
     for t in range(10, 12):
-        with open('myfile_{}.txt'.format(t)) as tmp:
+        with open(f'myfile_{t}.txt') as tmp:
             assert tmp.read().strip() == str(t)
 
 

test/test_target.py

@@ -218,12 +218,12 @@ def test_target_format():
     ]:
         if isinstance(res, str):
             assert interpolate(
-                "{{target:{}}}".format(fmt), globals(),
+                f"{{target:{fmt}}}", globals(),
                 locals()) == res, "Interpolation of {}:{} should be {}".format(
                     target, fmt, res)
 
         else:
-            assert interpolate("{{target:{}}}".format(fmt), globals(), locals(
+            assert interpolate(f"{{target:{fmt}}}", globals(), locals(
             )) in res, "Interpolation of {}:{} should be one of {}".format(
                 target, fmt, res)
 

test/test_task.py

@@ -413,7 +413,7 @@ def test_task_tags():
     """Test option tags of tasks"""
     import random
 
-    tag = "tag{}".format(random.randint(1, 100000))
+    tag = f"tag{random.randint(1, 100000)}"
     with open("test_tags.sos", "w") as tt:
         tt.write("""
 [10]
@@ -434,11 +434,11 @@ def test_task_tags():
         },
     ).run()
     ret = subprocess.check_output(
-        "sos status -t {}".format(tag), shell=True).decode()
-    assert len(ret.splitlines()) == 5, "Obtained {}".format(ret)
+        f"sos status -t {tag}", shell=True).decode()
+    assert len(ret.splitlines()) == 5, f"Obtained {ret}"
     # test multiple tags
-    tag1 = "tag{}".format(random.randint(1, 100000))
-    tag2 = "tag{}".format(random.randint(1, 100000))
+    tag1 = f"tag{random.randint(1, 100000)}"
+    tag2 = f"tag{random.randint(1, 100000)}"
     with open("test_tags.sos", "w") as tt:
         tt.write("""
 [10]
@@ -460,8 +460,8 @@ def test_task_tags():
         },
     ).run()
     ret = subprocess.check_output(
-        "sos status -t {}".format(tag2), shell=True).decode()
-    assert len(ret.splitlines()) == 2, "Obtained {}".format(ret)
+        f"sos status -t {tag2}", shell=True).decode()
+    assert len(ret.splitlines()) == 2, f"Obtained {ret}"
 
 
 @pytest.mark.skipif(not has_docker, reason="Docker container not usable")