1# SPDX-License-Identifier: Apache-2.0 2# 3# Copyright (C) 2015, ARM Limited and contributors. 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); you may 6# not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16# 17 18import os 19import unittest 20import logging 21 22from bart.sched.SchedAssert import SchedAssert 23from bart.sched.SchedMultiAssert import SchedMultiAssert 24from devlib.utils.misc import memoized 25import wrapt 26 27from env import TestEnv 28from executor import Executor 29from trace import Trace 30 31 32class LisaTest(unittest.TestCase): 33 """ 34 A base class for LISA tests 35 36 This class is intended to be subclassed in order to create automated tests 37 for LISA. It sets up the TestEnv and Executor and provides convenience 38 methods for making assertions on results. 39 40 Subclasses should provide a test_conf to configure the TestEnv and an 41 experiments_conf to configure the executor. 42 43 Tests whose behaviour is dependent on target parameters, for example 44 presence of cpufreq governors or number of CPUs, can override 45 _getExperimentsConf to generate target-dependent experiments. 46 47 Example users of this class can be found under LISA's tests/ directory. 48 49 :ivar experiments: List of :class:`Experiment` s executed for the test. Only 50 available after :meth:`init` has been called. 51 """ 52 53 test_conf = None 54 """Override this with a dictionary or JSON path to configure the TestEnv""" 55 56 experiments_conf = None 57 """Override this with a dictionary or JSON path to configure the Executor""" 58 59 @classmethod 60 def _getTestConf(cls): 61 if cls.test_conf is None: 62 raise NotImplementedError("Override `test_conf` attribute") 63 return cls.test_conf 64 65 @classmethod 66 def _getExperimentsConf(cls, test_env): 67 """ 68 Get the experiments_conf used to configure the Executor 69 70 This method receives the initialized TestEnv as a parameter, so 71 subclasses can override it to configure workloads or target confs in a 72 manner dependent on the target. If not overridden, just returns the 73 experiments_conf attribute. 74 """ 75 if cls.experiments_conf is None: 76 raise NotImplementedError("Override `experiments_conf` attribute") 77 return cls.experiments_conf 78 79 @classmethod 80 def runExperiments(cls): 81 """ 82 Set up logging and trigger running experiments 83 """ 84 cls.logger = logging.getLogger('LisaTest') 85 86 cls.logger.info('Setup tests execution engine...') 87 test_env = TestEnv(test_conf=cls._getTestConf()) 88 89 experiments_conf = cls._getExperimentsConf(test_env) 90 cls.executor = Executor(test_env, experiments_conf) 91 92 # Alias tests and workloads configurations 93 cls.wloads = cls.executor._experiments_conf["wloads"] 94 cls.confs = cls.executor._experiments_conf["confs"] 95 96 # Alias executor objects to make less verbose tests code 97 cls.te = cls.executor.te 98 cls.target = cls.executor.target 99 100 # Execute pre-experiments code defined by the test 101 cls._experimentsInit() 102 103 cls.logger.info('Experiments execution...') 104 cls.executor.run() 105 106 cls.experiments = cls.executor.experiments 107 108 # Execute post-experiments code defined by the test 109 cls._experimentsFinalize() 110 111 @classmethod 112 def _experimentsInit(cls): 113 """ 114 Code executed before running the experiments 115 """ 116 117 @classmethod 118 def _experimentsFinalize(cls): 119 """ 120 Code executed after running the experiments 121 """ 122 123 @memoized 124 def get_sched_assert(self, experiment, task): 125 """ 126 Return a SchedAssert over the task provided 127 """ 128 return SchedAssert( 129 self.get_trace(experiment).ftrace, self.te.topology, execname=task) 130 131 @memoized 132 def get_multi_assert(self, experiment, task_filter=""): 133 """ 134 Return a SchedMultiAssert over the tasks whose names contain task_filter 135 136 By default, this includes _all_ the tasks that were executed for the 137 experiment. 138 """ 139 tasks = experiment.wload.tasks.keys() 140 return SchedMultiAssert(self.get_trace(experiment).ftrace, 141 self.te.topology, 142 [t for t in tasks if task_filter in t]) 143 144 def get_trace(self, experiment): 145 if not hasattr(self, "__traces"): 146 self.__traces = {} 147 if experiment.out_dir in self.__traces: 148 return self.__traces[experiment.out_dir] 149 150 if ('ftrace' not in experiment.conf['flags'] 151 or 'ftrace' not in self.test_conf): 152 raise ValueError( 153 'Tracing not enabled. If this test needs a trace, add "ftrace" ' 154 'to your test/experiment configuration flags') 155 156 events = self.test_conf['ftrace']['events'] 157 tasks = experiment.wload.tasks.keys() 158 trace = Trace(self.te.platform, experiment.out_dir, events, tasks) 159 160 self.__traces[experiment.out_dir] = trace 161 return trace 162 163 def get_start_time(self, experiment): 164 """ 165 Get the time at which the experiment workload began executing 166 """ 167 start_times_dict = self.get_multi_assert(experiment).getStartTime() 168 return min([t["starttime"] for t in start_times_dict.itervalues()]) 169 170 def get_end_time(self, experiment): 171 """ 172 Get the time at which the experiment workload finished executing 173 """ 174 end_times_dict = self.get_multi_assert(experiment).getEndTime() 175 return max([t["endtime"] for t in end_times_dict.itervalues()]) 176 177 def get_window(self, experiment): 178 return (self.get_start_time(experiment), self.get_end_time(experiment)) 179 180 def get_end_times(self, experiment): 181 """ 182 Get the time at which each task in the workload finished 183 184 Returned as a dict; {"task_name": finish_time, ...} 185 """ 186 187 end_times = {} 188 ftrace = self.get_trace(experiment).ftrace 189 for task in experiment.wload.tasks.keys(): 190 sched_assert = SchedAssert(ftrace, self.te.topology, execname=task) 191 end_times[task] = sched_assert.getEndTime() 192 193 return end_times 194 195 def _dummy_method(self): 196 pass 197 198 # In the Python unittest framework you instantiate TestCase objects passing 199 # the name of a test method that is going to be run to make assertions. We 200 # run our tests using nosetests, which automatically discovers these 201 # methods. However we also want to be able to instantiate LisaTest objects 202 # in notebooks without the inconvenience of having to provide a methodName, 203 # since we won't need any assertions. So we'll override __init__ with a 204 # default dummy test method that does nothing. 205 def __init__(self, methodName='_dummy_method', *args, **kwargs): 206 super(LisaTest, self).__init__(methodName, *args, **kwargs) 207 208@wrapt.decorator 209def experiment_test(wrapped_test, instance, args, kwargs): 210 """ 211 Convert a LisaTest test method to be automatically called for each experiment 212 213 The method will be passed the experiment object and a list of the names of 214 tasks that were run as the experiment's workload. 215 """ 216 for experiment in instance.executor.experiments: 217 tasks = experiment.wload.tasks.keys() 218 try: 219 wrapped_test(experiment, tasks, *args, **kwargs) 220 except AssertionError as e: 221 trace_relpath = os.path.join(experiment.out_dir, "trace.dat") 222 add_msg = "\n\tCheck trace file: " + os.path.abspath(trace_relpath) 223 orig_msg = e.args[0] if len(e.args) else "" 224 e.args = (orig_msg + add_msg,) + e.args[1:] 225 raise 226 227# Prevent nosetests from running experiment_test directly as a test case 228experiment_test.__test__ = False 229 230# vim :set tabstop=4 shiftwidth=4 expandtab 231