diff --git a/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/bin/gst_resources_generator.py b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/bin/gst_resources_generator.py new file mode 100755 index 000000000..29b3085ff --- /dev/null +++ b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/bin/gst_resources_generator.py @@ -0,0 +1,174 @@ +#!/usr/bin/env python3 +# This file is part of Checkbox. +# +# Copyright 2024 Canonical Ltd. +# Written by: +# Patrick Chang +# +# Checkbox is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 3, +# as published by the Free Software Foundation. +# +# Checkbox is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with Checkbox. If not, see . + +import argparse +import os +import json +import logging +from pathlib import Path +from typing import Dict, Any, List + +logging.basicConfig(level=logging.INFO) + + +def load_json_config(file_path: str) -> Dict: + """ + Reads a JSON file from the specified file path. + + :param file_path: + The path to the JSON file. + :returns: + dict: The loaded configuration dictionary. + :raises FileNotFoundError: + """ + file_path = Path(file_path) + if not file_path.exists(): + raise FileNotFoundError( + "The file '{}' does not exist.".format(file_path) + ) + + with file_path.open("r") as f: + data = json.load(f) + + return data + + +def register_arguments(): + parser = argparse.ArgumentParser( + formatter_class=argparse.RawDescriptionHelpFormatter, + description=( + "Script helps verify the MD5 checksum from specific Gstreamer" + " Decoder with different resolutions and color spaces is exactly" + " match golden reference" + ), + ) + + parser.add_argument( + "gst_conf_name", + type=str, + help="Path of the specified test configuration file.", + ) + + parser.add_argument( + "-gtdp", + "--gst_testing_data_path", + type=str, + help="Path of the testing data.", + ) + + args = parser.parse_args() + return args + + +class GstResources: + def __init__(self, args: Any, scenarios: dict): + self._scenarios = scenarios + self._current_scenario_name = "" + self._args = args + self._resource_items = [] + + def _gst_v4l2_video_decoder_md5_checksum_comparison_helper( + self, + decoder_plugin: str, + width: dict, + height: dict, + color_space: str, + source_format: str, + ) -> dict: + """ + Generate a resource item dictionary for + gst_v4l2_video_decoder_md5_checksum_comparison scenario + """ + name = "{}x{}-{}-{}".format(width, height, decoder_plugin, color_space) + golden_sample_file = "{}/video_golden_samples/{}.{}".format( + self._args.gst_testing_data_path, name, source_format + ) + golden_md5_checkum_file = "{}/{}/golden_md5_checksum/{}/{}.md5".format( + self._args.gst_testing_data_path, + self._current_scenario_name, + self._args.gst_conf_name, + name, + ) + + returned_dict = { + "scenario": self._current_scenario_name, + "decoder_plugin": decoder_plugin, + "width": width, + "height": height, + "color_space": color_space, + "golden_sample_file": golden_sample_file, + "golden_md5_checkum_file": golden_md5_checkum_file, + } + + return returned_dict + + def gst_v4l2_video_decoder_md5_checksum_comparison( + self, scenario_data: List[Dict] + ) -> None: + for item in scenario_data: + self._resource_items.extend([ + self._gst_v4l2_video_decoder_md5_checksum_comparison_helper( + decoder_plugin=item["decoder_plugin"], + width=resolution["width"], + height=resolution["height"], + color_space=color_space, + source_format=item["source_format"], + ) + for resolution in item["resolutions"] + for color_space in item["color_spaces"] + ]) + + def main(self): + for scenario in self._scenarios: + self._current_scenario_name = scenario + getattr(self, scenario)(self._scenarios[scenario]) + self._dump_resources() + + def _dump_resources(self): + """ + Prints each key-value pair from the self._resource_items in the format + "key": "value". + """ + for item in self._resource_items: + for key, value in item.items(): + print("{}: {}".format(key, value)) + print() + print() + # Renew for next scenario + self._resource_items = [] + + +def main() -> None: + try: + data_dir = os.environ["PLAINBOX_PROVIDER_DATA"] + except KeyError: + raise SystemExit("PLAINBOX_PROVIDER_DATA variable not set") + args = register_arguments() + + # All gstreamer related configs should be put into gstreamer-test-confs + # direcotry by design + conf_path = os.path.join( + data_dir, "gstreamer-test-confs", "{}.json".format(args.gst_conf_name) + ) + scenarios = load_json_config(file_path=conf_path) + GstResources(args, scenarios).main() + + +if __name__ == "__main__": + main() diff --git a/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/bin/gst_v4l2_video_decoder_md5_checksum_comparison.py b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/bin/gst_v4l2_video_decoder_md5_checksum_comparison.py new file mode 100755 index 000000000..f46b38392 --- /dev/null +++ b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/bin/gst_v4l2_video_decoder_md5_checksum_comparison.py @@ -0,0 +1,238 @@ +#!/usr/bin/env python3 +# This file is part of Checkbox. +# +# Copyright 2024 Canonical Ltd. +# Written by: +# Patrick Chang +# +# Checkbox is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 3, +# as published by the Free Software Foundation. +# +# Checkbox is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with Checkbox. If not, see . + +import argparse +import logging +import os +import re +import shlex +import subprocess +from typing import Any + +logging.basicConfig(level=logging.INFO) + + +def register_arguments(): + parser = argparse.ArgumentParser( + formatter_class=argparse.RawDescriptionHelpFormatter, + description=( + "Script helps verify the MD5 checksum from specific Gstreamer" + " Decoder with different resolutions and color spaces is exactly" + " match golden reference" + ), + ) + + parser.add_argument( + "-gp", + "--golden_sample_path", + required=True, + type=str, + help="Path of Golden Sample file", + ) + + parser.add_argument( + "-gmp", + "--golden_sample_md5_checksum_path", + required=True, + type=str, + help="Path of Golden Sample's MD5 chekcusm", + ) + + parser.add_argument( + "-dp", + "--decoder_plugin", + required=True, + type=str, + help="Decoder plugin be used in gstreamer pipeline e.g. v4l2h264dec", + ) + + parser.add_argument( + "-cs", + "--color_space", + required=True, + type=str, + help="Color space be used in gstreamer format e.g. I420 or NV12", + ) + + args = parser.parse_args() + return args + + +def build_gst_command( + gst_bin: str, golden_sample_path: str, decoder: str, color_sapce: str +) -> str: + """ + Builds a GStreamer command to process the golden sample. + + :param gst_bin: + The binary name of gstreamer. Default is "gst-launch-1.0" + You can assign the snap name to GST_LAUNCH_BIN env variable if you + want to using snap. + :param golden_sample: + The path to the golden sample file. + :param decoder: + The decoder to use for the video, e.g., "v4l2vp8dec", "v4l2vp9dec". + :param color_space: + The desired color space format for the output, e.g., "I420", "NV12". + + :returns: + The GStreamer command to execute. + """ + x_raw_format_str = "video/x-raw,format={} ! ".format(color_sapce) + + if decoder in ["v4l2vp8dec", "v4l2vp9dec"]: + x_raw_format_str = "" + + cmd = ( + "{} -v filesrc location={} ! parsebin ! {} ! v4l2convert ! {}" + "checksumsink hash=0 sync=false" + ).format(gst_bin, golden_sample_path, decoder, x_raw_format_str) + + return cmd + + +def get_md5_checksum_from_command(cmd: str) -> str: + """ + Executes the GStreamer command and extracts the MD5 checksums. + + :param cmd: + The GStreamer command to execute. + + :returns: + The extracted MD5 checksums. + """ + try: + logging.info("Starting command: '{}'".format(cmd)) + ret = subprocess.run( + shlex.split(cmd), + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + encoding="utf-8", + timeout=30, + ) + md5_data = extract_the_md5_checksum(ret.stdout) + return md5_data + except Exception as e: + logging.error(e.stderr) + raise SystemExit(e.returncode) + + +def extract_the_md5_checksum(intput: str) -> str: + """ + Extracts the MD5 checksums from the given input string. + + :param intput: + The input string containing the MD5 checksums. + + :returns: + The extracted MD5 checksums. + """ + pattern = r"^(\d+:\d+:\d+\.\d+) ([0-9a-f]+)$" + checksums = re.findall(pattern, intput, re.MULTILINE) + output = "" + for checksum in checksums: + output += checksum[1] + os.linesep + return output + + +def validate_video_decoder_md5_checksum(args: Any) -> None: + """ + Validates the MD5 checksum of the output generated by a video decoder with + specific resolution and color space. + + This function performs the following steps: + + 1. Checks if the golden sample file and the golden MD5 checksum file exist. + 2. Builds a GStreamer command to process the golden sample using the + specified decoder and color space. + 3. Executes the GStreamer command and extracts the MD5 checksums from the + output. + 4. Reads the golden MD5 checksum from the file. + 5. Compares the extracted MD5 checksum with the golden MD5 checksum. + 6. If the checksums match, logs a "Pass" message. If they don't match, + logs the golden MD5 checksum and raises a `SystemExit` exception. + + :param args: + An object containing the following attributes: + - `golden_sample_path` (str): The path to the golden sample file. + - `golden_sample_md5_checksum_path` (str): The path to the file + containing the golden MD5 checksum. + - `decoder_plugin` (str): The video decoder to use, e.g., + "v4l2vp8dec", "v4l2vp9dec". + - `color_space` (str): The desired color space format for the + output, e.g., "I420", "NV12". + + :raises SystemExit: + If the golden sample file or the golden MD5 checksum file does not + exist, or if the extracted MD5 checksum does not match the golden MD5 + checksum. + """ + # Check the golden sample and golden MD5 checksum exixt + if not os.path.exists(args.golden_sample_path): + raise SystemExit( + "Golden Sample '{}' doesn't exist".format(args.golden_sample_path) + ) + if not os.path.exists(args.golden_sample_md5_checksum_path): + raise SystemExit( + "Golden Sample's MD5 checksum '{}' doesn't exist".format( + args.golden_sample_md5_checksum_path + ) + ) + # Run command to get comapred md5 checksum by consuming golden sample + 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, + color_sapce=args.color_space, + ) + compared_md5_data = get_md5_checksum_from_command(cmd).rstrip(os.linesep) + + logging.info( + "===== MD5 Checksum: {} ====\n{}\n".format( + args.golden_sample_path, compared_md5_data + ) + ) + # Read the Golden Sample's MD5 checksum and compare it + # with compared_md5_data data + with open( + args.golden_sample_md5_checksum_path, mode="r", encoding="UTF-8" + ) as gf: + golden_content = gf.read().rstrip(os.linesep) + if golden_content == compared_md5_data: + logging.info("Pass. MD5 checksum is same as Golden Sample") + else: + logging.info( + "===== Golden MD5 Checksum: {} ====\n{}\n".format( + args.golden_sample_md5_checksum_path, golden_content + ) + ) + raise SystemExit( + "Failed. MD5 checksum is not same as Golden Sample" + ) + + +def main() -> None: + args = register_arguments() + validate_video_decoder_md5_checksum(args) + + +if __name__ == "__main__": + main() diff --git a/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/data/gstreamer-test-confs/genio-1200.json b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/data/gstreamer-test-confs/genio-1200.json new file mode 100644 index 000000000..40d45a515 --- /dev/null +++ b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/data/gstreamer-test-confs/genio-1200.json @@ -0,0 +1,29 @@ +{ + "gst_v4l2_video_decoder_md5_checksum_comparison": [ + { + "decoder_plugin": "v4l2h264dec", + "resolutions": [{"width": 320, "height": 320}, {"width": 1920, "height": 1080}], + "color_spaces": ["NV12", "NV21", "ARGB", "RGB", "I420", "YV12"], + "source_format": "mp4" + }, + { + "decoder_plugin": "v4l2h265dec", + "resolutions": [{"width": 320, "height": 320}, {"width": 1920, "height": 1080}], + "color_spaces": ["NV12", "NV21", "ARGB", "RGB", "I420", "YV12"], + "source_format": "mp4" + }, + { + "decoder_plugin": "v4l2vp8dec", + "resolutions": [{"width": 320, "height": 320}, {"width": 1920, "height": 1080}], + "color_spaces": ["YUV"], + "source_format": "webm" + }, + { + "decoder_plugin": "v4l2vp9dec", + "resolutions": [{"width": 320, "height": 320}, {"width": 1920, "height": 1080}], + "color_spaces": ["YUV"], + "source_format": "webm" + } + ] +} + diff --git a/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/tests/test_gst_resources_generator.py b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/tests/test_gst_resources_generator.py new file mode 100755 index 000000000..e47947daf --- /dev/null +++ b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/tests/test_gst_resources_generator.py @@ -0,0 +1,228 @@ +import io +import unittest +from unittest.mock import patch, mock_open, MagicMock, call +from gst_resources_generator import ( + load_json_config, + GstResources +) + + +class TestLoadJsonConfig(unittest.TestCase): + @patch('pathlib.Path.exists') + @patch('pathlib.Path.open', new_callable=mock_open) + @patch('json.load') + def test_load_json_config( + self, + mock_json_load, + mock_open, + mock_path_exists + ): + mock_path_exists.return_value = True + mock_json_load.return_value = {'key': 'value'} + json_path = "/fake.json" + data = load_json_config(json_path) + mock_path_exists.assert_called_once_with() + mock_open.assert_called_once_with('r') + mock_json_load.assert_called_once_with( + mock_open.return_value.__enter__.return_value) + self.assertEqual(data, {'key': 'value'}) + + @patch('pathlib.Path.open', new_callable=mock_open) + @patch('pathlib.Path.exists') + def test_load_json_config_file_not_found(self, mock_exists, mock_open): + mock_exists.return_value = False + with self.assertRaises(FileNotFoundError): + load_json_config('/fake.json') + mock_open.assert_not_called() + + +class TestGstResources(unittest.TestCase): + def test_gst_v4l2_video_decoder_md5_checksum_comparison_helper(self): + args = MagicMock() + args.gst_testing_data_path = "/fake-path" + args.gst_conf_name = "platform_conf" + + ins = GstResources(args, {}) + ins._current_scenario_name = "scenario1" + + expected_result = { + "scenario": "scenario1", + "decoder_plugin": "plugin1", + "width": 1920, + "height": 1080, + "color_space": "rgb", + "golden_sample_file": ( + "/fake-path/video_golden_samples" + "/1920x1080-plugin1-rgb.mp4" + ), + "golden_md5_checkum_file": ( + "/fake-path/scenario1/" + "golden_md5_checksum/platform_conf/1920x1080-plugin1-rgb.md5" + ) + } + + result = ins._gst_v4l2_video_decoder_md5_checksum_comparison_helper( + decoder_plugin="plugin1", + width=1920, + height=1080, + color_space="rgb", + source_format="mp4" + ) + + self.assertEqual(result, expected_result) + + @patch.object( + GstResources, + '_gst_v4l2_video_decoder_md5_checksum_comparison_helper') + def test_gst_v4l2_video_decoder_md5_checksum_comparison_multi_color_space( + self, + mock_helper + ): + args = MagicMock() + ins = GstResources(args, {}) + + scenario_data = [ + { + "decoder_plugin": "plugin1", + "resolutions": [{"width": 1920, "height": 1080}], + "color_spaces": ["rgb", "yuv"], + "source_format": "mp4" + } + ] + + ins.gst_v4l2_video_decoder_md5_checksum_comparison(scenario_data) + + mock_helper.assert_has_calls([ + call( + decoder_plugin="plugin1", + width=1920, + height=1080, + color_space="rgb", + source_format="mp4" + ), + call( + decoder_plugin="plugin1", + width=1920, + height=1080, + color_space="yuv", + source_format="mp4" + ) + ]) + + @patch.object( + GstResources, + '_gst_v4l2_video_decoder_md5_checksum_comparison_helper') + def test_gst_v4l2_video_decoder_md5_checksum_comparison_multi_resolutions( + self, + mock_helper + ): + args = MagicMock() + ins = GstResources(args, {}) + + scenario_data = [ + { + "decoder_plugin": "plugin1", + "resolutions": [ + {"width": 320, "height": 320}, + {"width": 1920, "height": 1080} + ], + "color_spaces": ["NV12"], + "source_format": "mp4" + } + ] + + ins.gst_v4l2_video_decoder_md5_checksum_comparison(scenario_data) + + mock_helper.assert_has_calls([ + call( + decoder_plugin="plugin1", + width=320, + height=320, + color_space="NV12", + source_format="mp4" + ), + call( + decoder_plugin="plugin1", + width=1920, + height=1080, + color_space="NV12", + source_format="mp4" + ) + ]) + + @patch.object( + GstResources, + '_gst_v4l2_video_decoder_md5_checksum_comparison_helper') + def test_gst_v4l2_video_decoder_md5_checksum_comparison_multi_decoders( + self, + mock_helper + ): + args = MagicMock() + ins = GstResources(args, {}) + + scenario_data = [ + { + "decoder_plugin": "plugin1", + "resolutions": [ + {"width": 320, "height": 320}, + ], + "color_spaces": ["NV12"], + "source_format": "mp4" + }, + { + "decoder_plugin": "plugin2", + "resolutions": [ + {"width": 3840, "height": 2160}, + ], + "color_spaces": ["ABC"], + "source_format": "webm" + } + ] + + ins.gst_v4l2_video_decoder_md5_checksum_comparison(scenario_data) + + mock_helper.assert_has_calls([ + call( + decoder_plugin="plugin1", + width=320, + height=320, + color_space="NV12", + source_format="mp4" + ), + call( + decoder_plugin="plugin2", + width=3840, + height=2160, + color_space="ABC", + source_format="webm" + ) + ]) + + @patch('sys.stdout', new_callable=io.StringIO) + def test_dump_resources(self, mock_stdout): + args = MagicMock() + ins = GstResources(args, {}) + ins._resource_items = [ + { + "key1": "value1", + "foo1": "bar1" + }, + { + "key2": "value2", + "foo2": "bar2", + "good2": "bad2", + }, + ] + + ins._dump_resources() + + expected_output = ( + "key1: value1\nfoo1: bar1\n\n" + "key2: value2\nfoo2: bar2\ngood2: bad2\n\n\n") + + self.assertEqual(mock_stdout.getvalue(), expected_output) + self.assertEqual(ins._resource_items, []) + + +if __name__ == '__main__': + unittest.main() diff --git a/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/tests/test_gst_v4l2_video_decoder_md5_checksum_comparison.py b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/tests/test_gst_v4l2_video_decoder_md5_checksum_comparison.py new file mode 100644 index 000000000..bc64260a3 --- /dev/null +++ b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/tests/test_gst_v4l2_video_decoder_md5_checksum_comparison.py @@ -0,0 +1,327 @@ +import logging +import sys +import subprocess +import unittest +from io import StringIO +from unittest.mock import patch, MagicMock, mock_open, call +from gst_v4l2_video_decoder_md5_checksum_comparison import ( + build_gst_command, + extract_the_md5_checksum, + get_md5_checksum_from_command, + validate_video_decoder_md5_checksum +) + + +class TestBuildGstCommand(unittest.TestCase): + def test_build_gst_command_vp8_vp9_decoder(self): + # Test case 1: VP8/VP9 decoder + gst_bin = "gst-launch-1.0" + golden_sample_path = "/path/to/golden/sample.mp4" + decoder = "v4l2vp8dec" + color_space = "I420" + + expected_command = ( + "gst-launch-1.0 -v filesrc location=/path/to/golden/sample.mp4 ! " + "parsebin ! v4l2vp8dec ! v4l2convert ! " + "checksumsink hash=0 sync=false" + ) + command = build_gst_command( + gst_bin, golden_sample_path, decoder, color_space + ) + self.assertEqual(command, expected_command) + + def test_build_gst_command_other_decoder(self): + gst_bin = "gst-launch-1.0" + golden_sample_path = "/path/to/golden/sample.mp4" + decoder = "v4l2h264dec" + color_space = "NV12" + + expected_command = ( + "gst-launch-1.0 -v filesrc location=/path/to/golden/sample.mp4 ! " + "parsebin ! v4l2h264dec ! v4l2convert ! " + "video/x-raw,format=NV12 ! checksumsink hash=0 sync=false" + ) + command = build_gst_command( + gst_bin, golden_sample_path, decoder, color_space + ) + self.assertEqual(command, expected_command) + + +class TestExtractTheMD5Checksum(unittest.TestCase): + def test_extract_the_md5_checksum_multiple(self): + input_str = """ +Pipeline is PREROLLED ... +Setting pipeline to PLAYING ... +Redistribute latency... +New clock: GstSystemClock +0:00:00.000000000 00402d37c1a1c9a887cf8c06e1046489 +0:00:00.033333333 52e92b8dbafc0ed038e89d0196326c57 +0:00:00.066666666 6f9edba153b10ffc17a14ac0b8eade4f +0:00:00.100000000 2981ace12393b0e89b1e0a44698c5df8 +0:00:00.133333333 6286c1207577e76dc690715669a4d890 +Got EOS from element "pipeline0". +Execution ended after 0:00:00.768310428 +Setting pipeline to NULL ... +Freeing pipeline ... +""" + expected_output = "\n".join([ + "00402d37c1a1c9a887cf8c06e1046489", + "52e92b8dbafc0ed038e89d0196326c57", + "6f9edba153b10ffc17a14ac0b8eade4f", + "2981ace12393b0e89b1e0a44698c5df8", + "6286c1207577e76dc690715669a4d890\n", + ]) + actual_output = extract_the_md5_checksum(input_str) + self.assertEqual(actual_output, expected_output) + + def test_extract_the_md5_checksum_none(self): + input_str = "This is a string without any MD5 checksums." + expected_output = "" + actual_output = extract_the_md5_checksum(input_str) + self.assertEqual(actual_output, expected_output) + + +class TestGetMD5ChecksumFromCommand(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.original_stdout = sys.stdout + suppress_text = StringIO() + sys.stdout = suppress_text + logging.disable(logging.CRITICAL) + + @classmethod + def tearDownClass(cls): + sys.stdout = cls.original_stdout + logging.disable(logging.NOTSET) + + @patch( + "gst_v4l2_video_decoder_md5_checksum_comparison." + "extract_the_md5_checksum" + ) + @patch("subprocess.run") + def test_get_md5_checksum_from_command_success( + self, + mock_subprocess_run, + mock_extract_the_md5_checksum + ): + cmd = "123 success command" + expected_md5_checksum = "fake md5 12345abcf35" + mock_extract_the_md5_checksum.return_value = expected_md5_checksum + + mock_subprocess_run.return_value = MagicMock( + returncode=0, + stdout="0:00:00.000000000 = fake md5 12345abcf35" + ) + + md5_checksum = get_md5_checksum_from_command(cmd) + + self.assertEqual(md5_checksum, expected_md5_checksum) + mock_extract_the_md5_checksum.assert_called_once_with( + mock_subprocess_run.return_value.stdout + ) + mock_subprocess_run.assert_called_once_with( + ["123", "success", "command"], + check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + encoding="utf-8", timeout=30 + ) + + @patch("subprocess.run") + def test_get_md5_checksum_from_command_failure(self, mock_subprocess_run): + cmd = "Failure Command" + + mock_subprocess_run.side_effect = subprocess.CalledProcessError( + returncode=100, cmd=cmd, stderr="Command failed" + ) + + with self.assertRaises(SystemExit) as cm: + get_md5_checksum_from_command(cmd) + self.assertEqual(cm.exception.code, 100) + mock_subprocess_run.assert_called_once_with( + ["Failure", "Command"], + check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + encoding="utf-8", timeout=30 + ) + + +class TestValidateVideoDecoderMD5Checksum(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.original_stdout = sys.stdout + suppress_text = StringIO() + sys.stdout = suppress_text + logging.disable(logging.CRITICAL) + + @classmethod + def tearDownClass(cls): + sys.stdout = cls.original_stdout + logging.disable(logging.NOTSET) + + @patch("os.path.exists") + @patch( + "gst_v4l2_video_decoder_md5_checksum_comparison." + "build_gst_command" + ) + @patch( + "gst_v4l2_video_decoder_md5_checksum_comparison." + "get_md5_checksum_from_command" + ) + @patch( + "builtins.open", + new_callable=mock_open, + read_data="golden_md5_checksum" + ) + def test_validate_video_decoder_md5_checksum_success( + self, + mock_open, + mock_get_md5_checksum_from_command, + mock_build_gst_command, + mock_os_path_exists + ): + args = type("", (), { + "golden_sample_path": "golden_sample.mp4", + "golden_sample_md5_checksum_path": "my_test.md5", + "decoder_plugin": "fake_decoder", + "color_space": "NN" + })() + + mock_os_path_exists.side_effect = [True, True] + mock_get_md5_checksum_from_command.return_value = "golden_md5_checksum" + mock_build_gst_command.return_value = "my testing command" + + validate_video_decoder_md5_checksum(args) + + mock_os_path_exists.assert_has_calls([ + call("golden_sample.mp4"), + call("my_test.md5") + ]) + mock_get_md5_checksum_from_command.assert_called_once_with( + "my testing command" + ) + mock_open.assert_called_once_with( + "my_test.md5", mode="r", encoding="UTF-8") + + @patch("os.path.exists") + @patch( + "gst_v4l2_video_decoder_md5_checksum_comparison." + "get_md5_checksum_from_command" + ) + @patch("builtins.open", new_callable=mock_open) + def test_validate_video_decoder_md5_checksum_golden_sample_not_found( + self, + mock_open, + mock_get_md5_checksum_from_command, + mock_os_path_exists + ): + # Arrange + args = type("", (), { + "golden_sample_path": "non_exist_golden_sample.mp4", + "golden_sample_md5_checksum_path": "golden_sample.md5", + "decoder_plugin": "fake_decoder", + "color_space": "NN" + })() + + mock_os_path_exists.side_effect = [False] + + # Act and Assert + with self.assertRaises(SystemExit) as cm: + validate_video_decoder_md5_checksum(args) + self.assertEqual( + cm.exception.code, + "Golden Sample 'non_exist_golden_sample.mp4' doesn't exist" + ) + mock_os_path_exists.assert_has_calls([ + call("non_exist_golden_sample.mp4") + ]) + mock_get_md5_checksum_from_command.assert_not_called() + mock_open.assert_not_called() + + @patch("os.path.exists") + @patch( + "gst_v4l2_video_decoder_md5_checksum_comparison." + "get_md5_checksum_from_command" + ) + @patch("builtins.open", new_callable=mock_open) + def test_validate_video_decoder_md5_checksum_golden_md5_checksum_not_found( + self, + mock_open, + mock_get_md5_checksum_from_command, + mock_os_path_exists + ): + args = type("", (), { + "golden_sample_path": "golden_sample.mp4", + "golden_sample_md5_checksum_path": "non_exist_golden_sample.md5", + "decoder_plugin": "fake_decoder", + "color_space": "NN" + })() + + mock_os_path_exists.side_effect = [True, False] + + with self.assertRaises(SystemExit) as cm: + validate_video_decoder_md5_checksum(args) + self.assertEqual( + cm.exception.code, + ("Golden Sample's MD5 checksum 'non_exist_golden_sample.md5'" + " doesn't exist") + ) + mock_os_path_exists.assert_has_calls([ + call("golden_sample.mp4"), + call("non_exist_golden_sample.md5") + ]) + mock_get_md5_checksum_from_command.assert_not_called() + mock_open.assert_not_called() + + @patch("os.path.exists") + @patch( + "gst_v4l2_video_decoder_md5_checksum_comparison." + "build_gst_command" + ) + @patch( + "gst_v4l2_video_decoder_md5_checksum_comparison." + "get_md5_checksum_from_command" + ) + @patch( + "builtins.open", + new_callable=mock_open, + read_data="different_golden_md5_checksum" + ) + def test_validate_video_decoder_md5_checksum_failure( + self, + mock_open, + mock_get_md5_checksum_from_command, + mock_build_gst_command, + mock_os_path_exists + ): + args = type("", (), { + "golden_sample_path": "golden_sample.mp4", + "golden_sample_md5_checksum_path": "golden_sample.md5", + "decoder_plugin": "fake_decoder", + "color_space": "NN" + })() + + mock_os_path_exists.side_effect = [True, True] + mock_get_md5_checksum_from_command.return_value = "md5_checksum" + mock_build_gst_command.return_value = "my testing command" + + # Act and Assert + with self.assertRaises(SystemExit) as cm: + validate_video_decoder_md5_checksum(args) + self.assertEqual( + cm.exception.code, + "Failed. MD5 checksum is not same as Golden Sample" + ) + mock_os_path_exists.assert_has_calls([ + call("golden_sample.mp4"), + call("golden_sample.md5") + ]) + mock_get_md5_checksum_from_command.assert_called_once_with( + "my testing command" + ) + mock_open.assert_called_once_with( + "golden_sample.md5", + mode="r", + encoding="UTF-8" + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gstreamer/README.md b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gstreamer/README.md new file mode 100644 index 000000000..83fbc5d5c --- /dev/null +++ b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gstreamer/README.md @@ -0,0 +1,17 @@ +# Readme for Gstreamer-Related Jobs + +This readme provides an overview of the different gstreamer tests available in this project + +## Usage + +Before starting the testing, please read the [OQ013 - Gstreamer Testing Document](https://docs.google.com/document/d/1yuAdse3u64QZGCL2VQ4_PpuPIC0i1yXqHxKI6660WFg/edit?usp=sharing) to understand the overall testing process and the scenarios that interest you. + +## Scenarios + +### Scenario: gst_v4l2_video_decoder_md5_checksum_comparison + +#### Goal + + The purpose of this scenario is to use MD5 checksum comparison to ensure that Gstreamer Video-related decoders, under different combinations (decoder plugin, resolution, and color space), produce MD5 checksums that match those of the Golden Sample. + + Please reference [v4l2_video_decoder_md5_checksum_comparison](https://docs.google.com/document/d/1yuAdse3u64QZGCL2VQ4_PpuPIC0i1yXqHxKI6660WFg/edit#heading=h.rh805u3vq3ig) to learn the detail. diff --git a/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gstreamer/category.pxu b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gstreamer/category.pxu new file mode 100644 index 000000000..3e8596897 --- /dev/null +++ b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gstreamer/category.pxu @@ -0,0 +1,3 @@ +unit: category +id: gstreamer +_name: Gstreamer tests diff --git a/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gstreamer/jobs.pxu b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gstreamer/jobs.pxu new file mode 100644 index 000000000..f499c8cca --- /dev/null +++ b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gstreamer/jobs.pxu @@ -0,0 +1,26 @@ +id: ce-oem-gstreamer/resource-gst_v4l2_video_decoder_md5_checksum_comparison +_summary: Generates mappings for gst_v4l2_video_decoder_md5_checksum_comparison +_description: + Generate resource for gst_v4l2_video_decoder_md5_checksum_comparison scenario. +estimated_duration: 0.02 +category_id: gstreamer +plugin: resource +environ: GSTREAMER_JSON_CONFIG_NAME GSTREAMER_TESTING_DATA PLAINBOX_PROVIDER_DATA +command: + gst_resources_generator.py "$GSTREAMER_JSON_CONFIG_NAME" -gtdp "$GSTREAMER_TESTING_DATA" + +unit: template +template-engine: jinja2 +template-resource: ce-oem-gstreamer/resource-gst_v4l2_video_decoder_md5_checksum_comparison +template-unit: job +template-id: ce-oem-gstreamer/gst_v4l2_video_decoder_md5_checksum_comparison +id: ce-oem-gstreamer/gst_v4l2_video_decoder_md5_checksum_comparison-{{ decoder_plugin }}-{{ width }}x{{ height }}-{{ color_space }} +_summary: Compare the MD5 checksum to golden reference under {{ decoder_plugin }} {{ width }}x{{ height }} {{ color_space }} +plugin: shell +category_id: gstreamer +estimated_duration: 1s +requires: manifest.has_gstreamer == 'True' +flags: also-after-suspend +environ: GST_LAUNCH_BIN +command: + gst_v4l2_video_decoder_md5_checksum_comparison.py -dp {{decoder_plugin}} -cs {{color_space}} -gp {{golden_sample_file}} -gmp {{golden_md5_checkum_file}} diff --git a/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gstreamer/manifest.pxu b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gstreamer/manifest.pxu new file mode 100644 index 000000000..e27857978 --- /dev/null +++ b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gstreamer/manifest.pxu @@ -0,0 +1,4 @@ +unit: manifest entry +id: has_gstreamer +_name: Does platform support Gstreamer? +value-type: bool diff --git a/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gstreamer/test-plan.pxu b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gstreamer/test-plan.pxu new file mode 100644 index 000000000..f8991ae08 --- /dev/null +++ b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gstreamer/test-plan.pxu @@ -0,0 +1,23 @@ +id: ce-oem-gstreamer-full +unit: test plan +_name: Gstreamer tests +_description: Full tests for Gstreamer +include: +nested_part: + ce-oem-gstreamer-manual + ce-oem-gstreamer-automated + +id: ce-oem-gstreamer-manual +unit: test plan +_name: Gstreamer manual tests +_description: Manual tests for Gstreamer in before suspend and post suspend stage +include: + +id: ce-oem-gstreamer-automated +unit: test plan +_name: Gstreamer auto tests +_description: Automated tests for Gstreamer in before suspend and post suspend stage +bootstrap_include: + ce-oem-gstreamer/resource-gst_v4l2_video_decoder_md5_checksum_comparison +include: + ce-oem-gstreamer/gst_v4l2_video_decoder_md5_checksum_comparison