xref: /aosp_15_r20/external/openscreen/cast/standalone_e2e.py (revision 3f982cf4871df8771c9d4abe6e9a6f8d829b2736)

#!/usr/bin/env python3
# Copyright 2021 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""
This script is intended to cover end to end testing for the standalone sender
and receiver executables in cast. This ensures that the basic functionality of
these executables is not impaired, such as the TLS/UDP connections and encoding
and decoding video.
"""

import argparse
import os
import pathlib
import logging
import subprocess
import sys
import time
import unittest
import ssl
from collections import namedtuple

from enum import IntEnum, IntFlag
from urllib import request

# Environment variables that can be overridden to set test properties.
ROOT_ENVVAR = 'OPENSCREEN_ROOT_DIR'
BUILD_ENVVAR = 'OPENSCREEN_BUILD_DIR'
LIBAOM_ENVVAR = 'OPENSCREEN_HAVE_LIBAOM'

TEST_VIDEO_NAME = 'Contador_Glam.mp4'
# NOTE: we use the HTTP protocol instead of HTTPS due to certificate issues
# in the legacy urllib.request API.
TEST_VIDEO_URL = ('https://storage.googleapis.com/openscreen_standalone/' +
                  TEST_VIDEO_NAME)

PROCESS_TIMEOUT = 15  # seconds

# Open Screen test certificates expire after 3 days. We crop this slightly (by
# 8 hours) to account for potential errors in time calculations.
CERT_EXPIRY_AGE = (3 * 24 - 8) * 60 * 60

# These properties are based on compiled settings in Open Screen, and should
# not change without updating this file.
TEST_CERT_NAME = 'generated_root_cast_receiver.crt'
TEST_KEY_NAME = 'generated_root_cast_receiver.key'
SENDER_BINARY_NAME = 'cast_sender'
RECEIVER_BINARY_NAME = 'cast_receiver'

EXPECTED_RECEIVER_MESSAGES = [
    "CastService is running.", "Found codec: opus (known to FFMPEG as opus)",
    "Successfully negotiated a session, creating SDL players.",
    "Receivers are currently destroying, resetting SDL players."
]

class VideoCodec(IntEnum):
  """There are different messages printed by the receiver depending on the codec
  chosen. """
  Vp8 = 0
  Vp9 = 1
  Av1 = 2

VIDEO_CODEC_SPECIFIC_RECEIVER_MESSAGES = [
  "Found codec: vp8 (known to FFMPEG as vp8)",
  "Found codec: vp9 (known to FFMPEG as vp9)",
  "Found codec: libaom-av1 (known to FFMPEG as av1)"
]

EXPECTED_SENDER_MESSAGES = [
    "Launching Mirroring App on the Cast Receiver",
    "Max allowed media bitrate (audio + video) will be",
    "Contador_Glam.mp4 (starts in one second)...",
    "The video capturer has reached the end of the media stream.",
    "The audio capturer has reached the end of the media stream.",
    "Video complete. Exiting...", "Shutting down..."
]

MISSING_LOG_MESSAGE = """Missing an expected message from either the sender
or receiver. This either means that one of the binaries misbehaved, or you
changed or deleted one of the log messages used for validation. Please ensure
that the necessary log messages are left unchanged, or update this
test suite's expectations."""

DESCRIPTION = """Runs end to end tests for the standalone Cast Streaming sender
and receiver. By default, this script assumes it is being ran from a current
working directory inside Open Screen's source directory, and uses
<root_dir>/out/Default as the build directory. To override these, set the
OPENSCREEN_ROOT_DIR and OPENSCREEN_BUILD_DIR environment variables. If the root
directory is set and the build directory is not,
<OPENSCREEN_ROOT_DIR>/out/Default will be used. In addition, if LibAOM is
installed, one can choose to run AV1 tests by defining the
OPENSCREEN_HAVE_LIBAOM environment variable.

See below for the the help output generated by the `unittest` package."""


def _set_log_level(is_verbose):
    """Sets the logging level, either DEBUG or ERROR as appropriate."""
    level = logging.DEBUG if is_verbose else logging.INFO
    logging.basicConfig(stream=sys.stdout, level=level)


def _get_loopback_adapter_name():
    """Retrieves the name of the loopback adapter (lo on Linux/lo0 on Mac)."""
    if sys.platform == 'linux' or sys.platform == 'linux2':
        return 'lo'
    if sys.platform == 'darwin':
        return 'lo0'
    return None


