#!/usr/bin/env python3 # # Copyright 2019, The Android Open Source Project # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # """Unit tests for the run_app_with_prefetch_test.py script. Install: $> sudo apt-get install python3-pytest ## OR $> pip install -U pytest See also https://docs.pytest.org/en/latest/getting-started.html Usage: $> ./run_app_with_prefetch_test.py $> pytest run_app_with_prefetch_test.py $> python -m pytest run_app_with_prefetch_test.py See also https://docs.pytest.org/en/latest/usage.html """ import io import os import shlex import sys import tempfile # global imports from contextlib import contextmanager # pip imports import pytest # local imports import run_app_with_prefetch as runner from mock import call, patch, Mock sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app_startup.lib.app_runner import AppRunner # # Argument Parsing Helpers # @contextmanager def ignore_stdout_stderr(): """Ignore stdout/stderr output for duration of this context.""" old_stdout = sys.stdout old_stderr = sys.stderr sys.stdout = io.StringIO() sys.stderr = io.StringIO() try: yield finally: sys.stdout = old_stdout sys.stderr = old_stderr @contextmanager def argparse_bad_argument(msg): """Asserts that a SystemExit is raised when executing this context. If the assertion fails, print the message 'msg'. """ with pytest.raises(SystemExit, message=msg): with ignore_stdout_stderr(): yield def assert_bad_argument(args, msg): """Asserts that the command line arguments in 'args' are malformed. Prints 'msg' if the assertion fails. """ with argparse_bad_argument(msg): parse_args(args) def parse_args(args): """ :param args: command-line like arguments as a single string :return: dictionary of parsed key/values """ # "-a b -c d" => ['-a', 'b', '-c', 'd'] return vars(runner.parse_options(shlex.split(args))) def default_dict_for_parsed_args(**kwargs): """Combines it with all of the "optional" parameters' default values.""" d = { 'readahead': 'cold', 'simulate': None, 'simulate': False, 'debug': False, 'input': 'TraceFile.pb', 'timeout': 10, 'compiler_filter': None, 'activity': None } d.update(kwargs) return d def default_mock_dict_for_parsed_args(include_optional=True, **kwargs): """Combines default dict with all optional parameters with some mock required parameters. """ d = {'package': 'com.fake.package'} if include_optional: d.update(default_dict_for_parsed_args()) d.update(kwargs) return d def parse_optional_args(str): """ Parses an argument string which already includes all the required arguments in default_mock_dict_for_parsed_args. """ req = '--package com.fake.package' return parse_args('%s %s' % (req, str)) def test_argparse(): # missing arguments assert_bad_argument('', '-p are required') # required arguments are parsed correctly ad = default_dict_for_parsed_args # assert dict assert parse_args('--package xyz') == ad(package='xyz') assert parse_args('-p xyz') == ad(package='xyz') assert parse_args('-p xyz -s') == ad(package='xyz', simulate=True) assert parse_args('-p xyz --simulate') == ad(package='xyz', simulate=True) # optional arguments are parsed correctly. mad = default_mock_dict_for_parsed_args # mock assert dict assert parse_optional_args('--input trace.pb') == mad(input='trace.pb') assert parse_optional_args('--compiler-filter speed') == \ mad(compiler_filter='speed') assert parse_optional_args('-d') == mad(debug=True) assert parse_optional_args('--debug') == mad(debug=True) assert parse_optional_args('--timeout 123') == mad(timeout=123) assert parse_optional_args('-t 456') == mad(timeout=456) assert parse_optional_args('-r warm') == mad(readahead='warm') assert parse_optional_args('--readahead warm') == mad(readahead='warm') assert parse_optional_args('-a act') == mad(activity='act') assert parse_optional_args('--activity act') == mad(activity='act') def test_main(): args = '--package com.fake.package --activity act -s' opts = runner.parse_options(shlex.split(args)) result = runner.PrefetchAppRunner(**vars(opts)).run() assert result == [('TotalTime', '123')] def _mocked_run_shell_command(*args, **kwargs): if args[0] == 'adb shell ps | grep "music" | awk \'{print $2;}\'': return (True, '9999') else: return (True, '') def test_preprocess_no_cache_drop(): with patch('lib.cmd_utils.run_shell_command', new_callable=Mock) as mock_run_shell_command: mock_run_shell_command.side_effect = _mocked_run_shell_command prefetch_app_runner = runner.PrefetchAppRunner(package='music', activity='MainActivity', readahead='warm', compiler_filter=None, timeout=None, simulate=False, debug=False, input=None) prefetch_app_runner.preprocess() calls = [call('adb root'), call('adb shell "getenforce"'), call('adb shell "setenforce 0"'), call('adb shell "stop"'), call('adb shell "start"'), call('adb wait-for-device'), call('adb shell ps | grep "music" | awk \'{print $2;}\''), call('adb shell "kill 9999"')] mock_run_shell_command.assert_has_calls(calls) def test_preprocess_with_cache_drop(): with patch('lib.cmd_utils.run_shell_command', new_callable=Mock) as mock_run_shell_command: mock_run_shell_command.side_effect = _mocked_run_shell_command prefetch_app_runner = runner.PrefetchAppRunner(package='music', activity='MainActivity', readahead='cold', compiler_filter=None, timeout=None, simulate=False, debug=False, input=None) prefetch_app_runner.preprocess() calls = [call('adb root'), call('adb shell "getenforce"'), call('adb shell "setenforce 0"'), call('adb shell "stop"'), call('adb shell "start"'), call('adb wait-for-device'), call('adb shell ps | grep "music" | awk \'{print $2;}\''), call('adb shell "kill 9999"'), call('adb shell "echo 3 > /proc/sys/vm/drop_caches"')] mock_run_shell_command.assert_has_calls(calls) def test_preprocess_with_cache_drop_and_iorapd_enabled(): with patch('lib.cmd_utils.run_shell_command', new_callable=Mock) as mock_run_shell_command: mock_run_shell_command.side_effect = _mocked_run_shell_command with tempfile.NamedTemporaryFile() as input: prefetch_app_runner = runner.PrefetchAppRunner(package='music', activity='MainActivity', readahead='fadvise', compiler_filter=None, timeout=None, simulate=False, debug=False, input=input.name) prefetch_app_runner.preprocess() calls = [call('adb root'), call('adb shell "getenforce"'), call('adb shell "setenforce 0"'), call('adb shell "stop"'), call('adb shell "start"'), call('adb wait-for-device'), call( 'adb shell ps | grep "music" | awk \'{print $2;}\''), call('adb shell "kill 9999"'), call('adb shell "echo 3 > /proc/sys/vm/drop_caches"'), call('bash -c "source {}; iorapd_readahead_enable"'. format(AppRunner.IORAP_COMMON_BASH_SCRIPT))] mock_run_shell_command.assert_has_calls(calls) @patch('lib.adb_utils.blocking_wait_for_logcat_displayed_time') @patch('lib.cmd_utils.run_shell_command') def test_postprocess_with_launch_cleanup( mock_run_shell_command, mock_blocking_wait_for_logcat_displayed_time): mock_run_shell_command.side_effect = _mocked_run_shell_command mock_blocking_wait_for_logcat_displayed_time.return_value = 123 with tempfile.NamedTemporaryFile() as input: prefetch_app_runner = runner.PrefetchAppRunner(package='music', activity='MainActivity', readahead='fadvise', compiler_filter=None, timeout=10, simulate=False, debug=False, input=input.name) prefetch_app_runner.postprocess('2019-07-02 23:20:06.972674825') calls = [ call('bash -c "source {script_path}; ' 'iorapd_readahead_wait_until_finished ' '\'{package}\' \'{activity}\' \'{timestamp}\' \'{timeout}\'"'. format(timeout=10, package='music', activity='MainActivity', timestamp='2019-07-02 23:20:06.972674825', script_path=AppRunner.IORAP_COMMON_BASH_SCRIPT)), call('bash -c "source {}; iorapd_readahead_disable"'. format(AppRunner.IORAP_COMMON_BASH_SCRIPT)), call('adb shell ps | grep "music" | awk \'{print $2;}\''), call('adb shell "kill 9999"')] mock_run_shell_command.assert_has_calls(calls) if __name__ == '__main__': pytest.main()