gstreamer/subprojects/gst-editing-services/tests/validate/geslaunch.py

#!/usr/bin/env python3
#
# Copyright (c) 2013, Thibault Saunier <thibault.saunier@collabora.com>
# Copyright (c) 2020, Igalia S.L
#    Author: 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 sys
import tempfile
import urllib.parse
import subprocess
from launcher import utils
from urllib.parse import unquote
import xml.etree.ElementTree as ET
from launcher.baseclasses import GstValidateTest, TestsManager, ScenarioManager, MediaFormatCombination, \
    MediaDescriptor, GstValidateEncodingTestInterface

GES_DURATION_TOLERANCE = utils.GST_SECOND / 2

GES_LAUNCH_COMMAND = "ges-launch-1.0"
if "win32" in sys.platform:
    GES_LAUNCH_COMMAND += ".exe"


GES_ENCODING_TARGET_COMBINATIONS = [
    MediaFormatCombination("ogg", "vorbis", "theora"),
    MediaFormatCombination("ogg", "opus", "theora"),
    MediaFormatCombination("webm", "vorbis", "vp8"),
    MediaFormatCombination("webm", "opus", "vp8"),
    MediaFormatCombination("mp4", "aac", "h264"),
    MediaFormatCombination("mp4", "ac3", "h264"),
    MediaFormatCombination("quicktime", "aac", "jpeg"),
    MediaFormatCombination("mkv", "opus", "h264"),
    MediaFormatCombination("mkv", "vorbis", "h264"),
    MediaFormatCombination("mkv", "opus", "jpeg"),
    MediaFormatCombination("mkv", "vorbis", "jpeg")
]


def quote_uri(uri):
    """
    Encode a URI/path according to RFC 2396, without touching the file:/// part.
    """
    # Split off the "file:///" part, if present.
    parts = urllib.parse.urlsplit(uri, allow_fragments=False)
    # Make absolutely sure the string is unquoted before quoting again!
    raw_path = unquote(parts.path)
    return utils.path2url(raw_path)


class XgesProjectDescriptor(MediaDescriptor):
    def __init__(self, uri):
        super(XgesProjectDescriptor, self).__init__()

        self._uri = uri
        self._xml_path = utils.url2path(uri)
        self._root = ET.parse(self._xml_path)
        self._duration = None

    def get_media_filepath(self):
        return self._xml_path

    def get_path(self):
        return self._xml_path

    def get_caps(self):
        raise NotImplemented

    def get_uri(self):
        return self._uri

    def get_duration(self):
        if self._duration:
            return self._duration

        for l in self._root.iter():
            if l.tag == "timeline":
                self._duration=int(l.attrib['metadatas'].split("duration=(guint64)")[1].split(" ")[0].split(";")[0])
                break

        if not self._duration:
            self.error("%s does not have duration! (setting 2mins)" % self._uri)
            self._duration = 2 * 60

        return self._duration

    def get_protocol(self):
        return Protocols.FILE

    def is_seekable(self):
        return True

    def is_image(self):
        return False

    def get_num_tracks(self, track_type):
        num_tracks = 0
        for l in self._root.iter():
            if l.tag == "track":
                if track_type in l.attrib["caps"]:
                    num_tracks += 1
        return num_tracks