def _get_file_age_in_seconds(path):
    """Get the age of a given file in seconds"""
    # Time is stored in seconds since epoch
    file_last_modified = 0
    if path.exists():
        file_last_modified = path.stat().st_mtime
    return time.time() - file_last_modified


def _get_build_paths():
    """Gets the root and build paths (either default or from the environment
    variables), and sets related paths to binaries and files."""
    root_path = pathlib.Path(
    os.environ[ROOT_ENVVAR] if os.getenv(ROOT_ENVVAR) else subprocess.
    getoutput('git rev-parse --show-toplevel'))
    assert root_path.exists(), 'Could not find openscreen root!'

    build_path = pathlib.Path(os.environ[BUILD_ENVVAR]) if os.getenv(
        BUILD_ENVVAR) else root_path.joinpath('out',
                                                    'Default').resolve()
    assert build_path.exists(), 'Could not find openscreen build!'

    BuildPaths = namedtuple("BuildPaths",
                            "root build test_video cast_receiver cast_sender")
    return BuildPaths(root = root_path,
        build = build_path,
        test_video = build_path.joinpath(TEST_VIDEO_NAME).resolve(),
        cast_receiver = build_path.joinpath(RECEIVER_BINARY_NAME).resolve(),
        cast_sender = build_path.joinpath(SENDER_BINARY_NAME).resolve()
        )


class TestFlags(IntFlag):
    """
    Test flags, primarily used to control sender and receiver configuration
    to test different features of the standalone libraries.
    """
    UseRemoting = 1
    UseAndroidHack = 2


