chromeos-config/cros_config_host/cros_config_test_schema.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright 2018 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.

"""Transforms and validates cros config test from source YAML to target JSON"""

from __future__ import print_function

import argparse
import json
import os
import sys

import yaml  # pylint: disable=import-error

# pylint: disable=wrong-import-position
this_dir = os.path.dirname(__file__)
sys.path.insert(0, this_dir)
import libcros_schema
sys.path.pop(0)

default_test_schema = os.path.join(this_dir, 'cros_config_test_schema.yaml')

CHROMEOS = 'chromeos'
DEVICES = 'devices'
ROOT_PATH = 'properties/chromeos/properties/devices/items/properties'


def ParseArgs(argv):
  """Parse the available arguments.

  Invalid arguments or -h cause this function to print a message and exit.

  Args:
    argv: List of string arguments (excluding program name / argv[0])

  Returns:
    argparse.Namespace object containing the attributes.
  """
  parser = argparse.ArgumentParser(description=__doc__)
  parser.add_argument(
      '-s',
      '--schema',
      type=str,
      help='Path to the schema file used to validate the config')
  parser.add_argument(
      '-c',
      '--config',
      type=str,
      help='Path to the YAML config file that will be validated/transformed',
      required=True)
  parser.add_argument(
      '-f',
      '--filter',
      type=str,
      help='Filter device by name')
  parser.add_argument(
      '-o',
      '--output',
      type=str,
      help='Output file that will be generated by the transform (system file)',
      required=True)
  return parser.parse_args(argv)


def TransformConfig(config, device_filter=None):
  """Transforms the source config (YAML) to the target system format (JSON)

  Applies consistent transforms to covert a source YAML configuration into
  JSON output that will be used by the tast test program.

  Args:
    config: Config that will be transformed.
    device_filter: Only returns configs that match the filter.

  Returns:
    Resulting JSON output from the transform.
  """
  config_yaml = yaml.load(config, Loader=yaml.SafeLoader)
  json_from_yaml = json.dumps(config_yaml, sort_keys=True, indent=2)
  json_config = json.loads(json_from_yaml)
  configs = []
  if DEVICES in json_config[CHROMEOS]:
    for device in json_config[CHROMEOS][DEVICES]:
      configs.append(device)

  if device_filter:
    configs = [
        config for config in configs if device_filter == config['device-name']
    ]

  # Drop everything except for devices since they were just used as shared
  # config in the source yaml.
  json_config = {
      CHROMEOS: {
          DEVICES: configs,
      },
  }

  return libcros_schema.FormatJson(json_config)


def MergeConfig(yaml_file, filter_name):
  """Evaluates and merges all config files into a single configuration.

  Args:
    yaml_file: List of source config files that will be transformed/merged.
    filter_name: Name of device to filter on.

  Returns:
    Final merged JSON result.
  """
  yaml_with_imports = libcros_schema.ApplyImports(yaml_file)
  json_transformed_file = TransformConfig(yaml_with_imports, filter_name)
  return json_transformed_file


def Start(config, filter_name, output, schema):
  """Transforms and validates a cros config test file for use on the system

  Applies consistent transforms to covert a source YAML configuration into
  a JSON file that will be used on the system by cros_config tast tests.

  Verifies that the file complies with the schema verification rules and
  performs additional verification checks for config consistency.

  Args:
    config: Source config file that will be transformed/verified.
    filter_name: Device name to filter on.
    output: Output file that will be generated by the transform.
    schema: Schema file used to verify the config.
  """
  json_transform = MergeConfig(config, filter_name)

  if schema is None:
    schema = default_test_schema
  with open(schema, 'r') as schema_stream:
    libcros_schema.ValidateConfigSchema(
        schema_stream.read(), libcros_schema.FormatJson(json_transform))

  if output:
    with open(output, 'w') as output_stream:
      # Using print function adds proper trailing newline.
      print(json_transform, file=output_stream)
  else:
    print(json_transform)


# The distutils generated command line wrappers will not pass us argv.
def main(argv=None):
  """Main program which parses sys.argv and runs

  Args:
    argv: List of command line arguments, if None uses sys.argv.
  """
  if argv is None:
    argv = sys.argv[1:]
  opts = ParseArgs(argv)
  Start(opts.config, opts.filter, opts.output, opts.schema)


if __name__ == '__main__':
  sys.exit(main(sys.argv[1:]))