recording into temp file first when live test run and replace the old…

… recording file after test run success. (#25055)
Azure · Jan 5, 2023 · 674625c · 674625c
1 parent f3910c3
commit 674625c
Showing 1 changed file with 20 additions and 3 deletions.
diff --git a/src/azure-cli-testsdk/azure/cli/testsdk/scenario_tests/base.py b/src/azure-cli-testsdk/azure/cli/testsdk/scenario_tests/base.py
@@ -124,8 +124,10 @@ def __init__(self,  # pylint: disable=too-many-arguments
             recording_dir,
             '{}.yaml'.format(recording_name or method_name)
         )
-        if self.is_live and os.path.exists(self.recording_file):
-            os.remove(self.recording_file)
+        self.temp_recording_file = os.path.join(
+            recording_dir,
+            '{}.temp.yaml'.format(recording_name or method_name)
+        )
 
         self.in_recording = self.is_live or not os.path.exists(self.recording_file)
         self.test_resources_count = 0
@@ -135,9 +137,11 @@ def setUp(self):
         super(ReplayableTest, self).setUp()
 
         # set up cassette
-        cm = self.vcr.use_cassette(self.recording_file)
+        cm = self.vcr.use_cassette(self.temp_recording_file if self.in_recording else self.recording_file)
         self.cassette = cm.__enter__()
         self.addCleanup(cm.__exit__)
+        if self.in_recording:
+            self.addCleanup(self._save_recording_file)
 
         # set up mock patches
         if self.in_recording:
@@ -156,6 +160,19 @@ def tearDown(self):
         assert not [t for t in threading.enumerate() if t.name.startswith("LROPoller")], \
             "You need to call 'result' or 'wait' on all LROPoller you have created"
 
+    def _save_recording_file(self, *args):  # pylint: disable=unused-argument
+        if self.in_recording and self.cassette.dirty:
+            self.cassette._save()  # pylint: disable=protected-access
+            if self._outcome.errors or self._outcome.skipped:   # pylint: disable=protected-access
+                # remove temp file
+                # to keep the temp_recording_file for debug, a switch logic can add here
+                os.remove(self.temp_recording_file)
+            else:
+                # replace recording_file by temp_recording_file
+                if os.path.exists(self.recording_file):
+                    os.remove(self.recording_file)
+                os.rename(self.temp_recording_file, self.recording_file)
+
     def _process_request_recording(self, request):
         if self.disable_recording:
             return None