Use context manager to handle test file's download and deletion

Author: rickwu666666

contrib/checkbox-ce-oem/checkbox-provider-ce-oem/bin/gst_encoder_psnr.py

@@ -30,11 +30,12 @@
     GStreamerMuxerType,
     GStreamerEncodePlugins,
     MetadataValidator,
-    get_big_bug_bunny_golden_sample,
     generate_artifact_name,
     compare_psnr,
     delete_file,
     execute_command,
+    get_test_file_path_by_params,
+    manage_test_file_by_params,
 )
 
 logging.basicConfig(level=logging.INFO)
@@ -200,8 +201,8 @@ def __init__(
         }
         # This sample video file will be consumed by any gstreamer piple as
         # input video.
-        self._golden_sample = get_big_bug_bunny_golden_sample(
-            width=self._width, height=self._height, framerate=self._framerate
+        self._golden_sample = get_test_file_path_by_params(
+            self._width, self._height, self._framerate
         )
         self._artifact_file = ""
 
@@ -332,9 +333,8 @@ def __init__(
         }
         # This sample video file will be consumed by any gstreamer piple as
         # input video.
-        self._golden_sample = os.path.join(
-            VIDEO_CODEC_TESTING_DATA,
-            "{}p_{}fps_h264.mp4".format(self._height, self._framerate),
+        self._golden_sample = get_test_file_path_by_params(
+            self._width, self._height, self._framerate, "h264"
         )
         self._artifact_file = ""
 
@@ -427,9 +427,8 @@ def _h264_pipeline_builder(self) -> str:
         """
         # This sample video file will be consumed by any gstreamer piple as
         # input video.
-        self._golden_sample = os.path.join(
-            VIDEO_CODEC_TESTING_DATA,
-            "{}p_{}fps_h264.mp4".format(self._height, self._framerate),
+        self._golden_sample = get_test_file_path_by_params(
+            self._width, self._height, self._framerate, "h264"
         )
         pipeline = (
             "{} filesrc location={} ! qtdemux ! decodebin !"
@@ -452,9 +451,8 @@ def _h265_pipeline_builder(self) -> str:
         """
         # This sample video file will be consumed by any gstreamer piple as
         # input video.
-        self._golden_sample = os.path.join(
-            VIDEO_CODEC_TESTING_DATA,
-            "{}p_{}fps_h265.mp4".format(self._height, self._framerate),
+        self._golden_sample = get_test_file_path_by_params(
+            self._width, self._height, self._framerate, "h265"
         )
         pipeline = (
             "{} filesrc location={} ! qtdemux ! decodebin !"
@@ -475,9 +473,8 @@ def _vp8_pipeline_builder(self) -> str:
         """
         # This sample video file will be consumed by any gstreamer piple as
         # input video.
-        self._golden_sample = os.path.join(
-            VIDEO_CODEC_TESTING_DATA,
-            "{}p_{}fps_vp8.webm".format(self._height, self._framerate),
+        self._golden_sample = get_test_file_path_by_params(
+            self._width, self._height, self._framerate, "vp8"
         )
         pipeline = (
             "{} filesrc location={} ! qtdemux ! decodebin !"
@@ -546,9 +543,8 @@ def _omxh264_pipeline_builder(self) -> str:
         """
         Build gstreamer pipeline for omxh264enc
         """
-        self._golden_sample = os.path.join(
-            VIDEO_CODEC_TESTING_DATA,
-            "{}p_{}fps_h264.mp4".format(self._height, self._framerate),
+        self._golden_sample = get_test_file_path_by_params(
+            self._width, self._height, self._framerate, "h264"
         )
         pipeline = (
             "{} filesrc location={} ! qtdemux ! h264parse !"
@@ -585,22 +581,27 @@ def build_pipeline(self) -> str:
 
 def main() -> None:
     args = register_arguments()
-    p = project_factory(args)
-    logging.info("Step 1: Generating artifact...")
-    cmd = p.build_pipeline()
-    # execute command
-    execute_command(cmd=cmd)
-    logging.info("\nStep 2: Checking metadata...")
-    mv = MetadataValidator(file_path=p.artifact_file)
-    mv.validate("width", args.width).validate("height", args.height).validate(
-        "frame_rate", args.framerate
-    ).validate("codec", args.encoder_plugin).is_valid()
-    logging.info("\nStep 3: Comparing PSNR...")
-    compare_psnr(
-        golden_reference_file=p.psnr_reference_file,
-        artifact_file=p.artifact_file,
-    )
-    delete_file(file_path=p.artifact_file)
+    with manage_test_file_by_params(
+        args.width, args.height, args.framerate, args.encoder_plugin
+    ):
+        p = project_factory(args)
+        logging.info("Step 1: Generating artifact...")
+        cmd = p.build_pipeline()
+        # execute command
+        execute_command(cmd=cmd)
+        logging.info("\nStep 2: Checking metadata...")
+        mv = MetadataValidator(file_path=p.artifact_file)
+        mv.validate("width", args.width).validate(
+            "height", args.height
+        ).validate("frame_rate", args.framerate).validate(
+            "codec", args.encoder_plugin
+        ).is_valid()
+        logging.info("\nStep 3: Comparing PSNR...")
+        compare_psnr(
+            golden_reference_file=p.psnr_reference_file,
+            artifact_file=p.artifact_file,
+        )
+        delete_file(file_path=p.artifact_file)
 
 
 if __name__ == "__main__":

contrib/checkbox-ce-oem/checkbox-provider-ce-oem/bin/gst_transform_resize.py

@@ -25,16 +25,15 @@
 
 from gst_utils import (
     GST_LAUNCH_BIN,
-    VIDEO_CODEC_TESTING_DATA,
-    SAMPLE_2_FOLDER,
     PipelineInterface,
     GStreamerEncodePlugins,
     MetadataValidator,
-    get_big_bug_bunny_golden_sample,
     generate_artifact_name,
     compare_psnr,
     delete_file,
     execute_command,
+    get_test_file_path_by_params,
+    manage_test_file_by_params,
 )
 
 logging.basicConfig(level=logging.INFO)
@@ -169,10 +168,8 @@ def __init__(
         }
         # This sample video file will be consumed by any gstreamer piple as
         # input video.
-        self._golden_sample = get_big_bug_bunny_golden_sample(
-            width=self._width_from,
-            height=self._height_from,
-            framerate=self._framerate,
+        self._golden_sample = get_test_file_path_by_params(
+            self._width_from, self._height_from, self._framerate
         )
         self._artifact_file = ""
 
@@ -188,17 +185,13 @@ def psnr_reference_file(self) -> str:
         A golden reference which has been transformed in advance. It's used to
         be the compared reference file for PSNR.
         """
-        golden_reference = get_big_bug_bunny_golden_sample(
+        golden_reference = get_test_file_path_by_params(
             self._width_to, self._height_to, self._framerate
         )
-
-        full_path = os.path.join(
-            VIDEO_CODEC_TESTING_DATA, SAMPLE_2_FOLDER, golden_reference
-        )
-        if not os.path.exists(full_path):
+        if not os.path.exists(golden_reference):
             raise SystemExit(
                 "Error: Golden PSNR reference '{}' doesn't exist".format(
-                    full_path
+                    golden_reference
                 )
             )
 
@@ -229,24 +222,30 @@ def build_pipeline(self) -> str:
 
 def main() -> None:
     args = register_arguments()
-    p = project_factory(args)
-    logging.info("Step 1: Generating artifact...")
-    cmd = p.build_pipeline()
-    # execute command
-    execute_command(cmd=cmd)
-    logging.info("\nStep 2: Checking metadata...")
-    mv = MetadataValidator(file_path=p.artifact_file)
-    mv.validate("width", args.width_to).validate(
-        "height", args.height_to
-    ).validate("frame_rate", args.framerate).validate(
-        "codec", args.encoder_plugin
-    ).is_valid()
-    logging.info("\nStep 3: Comparing PSNR...")
-    compare_psnr(
-        golden_reference_file=p.psnr_reference_file,
-        artifact_file=p.artifact_file,
-    )
-    delete_file(file_path=p.artifact_file)
+    with manage_test_file_by_params(
+        args.width_from, args.height_from, args.framerate, args.encoder_plugin
+    ):
+        with manage_test_file_by_params(
+            args.width_to, args.height_to, args.framerate, args.encoder_plugin
+        ):
+            p = project_factory(args)
+            logging.info("Step 1: Generating artifact...")
+            cmd = p.build_pipeline()
+            # execute command
+            execute_command(cmd=cmd)
+            logging.info("\nStep 2: Checking metadata...")
+            mv = MetadataValidator(file_path=p.artifact_file)
+            mv.validate("width", args.width_to).validate(
+                "height", args.height_to
+            ).validate("frame_rate", args.framerate).validate(
+                "codec", args.encoder_plugin
+            ).is_valid()
+            logging.info("\nStep 3: Comparing PSNR...")
+            compare_psnr(
+                golden_reference_file=p.psnr_reference_file,
+                artifact_file=p.artifact_file,
+            )
+            delete_file(file_path=p.artifact_file)
 
 
 if __name__ == "__main__":

contrib/checkbox-ce-oem/checkbox-provider-ce-oem/bin/gst_v4l2_audio_video_synchronization.py

@@ -23,6 +23,7 @@
 import shlex
 import subprocess
 from typing import Any
+from gst_utils import manage_test_file_by_name
 
 logging.basicConfig(level=logging.INFO)
 
@@ -169,21 +170,20 @@ def play_video_for_av_synchronization_test(args: Any) -> None:
         exist, or if the extracted MD5 checksum does not match the golden MD5
         checksum.
     """
-    # Check the golden sample exists
-    if not os.path.exists(args.golden_sample_path):
-        raise SystemExit(
-            "Golden Sample '{}' doesn't exist".format(args.golden_sample_path)
+    with manage_test_file_by_name(
+        file_name=os.path.basename(args.golden_sample_path),
+        target_dir=os.path.dirname(args.golden_sample_path),
+    ):
+        gst_launch_bin = os.getenv("GST_LAUNCH_BIN", "gst-launch-1.0")
+        cmd = build_gst_command(
+            gst_bin=gst_launch_bin,
+            golden_sample_path=args.golden_sample_path,
+            decoder=args.decoder_plugin,
+            video_sink=args.video_sink,
+            capssetter_pipeline=args.capssetter_pipeline,
         )
-    gst_launch_bin = os.getenv("GST_LAUNCH_BIN", "gst-launch-1.0")
-    cmd = build_gst_command(
-        gst_bin=gst_launch_bin,
-        golden_sample_path=args.golden_sample_path,
-        decoder=args.decoder_plugin,
-        video_sink=args.video_sink,
-        capssetter_pipeline=args.capssetter_pipeline,
-    )
-    # The video will be displayed on the real display
-    execute_command(cmd)
+        # The video will be displayed on the real display
+        execute_command(cmd)
 
 
 def main() -> None:

contrib/checkbox-ce-oem/checkbox-provider-ce-oem/bin/gst_video_decoder_performance.py

@@ -23,7 +23,7 @@
 import re
 
 from performance_mode_controller import get_performance_ctx_function
-from gst_utils import execute_command
+from gst_utils import execute_command, manage_test_file_by_name
 
 logging.basicConfig(level=logging.INFO)
 
@@ -304,50 +304,50 @@ def main() -> None:
         ),
         args.minimum_fps,
     )
-    # Check the golden sample exixt
-    if not os.path.exists(args.golden_sample_path):
-        raise SystemExit(
-            "Golden Sample '{}' doesn't exist".format(args.golden_sample_path)
-        )
-    gst_launch_bin = os.getenv("GST_LAUNCH_BIN", "gst-launch-1.0")
-    if "imx8m" in args.platform:
-        cmd = build_imx_gst_command(
-            gst_bin=gst_launch_bin,
-            golden_sample_path=args.golden_sample_path,
-            decoder=args.decoder_plugin,
-            sink=args.sink,
-            fpsdisplaysink_sync=args.fpsdisplaysink_sync,
-        )
-    elif "rz" in args.platform:
-        cmd = build_renesas_gst_command(
-            gst_bin=gst_launch_bin,
-            golden_sample_path=args.golden_sample_path,
-            decoder=args.decoder_plugin,
-            sink=args.sink,
-            fpsdisplaysink_sync=args.fpsdisplaysink_sync,
-            platform=args.platform,
-        )
-    else:
-        cmd = build_gst_command(
-            gst_bin=gst_launch_bin,
-            golden_sample_path=args.golden_sample_path,
-            decoder=args.decoder_plugin,
-            sink=args.sink,
-            fpsdisplaysink_sync=args.fpsdisplaysink_sync,
-        )
-
-    output = ""
-
-    if args.performance_mode_target:
-        performance_mode_ctx = get_performance_ctx_function()
-        with performance_mode_ctx(platform=args.performance_mode_target):
+    with manage_test_file_by_name(
+        file_name=os.path.basename(args.golden_sample_path),
+        target_dir=os.path.dirname(args.golden_sample_path),
+    ):
+        gst_launch_bin = os.getenv("GST_LAUNCH_BIN", "gst-launch-1.0")
+        if "imx8m" in args.platform:
+            cmd = build_imx_gst_command(
+                gst_bin=gst_launch_bin,
+                golden_sample_path=args.golden_sample_path,
+                decoder=args.decoder_plugin,
+                sink=args.sink,
+                fpsdisplaysink_sync=args.fpsdisplaysink_sync,
+            )
+        elif "rz" in args.platform:
+            cmd = build_renesas_gst_command(
+                gst_bin=gst_launch_bin,
+                golden_sample_path=args.golden_sample_path,
+                decoder=args.decoder_plugin,
+                sink=args.sink,
+                fpsdisplaysink_sync=args.fpsdisplaysink_sync,
+                platform=args.platform,
+            )
+        else:
+            cmd = build_gst_command(
+                gst_bin=gst_launch_bin,
+                golden_sample_path=args.golden_sample_path,
+                decoder=args.decoder_plugin,
+                sink=args.sink,
+                fpsdisplaysink_sync=args.fpsdisplaysink_sync,
+            )
+
+        output = ""
+
+        if args.performance_mode_target:
+            performance_mode_ctx = get_performance_ctx_function()
+            with performance_mode_ctx(platform=args.performance_mode_target):
+                output = execute_command(cmd).rstrip(os.linesep)
+        else:
             output = execute_command(cmd).rstrip(os.linesep)
-    else:
-        output = execute_command(cmd).rstrip(os.linesep)
 
-    if not is_valid_result(output, args.minimum_fps):
-        raise SystemExit(1)
-    logging.info("Pass")
+        is_valid = is_valid_result(output, args.minimum_fps)
+        if not is_valid:
+            raise SystemExit(1)
+        logging.info("Pass")
 
 
 if __name__ == "__main__":