lib/cts_helper.py

# -*- coding: utf-8 -*-
# Copyright 2017 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""CTS/GTS-related helper functions/classes."""
from __future__ import print_function

import os
import glob
import gzip
import shutil
import sys

from chromite.lib import constants
from chromite.lib import cros_logging as logging

# TODO(pwang): Move CTS upload logic from autotest to here for consistency.


assert sys.version_info >= (3, 6), 'This module requires Python 3.6+'


# Autotest test to collect list of CTS tests
TEST_LIST_COLLECTOR = 'tradefed-run-collect-tests-only'


def isCtsTest(test_name):
  """Check if the test is CTS/GTS tests."""
  for cts_prefix in constants.DEFAULT_CTS_TEST_XML_MAP.keys():
    if test_name.startswith(cts_prefix):
      return True
  return False


def getXMLPattern(test_name):
  """Return CTS result XML file pattern string."""
  for cts_prefix, resultXML in constants.DEFAULT_CTS_TEST_XML_MAP.items():
    if test_name.startswith(cts_prefix):
      return resultXML
  return None


def getXMLGZFiles(test_name, test_folder):
  """Get CTS/GTS XML file in gz format within test_folder.

  Args:
    test_name: name of the test.
    test_folder: path to test folder.

  Returns:
    A list of xml.gz files or []
  """
  if not isCtsTest(test_name):
    return []
  _D = '[0-9][0-9]'
  _TIMESTAMP_PATTERN = '%s%s.%s.%s_%s.%s.%s' % (_D, _D, _D, _D, _D, _D, _D)
  xml_pattern = getXMLPattern(test_name)
  xml_files = glob.glob(os.path.join(test_folder, '*', 'results', '*',
                                     _TIMESTAMP_PATTERN, xml_pattern))
  files = []
  for xml_file in xml_files:
    xml_file_gz = '%s.gz' % xml_file
    with open(xml_file, 'r') as f_in, (
        gzip.open(xml_file_gz, 'w')) as f_out:
      shutil.copyfileobj(f_in, f_out)
    files.append(xml_file_gz)
  return files


def getApfeFiles(test_name, test_folder):
  """Get CTS/GTS Apfe file within test_folder.

  Args:
    test_name: name of the test.
    test_folder: path to test folder.

  Returns:
    A list of APFE files or []
  """
  if not isCtsTest(test_name):
    return []
  _D = '[0-9][0-9]'
  _TIMESTAMP_PATTERN = '%s%s.%s.%s_%s.%s.%s' % (_D, _D, _D, _D, _D, _D, _D)
  apfe_files = glob.glob(os.path.join(test_folder, '*', 'results', '*',
                                      '%s.zip' % _TIMESTAMP_PATTERN))
  return apfe_files


def _is_test_collector(package):
  """Returns true if the test run is just to collect list of CTS tests.

  Args:
    package: Autotest package name. e.g. cheets_CTS_N.CtsGraphicsTestCase

  Returns:
    Bool flag indicating a test package is CTS list generator or not.
  """
  return TEST_LIST_COLLECTOR in package


def uploadFiles(dir_entry, build, apfe_id, job_id, package, uploader,
                *args, **kwargs):
  """Upload CTS/GTS tests result to gs buckets.

  Args:
    dir_entry: path to the test folder.
    build: build name such as samus-release, or kevin-release.
    apfe_id: id number used for apfe upload, typically we use autotest parent
             job id.
    job_id: id number, such as autotest_job_id or builder_id.
    package: CTS package name.
    uploader: upload function to upload to gs
  """
  xml_files = getXMLGZFiles(package, dir_entry)
  logging.info('Uploading CTS/GTS xml files: %s', xml_files)
  for xml in xml_files:
    timestamp = os.path.basename(os.path.dirname(xml))
    gs_url = os.path.join(constants.DEFAULT_CTS_RESULTS_GSURI, package,
                          build + '-' + job_id + '_' + timestamp)
    uploader(gs_url, xml, *args, **kwargs)

  # Results produced by CTS test list collector are dummy results.
  # They don't need to be copied to APFE bucket which is mainly being used for
  # CTS APFE submission.
  if not _is_test_collector(package):
    apfe_files = getApfeFiles(package, dir_entry)
    logging.info('Uploading CTS/GTS apfe files: %s', apfe_files)
    for apfe in apfe_files:
      timestamp = os.path.splitext(os.path.basename(apfe))[0]
      gs_url = os.path.join(constants.DEFAULT_CTS_APFE_GSURI, build,
                            apfe_id, package, job_id + '_' + timestamp)
      uploader(gs_url, apfe, *args, **kwargs)
  else:
    logging.debug('%s is a CTS Test collector Autotest test run.', package)
    logging.debug('Skipping CTS results upload to APFE gs:// bucket.')