class GESTest(GstValidateTest):
    def __init__(self, classname, options, reporter, project, scenario=None,
                 combination=None, expected_failures=None, nest=False, testfile=None):

        super(GESTest, self).__init__(GES_LAUNCH_COMMAND, classname, options, reporter,
                                      scenario=scenario)

        self.project = project
        self.nested = nest
        self.testfile = testfile

    def set_sample_paths(self):
        if self.testfile:
            # testfile should be self contained
            return

        if not self.options.paths:
            if self.options.disable_recurse:
                return
            if self.project:
                paths = [os.path.dirname(self.project.get_media_filepath())]
            else:
                paths = []
        else:
            paths = self.options.paths

        if not isinstance(paths, list):
            paths = [paths]

        for path in paths:
            # We always want paths separator to be cut with '/' for ges-launch
            path = path.replace("\\", "/")
            if not self.options.disable_recurse:
                self.add_arguments("--ges-sample-path-recurse", quote_uri(path))
                self.add_arguments("--ges-sample-path-recurse", quote_uri(self.options.projects_paths))
            else:
                self.add_arguments("--ges-sample-paths", quote_uri(path))

    def set_sink_args(self):
        if self.testfile:
            # testfile should be self contained and --mute should give required infos.
            if self.options.mute:
                self.add_arguments("--mute")
            return

        if self.options.mute:
            needs_clock = self.scenario.needs_clock_sync() \
                if self.scenario else False
            audiosink = utils.get_fakesink_for_media_type("audio", needs_clock)
            videosink = utils.get_fakesink_for_media_type("video", needs_clock)
        else:
            audiosink = 'autoaudiosink'
            videosink = 'autovideosink'
        self.add_arguments("--videosink", videosink + " name=videosink")
        self.add_arguments("--audiosink", audiosink + " name=audiosink")

    def build_arguments(self):
        GstValidateTest.build_arguments(self)

        self.set_sink_args()
        self.set_sample_paths()

        if self.project:
            assert self.testfile is None
            if self.nested:
                self.add_arguments("+clip", self.project.get_uri())
            else:
                self.add_arguments("-l", self.project.get_uri())
        elif self.testfile:
            self.add_arguments("--set-test-file", self.testfile)

class GESPlaybackTest(GESTest):
    def __init__(self, classname, options, reporter, project, scenario,nest):
        super(GESPlaybackTest, self).__init__(classname, options, reporter,
                                      project, scenario=scenario, nest=nest)

    def get_current_value(self):
        return self.get_current_position()

class GESScenarioTest(GESTest):
    def __init__(self, classname, options, reporter, scenario):
        super().__init__(classname, options, reporter, None, scenario=scenario)

    def build_arguments(self):
        super().build_arguments()
        self.add_arguments("--set-scenario", self.scenario.path)

    def get_subproc_env(self):
        scenario = self.scenario
        self.scenario = None
        res = super().get_subproc_env()
        self.scenario = scenario

        return res

    def get_current_value(self):
        return self.get_current_position()


class GESRenderTest(GESTest, GstValidateEncodingTestInterface):
    def __init__(self, classname, options, reporter, project, combination):
        GESTest.__init__(self, classname, options, reporter, project)

        GstValidateEncodingTestInterface.__init__(self, combination, self.project)

    def build_arguments(self):
        GESTest.build_arguments(self)
        self._set_rendering_info()

    def run_external_checks(self):
        reference_file_path = urllib.parse.urlsplit(self.media_descriptor.get_uri()).path + ".expected_result"
        if os.path.exists(reference_file_path):
            self.run_iqa_test(utils.path2url(reference_file_path))

    def _set_rendering_info(self):
        self.dest_file = path = os.path.join(self.options.dest,
                                             self.classname.replace(".render.", os.sep).
                                             replace(".", os.sep))
        utils.mkdir(os.path.dirname(urllib.parse.urlsplit(self.dest_file).path))
        if not utils.isuri(self.dest_file):
            self.dest_file = utils.path2url(self.dest_file)

        profile = self.get_profile()
        self.add_arguments("-f", profile, "-o", self.dest_file)

    def check_results(self):
        self.check_encoded_file()
        return GstValidateTest.check_results(self)

    def get_current_value(self):
        size = self.get_current_size()
        if size is None:
            return self.get_current_position()

        return size


