Fix broken unit tests

Author: ian-Liaozy

sdks/python/apache_beam/runners/interactive/recording_manager.py

@@ -492,7 +492,6 @@ def _run_async_computation(
         if not self._wait_for_dependencies(pcolls_to_compute, async_result):
           raise RuntimeError('Dependency computation failed or was cancelled.')
 
-      self._env.mark_pcollection_computing(pcolls_to_compute)
       _LOGGER.info(
           'Starting asynchronous computation for %d PCollections.',
           len(pcolls_to_compute))
@@ -696,6 +695,7 @@ def compute_async(
       async_result = AsyncComputationResult(
           future, pcolls_to_compute, self.user_pipeline, self)
       self._async_computations[async_result._display_id] = async_result
+      self._env.mark_pcollection_computing(pcolls_to_compute)
 
       def task():
         try:

sdks/python/apache_beam/runners/interactive/recording_manager_test.py

@@ -993,12 +993,6 @@ def capture_task(task):
 
     mock_submit.side_effect = capture_task
 
-    res = rm.compute_async({pcoll}, blocking=False)
-    self.assertIs(res, mock_async_res_instance)
-    mock_submit.assert_called_once()
-    self.assertIsNotNone(task_submitted)
-
-    # Patch dependencies of _run_async_computation
     with patch.object(
         rm, '_wait_for_dependencies', return_value=True
     ), patch.object(
@@ -1008,10 +1002,22 @@ def capture_task(task):
         'mark_pcollection_computing',
         wraps=ie.current_env().mark_pcollection_computing,
     ) as wrapped_mark:
-      # Run the task to trigger the marks
-      task_submitted()
+
+      res = rm.compute_async({pcoll}, blocking=False)
       wrapped_mark.assert_called_once_with({pcoll})
 
+    # Run the task to trigger the marks
+    self.assertIs(res, mock_async_res_instance)
+    mock_submit.assert_called_once()
+    self.assertIsNotNone(task_submitted)
+
+    with patch.object(
+        rm, '_wait_for_dependencies', return_value=True
+    ), patch.object(
+        rm, '_execute_pipeline_fragment'
+    ) as _:
+      task_submitted()
+
     self.assertTrue(pcoll in ie.current_env().computing_pcollections)
 
   def test_get_all_dependencies(self):

sdks/python/apache_beam/runners/interactive/utils_test.py

@@ -244,7 +244,9 @@ def test_child_module_logger_can_override_logging_level(self, mock_emit):
     reason='[interactive] dependency is not installed.')
 class ProgressIndicatorTest(unittest.TestCase):
   def setUp(self):
-    ie.new_env()
+    with patch(
+        'apache_beam.runners.interactive.cache_manager.CacheManager.cleanup'):
+      ie.new_env()
 
   @patch('IPython.get_ipython', new_callable=mock_get_ipython)
   @patch(