atest/tools/atest_tools_unittest.py

#!/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.

"""Unittest for atest_tools."""

# pylint: disable=line-too-long

import os
import pickle
import platform
import subprocess
import unittest

from unittest import mock

import constants
import unittest_constants as uc

from tools import atest_tools

SEARCH_ROOT = uc.TEST_DATA_DIR
PRUNEPATH = uc.TEST_CONFIG_DATA_DIR
LOCATE = atest_tools.LOCATE
UPDATEDB = atest_tools.UPDATEDB

class AtestToolsUnittests(unittest.TestCase):
    """"Unittest Class for atest_tools.py."""

    @mock.patch('constants.LOCATE_CACHE', uc.LOCATE_CACHE)
    @mock.patch('tools.atest_tools.SEARCH_TOP', uc.TEST_DATA_DIR)
    @mock.patch('module_info.ModuleInfo.get_testable_modules')
    @mock.patch('module_info.ModuleInfo.__init__')
    def test_index_targets(self, mock_mod_info, mock_testable_mod):
        """Test method index_targets."""
        mock_mod_info.return_value = None
        mock_testable_mod.return_value = {uc.MODULE_NAME, uc.MODULE2_NAME}
        if atest_tools.has_command(UPDATEDB) and atest_tools.has_command(LOCATE):
            # 1. Test run_updatedb() is functional.
            atest_tools.run_updatedb(SEARCH_ROOT, uc.LOCATE_CACHE,
                                     prunepaths=PRUNEPATH)
            # test_config/ is excluded so that a.xml won't be found.
            locate_cmd1 = [LOCATE, '-d', uc.LOCATE_CACHE, '/a.xml']
            # locate always return 0 when not found in Darwin, therefore,
            # check null return in Darwin and return value in Linux.
            if platform.system() == 'Darwin':
                output = subprocess.check_output(locate_cmd1).decode()
                self.assertEqual(output, "")
            else:
                self.assertEqual(subprocess.call(locate_cmd1), 1)
            # module-info.json can be found in the search_root.
            locate_cmd2 = [LOCATE, '-d', uc.LOCATE_CACHE, 'module-info.json']
            self.assertEqual(subprocess.call(locate_cmd2), 0)

            # 2. Test index_targets() is functional.
            atest_tools.index_targets(uc.LOCATE_CACHE,
                                      class_index=uc.CLASS_INDEX,
                                      cc_class_index=uc.CC_CLASS_INDEX,
                                      module_index=uc.MODULE_INDEX,
                                      package_index=uc.PACKAGE_INDEX,
                                      qclass_index=uc.QCLASS_INDEX)
            _cache = {}
            # Test finding a Java class.
            with open(uc.CLASS_INDEX, 'rb') as cache:
                _cache = pickle.load(cache)
            self.assertIsNotNone(_cache.get('PathTesting'))
            # Test finding a CC class.
            with open(uc.CC_CLASS_INDEX, 'rb') as cache:
                _cache = pickle.load(cache)
            self.assertIsNotNone(_cache.get('HelloWorldTest'))
            # Test finding a package.
            with open(uc.PACKAGE_INDEX, 'rb') as cache:
                _cache = pickle.load(cache)
            self.assertIsNotNone(_cache.get(uc.PACKAGE))
            # Test finding a fully qualified class name.
            with open(uc.QCLASS_INDEX, 'rb') as cache:
                _cache = pickle.load(cache)
            self.assertIsNotNone(_cache.get('android.jank.cts.ui.PathTesting'))
            _cache = set()
            # Test finding a module name.
            with open(uc.MODULE_INDEX, 'rb') as cache:
                _cache = pickle.load(cache)
            self.assertTrue(uc.MODULE_NAME in _cache)
            self.assertFalse(uc.CLASS_NAME in _cache)
            # Clean up.
            targets_to_delete = (uc.CC_CLASS_INDEX,
                                 uc.CLASS_INDEX,
                                 uc.LOCATE_CACHE,
                                 uc.MODULE_INDEX,
                                 uc.PACKAGE_INDEX,
                                 uc.QCLASS_INDEX)
            for idx in targets_to_delete:
                os.remove(idx)
        else:
            self.assertEqual(atest_tools.has_command(UPDATEDB), False)
            self.assertEqual(atest_tools.has_command(LOCATE), False)

    def test_get_report_file(self):
        """Test method get_report_file."""
        report_file = '/tmp/acloud_status.json'

        arg_with_equal = '-a --report-file={} --all'.format(report_file)
        self.assertEqual(atest_tools.get_report_file('/abc', arg_with_equal),
                         report_file)

        arg_with_equal = '-b --report_file={} --ball'.format(report_file)
        self.assertEqual(atest_tools.get_report_file('/abc', arg_with_equal),
                         report_file)

        arg_without_equal = '-c --report-file {} --call'.format(report_file)
        self.assertEqual(atest_tools.get_report_file('/abc', arg_without_equal),
                         report_file)

        arg_without_equal = '-d --report_file {} --dall'.format(report_file)
        self.assertEqual(atest_tools.get_report_file('/abc', arg_without_equal),
                         report_file)

        arg_without_report = '-e --build-id 1234567'
        self.assertEqual(atest_tools.get_report_file('/tmp', arg_without_report),
                         report_file)

    def test_probe_acloud_status(self):
        """Test method prob_acloud_status."""
        success = os.path.join(SEARCH_ROOT, 'acloud', 'create_success.json')
        self.assertEqual(atest_tools.probe_acloud_status(success),
                         constants.EXIT_CODE_SUCCESS)

        failure = os.path.join(SEARCH_ROOT, 'acloud', 'create_failure.json')
        self.assertEqual(atest_tools.probe_acloud_status(failure),
                         constants.EXIT_CODE_AVD_CREATE_FAILURE)

        inexistence = os.path.join(SEARCH_ROOT, 'acloud', 'inexistence.json')
        self.assertEqual(atest_tools.probe_acloud_status(inexistence),
                         constants.EXIT_CODE_AVD_INVALID_ARGS)

    def test_get_acloud_duration(self):
        """Test method get_acloud_duration."""
        success = os.path.join(SEARCH_ROOT, 'acloud', 'create_success.json')
        success_duration = 152.659824
        self.assertEqual(atest_tools.get_acloud_duration(success),
                         success_duration)

        failure = os.path.join(SEARCH_ROOT, 'acloud', 'create_failure.json')
        failure_duration = 178.621254
        self.assertEqual(atest_tools.get_acloud_duration(failure),
                         failure_duration)

if __name__ == "__main__":
    unittest.main()