class StandaloneCastTest(unittest.TestCase):
    """
    Test class for setting up and running end to end tests on the
    standalone sender and receiver binaries. This class uses the unittest
    package, so methods that are executed as tests all have named prefixed
    with "test_".

    This suite sets the current working directory to the root of the Open
    Screen repository, and references all files from the root directory.
    Generated certificates should always be in |cls.build_paths.root|.
    """

    @classmethod
    def setUpClass(cls):
        """Shared setup method for all tests, handles one-time updates."""
        cls.build_paths = _get_build_paths()
        os.chdir(cls.build_paths.root)
        cls.download_video()
        cls.generate_certificates()

    @classmethod
    def download_video(cls):
        """Downloads the test video from Google storage."""
        if os.path.exists(cls.build_paths.test_video):
            logging.debug('Video already exists, skipping download...')
            return

        logging.debug('Downloading video from %s', TEST_VIDEO_URL)
        with request.urlopen(TEST_VIDEO_URL, context=ssl.SSLContext()) as url:
            with open(cls.build_paths.test_video, 'wb') as file:
                file.write(url.read())

    @classmethod
    def generate_certificates(cls):
        """Generates test certificates using the cast receiver."""
        cert_age = _get_file_age_in_seconds(pathlib.Path(TEST_CERT_NAME))
        key_age = _get_file_age_in_seconds(pathlib.Path(TEST_KEY_NAME))
        if cert_age < CERT_EXPIRY_AGE and key_age < CERT_EXPIRY_AGE:
            logging.debug('Credentials are up to date...')
            return

        logging.debug('Credentials out of date, generating new ones...')
        try:
            subprocess.check_output(
                [
                    cls.build_paths.cast_receiver,
                    '-g',  # Generate certificate and private key.
                    '-v'  # Enable verbose logging.
                ],
                stderr=subprocess.STDOUT)
        except subprocess.CalledProcessError as e:
            print('Generation failed with output: ', e.output.decode())
            raise

    def launch_receiver(self):
        """Launches the receiver process with discovery disabled."""
        logging.debug('Launching the receiver application...')
        loopback = _get_loopback_adapter_name()
        self.assertTrue(loopback)

        #pylint: disable = consider-using-with
        return subprocess.Popen(
            [
                self.build_paths.cast_receiver,
                '-d',
                TEST_CERT_NAME,
                '-p',
                TEST_KEY_NAME,
                '-x',  # Skip discovery, only necessary on Mac OS X.
                '-v',  # Enable verbose logging.
                loopback
            ],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE)

    def launch_sender(self, flags, codec=None):
        """Launches the sender process, running the test video file once."""
        logging.debug('Launching the sender application...')
        command = [
            self.build_paths.cast_sender,
            '127.0.0.1:8010',
            self.build_paths.test_video,
            '-d',
            TEST_CERT_NAME,
            '-n'  # Only play the video once, and then exit.
        ]
        if TestFlags.UseAndroidHack in flags:
            command.append('-a')
        if TestFlags.UseRemoting in flags:
            command.append('-r')

        # The standalone sender sends VP8 if no codec command line argument is
        # passed.
        if codec:
          command.append('-c')
          if codec == VideoCodec.Vp8:
            command.append('vp8')
          elif codec == VideoCodec.Vp9:
              command.append('vp9')
          else:
              self.assertTrue(codec == VideoCodec.Av1)
              command.append('av1')

        #pylint: disable = consider-using-with
        return subprocess.Popen(command,
                                stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE)

    def check_logs(self, logs, codec=None):
        """Checks that the outputted logs contain expected behavior."""

        # If a codec was not provided, we should make sure that the standalone
        # sender sent VP8.
        if codec == None:
          codec = VideoCodec.Vp8

        for message in (EXPECTED_RECEIVER_MESSAGES +
                        [VIDEO_CODEC_SPECIFIC_RECEIVER_MESSAGES[codec]]):
            self.assertTrue(
                message in logs[0],
                'Missing log message: {}.\n{}'.format(message,
                                                      MISSING_LOG_MESSAGE))
        for message in EXPECTED_SENDER_MESSAGES:
            self.assertTrue(
                message in logs[1],
                'Missing log message: {}.\n{}'.format(message,
                                                      MISSING_LOG_MESSAGE))
        for log, prefix in logs, ["[ERROR:", "[FATAL:"]:
            self.assertTrue(prefix not in log, "Logs contained an error")
        logging.debug('Finished validating log output')

    def get_output(self, flags, codec=None):
        """Launches the sender and receiver, and handles exit output."""
        receiver_process = self.launch_receiver()
        logging.debug('Letting the receiver start up...')
        time.sleep(3)
        sender_process = self.launch_sender(flags, codec)

        logging.debug('Launched sender PID %i and receiver PID %i...',
            sender_process.pid, receiver_process.pid)
        logging.debug('collating output...')
        output = (receiver_process.communicate(
            timeout=PROCESS_TIMEOUT)[1].decode('utf-8'),
                  sender_process.communicate(
                      timeout=PROCESS_TIMEOUT)[1].decode('utf-8'))

        # TODO(issuetracker.google.com/194292855): standalones should exit zero.
        # Remoting causes the sender to exit with code -4.
        if not TestFlags.UseRemoting in flags:
            self.assertEqual(sender_process.returncode, 0,
                             'sender had non-zero exit code')
        return output

    def test_golden_case(self):
        """Tests that when settings are normal, things work end to end."""
        output = self.get_output([])
        self.check_logs(output)

    def test_remoting(self):
        """Tests that basic remoting works."""
        output = self.get_output(TestFlags.UseRemoting)
        self.check_logs(output)

    def test_with_android_hack(self):
        """Tests that things work when the Android RTP hack is enabled."""
        output = self.get_output(TestFlags.UseAndroidHack)
        self.check_logs(output)

    def test_vp8_flag(self):
      """Tests that the VP8 flag works with standard settings."""
      output = self.get_output([], VideoCodec.Vp8)
      self.check_logs(output, VideoCodec.Vp8)

    def test_vp9_flag(self):
      """Tests that the VP9 flag works with standard settings."""
      output = self.get_output([], VideoCodec.Vp9)
      self.check_logs(output, VideoCodec.Vp9)

    @unittest.skipUnless(os.getenv(LIBAOM_ENVVAR),
                        'Skipping AV1 test since LibAOM not installed.')
    def test_av1_flag(self):
      """Tests that the AV1 flag works with standard settings."""
      output = self.get_output([], VideoCodec.Av1)
      self.check_logs(output, VideoCodec.Av1)


def parse_args():
    """Parses the command line arguments and sets up the logging module."""
    # NOTE for future developers: the `unittest` module will complain if it is
    # passed any args that it doesn't understand. If any Open Screen-specific
    # command line arguments are added in the future, they should be cropped
    # from sys.argv before |unittest.main()| is called.
    parser = argparse.ArgumentParser(description=DESCRIPTION)
    parser.add_argument('-v',
                        '--verbose',
                        help='enable debug logging',
                        action='store_true')

    parsed_args = parser.parse_args(sys.argv[1:])
    _set_log_level(parsed_args.verbose)


if __name__ == '__main__':
    parse_args()
    unittest.main()