#
# Copyright (C) 2017 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import cmd
import ctypes
import datetime
import imp # Python v2 compatibility
import logging
import multiprocessing
import multiprocessing.pool
import os
import re
import shutil
import signal
import socket
import sys
import tempfile
import threading
import time
import urlparse
from host_controller import common
from host_controller.command_processor import command_adb
from host_controller.command_processor import command_build
from host_controller.command_processor import command_config
from host_controller.command_processor import command_config_local
from host_controller.command_processor import command_copy
from host_controller.command_processor import command_device
from host_controller.command_processor import command_dut
from host_controller.command_processor import command_exit
from host_controller.command_processor import command_fastboot
from host_controller.command_processor import command_fetch
from host_controller.command_processor import command_flash
from host_controller.command_processor import command_gsispl
from host_controller.command_processor import command_info
from host_controller.command_processor import command_lease
from host_controller.command_processor import command_list
from host_controller.command_processor import command_password
from host_controller.command_processor import command_release
from host_controller.command_processor import command_retry
from host_controller.command_processor import command_request
from host_controller.command_processor import command_repack
from host_controller.command_processor import command_sheet
from host_controller.command_processor import command_shell
from host_controller.command_processor import command_sleep
from host_controller.command_processor import command_test
from host_controller.command_processor import command_reproduce
from host_controller.command_processor import command_upload
from host_controller.build import build_info
from host_controller.build import build_provider_ab
from host_controller.build import build_provider_gcs
from host_controller.build import build_provider_local_fs
from host_controller.build import build_provider_pab
from host_controller.utils.ipc import file_lock
from host_controller.utils.ipc import shared_dict
from host_controller.vti_interface import vti_endpoint_client
from vts.runners.host import logger
from vts.utils.python.common import cmd_utils
COMMAND_PROCESSORS = [
command_adb.CommandAdb,
command_build.CommandBuild,
command_config.CommandConfig,
command_config_local.CommandConfigLocal,
command_copy.CommandCopy,
command_device.CommandDevice,
command_dut.CommandDUT,
command_exit.CommandExit,
command_fastboot.CommandFastboot,
command_fetch.CommandFetch,
command_flash.CommandFlash,
command_gsispl.CommandGsispl,
command_info.CommandInfo,
command_lease.CommandLease,
command_list.CommandList,
command_password.CommandPassword,
command_release.CommandRelease,
command_retry.CommandRetry,
command_request.CommandRequest,
command_repack.CommandRepack,
command_sheet.CommandSheet,
command_shell.CommandShell,
command_sleep.CommandSleep,
command_test.CommandTest,
command_reproduce.CommandReproduce,
command_upload.CommandUpload,
]
class NonDaemonizedProcess(multiprocessing.Process):
"""Process class which is not daemonized."""
def _get_daemon(self):
return False
def _set_daemon(self, value):
pass
daemon = property(_get_daemon, _set_daemon)
class NonDaemonizedPool(multiprocessing.pool.Pool):
"""Pool class which is not daemonized."""
Process = NonDaemonizedProcess
def JobMain(vti_address, in_queue, out_queue, device_status, password, hosts):
"""Main() for a child process that executes a leased job.
Currently, lease jobs must use VTI (not TFC).
Args:
vti_client: VtiEndpointClient needed to create Console.
in_queue: Queue to get new jobs.
out_queue: Queue to put execution results.
device_status: SharedDict, contains device status information.
shared between processes.
password: multiprocessing.managers.ValueProxy, a proxy instance of a
string(ctypes.c_char_p) represents the password which is
to be passed to the prompt when executing certain command
as root user.
hosts: A list of HostController objects. Needed for the device command.
"""
def SigTermHandler(signum, frame):
"""Signal handler for exiting pool process explicitly.
Added to resolve orphaned pool process issue.
"""
sys.exit(0)
signal.signal(signal.SIGTERM, SigTermHandler)
vti_client = vti_endpoint_client.VtiEndpointClient(vti_address)
console = Console(vti_client, None, None, hosts, job_pool=True)
console.device_status = device_status
console.password = password
multiprocessing.util.Finalize(console, console.__exit__, exitpriority=0)
while True:
command = in_queue.get()
if command == "exit":
break
elif command == "lease":
filepath, kwargs = vti_client.LeaseJob(socket.gethostname(), True)
logging.debug("Job %s -> %s" % (os.getpid(), kwargs))
if filepath is not None:
# TODO: redirect console output and add
# console command to access them.
console._build_provider[
"pab"] = build_provider_pab.BuildProviderPAB()
console._build_provider[
"gcs"] = build_provider_gcs.BuildProviderGCS()
for serial in kwargs["serial"]:
console.ChangeDeviceState(
serial, common._DEVICE_STATUS_DICT["use"])
print_to_console = True
if not print_to_console:
sys.stdout = out
sys.stderr = err
ret, gcs_log_url = console.ProcessConfigurableScript(
os.path.join(os.getcwd(), "host_controller", "campaigns",
filepath), **kwargs)
if ret:
job_status = "complete"
else:
job_status = "infra-err"
vti_client.StopHeartbeat(job_status, gcs_log_url)
logging.info("Job execution complete. "
"Setting job status to {}".format(job_status))
if not print_to_console:
sys.stdout = sys.__stdout__
sys.stderr = sys.__stderr__
for serial in kwargs["serial"]:
console.ChangeDeviceState(
serial, common._DEVICE_STATUS_DICT["ready"])
del console._build_provider["pab"]
del console._build_provider["gcs"]
console.fetch_info = {}
console._detailed_fetch_info = {}
else:
logging.error("Unknown job command %s", command)
out_queue.put("exit")
class Console(cmd.Cmd):
"""The console for host controllers.
Attributes:
command_processors: dict of string:BaseCommandProcessor,
map between command string and command processors.
device_image_info: dict containing info about device image files.
prompt: The prompt string at the beginning of each command line.
test_result: dict containing info about the last test result.
test_suite_info: dict containing info about test suite package files.
tools_info: dict containing info about custom tool files.
scheduler_thread: dict containing threading.Thread instances(s) that
update configs regularly.
_build_provider_pab: The BuildProviderPAB used to download artifacts.
_vti_address: string, VTI service URI.
_vti_client: VtiEndpoewrClient, used to upload data to a test
scheduling infrastructure.
_tfc_client: The TfcClient that the host controllers connect to.
_hosts: A list of HostController objects.
_in_file: The input file object.
_out_file: The output file object.
_serials: A list of string where each string is a device serial.
_device_status: SharedDict, shared with process pool.
contains status data on each devices.
_job_pool: bool, True if Console is created from job pool process
context.
_password: multiprocessing.managers.ValueProxy, a proxy instance of a
string(ctypes.c_char_p) represents the password which is
to be passed to the prompt when executing certain command
as root user.
_manager: SyncManager. an instance of a manager for shared objects and
values between processes.
_vtslab_version: string, contains version information of vtslab package.
(<git commit timestamp>:<git commit hash value>)
_detailed_fetch_info: A nested dict, holds the branch and target value
of the device, gsi, or test suite artifact.
_file_lock: FileLock, an instance used for synchronizing the devices'
use when the automated self-update happens.
"""
def __init__(self,
vti_endpoint_client,
tfc,
pab,
host_controllers,
vti_address=None,
in_file=sys.stdin,
out_file=sys.stdout,
job_pool=False,
password=None):
"""Initializes the attributes and the parsers."""
# cmd.Cmd is old-style class.
cmd.Cmd.__init__(self, stdin=in_file, stdout=out_file)
self._build_provider = {}
self._job_pool = job_pool
if not self._job_pool:
self._build_provider["pab"] = pab
self._build_provider["gcs"] = build_provider_gcs.BuildProviderGCS()
self._build_provider[
"local_fs"] = build_provider_local_fs.BuildProviderLocalFS()
self._build_provider["ab"] = build_provider_ab.BuildProviderAB()
self._manager = multiprocessing.Manager()
self._device_status = shared_dict.SharedDict(self._manager)
self._password = self._manager.Value(ctypes.c_char_p, password)
try:
with open(common._VTSLAB_VERSION_TXT, "r") as file:
self._vtslab_version = file.readline().strip()
file.close()
logging.info("VTSLAB version: %s" % self._vtslab_version)
except IOError as e:
logging.exception(e)
logging.error("Version info missing in vtslab package. "
"Setting version as %s",
common._VTSLAB_VERSION_DEFAULT_VALUE)
self._vtslab_version = common._VTSLAB_VERSION_DEFAULT_VALUE
self._logfile_upload_path = ""
self._vti_endpoint_client = vti_endpoint_client
self._vti_address = vti_address
self._tfc_client = tfc
self._hosts = host_controllers
self._in_file = in_file
self._out_file = out_file
self.prompt = "> "
self.command_processors = {}
self.device_image_info = build_info.BuildInfo()
self.test_result = {}
self.test_suite_info = build_info.BuildInfo()
self.tools_info = build_info.BuildInfo()
self.fetch_info = {}
self._detailed_fetch_info = {}
self.test_results = {}
self._file_lock = file_lock.FileLock()
self.repack_dest_path = ""
if common._ANDROID_SERIAL in os.environ:
self._serials = [os.environ[common._ANDROID_SERIAL]]
else:
self._serials = []
self.InitCommandModuleParsers()
self.SetUpCommandProcessors()
tempdir_base = os.path.join(os.getcwd(), "tmp")
if not os.path.exists(tempdir_base):
os.mkdir(tempdir_base)
self._tmpdir_default = tempfile.mkdtemp(dir=tempdir_base)
self._tmp_logdir = tempfile.mkdtemp(dir=tempdir_base)
if not self._job_pool:
self._logfile_path = logger.setupTestLogger(
self._tmp_logdir, create_symlink=False)
def __exit__(self):
"""Finalizes the build provider attributes explicitly when exited."""
for bp in self._build_provider:
self._build_provider[bp].__del__()
if os.path.exists(self._tmp_logdir):
shutil.rmtree(self._tmp_logdir)
@property
def job_pool(self):
"""getter for self._job_pool"""
return self._job_pool
@property
def device_status(self):
"""getter for self._device_status"""
return self._device_status
@device_status.setter
def device_status(self, device_status):
"""setter for self._device_status"""
self._device_status = device_status
@property
def build_provider(self):
"""getter for self._build_provider"""
return self._build_provider
@property
def tmpdir_default(self):
"""getter for self._password"""
return self._tmpdir_default
@tmpdir_default.setter
def tmpdir_default(self, tmpdir):
"""getter for self._password"""
self._tmpdir_default = tmpdir
@property
def password(self):
"""getter for self._password"""
return self._password
@password.setter
def password(self, password):
"""getter for self._password"""
self._password = password
@property
def logfile_path(self):
"""getter for self._logfile_path"""
return self._logfile_path
@property
def tmp_logdir(self):
"""getter for self._tmp_logdir"""
return self._tmp_logdir
@property
def vti_endpoint_client(self):
"""getter for self._vti_endpoint_client"""
return self._vti_endpoint_client
@property
def vtslab_version(self):
"""getter for self._vtslab_version"""
return self._vtslab_version
@property
def detailed_fetch_info(self):
return self._detailed_fetch_info
def UpdateFetchInfo(self, artifact_type):
if artifact_type in common._ARTIFACT_TYPE_LIST:
self._detailed_fetch_info[artifact_type] = {}
self._detailed_fetch_info[artifact_type].update(self.fetch_info)
else:
logging.error("Unrecognized artifact type: %s", artifact_type)
@property
def file_lock(self):
"""getter for self._file_lock"""
return self._file_lock
def ChangeDeviceState(self, serial, state):
"""Changes a device's state and (un)locks the file lock if necessary.
Args:
serial: string, serial number of a device.
state: int, devices' status value pre-defined in
common._DEVICE_STATUS_DICT.
Returns:
True if the state change and locking/unlocking are successful.
False otherwise.
"""
if state == common._DEVICE_STATUS_DICT["use"]:
ret = self._file_lock.LockDevice(serial)
if ret == False:
return False
current_status = self.device_status[serial]
self.device_status[serial] = state
if (current_status in (common._DEVICE_STATUS_DICT["use"],
common._DEVICE_STATUS_DICT["error"])
and current_status != state):
self._file_lock.UnlockDevice(serial)
def InitCommandModuleParsers(self):
"""Init all console command modules"""
for name in dir(self):
if name.startswith('_Init') and name.endswith('Parser'):
attr_func = getattr(self, name)
if hasattr(attr_func, '__call__'):
attr_func()
def SetUpCommandProcessors(self):
"""Sets up all command processors."""
for command_processor in COMMAND_PROCESSORS:
cp = command_processor()
cp._SetUp(self)
do_text = "do_%s" % cp.command
help_text = "help_%s" % cp.command
setattr(self, do_text, cp._Run)
setattr(self, help_text, cp._Help)
self.command_processors[cp.command] = cp
def TearDown(self):
"""Removes all command processors."""
for command_processor in self.command_processors.itervalues():
command_processor._TearDown()
self.command_processors.clear()
self.__exit__()
def FormatString(self, format_string):
"""Replaces variables with the values in the console's dictionaries.
Args:
format_string: The string containing variables enclosed in {}.
Returns:
The formatted string.
Raises:
KeyError if a variable is not found in the dictionaries or the
value is empty.
"""
def ReplaceVariable(match):
"""Replacement functioon for re.sub().
replaces string encased in braces with values in the console's dict.
Args:
match: regex, used for extracting the variable name.
Returns:
string value corresponding to the input variable name.
"""
name = match.group(1)
if name in ("build_id", "branch", "target", "account_id"):
value = self.fetch_info[name]
elif name in ("result_full", "result_zip", "suite_plan",
"suite_name"):
value = self.test_result[name]
elif "timestamp" in name:
current_datetime = datetime.datetime.now()
value_date = current_datetime.strftime("%Y%m%d")
value_time = current_datetime.strftime("%H%M%S")
if "_date" in name:
value = value_date
elif "_time" in name:
value = value_time
elif "_year" in name:
value = value_date[0:4]
elif "_month" in name:
value = value_date[4:6]
elif "_day" in name:
value = value_date[6:8]
else:
value = "%s-%s" % (value_date, value_time)
elif name in ("hc_log", "hc_log_file", "hc_log_upload_path"):
# hc_log: full abs path to the current process's infra log.
# hc_log_file: infra log file name, with no path information.
# hc_log_upload_path: path of the infra log file in GCS.
value = self._logfile_path
if name == "hc_log_file":
value = os.path.basename(value)
elif name == "hc_log_upload_path":
value = self._logfile_upload_path
elif name in ("repack_path"):
value = self.repack_dest_path
self.repack_dest_path = ""
elif name in ("hostname"):
value = socket.gethostname()
elif "." in name and name.split(".")[0] in self.command_processors:
command, arg = name.split(".")
try:
value = self.command_processors[command].arg_buffer[arg]
except KeyError as e:
logging.exception(e)
value = ""
if value is None:
value = ""
else:
value = None
if value is None:
raise KeyError(name)
return value
return re.sub("{([^}]+)}", ReplaceVariable, format_string)
def ProcessScript(self, script_file_path):
"""Processes a .py script file.
A script file implements a function which emits a list of console
commands to execute. That function emits an empty list or None if
no more command needs to be processed.
Args:
script_file_path: string, the path of a script file (.py file).
Returns:
True if successful; False otherwise
"""
if not script_file_path.endswith(".py"):
logging.error("Script file is not .py file: %s" % script_file_path)
return False
script_module = imp.load_source('script_module', script_file_path)
commands = script_module.EmitConsoleCommands()
if commands:
for command in commands:
ret = self.onecmd(command)
if ret == False:
return False
return True
def ProcessConfigurableScript(self, script_file_path, **kwargs):
"""Processes a .py script file.
A script file implements a function which emits a list of console
commands to execute. That function emits an empty list or None if
no more command needs to be processed.
Args:
script_file_path: string, the path of a script file (.py file).
kwargs: extra args for the interface function defined in
the script file.
Returns:
True if successful; False otherwise
String which represents URL to the upload infra log file.
"""
if script_file_path and not script_file_path.endswith(".py"):
script_file_path += ".py"
if not script_file_path.endswith(".py"):
logging.error("Script file is not .py file: %s", script_file_path)
return False
ret = True
self._logfile_path, file_handler = logger.addLogFile(self._tmp_logdir)
src = self.FormatString("{hc_log}")
dest = self.FormatString(
"gs://vts-report/infra_log/{hostname}/%s_{timestamp}/{hc_log_file}"
% kwargs["build_target"])
self._logfile_upload_path = dest
script_module = imp.load_source('script_module', script_file_path)
commands = script_module.EmitConsoleCommands(**kwargs)
logging.info("Command list: %s", commands)
if commands:
logging.info("Console commands: %s", commands)
for command in commands:
ret = self.onecmd(command)
if ret == False:
break
else:
ret = False
file_handler.flush()
infra_log_upload_command = "upload"
infra_log_upload_command += " --src=%s" % src
infra_log_upload_command += " --dest=%s" % dest
for serial in kwargs["serial"]:
if self.device_status[serial] == common._DEVICE_STATUS_DICT[
"error"]:
self.vti_endpoint_client.SetJobStatusFromLeasedTo("bootup-err")
break
if not self.vti_endpoint_client.CheckBootUpStatus():
infra_log_upload_command += (" --report_path=gs://vts-report/"
"suite_result/{timestamp_year}/"
"{timestamp_month}/{timestamp_day}")
suite_name, plan_name = kwargs["test_name"].split("/")
infra_log_upload_command += (
" --result_from_suite=%s" % suite_name)
infra_log_upload_command += (" --result_from_plan=%s" % plan_name)
self.onecmd(infra_log_upload_command)
if self.GetSerials():
self.onecmd("device --update=stop")
logging.getLogger().removeHandler(file_handler)
os.remove(self._logfile_path)
return (ret != False), dest
def _Print(self, string):
"""Prints a string and a new line character.
Args:
string: The string to be printed.
"""
self._out_file.write(string + "\n")
def _PrintObjects(self, objects, attr_names):
"""Shows objects as a table.
Args:
object: The objects to be shown, one object in a row.
attr_names: The attributes to be shown, one attribute in a column.
"""
width = [len(name) for name in attr_names]
rows = [attr_names]
for dev_info in objects:
attrs = [
_ToPrintString(getattr(dev_info, name, ""))
for name in attr_names
]
rows.append(attrs)
for index, attr in enumerate(attrs):
width[index] = max(width[index], len(attr))
for row in rows:
self._Print(" ".join(
attr.ljust(width[index]) for index, attr in enumerate(row)))
def DownloadTestResources(self, request_id):
"""Download all of the test resources for a TFC request id.
Args:
request_id: int, TFC request id
"""
resources = self._tfc_client.TestResourceList(request_id)
for resource in resources:
self.DownloadTestResource(resource['url'])
def DownloadTestResource(self, url):
"""Download a test resource with build provider, given a url.
Args:
url: a resource locator (not necessarily HTTP[s])
with the scheme specifying the build provider.
"""
parsed = urlparse.urlparse(url)
path = (parsed.netloc + parsed.path).split('/')
if parsed.scheme == "pab":
if len(path) != 5:
logging.error("Invalid pab resource locator: %s", url)
return
account_id, branch, target, build_id, artifact_name = path
cmd = ("fetch"
" --type=pab"
" --account_id=%s"
" --branch=%s"
" --target=%s"
" --build_id=%s"
" --artifact_name=%s") % (account_id, branch, target,
build_id, artifact_name)
self.onecmd(cmd)
elif parsed.scheme == "ab":
if len(path) != 4:
logging.error("Invalid ab resource locator: %s", url)
return
branch, target, build_id, artifact_name = path
cmd = ("fetch"
"--type=ab"
" --branch=%s"
" --target=%s"
" --build_id=%s"
" --artifact_name=%s") % (branch, target, build_id,
artifact_name)
self.onecmd(cmd)
elif parsed.scheme == gcs:
cmd = "fetch --type=gcs --path=%s" % url
self.onecmd(cmd)
else:
logging.error("Invalid URL: %s", url)
def SetSerials(self, serials):
"""Sets the default serial numbers for flashing and testing.
Args:
serials: A list of strings, the serial numbers.
"""
self._serials = serials
def FlashImgPackage(self, package_path_gcs):
"""Fetches a repackaged image set from GCS and flashes to the device(s).
Args:
package_path_gcs: GCS URL to the packaged img zip file. May contain
the GSI imgs.
"""
self.onecmd("fetch --type=gcs --path=%s --full_device_images=True" %
package_path_gcs)
if common.FULL_ZIPFILE not in self.device_image_info:
logging.error("Failed to fetch the given file: %s",
package_path_gcs)
return False
if not self._serials:
logging.error("Please specify the serial number(s) of target "
"device(s) for flashing.")
return False
campaign_common = imp.load_source(
'campaign_common',
os.path.join(os.getcwd(), "host_controller", "campaigns",
"campaign_common.py"))
flash_command_list = []
for serial in self._serials:
flash_commands = []
cmd_utils.ExecuteOneShellCommand(
"adb -s %s reboot bootloader" % serial)
_, stderr, retcode = cmd_utils.ExecuteOneShellCommand(
"fastboot -s %s getvar product" % serial)
if retcode == 0:
res = stderr.splitlines()[0].rstrip()
if ":" in res:
product = res.split(":")[1].strip()
elif "waiting for %s" % serial in res:
res = stderr.splitlines()[1].rstrip()
product = res.split(":")[1].strip()
else:
product = "error"
else:
product = "error"
logging.info("Device %s product type: %s", serial, product)
if product in campaign_common.FLASH_COMMAND_EMITTER:
flash_commands.append(
campaign_common.FLASH_COMMAND_EMITTER[product](
serial, repacked_imageset=True))
elif product != "error":
flash_commands.append(
"flash --current --serial %s --skip-vbmeta=True" % serial)
else:
logging.error(
"Device %s does not exist. Omitting the flashing "
"to the device.", serial)
continue
flash_command_list.append(flash_commands)
ret = self.onecmd(flash_command_list)
if ret == False:
logging.error("Flash failed on device %s.", self._serials)
else:
logging.info("Flash succeeded on device %s.", self._serials)
return ret
def GetSerials(self):
"""Returns the serial numbers saved in the console.
Returns:
A list of strings, the serial numbers.
"""
return self._serials
def ResetSerials(self):
"""Clears all the serial numbers set to this console obj."""
self._serials = []
def JobThread(self):
"""Job thread which monitors and uploads results."""
thread = threading.currentThread()
while getattr(thread, "keep_running", True):
time.sleep(1)
if self._job_pool:
self._job_pool.close()
self._job_pool.terminate()
self._job_pool.join()
def StartJobThreadAndProcessPool(self):
"""Starts a background thread to control leased jobs."""
self._job_in_queue = multiprocessing.Queue()
self._job_out_queue = multiprocessing.Queue()
self._job_pool = NonDaemonizedPool(
common._MAX_LEASED_JOBS, JobMain,
(self._vti_address, self._job_in_queue, self._job_out_queue,
self._device_status, self._password, self._hosts))
self._job_thread = threading.Thread(target=self.JobThread)
self._job_thread.daemon = True
self._job_thread.start()
def StopJobThreadAndProcessPool(self):
"""Terminates the thread and processes that runs the leased job."""
if hasattr(self, "_job_thread"):
self._job_thread.keep_running = False
self._job_thread.join()
def WaitForJobsToExit(self):
"""Wait for the running jobs to complete before exiting HC."""
if self._job_pool:
pool_process_count = common._MAX_LEASED_JOBS
for _ in range(common._MAX_LEASED_JOBS):
self._job_in_queue.put("exit")
while True:
response = self._job_out_queue.get()
if response == "exit":
pool_process_count -= 1
if pool_process_count <= 0:
break
# @Override
def onecmd(self, line, depth=1, ret_out_queue=None):
"""Executes command(s) and prints any exception.
Parallel execution only for 2nd-level list element.
Args:
line: a list of string or string which keeps the command to run.
"""
if not line:
return
if type(line) == list:
if depth == 1: # 1 to use multi-threading
jobs = []
ret_queue = multiprocessing.Queue()
for sub_command in line:
p = multiprocessing.Process(
target=self.onecmd,
args=(
sub_command,
depth + 1,
ret_queue,
))
jobs.append(p)
p.start()
for job in jobs:
job.join()
ret_cmd_list = True
while not ret_queue.empty():
ret_from_subprocess = ret_queue.get()
ret_cmd_list = ret_cmd_list and ret_from_subprocess
if ret_cmd_list == False:
return False
else:
for sub_command in line:
ret_cmd_list = self.onecmd(sub_command, depth + 1)
if ret_cmd_list == False and ret_out_queue:
ret_out_queue.put(False)
return False
return
logging.info("Command: %s", line)
try:
ret_cmd = cmd.Cmd.onecmd(self, line)
if ret_cmd == False and ret_out_queue:
ret_out_queue.put(ret_cmd)
return ret_cmd
except Exception as e:
self._Print("%s: %s" % (type(e).__name__, e))
if ret_out_queue:
ret_out_queue.put(False)
return False
# @Override
def emptyline(self):
"""Ignores empty lines."""
pass
# @Override
def default(self, line):
"""Handles unrecognized commands.
Returns:
True if receives EOF; otherwise delegates to default handler.
"""
if line == "EOF":
return self.do_exit(line)
return cmd.Cmd.default(self, line)
def _ToPrintString(obj):
"""Converts an object to printable string on console.
Args:
obj: The object to be printed.
"""
if isinstance(obj, (list, tuple, set)):
return ",".join(str(x) for x in obj)
return str(obj)