text: #!/usr/bin/env python
  # Copyright 2015 gRPC authors.
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #     http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  """Run interop (cross-language) tests in parallel."""
  
  from __future__ import print_function
  
  import argparse
  import atexit
  import itertools
  import json
  import multiprocessing
  import os
  import re
  import subprocess
  import sys
  import tempfile
  import time
  import uuid
  import six
  import traceback
  
  import python_utils.dockerjob as dockerjob
  import python_utils.jobset as jobset
  import python_utils.report_utils as report_utils
  
  # Docker doesn't clean up after itself, so we do it on exit.
  atexit.register(lambda: subprocess.call(['stty', 'echo']))
  
  ROOT = os.path.abspath(os.path.join(os.path.dirname(sys.argv[0]), '../..'))
  os.chdir(ROOT)
  
  _FALLBACK_SERVER_PORT = 443
  _BALANCER_SERVER_PORT = 12000
  _BACKEND_SERVER_PORT = 8080
  
  _TEST_TIMEOUT = 30
  
  _FAKE_SERVERS_SAFENAME = 'fake_servers'
  
  # Use a name that's verified by the test certs
  _SERVICE_NAME = 'server.test.google.fr'
  
  
  class CXXLanguage:
  
      def __init__(self):
          self.client_cwd = '/var/local/git/grpc'
          self.safename = 'cxx'
  
      def client_cmd(self, args):
          return ['bins/opt/interop_client'] + args
  
      def global_env(self):
          # 1) Set c-ares as the resolver, to
          #    enable grpclb.
          # 2) Turn on verbose logging.
          # 3) Set the ROOTS_PATH env variable
          #    to the test CA in order for
          #    GoogleDefaultCredentials to be
          #    able to use the test CA.
          return {
              'GRPC_DNS_RESOLVER':
              'ares',
              'GRPC_VERBOSITY':
              'DEBUG',
              'GRPC_TRACE':
              'client_channel,glb',
              'GRPC_DEFAULT_SSL_ROOTS_FILE_PATH':
              '/var/local/git/grpc/src/core/tsi/test_creds/ca.pem',
          }
  
      def __str__(self):
          return 'c++'
  
  
  class JavaLanguage:
  
      def __init__(self):
          self.client_cwd = '/var/local/git/grpc-java'
          self.safename = str(self)
  
      def client_cmd(self, args):
          # Take necessary steps to import our test CA into
          # the set of test CA's that the Java runtime of the
          # docker container will pick up, so that
          # Java GoogleDefaultCreds can use it.
          pem_to_der_cmd = ('openssl x509 -outform der '
                            '-in /external_mount/src/core/tsi/test_creds/ca.pem '
                            '-out /tmp/test_ca.der')
          keystore_import_cmd = (
              'keytool -import '
              '-keystore /usr/lib/jvm/java-8-oracle/jre/lib/security/cacerts '
              '-file /tmp/test_ca.der '
              '-deststorepass changeit '
              '-noprompt')
          return [
              'bash', '-c', ('{pem_to_der_cmd} && '
                             '{keystore_import_cmd} && '
                             './run-test-client.sh {java_client_args}').format(
                                 pem_to_der_cmd=pem_to_der_cmd,
                                 keystore_import_cmd=keystore_import_cmd,
                                 java_client_args=' '.join(args))
          ]
  
      def global_env(self):
          # 1) Enable grpclb
          # 2) Enable verbose logging
          return {
              'JAVA_OPTS':
              ('-Dio.grpc.internal.DnsNameResolverProvider.enable_grpclb=true '
               '-Djava.util.logging.config.file=/var/local/grpc_java_logging/logconf.txt'
              )
          }
  
      def __str__(self):
          return 'java'
  
  
  class GoLanguage:
  
      def __init__(self):
          self.client_cwd = '/go/src/google.golang.org/grpc/interop/client'
          self.safename = str(self)
  
      def client_cmd(self, args):
          # Copy the test CA file into the path that
          # the Go runtime in the docker container will use, so
          # that Go's GoogleDefaultCredentials can use it.
          # See https://golang.org/src/crypto/x509/root_linux.go.
          return [
              'bash', '-c', ('cp /external_mount/src/core/tsi/test_creds/ca.pem '
                             '/etc/ssl/certs/ca-certificates.crt && '
                             '/go/bin/client {go_client_args}'
                            ).format(go_client_args=' '.join(args))
          ]
  
      def global_env(self):
          return {
              'GRPC_GO_LOG_VERBOSITY_LEVEL': '3',
              'GRPC_GO_LOG_SEVERITY_LEVEL': 'INFO'
          }
  
      def __str__(self):
          return 'go'
  
  
  _LANGUAGES = {
      'c++': CXXLanguage(),
      'go': GoLanguage(),
      'java': JavaLanguage(),
  }
  
  
  def docker_run_cmdline(cmdline, image, docker_args, cwd, environ=None):
      """Wraps given cmdline array to create 'docker run' cmdline from it."""
      # turn environ into -e docker args
      docker_cmdline = 'docker run -i --rm=true'.split()
      if environ:
          for k, v in environ.items():
              docker_cmdline += ['-e', '%s=%s' % (k, v)]
      return docker_cmdline + ['-w', cwd] + docker_args + [image] + cmdline
  
  
  def _job_kill_handler(job):
      assert job._spec.container_name
      dockerjob.docker_kill(job._spec.container_name)
  
  
  def transport_security_to_args(transport_security):
      args = []
      if transport_security == 'tls':
          args += ['--use_tls=true']
      elif transport_security == 'alts':
          args += ['--use_tls=false', '--use_alts=true']
      elif transport_security == 'insecure':
          args += ['--use_tls=false']
      elif transport_security == 'google_default_credentials':
          args += ['--custom_credentials_type=google_default_credentials']
      else:
          print('Invalid transport security option.')
          sys.exit(1)
      return args
  
  
  def lb_client_interop_jobspec(language,
                                dns_server_ip,
                                docker_image,
                                transport_security='tls'):
      """Runs a gRPC client under test in a docker container"""
      interop_only_options = [
          '--server_host=%s' % _SERVICE_NAME,
          '--server_port=%d' % _FALLBACK_SERVER_PORT
      ] + transport_security_to_args(transport_security)
      # Don't set the server host override in any client;
      # Go and Java default to no override.
      # We're using a DNS server so there's no need.
      if language.safename == 'c++':
          interop_only_options += ['--server_host_override=""']
      # Don't set --use_test_ca; we're configuring
      # clients to use test CA's via alternate means.
      interop_only_options += ['--use_test_ca=false']
      client_args = language.client_cmd(interop_only_options)
      container_name = dockerjob.random_name(
          'lb_interop_client_%s' % language.safename)
      docker_cmdline = docker_run_cmdline(
          client_args,
          environ=language.global_env(),
          image=docker_image,
          cwd=language.client_cwd,
          docker_args=[
              '--dns=%s' % dns_server_ip,
              '--net=host',
              '--name=%s' % container_name,
              '-v',
              '{grpc_grpc_root_dir}:/external_mount:ro'.format(
                  grpc_grpc_root_dir=ROOT),
          ])
      jobset.message(
          'IDLE',
          'docker_cmdline:\b|%s|' % ' '.join(docker_cmdline),
          do_newline=True)
      test_job = jobset.JobSpec(
          cmdline=docker_cmdline,
          shortname=('lb_interop_client:%s' % language),
          timeout_seconds=_TEST_TIMEOUT,
          kill_handler=_job_kill_handler)
      test_job.container_name = container_name
      return test_job
  
  
  def fallback_server_jobspec(transport_security, shortname):
      """Create jobspec for running a fallback server"""
      cmdline = [
          'bin/server',
          '--port=%d' % _FALLBACK_SERVER_PORT,
      ] + transport_security_to_args(transport_security)
      return grpc_server_in_docker_jobspec(
          server_cmdline=cmdline, shortname=shortname)
  
  
  def backend_server_jobspec(transport_security, shortname):
      """Create jobspec for running a backend server"""
      cmdline = [
          'bin/server',
          '--port=%d' % _BACKEND_SERVER_PORT,
      ] + transport_security_to_args(transport_security)
      return grpc_server_in_docker_jobspec(
          server_cmdline=cmdline, shortname=shortname)
  
  
  def grpclb_jobspec(transport_security, short_stream, backend_addrs, shortname):
      """Create jobspec for running a balancer server"""
      cmdline = [
          'bin/fake_grpclb',
          '--backend_addrs=%s' % ','.join(backend_addrs),
          '--port=%d' % _BALANCER_SERVER_PORT,
          '--short_stream=%s' % short_stream,
          '--service_name=%s' % _SERVICE_NAME,
      ] + transport_security_to_args(transport_security)
      return grpc_server_in_docker_jobspec(
          server_cmdline=cmdline, shortname=shortname)
  
  
  def grpc_server_in_docker_jobspec(server_cmdline, shortname):
      container_name = dockerjob.random_name(shortname)
      environ = {
          'GRPC_GO_LOG_VERBOSITY_LEVEL': '3',
          'GRPC_GO_LOG_SEVERITY_LEVEL': 'INFO ',
      }
      docker_cmdline = docker_run_cmdline(
          server_cmdline,
          cwd='/go',
          image=docker_images.get(_FAKE_SERVERS_SAFENAME),
          environ=environ,
          docker_args=['--name=%s' % container_name])
      jobset.message(
          'IDLE',
          'docker_cmdline:\b|%s|' % ' '.join(docker_cmdline),
          do_newline=True)
      server_job = jobset.JobSpec(
          cmdline=docker_cmdline, shortname=shortname, timeout_seconds=30 * 60)
      server_job.container_name = container_name
      return server_job
  
  
  def dns_server_in_docker_jobspec(grpclb_ips, fallback_ips, shortname,
                                   cause_no_error_no_data_for_balancer_a_record):
      container_name = dockerjob.random_name(shortname)
      run_dns_server_cmdline = [
          'python',
          'test/cpp/naming/utils/run_dns_server_for_lb_interop_tests.py',
          '--grpclb_ips=%s' % ','.join(grpclb_ips),
          '--fallback_ips=%s' % ','.join(fallback_ips),
      ]
      if cause_no_error_no_data_for_balancer_a_record:
          run_dns_server_cmdline.append(
              '--cause_no_error_no_data_for_balancer_a_record')
      docker_cmdline = docker_run_cmdline(
          run_dns_server_cmdline,
          cwd='/var/local/git/grpc',
          image=docker_images.get(_FAKE_SERVERS_SAFENAME),
          docker_args=['--name=%s' % container_name])
      jobset.message(
          'IDLE',
          'docker_cmdline:\b|%s|' % ' '.join(docker_cmdline),
          do_newline=True)
      server_job = jobset.JobSpec(
          cmdline=docker_cmdline, shortname=shortname, timeout_seconds=30 * 60)
      server_job.container_name = container_name
      return server_job
  
  
  def build_interop_image_jobspec(lang_safename, basename_prefix='grpc_interop'):
      """Creates jobspec for building interop docker image for a language"""
      tag = '%s_%s:%s' % (basename_prefix, lang_safename, uuid.uuid4())
      env = {
          'INTEROP_IMAGE': tag,
          'BASE_NAME': '%s_%s' % (basename_prefix, lang_safename),
      }
      build_job = jobset.JobSpec(
          cmdline=['tools/run_tests/dockerize/build_interop_image.sh'],
          environ=env,
          shortname='build_docker_%s' % lang_safename,
          timeout_seconds=30 * 60)
      build_job.tag = tag
      return build_job
  
  
  argp = argparse.ArgumentParser(description='Run interop tests.')
  argp.add_argument(
      '-l',
      '--language',
      choices=['all'] + sorted(_LANGUAGES),
      nargs='+',
      default=['all'],
      help='Clients to run.')
  argp.add_argument('-j', '--jobs', default=multiprocessing.cpu_count(), type=int)
  argp.add_argument(
      '-s',
      '--scenarios_file',
      default=None,
      type=str,
      help='File containing test scenarios as JSON configs.')
  argp.add_argument(
      '-n',
      '--scenario_name',
      default=None,
      type=str,
      help=(
          'Useful for manual runs: specify the name of '
          'the scenario to run from scenarios_file. Run all scenarios if unset.'))
  argp.add_argument(
      '--cxx_image_tag',
      default=None,
      type=str,
      help=('Setting this skips the clients docker image '
            'build step and runs the client from the named '
            'image. Only supports running a one client language.'))
  argp.add_argument(
      '--go_image_tag',
      default=None,
      type=str,
      help=('Setting this skips the clients docker image build '
            'step and runs the client from the named image. Only '
            'supports running a one client language.'))
  argp.add_argument(
      '--java_image_tag',
      default=None,
      type=str,
      help=('Setting this skips the clients docker image build '
            'step and runs the client from the named image. Only '
            'supports running a one client language.'))
  argp.add_argument(
      '--servers_image_tag',
      default=None,
      type=str,
      help=('Setting this skips the fake servers docker image '
            'build step and runs the servers from the named image.'))
  argp.add_argument(
      '--no_skips',
      default=False,
      type=bool,
      nargs='?',
      const=True,
      help=('Useful for manual runs. Setting this overrides test '
            '"skips" configured in test scenarios.'))
  argp.add_argument(
      '--verbose',
      default=False,
      type=bool,
      nargs='?',
      const=True,
      help='Increase logging.')
  args = argp.parse_args()
  
  docker_images = {}
  
  build_jobs = []
  if len(args.language) and args.language[0] == 'all':
      languages = _LANGUAGES.keys()
  else:
      languages = args.language
  for lang_name in languages:
      l = _LANGUAGES[lang_name]
      # First check if a pre-built image was supplied, and avoid
      # rebuilding the particular docker image if so.
      if lang_name == 'c++' and args.cxx_image_tag:
          docker_images[str(l.safename)] = args.cxx_image_tag
      elif lang_name == 'go' and args.go_image_tag:
          docker_images[str(l.safename)] = args.go_image_tag
      elif lang_name == 'java' and args.java_image_tag:
          docker_images[str(l.safename)] = args.java_image_tag
      else:
          # Build the test client in docker and save the fully
          # built image.
          job = build_interop_image_jobspec(l.safename)
          build_jobs.append(job)
          docker_images[str(l.safename)] = job.tag
  
  # First check if a pre-built image was supplied.
  if args.servers_image_tag:
      docker_images[_FAKE_SERVERS_SAFENAME] = args.servers_image_tag
  else:
      # Build the test servers in docker and save the fully
      # built image.
      job = build_interop_image_jobspec(
          _FAKE_SERVERS_SAFENAME, basename_prefix='lb_interop')
      build_jobs.append(job)
      docker_images[_FAKE_SERVERS_SAFENAME] = job.tag
  
  if build_jobs:
      jobset.message('START', 'Building interop docker images.', do_newline=True)
      print('Jobs to run: \n%s\n' % '\n'.join(str(j) for j in build_jobs))
      num_failures, _ = jobset.run(
          build_jobs, newline_on_success=True, maxjobs=args.jobs)
      if num_failures == 0:
          jobset.message(
              'SUCCESS', 'All docker images built successfully.', do_newline=True)
      else:
          jobset.message(
              'FAILED', 'Failed to build interop docker images.', do_newline=True)
          sys.exit(1)
  
  
  def wait_until_dns_server_is_up(dns_server_ip):
      """Probes the DNS server until it's running and safe for tests."""
      for i in range(0, 30):
          print('Health check: attempt to connect to DNS server over TCP.')
          tcp_connect_subprocess = subprocess.Popen([
              os.path.join(os.getcwd(), 'test/cpp/naming/utils/tcp_connect.py'),
              '--server_host', dns_server_ip, '--server_port',
              str(53), '--timeout',
              str(1)
          ])
          tcp_connect_subprocess.communicate()
          if tcp_connect_subprocess.returncode == 0:
              print(('Health check: attempt to make an A-record '
                     'query to DNS server.'))
              dns_resolver_subprocess = subprocess.Popen(
                  [
                      os.path.join(os.getcwd(),
                                   'test/cpp/naming/utils/dns_resolver.py'),
                      '--qname', ('health-check-local-dns-server-is-alive.'
                                  'resolver-tests.grpctestingexp'),
                      '--server_host', dns_server_ip, '--server_port',
                      str(53)
                  ],
                  stdout=subprocess.PIPE)
              dns_resolver_stdout, _ = dns_resolver_subprocess.communicate()
              if dns_resolver_subprocess.returncode == 0:
                  if '123.123.123.123' in dns_resolver_stdout:
                      print(('DNS server is up! '
                             'Successfully reached it over UDP and TCP.'))
                      return
          time.sleep(0.1)
      raise Exception(('Failed to reach DNS server over TCP and/or UDP. '
                       'Exitting without running tests.'))
  
  
  def shortname(shortname_prefix, shortname, index):
      return '%s_%s_%d' % (shortname_prefix, shortname, index)
  
  
  def run_one_scenario(scenario_config):
      jobset.message('START', 'Run scenario: %s' % scenario_config['name'])
      server_jobs = {}
      server_addresses = {}
      suppress_server_logs = True
      try:
          backend_addrs = []
          fallback_ips = []
          grpclb_ips = []
          shortname_prefix = scenario_config['name']
          # Start backends
          for i in xrange(len(scenario_config['backend_configs'])):
              backend_config = scenario_config['backend_configs'][i]
              backend_shortname = shortname(shortname_prefix, 'backend_server', i)
              backend_spec = backend_server_jobspec(
                  backend_config['transport_sec'], backend_shortname)
              backend_job = dockerjob.DockerJob(backend_spec)
              server_jobs[backend_shortname] = backend_job
              backend_addrs.append('%s:%d' % (backend_job.ip_address(),
                                              _BACKEND_SERVER_PORT))
          # Start fallbacks
          for i in xrange(len(scenario_config['fallback_configs'])):
              fallback_config = scenario_config['fallback_configs'][i]
              fallback_shortname = shortname(shortname_prefix, 'fallback_server',
                                             i)
              fallback_spec = fallback_server_jobspec(
                  fallback_config['transport_sec'], fallback_shortname)
              fallback_job = dockerjob.DockerJob(fallback_spec)
              server_jobs[fallback_shortname] = fallback_job
              fallback_ips.append(fallback_job.ip_address())
          # Start balancers
          for i in xrange(len(scenario_config['balancer_configs'])):
              balancer_config = scenario_config['balancer_configs'][i]
              grpclb_shortname = shortname(shortname_prefix, 'grpclb_server', i)
              grpclb_spec = grpclb_jobspec(balancer_config['transport_sec'],
                                           balancer_config['short_stream'],
                                           backend_addrs, grpclb_shortname)
              grpclb_job = dockerjob.DockerJob(grpclb_spec)
              server_jobs[grpclb_shortname] = grpclb_job
              grpclb_ips.append(grpclb_job.ip_address())
          # Start DNS server
          dns_server_shortname = shortname(shortname_prefix, 'dns_server', 0)
          dns_server_spec = dns_server_in_docker_jobspec(
              grpclb_ips, fallback_ips, dns_server_shortname,
              scenario_config['cause_no_error_no_data_for_balancer_a_record'])
          dns_server_job = dockerjob.DockerJob(dns_server_spec)
          server_jobs[dns_server_shortname] = dns_server_job
          # Get the IP address of the docker container running the DNS server.
          # The DNS server is running on port 53 of that IP address. Note we will
          # point the DNS resolvers of grpc clients under test to our controlled
          # DNS server by effectively modifying the /etc/resolve.conf "nameserver"
          # lists of their docker containers.
          dns_server_ip = dns_server_job.ip_address()
          wait_until_dns_server_is_up(dns_server_ip)
          # Run clients
          jobs = []
          for lang_name in languages:
              # Skip languages that are known to not currently
              # work for this test.
              if not args.no_skips and lang_name in scenario_config.get(
                      'skip_langs', []):
                  jobset.message('IDLE',
                                 'Skipping scenario: %s for language: %s\n' %
                                 (scenario_config['name'], lang_name))
                  continue
              lang = _LANGUAGES[lang_name]
              test_job = lb_client_interop_jobspec(
                  lang,
                  dns_server_ip,
                  docker_image=docker_images.get(lang.safename),
                  transport_security=scenario_config['transport_sec'])
              jobs.append(test_job)
          jobset.message('IDLE', 'Jobs to run: \n%s\n' % '\n'.join(
              str(job) for job in jobs))
          num_failures, resultset = jobset.run(
              jobs, newline_on_success=True, maxjobs=args.jobs)
          report_utils.render_junit_xml_report(resultset, 'sponge_log.xml')
          if num_failures:
              suppress_server_logs = False
              jobset.message(
                  'FAILED',
                  'Scenario: %s. Some tests failed' % scenario_config['name'],
                  do_newline=True)
          else:
              jobset.message(
                  'SUCCESS',
                  'Scenario: %s. All tests passed' % scenario_config['name'],
                  do_newline=True)
          return num_failures
      finally:
          # Check if servers are still running.
          for server, job in server_jobs.items():
              if not job.is_running():
                  print('Server "%s" has exited prematurely.' % server)
          suppress_failure = suppress_server_logs and not args.verbose
          dockerjob.finish_jobs(
              [j for j in six.itervalues(server_jobs)],
              suppress_failure=suppress_failure)
  
  
  num_failures = 0
  with open(args.scenarios_file, 'r') as scenarios_input:
      all_scenarios = json.loads(scenarios_input.read())
      for scenario in all_scenarios:
          if args.scenario_name:
              if args.scenario_name != scenario['name']:
                  jobset.message('IDLE',
                                 'Skipping scenario: %s' % scenario['name'])
                  continue
          num_failures += run_one_scenario(scenario)
  if num_failures == 0:
      sys.exit(0)
  else:
      sys.exit(1)
  