test.py revision 968e72e29bfae2fe1329fa77cdc978dd92930b37 
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    # To instantiate unittest.TestCase you need to provide a test method name,
196    # the default being 'runTest'. If this method doesn't exist then an error is
197    # raised. This method isn't needed when running via nosetests, which detects
198    # test methods via its own mechanism. However it can be useful to
199    # instantiate test objects directly by hand in notebooks, even if you have
200    # no intention of using the test assertions. So define an empty default
201    # runTest method.
202    def runTest(self, *args, **kwargs):
203        pass
204
205
206@wrapt.decorator
207def experiment_test(wrapped_test, instance, args, kwargs):
208    """
209    Convert a LisaTest test method to be automatically called for each experiment
210
211    The method will be passed the experiment object and a list of the names of
212    tasks that were run as the experiment's workload.
213    """
214    for experiment in instance.executor.experiments:
215        tasks = experiment.wload.tasks.keys()
216        try:
217            wrapped_test(experiment, tasks, *args, **kwargs)
218        except AssertionError as e:
219            trace_relpath = os.path.join(experiment.out_dir, "trace.dat")
220            add_msg = "\n\tCheck trace file: " + os.path.abspath(trace_relpath)
221            orig_msg = e.args[0] if len(e.args) else ""
222            e.args = (orig_msg + add_msg,) + e.args[1:]
223            raise
224
225# Prevent nosetests from running experiment_test directly as a test case
226experiment_test.__test__ = False
227
228# vim :set tabstop=4 shiftwidth=4 expandtab
229