1#!/usr/bin/env python3
2#
3#   Copyright 2018 - The Android Open Source Project
4#
5#   Licensed under the Apache License, Version 2.0 (the "License");
6#   you may 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,
13#   WITHOUT 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
17import os
18import shutil
19import tempfile
20
21from acts import logger
22from acts.libs.proc import job
23
24_UICD_JAR_CMD = 'java -jar %s/uicd-commandline.jar'
25_UNZIP_CMD = 'tar -xzf %s -C %s'
26
27
28class UicdError(Exception):
29    """Raised for exceptions that occur in UIConductor-related tasks"""
30
31
32class UicdCli(object):
33    """Provides an interface for running UIConductor (Uicd) workflows under its
34    CLI.
35
36    This class does not handle workflow creation, which requires the Uicd
37    frontend.
38    """
39    def __init__(self, uicd_zip, workflow_paths, log_path=None):
40        """Creates a UicdCli object. Extracts the required uicd-cli binaries.
41
42        Args:
43            uicd_zip: The path to uicd_cli.tar.gz
44            workflow_paths: List of paths to uicd workflows and/or directories
45                containing them.
46            log_path: Directory for storing logs generated by Uicd.
47        """
48        self._uicd_zip = uicd_zip[0] if isinstance(uicd_zip, list) else uicd_zip
49        self._uicd_path = tempfile.mkdtemp(prefix='uicd')
50        self._log_path = log_path
51        if self._log_path:
52            os.makedirs(self._log_path, exist_ok=True)
53        self._log = logger.create_tagged_trace_logger(tag='Uicd')
54        self._set_workflows(workflow_paths)
55        self._setup_cli()
56
57    def _set_workflows(self, workflow_paths):
58        """Set up a dictionary that maps workflow name to its file location.
59        This allows the user to specify workflows to run without having to
60        provide the full path.
61
62        Args:
63            workflow_paths: List of paths to uicd workflows and/or directories
64                containing them.
65
66        Raises:
67            UicdError if two or more Uicd workflows share the same file name
68        """
69        if isinstance(workflow_paths, str):
70            workflow_paths = [workflow_paths]
71
72        # get a list of workflow files from specified paths
73        def _raise(e):
74            raise e
75        workflow_files = []
76        for path in workflow_paths:
77            if os.path.isfile(path):
78                workflow_files.append(path)
79            else:
80                for (root, _, files) in os.walk(path, onerror=_raise):
81                    for file in files:
82                        workflow_files.append(os.path.join(root, file))
83
84        # populate the dictionary
85        self._workflows = {}
86        for path in workflow_files:
87            workflow_name = os.path.basename(path)
88            if workflow_name in self._workflows.keys():
89                raise UicdError('Uicd workflows may not share the same name.')
90            self._workflows[workflow_name] = path
91
92    def _setup_cli(self):
93        """Extract tar from uicd_zip and place unzipped files in uicd_path.
94
95        Raises:
96            Exception if the extraction fails.
97        """
98        self._log.debug('Extracting uicd-cli binaries from %s' % self._uicd_zip)
99        unzip_cmd = _UNZIP_CMD % (self._uicd_zip, self._uicd_path)
100        try:
101            job.run(unzip_cmd.split())
102        except job.Error:
103            self._log.exception('Failed to extract uicd-cli binaries.')
104            raise
105
106    def run(self, serial, workflows, timeout=120):
107        """Run specified workflows on the UIConductor CLI.
108
109        Args:
110            serial: Device serial
111            workflows: List or str of workflows to run.
112            timeout: Number seconds to wait for command to finish.
113        """
114        base_cmd = _UICD_JAR_CMD % self._uicd_path
115        if isinstance(workflows, str):
116            workflows = [workflows]
117        for workflow_name in workflows:
118            self._log.info('Running workflow "%s"' % workflow_name)
119            if workflow_name in self._workflows:
120                args = '-d %s -i %s' % (serial, self._workflows[workflow_name])
121            else:
122                self._log.error(
123                    'The workflow "%s" does not exist.' % workflow_name)
124                continue
125            if self._log_path:
126                args = '%s -o %s' % (args, self._log_path)
127            cmd = '%s %s' % (base_cmd, args)
128            try:
129                result = job.run(cmd.split(), timeout=timeout)
130            except job.Error:
131                self._log.exception(
132                    'Failed to run workflow "%s"' % workflow_name)
133                continue
134            if result.stdout:
135                stdout_split = result.stdout.splitlines()
136                if len(stdout_split) > 2:
137                    self._log.debug('Uicd logs stored at %s' % stdout_split[2])
138
139    def __del__(self):
140        """Delete the temp directory to Uicd CLI binaries upon ACTS exit."""
141        shutil.rmtree(self._uicd_path)
142