class GESTestsManager(TestsManager):
    name = "ges"

    _scenarios = ScenarioManager()

    def __init__(self):
        super(GESTestsManager, self).__init__()

    def init(self):
        try:
            with tempfile.NamedTemporaryFile() as f:
                if "--set-scenario=" in subprocess.check_output([GES_LAUNCH_COMMAND, "--help"], stderr=f).decode():
                    return True
                else:
                    self.warning("Can not use ges-launch, it seems not to be compiled against"
                                " gst-validate")
        except subprocess.CalledProcessError as e:
            self.warning("Can not use ges-launch: %s" % e)
        except OSError as e:
            self.warning("Can not use ges-launch: %s" % e)

    def add_options(self, parser):
        group = parser.add_argument_group("GStreamer Editing Services specific option"
                            " and behaviours",
                            description="""
The GStreamer Editing Services launcher will be usable only if GES has been compiled against GstValidate
You can simply run scenarios specifying project as args. For example the following will run all available
and activated scenarios on project.xges:

    $gst-validate-launcher ges /some/ges/project.xges


Available options:""")
        group.add_argument("-P", "--projects-paths", dest="projects_paths",
                         default=os.path.join(utils.DEFAULT_GST_QA_ASSETS,
                                              "ges",
                                              "ges-projects"),
                         help="Paths in which to look for moved medias")
        group.add_argument("--ges-scenario-paths", dest="scenarios_path",
                         default=None,
                         help="Paths in which to look for moved medias")
        group.add_argument("-r", "--disable-recurse-paths", dest="disable_recurse",
                         default=False, action="store_true",
                         help="Whether to recurse into paths to find medias")

    def set_settings(self, options, args, reporter):
        TestsManager.set_settings(self, options, args, reporter)
        self._scenarios.config = self.options

        try:
            os.makedirs(utils.url2path(options.dest)[0])
        except OSError:
            pass

    def list_tests(self):
        return self.tests

    def register_defaults(self, project_paths=None, scenarios_path=None):
        projects = list()
        all_scenarios = {}
        if not self.args:
            if project_paths == None:
                path = self.options.projects_paths
            else:
                path = project_paths

            for root, dirs, files in os.walk(path):
                for f in files:
                    if not f.endswith(".xges"):
                        continue
                    projects.append(utils.path2url(os.path.join(path, root, f)))

            if self.options.scenarios_path:
                scenarios_path = self.options.scenarios_path

            if scenarios_path:
                for root, dirs, files in os.walk(scenarios_path):
                    for f in files:
                        name, ext = os.path.splitext(f)
                        f = os.path.join(root, f)
                        if ext == ".validatetest":
                            fpath = os.path.abspath(os.path.join(root, f))
                            pathname = os.path.abspath(os.path.join(root, name))
                            name = pathname.replace(os.path.commonpath([scenarios_path, root]), '').replace('/', '.')
                            self.add_test(GESTest('test' + name,
                                                  self.options,
                                                  self.reporter,
                                                  None,
                                                  testfile=fpath))
                            continue
                        elif ext != ".scenario":
                            continue
                        config = f + ".config"
                        if not os.path.exists(config):
                            config = None
                        all_scenarios[f] = config
        else:
            for proj_uri in self.args:
                if not utils.isuri(proj_uri):
                    proj_uri = utils.path2url(proj_uri)

                if os.path.exists(proj_uri):
                    projects.append(proj_uri)

        if self.options.long_limit != 0:
            scenarios = ["none",
                         "scrub_forward_seeking",
                         "scrub_backward_seeking"]
        else:
            scenarios = ["play_15s",
                         "scrub_forward_seeking_full",
                         "scrub_backward_seeking_full"]
        for proj_uri in projects:
            # First playback casses
            project = XgesProjectDescriptor(proj_uri)
            for scenario_name in scenarios:
                scenario = self._scenarios.get_scenario(scenario_name)
                if scenario is None:
                    continue

                if scenario.get_min_media_duration() >= (project.get_duration() / utils.GST_SECOND):
                    continue

                classname = "playback.%s.%s" % (scenario.name,
                                                    os.path.basename(proj_uri).replace(".xges", ""))
                self.add_test(GESPlaybackTest(classname,
                                              self.options,
                                              self.reporter,
                                              project,
                                              scenario=scenario,
                                              nest=False))
                #For nested timelines
                classname = "playback.nested.%s.%s" % (scenario.name,
                                                    os.path.basename(proj_uri).replace(".xges", ""))
                self.add_test(GESPlaybackTest(classname,
                                              self.options,
                                              self.reporter,
                                              project,
                                              scenario=scenario,
                                              nest=True))

            # And now rendering casses
            for comb in GES_ENCODING_TARGET_COMBINATIONS:
                classname = "render.%s.%s" % (str(comb).replace(' ', '_'),
                                                  os.path.splitext(os.path.basename(proj_uri))[0])
                self.add_test(GESRenderTest(classname, self.options,
                                            self.reporter, project,
                                            combination=comb)
                                  )
        if all_scenarios:
            for scenario in self._scenarios.discover_scenarios(list(all_scenarios.keys())):
                config = all_scenarios[scenario.path]
                classname = "scenario.%s" % scenario.name
                test = GESScenarioTest(classname, self.options, self.reporter, scenario=scenario)
                if config:
                    test.add_validate_config(config)
                self.add_test(test)