Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def output_path(self):
utils.create_dir(self._output_dir_path)
return self._output_dir_path
def run(self):
"""Executes tests.
This will instantiate controller and test classes, execute tests, and
print a summary.
Raises:
Error: if no tests have previously been added to this runner using
add_test_class(...).
"""
if not self._test_run_infos:
raise Error('No tests to execute.')
# Ensure the log path exists. Necessary if `run` is used outside of the
# `mobly_logger` context.
utils.create_dir(self._root_output_path)
summary_writer = records.TestSummaryWriter(
os.path.join(self._root_output_path, records.OUTPUT_FILE_SUMMARY))
try:
for test_run_info in self._test_run_infos:
# Set up the test-specific config
test_config = test_run_info.config.copy()
test_config.log_path = self._root_output_path
test_config.summary_writer = summary_writer
test_config.test_class_name_suffix = test_run_info.test_class_name_suffix
try:
self._run_test_class(config=test_config,
test_class=test_run_info.test_class,
tests=test_run_info.tests)
except signals.TestAbortAll as e:
logging.warning(
begin_time = mobly_logger.get_log_file_timestamp()
new_br = True
try:
stdout = self.adb.shell('bugreportz -v').decode('utf-8')
# This check is necessary for builds before N, where adb shell's ret
# code and stderr are not propagated properly.
if 'not found' in stdout:
new_br = False
except adb.AdbError:
new_br = False
if destination is None:
destination = os.path.join(self.log_path, 'BugReports')
br_path = utils.abs_path(destination)
utils.create_dir(br_path)
filename = self.generate_filename(prefix, str(begin_time), 'txt')
if new_br:
filename = filename.replace('.txt', '.zip')
full_out_path = os.path.join(br_path, filename)
# in case device restarted, wait for adb interface to return
self.wait_for_boot_completion()
self.log.debug('Start taking bugreport.')
if new_br:
out = self.adb.shell('bugreportz', timeout=timeout).decode('utf-8')
if not out.startswith('OK'):
raise DeviceError(self, 'Failed to take bugreport: %s' % out)
br_out_path = out.split(':')[1].strip()
self.adb.pull([br_out_path, full_out_path])
else:
# shell=True as this command redirects the stdout to a local file
# using shell redirection.
def save_to_text_file(monsoon_data, file_path):
"""Save multiple MonsoonData objects to a text file.
Args:
monsoon_data: A list of MonsoonData objects to write to a text
file.
file_path: The full path of the file to save to, including the file
name.
"""
if not monsoon_data:
raise MonsoonError("Attempting to write empty Monsoon data to "
"file, abort")
utils.create_dir(os.path.dirname(file_path))
with io.open(file_path, 'w', encoding='utf-8') as f:
for md in monsoon_data:
f.write(str(md))
f.write(MonsoonData.delimiter)
def _start(self):
"""The actual logic of starting logcat."""
self._enable_logpersist()
if self._config.output_file_path:
self._close_logcat_file()
self.adb_logcat_file_path = self._config.output_file_path
if not self.adb_logcat_file_path:
f_name = self._ad.generate_filename(self.OUTPUT_FILE_TYPE,
extension_name='txt')
logcat_file_path = os.path.join(self._ad.log_path, f_name)
self.adb_logcat_file_path = logcat_file_path
utils.create_dir(os.path.dirname(self.adb_logcat_file_path))
cmd = '"%s" -s %s logcat -v threadtime -T 1 %s >> "%s"' % (
adb.ADB, self._ad.serial, self._config.logcat_params,
self.adb_logcat_file_path)
process = utils.start_standing_subprocess(cmd, shell=True)
self._adb_logcat_process = process
"""Convenient method for creating excerpts of adb logcat.
This copies logcat lines from self.adb_logcat_file_path to an excerpt
file, starting from the location where the previous excerpt ended.
Call this method at the end of: `setup_class`, `teardown_test`, and
`teardown_class`.
Args:
test_info: `self.current_test_info` in a Mobly test.
Returns:
List of strings, the absolute paths to excerpt files.
"""
dest_path = test_info.output_path
utils.create_dir(dest_path)
filename = self._ad.generate_filename(self.OUTPUT_FILE_TYPE, test_info,
'txt')
excerpt_file_path = os.path.join(dest_path, filename)
with io.open(excerpt_file_path, 'w', encoding='utf-8',
errors='replace') as out:
while True:
line = self._adb_logcat_file_obj.readline()
if not line:
break
out.write(line)
self._ad.log.debug('logcat excerpt created at: %s', excerpt_file_path)
return [excerpt_file_path]
Args:
tag: An identifier of the time period, usualy the name of a test.
begin_time: Logline format timestamp of the beginning of the time
period.
Returns:
String, full path to the excerpt file created.
"""
if not self.adb_logcat_file_path:
raise Error(
self._ad,
'Attempting to cat adb log when none has been collected.')
end_time = mobly_logger.get_log_line_timestamp()
self._ad.log.debug('Extracting adb log from logcat.')
adb_excerpt_path = os.path.join(self._ad.log_path, 'AdbLogExcerpts')
utils.create_dir(adb_excerpt_path)
out_name = '%s,%s.txt' % (tag, begin_time)
out_name = mobly_logger.sanitize_filename(out_name)
full_adblog_path = os.path.join(adb_excerpt_path, out_name)
with io.open(full_adblog_path, 'w', encoding='utf-8') as out:
in_file = self.adb_logcat_file_path
with io.open(in_file, 'r', encoding='utf-8',
errors='replace') as f:
in_range = False
while True:
line = None
try:
line = f.readline()
if not line:
break
except:
continue
root_output_path: path to the directory for the entire test run.
log_path: same as `root_output_path` outside of a test class run. In the
context of a test class run, this is the output directory for files
specific to a test class.
Args:
log_path: string, the location of the report file.
prefix: optional string, a prefix for each log line in terminal.
alias: optional string, The name of the alias to use for the latest log
directory. If a falsy value is provided, then the alias directory
will not be created, which is useful to save storage space when the
storage system (e.g. ZIP files) does not properly support
shortcut/symlinks.
"""
utils.create_dir(log_path)
_setup_test_logger(log_path, prefix)
logging.info('Test output folder: "%s"', log_path)
if alias:
create_latest_log_alias(log_path, alias=alias)