src/bin/config_parser/test_source_generator_for_all_projects.py

#!/usr/bin/env python3
# Copyright 2018 The Fuchsia Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Checks that the output of the config parser is stable across runs."""

import os
import subprocess
import sys
import tempfile

THIS_DIR = os.path.abspath(os.path.dirname(__file__))
SRC_ROOT_DIR = os.path.abspath(
    os.path.join(THIS_DIR, os.pardir, os.pardir, os.pardir)
)

sys.path += [os.path.join(SRC_ROOT_DIR, 'third_party', 'pyyaml', 'lib')]
import yaml

OUT_DIR = os.path.abspath(os.path.join(SRC_ROOT_DIR, 'out'))
CONFIG_PARSER_BIN = os.path.join(OUT_DIR, 'config_parser')
CONFIG_DIR = os.path.join(SRC_ROOT_DIR, 'third_party', 'cobalt_config')

SYSROOT_BIN = os.path.join(SRC_ROOT_DIR, 'sysroot', 'bin')
CLANG = os.path.join(SYSROOT_BIN, 'clang++')
RUSTC = os.path.join(SYSROOT_BIN, 'rustc')
GO_BIN = os.path.join(SRC_ROOT_DIR, 'sysroot', 'golang', 'bin', 'go')
LINUX_SYSROOT = os.path.join(SRC_ROOT_DIR, 'sysroot')


def main():
  registry = []
  with open(os.path.join(CONFIG_DIR, 'projects.yaml'), 'r') as stream:
    registry = yaml.safe_load(stream)

  running_compiles = []
  for customer in registry['customers']:
    customer_id = customer['customer_id']
    if customer['projects'] is None:
      # There are no projects defined for the customer.
      continue
    for project in customer['projects']:
      project_id = project['project_id']
      for format in ['cpp', 'rust', 'go', 'json']:
        cmd = [
            CONFIG_PARSER_BIN,
            '-config_dir',
            CONFIG_DIR,
            '-out_format',
            format,
            '-features',
            'generate-dimension-name-maps',
        ]
        cmd.extend(['-customer_id', str(customer_id)])
        cmd.extend(['-project_id', str(project_id)])
        if format == 'go':
          cmd.append('-go_package')
          cmd.append(
              '%s_%s' % (customer['customer_name'], project['project_name'])
          )
        out1 = subprocess.run(cmd, capture_output=True, check=True).stdout
        compile = None
        src = None
        tmpdir = None
        if format == 'cpp':
          src = tempfile.NamedTemporaryFile(suffix='.hpp')
          src.write(out1)
          src.flush()
          cmd = [CLANG, '-std=c++17']
          if sys.platform == 'linux':
            cmd += ['--sysroot=' + LINUX_SYSROOT]
          cmd += [src.name]
          compile = subprocess.Popen(cmd)
        elif format == 'rust':
          src = tempfile.NamedTemporaryFile(
              prefix=project['project_name'], suffix='.rs'
          )
          src.write(out1)
          src.flush()
          tmpdir = tempfile.TemporaryDirectory()
          cmd = [
              RUSTC,
              '--out-dir',
              tmpdir.name,
              '--edition=2018',
              '--crate-type=lib',
              '--crate-name=cobalt_client',
          ]
          if sys.platform == 'linux':
            cmd += ['-Clink-arg=--sysroot=' + LINUX_SYSROOT]
          cmd += [
              os.path.join(
                  SRC_ROOT_DIR, 'src', 'lib', 'client', 'rust', 'src', 'lib.rs'
              )
          ]
          subprocess.run(cmd, check=True)
          compile_cmd = [
              RUSTC,
              '--out-dir',
              tmpdir.name,
              '--edition=2018',
              '--extern',
              'cobalt_client=%s/libcobalt_client.rlib' % tmpdir.name,
              '--crate-type=lib',
          ]
          if sys.platform == 'linux':
            compile_cmd += ['-Clink-arg=--sysroot=' + LINUX_SYSROOT]
          compile_cmd += [src.name]
          compile = subprocess.Popen(
              compile_cmd, stderr=subprocess.STDOUT, stdout=subprocess.PIPE
          )
        elif format == 'go':
          src = tempfile.NamedTemporaryFile(suffix='.go')
          src.write(out1)
          src.flush()
          env = os.environ.copy()
          env['CGO_ENABLED'] = '0'
          compile = subprocess.Popen([GO_BIN, 'build', '-a', src.name], env=env)

        if compile:
          running_compiles.append((
              customer['customer_name'],
              project['project_name'],
              format,
              src,
              tmpdir,
              compile,
          ))

  failures = []
  for running_compile in running_compiles:
    customer_name, project_name, format, src, tmpdir, compile = running_compile
    compile.wait()
    if compile.returncode != 0:
      stdout, _ = compile.communicate()
      print(stdout)
      failures.append(
          'Failed to compile generated source for project: %s/%s (language: %s)'
          % (customer_name, project_name, format)
      )

  return failures


if __name__ == '__main__':
  errors = main()
  for error in errors:
    print(error)
  sys.exit(len(errors))