#
# Copyright (C) 2016 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.
#
import copy
import logging
import itertools
import operator
from vts.runners.host import const
from vts.utils.python.common import cmd_utils
from vts.utils.python.os import path_utils
from vts.testcases.kernel.ltp.shell_environment import shell_environment
from vts.testcases.kernel.ltp import ltp_enums
from vts.testcases.kernel.ltp import ltp_configs
from vts.testcases.kernel.ltp import requirements
class EnvironmentRequirementChecker(object):
"""LTP testcase environment checker.
This class contains a dictionary for some known environment
requirements for a set of test cases and several environment
check functions to be mapped with. All check functions' results
are cached in a dictionary for multiple use.
Attributes:
_REQUIREMENT_DEFINITIONS: dictionary {string, obj}, a map between
requirement name and the actual definition class object
_result_cache: dictionary {requirement_check_method_name:
(bool, string)}, a map between check method name and cached result
tuples (boolean, note)
_executable_available: dict {string, bool}, a map between executable
path and its existance on target
_shell_env: ShellEnvironment object, which checks and sets
shell environments given a shell mirror
shell: shell mirror object, can be used to execute shell
commands on target side through runner
ltp_bin_host_path: string, host path of ltp binary
"""
def __init__(self, shell):
self.shell = shell
self._result_cache = {}
self._executable_available = {}
self._shell_env = shell_environment.ShellEnvironment(self.shell)
self._REQUIREMENT_DEFINITIONS = requirements.GetRequrementDefinitions()
@property
def shell(self):
"""Get the runner's shell mirror object to execute commands"""
return self._shell
@shell.setter
def shell(self, shell):
"""Set the runner's shell mirror object to execute commands"""
self._shell = shell
def Cleanup(self):
"""Run all cleanup jobs at the end of tests"""
self._shell_env.Cleanup()
def GetRequirements(self, test_case):
"""Get a list of requirements for a fiven test case
Args:
test_case: TestCase object, the test case to query
"""
result = copy.copy(ltp_configs.REQUIREMENT_FOR_ALL)
result.extend(
rule
for rule, tests in ltp_configs.REQUIREMENTS_TO_TESTCASE.iteritems()
if test_case.fullname in tests)
result.extend(
rule
for rule, tests in ltp_configs.REQUIREMENT_TO_TESTSUITE.iteritems()
if test_case.testsuite in tests)
return list(set(result))
def Check(self, test_case):
"""Check whether a given test case's requirement has been satisfied.
Skip the test if not.
If check failed, this method returns False and the reason is set
to test_case.note.
Args:
test_case: TestCase object, a given test case to check
Returns:
True if check pass; False otherwise
"""
if (test_case.requirement_state ==
ltp_enums.RequirementState.UNSATISFIED or
not self.TestBinaryExists(test_case)):
return False
for requirement in self.GetRequirements(test_case):
if requirement not in self._result_cache:
definitions = self._REQUIREMENT_DEFINITIONS[requirement]
self._result_cache[
requirement] = self._shell_env.ExecuteDefinitions(
definitions)
result, note = self._result_cache[requirement]
logging.info("Result for %s's requirement %s is %s", test_case,
requirement, result)
if result is False:
test_case.requirement_state = ltp_enums.RequirementState.UNSATISFIED
test_case.note = note
return False
test_case.requirement_state = ltp_enums.RequirementState.SATISFIED
return True
def CheckAllTestCaseExecutables(self, test_cases):
"""Run a batch job to check executable exists and set permissions.
The result will be stored in self._executable_available for use in
TestBinaryExists method.
Args:
test_case: list of TestCase objects.
"""
executables_generators = (
test_case.GetRequiredExecutablePaths(self.ltp_bin_host_path)
for test_case in test_cases)
executables = list(
set(itertools.chain.from_iterable(executables_generators)))
# Set all executables executable permission using chmod.
logging.info("Setting permissions on device")
permission_command = "chmod 775 %s" % path_utils.JoinTargetPath(
ltp_configs.LTPBINPATH, '*')
permission_result = self.shell.Execute(permission_command)
if permission_result[const.EXIT_CODE][0]:
logging.error("Permission command '%s' failed.",
permission_command)
# Check existence of all executables used in test definition.
# Some executables needed by test cases but not listed in test
# definition will not be checked here
executable_exists_commands = [
"ls %s" % executable for executable in executables
if executable not in ltp_configs.INTERNAL_BINS
]
logging.info("Checking binary existence on host: %s",
executable_exists_commands)
cmd_results = cmd_utils.ExecuteShellCommand(executable_exists_commands)
executable_exists_results = map(operator.not_,
cmd_results[cmd_utils.EXIT_CODE])
logging.info("Finished checking binary existence on host: %s",
cmd_results)
self._executable_available = dict(
zip(executables, executable_exists_results))
# Check whether all the internal binaries in path needed exist
bin_path_exist_commands = ["which %s" % bin
for bin in ltp_configs.INTERNAL_BINS]
bin_path_results = map(
operator.not_,
self.shell.Execute(bin_path_exist_commands)[const.EXIT_CODE])
bin_path_results = map(
operator.not_,
self.shell.Execute(bin_path_exist_commands)[const.EXIT_CODE])
self._executable_available.update(
dict(zip(ltp_configs.INTERNAL_BINS, bin_path_results)))
def TestBinaryExists(self, test_case):
"""Check whether the given test case's binary exists.
Args:
test_case: TestCase, the object representing the test case
Return:
True if exists, False otherwise
"""
if test_case.requirement_state == ltp_enums.RequirementState.UNSATISFIED:
logging.warn("[Checker] Attempting to run test case that has "
"already been checked and requirement not satisfied."
"%s" % test_case)
return False
executables = test_case.GetRequiredExecutablePaths(
self.ltp_bin_host_path)
results = [self._executable_available[executable]
for executable in executables]
if not all(results):
test_case.requirement_state = ltp_enums.RequirementState.UNSATISFIED
test_case.note = "Some executables not exist: {}".format(
zip(executables, results))
logging.error("[Checker] Binary existance check failed for {}. "
"Reason: {}".format(test_case, test_case.note))
return False
else:
return True