py/test/session.py

# Copyright 2016 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.

"""Utilities for test session and invocation.

Test harness (usually Goofy in current implementation) should follow the
convention here to invoke a test.
"""

import logging
import os
import subprocess
import uuid

import factory_common  # pylint: disable=unused-import
from cros.factory.test.env import paths
from cros.factory.utils import file_utils
from cros.factory.utils import log_utils
from cros.factory.utils import type_utils


# Environment variable names.
ENV_TEST_PATH = 'CROS_FACTORY_TEST_PATH'
ENV_TEST_INVOCATION = 'CROS_FACTORY_TEST_INVOCATION'
ENV_TEST_METADATA = 'CROS_FACTORY_TEST_METADATA'
ENV_TEST_FILE_PATH = 'CROS_FACTORY_TEST_FILE_PATH'

LOG_ROOT = paths.DATA_LOG_DIR

DEVICE_ID_PATH = os.path.join(paths.DATA_DIR, '.device_id')
"""File containing a stable 'Device identifier'.

The file should be located somewhere less likely to be deleted.
The content should be the same even if system image is re-installed.
"""

INSTALLATION_ID_PATH = os.path.join(LOG_ROOT, 'installation_id')
"""File containing a GUID as identifier for software (image) installation.

The file should be deleted when the system software is re-installed,
and the content should be different when re-generated.
"""

INIT_COUNT_PATH = os.path.join(LOG_ROOT, 'init_count')
"""File containing the number of times session (Goofy) has been initialized."""


console = type_utils.LazyObject(
    log_utils.FileLogger, 'console', paths.CONSOLE_LOG_PATH,
    os.environ.get(ENV_TEST_PATH))
"""A wrapper for sending messages to global (UI) console using logging API."""


@type_utils.CachedGetter
def GetDeviceID():
  """Returns the device ID.

  The device ID is created and stored by init/goofy.d/device/device_id.sh
  calling bin/device_id on system startup.
  """
  if os.path.exists(DEVICE_ID_PATH):
    return file_utils.ReadFile(DEVICE_ID_PATH).strip()
  else:
    # The device_id file doesn't exist, we probably are not on DUT, just
    # run bin/device_id once and return the result.
    device_id_bin = os.path.join(paths.FACTORY_DIR, 'bin', 'device_id')
    return subprocess.check_output(device_id_bin).strip()


@type_utils.CachedGetter
def GetInstallationID():
  """Returns the installation ID.

  This is stored in INSTALLATION_ID_PATH; one is generated if not available.
  """
  value = None
  if os.path.exists(INSTALLATION_ID_PATH):
    value = file_utils.ReadFile(INSTALLATION_ID_PATH).strip()
  if not value:
    value = str(uuid.uuid4())
    file_utils.TryMakeDirs(os.path.dirname(INSTALLATION_ID_PATH))
    # There may be race condition here, but that is unlikely to happen due to
    # how we use GetInstallationID today.
    with file_utils.AtomicWrite(INSTALLATION_ID_PATH) as f:
      f.write(value)
    logging.info('No installation_id available yet: generated %s', value)
  return value


def GetInitCount(path=INIT_COUNT_PATH):
  """Returns the current session init count (or -1 if not available)."""
  # TODO(itspeter): Remove the path argument once event_log.py is phased out.
  try:
    return int(file_utils.ReadFile(path))
  except (IOError, ValueError):
    return -1


def IncrementInitCount(path=INIT_COUNT_PATH):
  """Increments the session init count.

  Creates the init count file if it does not already exist.
  """
  # TODO(itspeter): Remove the path argument once event_log.py is phased out.
  init_count = GetInitCount(path) + 1

  logging.info('Session (Goofy) init count = %d', init_count)

  file_utils.TryMakeDirs(os.path.dirname(path))
  with file_utils.AtomicWrite(path) as f:
    f.write('%d' % init_count)


@type_utils.CachedGetter
def GetBootID():
  """Returns the boot ID."""
  return file_utils.ReadFile('/proc/sys/kernel/random/boot_id').strip()


def GetToolkitVersion():
  """Returns TOOLKIT_VERSION of the factory directory."""
  return file_utils.ReadFile(paths.FACTORY_TOOLKIT_VERSION_PATH).rstrip()


@type_utils.CachedGetter
def GetCurrentTestPath():
  """Returns the path of the currently executing test, if any.

  This function may be cached because each invocation process should not have
  test path changed during execution.
  """
  return os.environ.get(ENV_TEST_PATH)


@type_utils.CachedGetter
def GetCurrentTestInvocation():
  """Returns the invocation UUID of current running test, if any.

  This function may be cached because each invocation process should not have
  test invocation changed during execution.
  """
  return os.environ.get(ENV_TEST_INVOCATION)


@type_utils.CachedGetter
def GetCurrentTestFilePath():
  """Returns the file path of the currently executing test, if any.

  This function may be cached because each invocation process should not have
  test file changed during execution.
  """
  return os.environ.get(ENV_TEST_FILE_PATH)


@type_utils.CachedGetter
def GetVerboseTestLogPath():
  """Returns a path for verbose logging of current test.

  The 'verbose test log' is a special log file that will be kept in log
  directory, and not merged into ``factory.log`` or ``testlog``.
  It was introduced to reduce log size sent to factory servers - helpful for
  debugging locally, but not meant for being stored if nothing goes wrong.

  The file name will contain test invocation ID and thus this method
  can only be called from a test.
  """
  log_name = '%s-log-%s' % (GetCurrentTestPath(), GetCurrentTestInvocation())
  return os.path.join(paths.DATA_LOG_DIR, log_name)