• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2017 The Chromium OS Authors. All rights reserved.
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5import importlib
6import logging
7import os
8import re
9
10import yaml
11
12from autotest_lib.client.common_lib import error
13
14class DeviceCapability(object):
15    """
16    Generate capabilities status on DUT from yaml files in a given path.
17    Answer from the capabilities whether some capability is satisfied on DUT.
18    """
19
20    def __init__(self, settings_path='/usr/local/etc/autotest-capability'):
21        """
22        @param settings_path: string, the base directory for autotest
23                              capability. There should be yaml files.
24        """
25        self.capabilities = self.__get_autotest_capability(settings_path)
26        logging.info("Capabilities:\n%r", self.capabilities)
27
28
29    def __get_autotest_capability(self, settings_path):
30        """
31        Generate and summarize capabilities from yaml files in
32        settings_path with detectors.
33
34        @param settings_path: string, the base directory for autotest
35                              capability. There should be yaml files.
36        @returns dict:
37            The capabilities on DUT.
38            Its key is string denoting a capability. Its value is 'yes', 'no' or
39            'disable.'
40        """
41
42        def run_detector(name):
43            """
44            Run a detector in the detector directory. (i.e.
45            autotest/files/client/cros/video/detectors)
46            Return the result of the detector.
47
48            @param name: string, the name of running detector.
49            @returns string, a result of detect() in the detector script.
50            """
51            if name not in detect_results:
52                detector = importlib.import_module(
53                    "autotest_lib.client.cros.video.detectors.%s"
54                    % name)
55                detect_results[name] = detector.detect()
56                logging.info("Detector result (%s): %s",
57                             name, detect_results[name])
58            return detect_results[name]
59
60        managed_cap_fpath = os.path.join(settings_path,
61                                         'managed-capabilities.yaml')
62        if not os.path.exists(managed_cap_fpath):
63            raise error.TestFail("%s is not installed" % managed_cap_fpath)
64        managed_caps = yaml.load(file(managed_cap_fpath))
65
66        cap_files = [f for f in os.listdir(settings_path)
67                     if re.match(r'^[0-9]+-.*\.yaml$', f)]
68        cap_files.sort(key=lambda f: int(f.split('-', 1)[0]))
69
70        detect_results = {}
71        autotest_caps = dict.fromkeys(managed_caps, 'no')
72        for fname in cap_files:
73            logging.debug('Processing caps: %s', fname)
74            fname = os.path.join(settings_path, fname)
75            for rule in yaml.load(file(fname)):
76                # The type of rule is string or dict
77                # If the type is a string, it is a capability (e.g. webcam).
78                # If a specific condition (e.g. kepler, cpu type) is required,
79                # rule would be dict, for example,
80                # {'detector': 'intel_cpu',
81                #  'match': ['intel_celeron_1007U'],
82                #  'capabilities': ['no hw_h264_enc_1080_30'] }.
83                logging.debug("%r", rule)
84                caps = []
85                if isinstance(rule, dict):
86                    if run_detector(rule['detector']) in rule['match']:
87                        caps = rule['capabilities']
88                else:
89                    caps = [rule]
90
91                for capability in caps:
92                    m = re.match(r'(?:(disable|no)\s+)?([\w\-]+)$', capability)
93                    prefix, capability = m.groups()
94                    if capability in managed_caps:
95                        autotest_caps[capability] = prefix or 'yes'
96                    else:
97                        raise error.TestFail(
98                            "Unexpected capability: %s" % capability)
99
100        return autotest_caps
101
102
103    def get_managed_caps(self):
104        return self.capabilities.keys()
105
106
107    def get_capability_results(self):
108        return self.capabilities
109
110
111    def get_capability(self, cap):
112        """
113        Decide if a device satisfies a required capability for an autotest.
114
115        @param cap: string, denoting one capability. It must be one in
116                    settings_path + 'managed-capabilities.yaml.'
117        @returns 'yes', 'no', or 'disable.'
118        """
119        try:
120            return self.capabilities[cap]
121        except KeyError:
122            raise error.TestFail("Unexpected capability: %s" % cap)
123
124
125    def ensure_capability(self, cap):
126        """
127        Raise TestNAError if a device doesn't satisfy cap.
128        """
129        if self.get_capability(cap) != 'yes':
130            raise error.TestNAError("Missing Capability: %s" % cap)
131
132
133    def have_capability(self, cap):
134        """
135        Return whether cap is available.
136        """
137        return self.get_capability(cap) == 'yes'
138