validate: Add a test manager to run itu h264 conformance testsuites

validate/launcher/apps/gstituh26xconformance.py

+#!/usr/bin/env python3
+#
+# Copyright (c) 2020,Thibault Saunier <tsaunier@igalia.com>
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This program 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
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this program; if not, write to the
+# Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+# Boston, MA 02110-1301, USA.
+import os
+import glob
+import hashlib
+from .apps.gstvalidate import GstValidateTestManager, GstValidateLaunchTest, GstValidatePipelineTestsGenerator
+from launcher.utils import get_fakesink_for_media_type
+
+
+H264_PIPELINE = 'filesrc location=%s ! h264parse ! %s ! video/x-raw,format=I420 ! %s'
+
+
+class GstITUH26XConformanceTest(GstValidateLaunchTest):
+    def get_subproc_env(self):
+        # For some reason doting the pipeline changes the md5sum oO
+        env = super().get_subproc_env()
+        del env['GST_DEBUG_DUMP_DOT_DIR']
+
+        return env
+
+
+class GstITUH26XConformance(GstValidateTestManager):
+
+    name = "ituh26x"
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.extra_data = {}
+        self.h264_dirs = []
+
+    def init(self):
+        return True
+
+    def add_options(self, parser):
+        group = parser.add_argument_group()
+        group.add_argument("--ituh26x-h264-decoder", dest="ituh26x_h264_decoder", default="avdec_h264")
+
+    def get_checksum(self, f):
+        _dir = os.path.dirname(f)
+        name, _ = os.path.splitext(os.path.basename(f))
+        files = []
+        for f in glob.glob(os.path.join(_dir, "*.yuv")):
+            if os.path.basename(f).startswith(name):
+                files.append(f)
+
+        if len(files) != 1:
+            return None
+
+        hash_md5 = hashlib.md5()
+        with open(files[0], "rb") as f:
+            for chunk in iter(lambda: f.read(4096), b""):
+                hash_md5.update(chunk)
+        return hash_md5.hexdigest()
+
+    def list_tests(self):
+        if self.tests:
+            return self.tests
+
+        for _dir in self.h264_dirs:
+            for root, _, files in os.walk(_dir):
+                for f in files:
+                    fname, ext = os.path.splitext(f)
+                    is_h264 = ext in ['.264', '.jsv', '.26l', '.avc', '.h264']
+                    file_path = os.path.join(root, f)
+                    if not is_h264:
+                        continue
+                    md5 = self.get_checksum(file_path)
+                    if not md5:
+                        if self.options.mute:
+                            sink = get_fakesink_for_media_type("video", False)
+                        else:
+                            sink = 'autovideosink'
+                    else:
+                        sink = "testsink expected-md5=" + md5
+
+                    sink += " name=sink"
+
+                    pipe = H264_PIPELINE % (file_path, self.options.ituh26x_h264_decoder, sink)
+                    test_private_dir = os.path.join(self.options.privatedir, self.name, fname)
+
+                    test = GstITUH26XConformanceTest("%s" % fname, self.options, self.reporter, pipe,)
+                    config_file = GstValidatePipelineTestsGenerator.get_config_file(
+                        ["%(validateflow)s, pad=sink:sink, record-buffers=true, buffers-checksum=true"],
+                        test_private_dir, fname, self.extra_data)
+                    test.add_validate_config(config_file)
+                    self.add_test(test)
+
+        return self.tests