test.py revision c278c2343c7f25083a80cb164b6bdc761d50050b 
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
29
30
31class LisaTest(unittest.TestCase):
32    """
33    A base class for LISA tests
34
35    This class is intended to be subclassed in order to create automated tests
36    for LISA. It sets up the TestEnv and Executor and provides convenience
37    methods for making assertions on results.
38
39    Subclasses should provide a test_conf to configure the TestEnv and an
40    experiments_conf to configure the executor.
41
42    Tests whose behaviour is dependent on target parameters, for example
43    presence of cpufreq governors or number of CPUs, can override
44    _getExperimentsConf to generate target-dependent experiments.
45
46    Example users of this class can be found under LISA's tests/ directory.
47    """
48
49    test_conf = None
50    """Override this with a dictionary or JSON path to configure the TestEnv"""
51
52    experiments_conf = None
53    """Override this with a dictionary to configure the Executor"""
54
55    @classmethod
56    def _init(cls, *args, **kwargs):
57        """
58        Set up logging and trigger running experiments
59        """
60
61        cls.logger = logging.getLogger('LisaTest')
62        cls._runExperiments()
63
64    @classmethod
65    def _getTestConf(cls):
66        if cls.test_conf is None:
67            raise NotImplementedError("Override `test_conf` attribute")
68        return cls.test_conf
69
70    @classmethod
71    def _getExperimentsConf(cls, test_env):
72        """
73        Get the experiments_conf used to configure the Executor
74
75        This method receives the initialized TestEnv as a parameter, so
76        subclasses can override it to configure workloads or target confs in a
77        manner dependent on the target. If not overridden, just returns the
78        experiments_conf attribute.
79        """
80        if cls.experiments_conf is None:
81            raise NotImplementedError("Override `experiments_conf` attribute")
82        return cls.experiments_conf
83
84    @classmethod
85    def _runExperiments(cls):
86        """
87        Default experiments execution engine
88        """
89
90        cls.logger.info('Setup tests execution engine...')
91        test_env = TestEnv(test_conf=cls._getTestConf())
92
93        experiments_conf = cls._getExperimentsConf(test_env)
94        cls.executor = Executor(test_env, experiments_conf)
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        # Execute post-experiments code defined by the test
107        cls._experimentsFinalize()
108
109    @classmethod
110    def _experimentsInit(cls):
111        """
112        Code executed before running the experiments
113        """
114
115    @classmethod
116    def _experimentsFinalize(cls):
117        """
118        Code executed after running the experiments
119        """
120
121    @memoized
122    def get_sched_assert(self, experiment, task):
123        """
124        Return a SchedAssert over the task provided
125        """
126        return SchedAssert(experiment.out_dir, self.te.topology, execname=task)
127
128    @memoized
129    def get_multi_assert(self, experiment, task_filter=""):
130        """
131        Return a SchedMultiAssert over the tasks whose names contain task_filter
132
133        By default, this includes _all_ the tasks that were executed for the
134        experiment.
135        """
136        tasks = experiment.wload.tasks.keys()
137        return SchedMultiAssert(experiment.out_dir,
138                                self.te.topology,
139                                [t for t in tasks if task_filter in t])
140
141    def get_start_time(self, experiment):
142        """
143        Get the time at which the experiment workload began executing
144        """
145        start_times_dict = self.get_multi_assert(experiment).getStartTime()
146        return min([t["starttime"] for t in start_times_dict.itervalues()])
147
148    def get_end_time(self, experiment):
149        """
150        Get the time at which the experiment workload finished executing
151        """
152        end_times_dict = self.get_multi_assert(experiment).getEndTime()
153        return max([t["endtime"] for t in end_times_dict.itervalues()])
154
155    def get_window(self, experiment):
156        return (self.get_start_time(experiment), self.get_end_time(experiment))
157
158    def get_end_times(self, experiment):
159        """
160        Get the time at which each task in the workload finished
161
162        Returned as a dict; {"task_name": finish_time, ...}
163        """
164
165        end_times = {}
166        for task in experiment.wload.tasks.keys():
167            sched_assert = SchedAssert(experiment.out_dir, self.te.topology,
168                                       execname=task)
169            end_times[task] = sched_assert.getEndTime()
170
171        return end_times
172
173
174@wrapt.decorator
175def experiment_test(wrapped_test, instance, args, kwargs):
176    """
177    Convert a LisaTest test method to be automatically called for each experiment
178
179    The method will be passed the experiment object and a list of the names of
180    tasks that were run as the experiment's workload.
181    """
182    for experiment in instance.executor.experiments:
183        tasks = experiment.wload.tasks.keys()
184        try:
185            wrapped_test(experiment, tasks, *args, **kwargs)
186        except AssertionError as e:
187            trace_relpath = os.path.join(experiment.out_dir, "trace.dat")
188            add_msg = "\n\tCheck trace file: " + os.path.abspath(trace_relpath)
189            orig_msg = e.args[0] if len(e.args) else ""
190            e.args = (orig_msg + add_msg,) + e.args[1:]
191            raise
192
193# Prevent nosetests from running experiment_test directly as a test case
194experiment_test.__test__ = False
195
196# vim :set tabstop=4 shiftwidth=4 expandtab
197