#!/usr/bin/env python3

# 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."""

import argparse

import atexit

import itertools

import json

import multiprocessing

import os

import re

import subprocess

import sys

import time

import uuid

import python_utils.dockerjob as dockerjob

import python_utils.jobset as jobset

import python_utils.report_utils as report_utils

# It's ok to not import because this is only necessary to upload results to BQ.

try:

from python_utils.upload_test_results import upload_interop_results_to_bq

except ImportError as e:

print(e)

# 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)

_DEFAULT_SERVER_PORT = 8080

_SKIP_CLIENT_COMPRESSION = [

"client_compressed_unary",

"client_compressed_streaming",

]

_SKIP_SERVER_COMPRESSION = [

"server_compressed_unary",

"server_compressed_streaming",

]

_SKIP_COMPRESSION = _SKIP_CLIENT_COMPRESSION + _SKIP_SERVER_COMPRESSION

_SKIP_ADVANCED = [

"status_code_and_message",

"custom_metadata",

"unimplemented_method",

"unimplemented_service",

]

_SKIP_SPECIAL_STATUS_MESSAGE = ["special_status_message"]

_ORCA_TEST_CASES = ["orca_per_rpc", "orca_oob"]

_GOOGLE_DEFAULT_CREDS_TEST_CASE = "google_default_credentials"

_SKIP_GOOGLE_DEFAULT_CREDS = [

_GOOGLE_DEFAULT_CREDS_TEST_CASE,

]

_COMPUTE_ENGINE_CHANNEL_CREDS_TEST_CASE = "compute_engine_channel_credentials"

_SKIP_COMPUTE_ENGINE_CHANNEL_CREDS = [

_COMPUTE_ENGINE_CHANNEL_CREDS_TEST_CASE,

]

_TEST_TIMEOUT = 3 * 60

# disable this test on core-based languages,

# see https://github.com/grpc/grpc/issues/9779

_SKIP_DATA_FRAME_PADDING = ["data_frame_padding"]

# report suffix "sponge_log.xml" is important for reports to get picked up by internal CI

_DOCKER_BUILD_XML_REPORT = "interop_docker_build/sponge_log.xml"

_TESTS_XML_REPORT = "interop_test/sponge_log.xml"

class CXXLanguage:

def __init__(self):

self.client_cwd = None

self.server_cwd = None

self.http2_cwd = None

self.safename = "cxx"

def client_cmd(self, args):

return ["cmake/build/interop_client"] + args

def client_cmd_http2interop(self, args):

return ["cmake/build/http2_client"] + args

def cloud_to_prod_env(self):

return {}

def server_cmd(self, args):

return ["cmake/build/interop_server"] + args

def global_env(self):

return {}

def unimplemented_test_cases(self):

return (

_SKIP_DATA_FRAME_PADDING

+ _SKIP_SPECIAL_STATUS_MESSAGE

+ _SKIP_COMPUTE_ENGINE_CHANNEL_CREDS

)

def unimplemented_test_cases_server(self):

return []

def __str__(self):

return "c++"

class AspNetCoreLanguage:

def __init__(self):

self.client_cwd = "../grpc-dotnet/output/InteropTestsClient"

self.server_cwd = "../grpc-dotnet/output/InteropTestsWebsite"

self.safename = str(self)

def cloud_to_prod_env(self):

return {}

def client_cmd(self, args):

return ["dotnet", "exec", "InteropTestsClient.dll"] + args

def server_cmd(self, args):

return ["dotnet", "exec", "InteropTestsWebsite.dll"] + args

def global_env(self):

return {}

def unimplemented_test_cases(self):

return (

_SKIP_GOOGLE_DEFAULT_CREDS

+ _SKIP_COMPUTE_ENGINE_CHANNEL_CREDS

+ _ORCA_TEST_CASES

)

def unimplemented_test_cases_server(self):

return _ORCA_TEST_CASES

def __str__(self):

return "aspnetcore"

class DartLanguage:

def __init__(self):

self.client_cwd = "../grpc-dart/interop"

self.server_cwd = "../grpc-dart/interop"

self.http2_cwd = "../grpc-dart/interop"

self.safename = str(self)

def client_cmd(self, args):

return ["dart", "bin/client.dart"] + args

def cloud_to_prod_env(self):

return {}

def server_cmd(self, args):

return ["dart", "bin/server.dart"] + args

def global_env(self):

return {}

def unimplemented_test_cases(self):

return (

_SKIP_COMPRESSION

+ _SKIP_SPECIAL_STATUS_MESSAGE

+ _SKIP_GOOGLE_DEFAULT_CREDS

+ _SKIP_COMPUTE_ENGINE_CHANNEL_CREDS

+ _ORCA_TEST_CASES

)

def unimplemented_test_cases_server(self):

return (

_SKIP_COMPRESSION + _SKIP_SPECIAL_STATUS_MESSAGE + _ORCA_TEST_CASES

)

def __str__(self):

return "dart"

class JavaLanguage:

def __init__(self):

self.client_cwd = "../grpc-java"

self.server_cwd = "../grpc-java"

self.http2_cwd = "../grpc-java"

self.safename = str(self)

def client_cmd(self, args):

return ["./run-test-client.sh"] + args

def client_cmd_http2interop(self, args):

return [

"./interop-testing/build/install/grpc-interop-testing/bin/http2-client"

] + args

def cloud_to_prod_env(self):

return {}

def server_cmd(self, args):

return ["./run-test-server.sh"] + args

def global_env(self):

return {}

def unimplemented_test_cases(self):

return []

def unimplemented_test_cases_server(self):

# Does not support CompressedRequest feature.

# Only supports CompressedResponse feature for unary.

return _SKIP_CLIENT_COMPRESSION + ["server_compressed_streaming"]

def __str__(self):

return "java"

class JavaOkHttpClient:

def __init__(self):

self.client_cwd = "../grpc-java"

self.safename = "java"

def client_cmd(self, args):

return ["./run-test-client.sh", "--use_okhttp=true"] + args

def cloud_to_prod_env(self):

return {}

def global_env(self):

return {}

def unimplemented_test_cases(self):

return _SKIP_DATA_FRAME_PADDING

def __str__(self):

return "javaokhttp"

class GoLanguage:

def __init__(self):

# TODO: this relies on running inside docker

self.client_cwd = "/go/src/google.golang.org/grpc/interop/client"

self.server_cwd = "/go/src/google.golang.org/grpc/interop/server"

self.http2_cwd = "/go/src/google.golang.org/grpc/interop/http2"

self.safename = str(self)

def client_cmd(self, args):

return ["go", "run", "client.go"] + args

def client_cmd_http2interop(self, args):

return ["go", "run", "negative_http2_client.go"] + args

def cloud_to_prod_env(self):

return {}

def server_cmd(self, args):

return ["go", "run", "server.go"] + args

def global_env(self):

return {"GO111MODULE": "on"}

def unimplemented_test_cases(self):

return _SKIP_COMPRESSION

def unimplemented_test_cases_server(self):

return _SKIP_COMPRESSION

def __str__(self):

return "go"

class Http2Server:

"""Represents the HTTP/2 Interop Test server

This pretends to be a language in order to be built and run, but really it

isn't.

"""

def __init__(self):

self.server_cwd = None

self.safename = str(self)

def server_cmd(self, args):

return ["python test/http2_test/http2_test_server.py"]

def cloud_to_prod_env(self):

return {}

def global_env(self):

return {}

def unimplemented_test_cases(self):

return (

_TEST_CASES

+ _SKIP_DATA_FRAME_PADDING

+ _SKIP_SPECIAL_STATUS_MESSAGE

+ _SKIP_GOOGLE_DEFAULT_CREDS

+ _SKIP_COMPUTE_ENGINE_CHANNEL_CREDS

)

def unimplemented_test_cases_server(self):

return _TEST_CASES

def __str__(self):

return "http2"

class Http2Client:

"""Represents the HTTP/2 Interop Test

This pretends to be a language in order to be built and run, but really it

isn't.

"""

def __init__(self):

self.client_cwd = None

self.safename = str(self)

def client_cmd(self, args):

return ["tools/http2_interop/http2_interop.test", "-test.v"] + args

def cloud_to_prod_env(self):

return {}

def global_env(self):

return {}

def unimplemented_test_cases(self):

return (

_TEST_CASES

+ _SKIP_SPECIAL_STATUS_MESSAGE

+ _SKIP_GOOGLE_DEFAULT_CREDS

+ _SKIP_COMPUTE_ENGINE_CHANNEL_CREDS

)

def unimplemented_test_cases_server(self):

return _TEST_CASES

def __str__(self):

return "http2"

class NodeLanguage:

def __init__(self):

self.client_cwd = "../../../../home/appuser/grpc-node"

self.server_cwd = "../../../../home/appuser/grpc-node"

self.safename = str(self)

def client_cmd(self, args):

return [

"packages/grpc-native-core/deps/grpc/tools/run_tests/interop/with_nvm.sh",

"node",

"--require",

"./test/fixtures/native_native",

"test/interop/interop_client.js",

] + args

def cloud_to_prod_env(self):

return {}

def server_cmd(self, args):

return [

"packages/grpc-native-core/deps/grpc/tools/run_tests/interop/with_nvm.sh",

"node",

"--require",

"./test/fixtures/native_native",

"test/interop/interop_server.js",

] + args

def global_env(self):

return {}

def unimplemented_test_cases(self):

return (

_SKIP_COMPRESSION

+ _SKIP_DATA_FRAME_PADDING

+ _SKIP_GOOGLE_DEFAULT_CREDS

+ _SKIP_COMPUTE_ENGINE_CHANNEL_CREDS

+ _ORCA_TEST_CASES

)

def unimplemented_test_cases_server(self):

return _SKIP_COMPRESSION + _ORCA_TEST_CASES

def __str__(self):

return "node"

class NodePureJSLanguage:

def __init__(self):

self.client_cwd = "../../../../home/appuser/grpc-node"

self.server_cwd = "../../../../home/appuser/grpc-node"

self.safename = str(self)

def client_cmd(self, args):

return [

"packages/grpc-native-core/deps/grpc/tools/run_tests/interop/with_nvm.sh",

"node",

"--require",

"./test/fixtures/js_js",

"test/interop/interop_client.js",

] + args

def cloud_to_prod_env(self):

return {}

def global_env(self):

return {}

def unimplemented_test_cases(self):

return (

_SKIP_COMPRESSION

+ _SKIP_DATA_FRAME_PADDING

+ _SKIP_GOOGLE_DEFAULT_CREDS

+ _SKIP_COMPUTE_ENGINE_CHANNEL_CREDS

+ _ORCA_TEST_CASES

)

def unimplemented_test_cases_server(self):

return _ORCA_TEST_CASES

def __str__(self):

return "nodepurejs"

class PHP7Language:

def __init__(self):

self.client_cwd = None

self.server_cwd = None

self.safename = str(self)

def client_cmd(self, args):

return ["src/php/bin/interop_client.sh"] + args

def cloud_to_prod_env(self):

return {}

def server_cmd(self, args):

return ["src/php/bin/interop_server.sh"] + args

def global_env(self):

return {}

def unimplemented_test_cases(self):

return (

_SKIP_SERVER_COMPRESSION

+ _SKIP_DATA_FRAME_PADDING

+ _SKIP_GOOGLE_DEFAULT_CREDS

+ _SKIP_COMPUTE_ENGINE_CHANNEL_CREDS

+ _ORCA_TEST_CASES

)

def unimplemented_test_cases_server(self):

return _SKIP_COMPRESSION + _ORCA_TEST_CASES

def __str__(self):

return "php7"

class ObjcLanguage:

def __init__(self):

self.client_cwd = "src/objective-c/tests"

self.safename = str(self)

def client_cmd(self, args):

# from args, extract the server port and craft xcodebuild command out of it

for arg in args:

port = re.search("--server_port=(\d+)", arg)

if port:

portnum = port.group(1)

cmdline = (

"pod install && xcodebuild -workspace Tests.xcworkspace"

' -scheme InteropTestsLocalSSL -destination name="iPhone 6"'

" HOST_PORT_LOCALSSL=localhost:%s test" % portnum

)

return [cmdline]

def cloud_to_prod_env(self):

return {}

def global_env(self):

return {}

def unimplemented_test_cases(self):

# ObjC test runs all cases with the same command. It ignores the testcase

# cmdline argument. Here we return all but one test cases as unimplemented,

# and depend upon ObjC test's behavior that it runs all cases even when

# we tell it to run just one.

return (

_TEST_CASES[1:]

+ _SKIP_COMPRESSION

+ _SKIP_DATA_FRAME_PADDING

+ _SKIP_SPECIAL_STATUS_MESSAGE

+ _SKIP_GOOGLE_DEFAULT_CREDS

+ _SKIP_COMPUTE_ENGINE_CHANNEL_CREDS

+ _ORCA_TEST_CASES

)

def unimplemented_test_cases_server(self):

return _SKIP_COMPRESSION + _ORCA_TEST_CASES

def __str__(self):

return "objc"

class RubyLanguage:

def __init__(self):

self.client_cwd = None

self.server_cwd = None

self.safename = str(self)

def client_cmd(self, args):

return [

"tools/run_tests/interop/with_rvm.sh",

"ruby",

"src/ruby/pb/test/client.rb",

] + args

def cloud_to_prod_env(self):

return {}

def server_cmd(self, args):

return [

"tools/run_tests/interop/with_rvm.sh",

"ruby",

"src/ruby/pb/test/server.rb",

] + args

def global_env(self):

return {}

def unimplemented_test_cases(self):

return (

_SKIP_SERVER_COMPRESSION

+ _SKIP_DATA_FRAME_PADDING

+ _SKIP_GOOGLE_DEFAULT_CREDS

+ _SKIP_COMPUTE_ENGINE_CHANNEL_CREDS

+ _ORCA_TEST_CASES

)

def unimplemented_test_cases_server(self):

return _SKIP_COMPRESSION + _ORCA_TEST_CASES

def __str__(self):

return "ruby"

_PYTHON_BINARY = "py39/bin/python"

class PythonLanguage:

def __init__(self):

self.client_cwd = None

self.server_cwd = None

self.http2_cwd = None

self.safename = str(self)

def client_cmd(self, args):

return [

_PYTHON_BINARY,

"src/python/grpcio_tests/setup.py",

"run_interop",

"--client",

'--args="{}"'.format(" ".join(args)),

]

def client_cmd_http2interop(self, args):

return [

_PYTHON_BINARY,

"src/python/grpcio_tests/tests/http2/negative_http2_client.py",

] + args

def cloud_to_prod_env(self):

return {}

def server_cmd(self, args):

return [

_PYTHON_BINARY,

"src/python/grpcio_tests/setup.py",

"run_interop",

"--server",

'--args="{}"'.format(" ".join(args)),

]

def global_env(self):

return {

"LD_LIBRARY_PATH": "{}/libs/opt".format(DOCKER_WORKDIR_ROOT),

"PYTHONPATH": "{}/src/python/gens".format(DOCKER_WORKDIR_ROOT),

}

def unimplemented_test_cases(self):

return (

_SKIP_COMPRESSION

+ _SKIP_DATA_FRAME_PADDING

+ _SKIP_GOOGLE_DEFAULT_CREDS

+ _SKIP_COMPUTE_ENGINE_CHANNEL_CREDS

+ _ORCA_TEST_CASES

)

def unimplemented_test_cases_server(self):

return _SKIP_COMPRESSION + _ORCA_TEST_CASES

def __str__(self):

return "python"

class PythonAsyncIOLanguage:

def __init__(self):

self.client_cwd = None

self.server_cwd = None

self.http2_cwd = None

self.safename = str(self)

def client_cmd(self, args):

return [

_PYTHON_BINARY,

"src/python/grpcio_tests/setup.py",

"run_interop",

"--use-asyncio",

"--client",

'--args="{}"'.format(" ".join(args)),

]

def client_cmd_http2interop(self, args):

return [

_PYTHON_BINARY,

"src/python/grpcio_tests/tests/http2/negative_http2_client.py",

] + args

def cloud_to_prod_env(self):

return {}

def server_cmd(self, args):

return [

_PYTHON_BINARY,

"src/python/grpcio_tests/setup.py",

"run_interop",

"--use-asyncio",

"--server",

'--args="{}"'.format(" ".join(args)),

]

def global_env(self):

return {

"LD_LIBRARY_PATH": "{}/libs/opt".format(DOCKER_WORKDIR_ROOT),

"PYTHONPATH": "{}/src/python/gens".format(DOCKER_WORKDIR_ROOT),

}

def unimplemented_test_cases(self):

# TODO(https://github.com/grpc/grpc/issues/21707)

return (

_SKIP_COMPRESSION

+ _SKIP_DATA_FRAME_PADDING

+ _AUTH_TEST_CASES

+ ["timeout_on_sleeping_server"]

+ _ORCA_TEST_CASES

)

def unimplemented_test_cases_server(self):

# TODO(https://github.com/grpc/grpc/issues/21749)

return (

_TEST_CASES

+ _AUTH_TEST_CASES

+ _HTTP2_TEST_CASES

+ _HTTP2_SERVER_TEST_CASES

)

def __str__(self):

return "pythonasyncio"

_LANGUAGES = {

"c++": CXXLanguage(),

"aspnetcore": AspNetCoreLanguage(),

"dart": DartLanguage(),

"go": GoLanguage(),

"java": JavaLanguage(),

"javaokhttp": JavaOkHttpClient(),

"node": NodeLanguage(),

"nodepurejs": NodePureJSLanguage(),

"php7": PHP7Language(),

"objc": ObjcLanguage(),

"ruby": RubyLanguage(),

"python": PythonLanguage(),

"pythonasyncio": PythonAsyncIOLanguage(),

}

# languages supported as cloud_to_cloud servers

_SERVERS = [

"c++",

"node",

"aspnetcore",

"java",

"go",

"ruby",

"python",

"dart",

"pythonasyncio",

"php7",

]

_TEST_CASES = [

"large_unary",

"empty_unary",

"ping_pong",

"empty_stream",

"client_streaming",

"server_streaming",

"cancel_after_begin",

"cancel_after_first_response",

"timeout_on_sleeping_server",

"custom_metadata",

"status_code_and_message",

"unimplemented_method",

"client_compressed_unary",

"server_compressed_unary",

"client_compressed_streaming",

"server_compressed_streaming",

"unimplemented_service",

"special_status_message",

"orca_per_rpc",

"orca_oob",

]

_AUTH_TEST_CASES = [

"compute_engine_creds",

"jwt_token_creds",

"oauth2_auth_token",

"per_rpc_creds",

_GOOGLE_DEFAULT_CREDS_TEST_CASE,

_COMPUTE_ENGINE_CHANNEL_CREDS_TEST_CASE,

]

_HTTP2_TEST_CASES = ["tls", "framing"]

_HTTP2_SERVER_TEST_CASES = [

"rst_after_header",

"rst_after_data",

"rst_during_data",

"goaway",

"ping",

"max_streams",

"data_frame_padding",

"no_df_padding_sanity_test",

]

_GRPC_CLIENT_TEST_CASES_FOR_HTTP2_SERVER_TEST_CASES = {

"data_frame_padding": "large_unary",

"no_df_padding_sanity_test": "large_unary",

}

_HTTP2_SERVER_TEST_CASES_THAT_USE_GRPC_CLIENTS = list(

_GRPC_CLIENT_TEST_CASES_FOR_HTTP2_SERVER_TEST_CASES.keys()

)

_LANGUAGES_WITH_HTTP2_CLIENTS_FOR_HTTP2_SERVER_TEST_CASES = [

"java",

"go",

"python",

"c++",

]

_LANGUAGES_FOR_ALTS_TEST_CASES = ["java", "go", "c++", "python"]

_SERVERS_FOR_ALTS_TEST_CASES = ["java", "go", "c++", "python"]

_TRANSPORT_SECURITY_OPTIONS = ["tls", "alts", "insecure"]

_CUSTOM_CREDENTIALS_TYPE_OPTIONS = [

"tls",

"google_default_credentials",

"compute_engine_channel_creds",

]

DOCKER_WORKDIR_ROOT = "/var/local/git/grpc"

def docker_run_cmdline(cmdline, image, docker_args=[], cwd=None, environ=None):

"""Wraps given cmdline array to create 'docker run' cmdline from it."""

# don't use '-t' even when TTY is available, since that would break

# the testcases generated by tools/interop_matrix/create_testcases.sh

docker_cmdline = ["docker", "run", "-i", "--rm=true"]

# turn environ into -e docker args

if environ:

for k, v in list(environ.items()):

docker_cmdline += ["-e", "%s=%s" % (k, v)]

# set working directory

workdir = DOCKER_WORKDIR_ROOT

if cwd:

workdir = os.path.join(workdir, cwd)

docker_cmdline += ["-w", workdir]

docker_cmdline += docker_args + [image] + cmdline

return docker_cmdline

def manual_cmdline(docker_cmdline, docker_image):

"""Returns docker cmdline adjusted for manual invocation."""

print_cmdline = []

for item in docker_cmdline:

if item.startswith("--name="):

continue

if item == docker_image:

item = "$docker_image"

item = item.replace('"', '\\"')

# add quotes when necessary

if any(character.isspace() for character in item):

item = '"%s"' % item

print_cmdline.append(item)

return " ".join(print_cmdline)

def write_cmdlog_maybe(cmdlog, filename):

"""Returns docker cmdline adjusted for manual invocation."""

if cmdlog:

with open(filename, "w") as logfile:

logfile.write("#!/bin/bash\n")

logfile.write("# DO NOT MODIFY\n")

logfile.write(

"# This file is generated by"

" run_interop_tests.py/create_testcases.sh\n"

)

logfile.writelines("%s\n" % line for line in cmdlog)

print("Command log written to file %s" % filename)

def bash_cmdline(cmdline):

"""Creates bash -c cmdline from args list."""

# Use login shell:

# * makes error messages clearer if executables are missing

return ["bash", "-c", " ".join(cmdline)]

def compute_engine_creds_required(language, test_case):

"""Returns True if given test requires access to compute engine creds."""

language = str(language)

if test_case == "compute_engine_creds":

return True

if test_case == "oauth2_auth_token" and language == "c++":

# C++ oauth2 test uses GCE creds because C++ only supports JWT

return True

return False

def auth_options(

language,

test_case,

google_default_creds_use_key_file,

service_account_key_file,

default_service_account,

):

"""Returns (cmdline, env) tuple with cloud_to_prod_auth test options."""

language = str(language)

cmdargs = []

env = {}

oauth_scope_arg = "--oauth_scope=https://www.googleapis.com/auth/xapi.zoo"

key_file_arg = "--service_account_key_file=%s" % service_account_key_file

default_account_arg = (

"--default_service_account=%s" % default_service_account

)

if test_case in ["jwt_token_creds", "per_rpc_creds", "oauth2_auth_token"]:

if language in [

"aspnetcore",

"node",

"php7",

"python",

"ruby",

"nodepurejs",

]:

env["GOOGLE_APPLICATION_CREDENTIALS"] = service_account_key_file

else:

cmdargs += [key_file_arg]

if test_case in ["per_rpc_creds", "oauth2_auth_token"]:

cmdargs += [oauth_scope_arg]

if test_case == "oauth2_auth_token" and language == "c++":

# C++ oauth2 test uses GCE creds and thus needs to know the default account

cmdargs += [default_account_arg]

if test_case == "compute_engine_creds":

cmdargs += [oauth_scope_arg, default_account_arg]

if test_case == _GOOGLE_DEFAULT_CREDS_TEST_CASE:

if google_default_creds_use_key_file:

env["GOOGLE_APPLICATION_CREDENTIALS"] = service_account_key_file

cmdargs += [default_account_arg]

if test_case == _COMPUTE_ENGINE_CHANNEL_CREDS_TEST_CASE:

cmdargs += [default_account_arg]

return (cmdargs, env)

def _job_kill_handler(job):

if job._spec.container_name:

dockerjob.docker_kill(job._spec.container_name)

# When the job times out and we decide to kill it,

# we need to wait a before restarting the job

# to prevent "container name already in use" error.

# TODO(jtattermusch): figure out a cleaner way to this.

time.sleep(2)

def cloud_to_prod_jobspec(

language,

test_case,

server_host_nickname,

server_host,

google_default_creds_use_key_file,

docker_image=None,

auth=False,

manual_cmd_log=None,

service_account_key_file=None,

default_service_account=None,

transport_security="tls",

):

"""Creates jobspec for cloud-to-prod interop test"""

container_name = None

cmdargs = [

"--server_host=%s" % server_host,

"--server_port=443",

"--test_case=%s" % test_case,

]

if transport_security == "tls":

transport_security_options = ["--use_tls=true"]

elif transport_security == "google_default_credentials" and str(

language

) in ["c++", "go", "java", "javaokhttp"]:

transport_security_options = [

"--custom_credentials_type=google_default_credentials"

]

elif transport_security == "compute_engine_channel_creds" and str(

language

) in ["go", "java", "javaokhttp"]:

transport_security_options = [

"--custom_credentials_type=compute_engine_channel_creds"

]

else:

print(

"Invalid transport security option %s in cloud_to_prod_jobspec."

" Lang: %s" % (str(language), transport_security)

)

sys.exit(1)

cmdargs = cmdargs + transport_security_options

environ = dict(language.cloud_to_prod_env(), **language.global_env())

if auth:

auth_cmdargs, auth_env = auth_options(

language,

test_case,

google_default_creds_use_key_file,

service_account_key_file,

default_service_account,

)

cmdargs += auth_cmdargs

environ.update(auth_env)

cmdline = bash_cmdline(language.client_cmd(cmdargs))

cwd = language.client_cwd

if docker_image:

container_name = dockerjob.random_name(

"interop_client_%s" % language.safename

)

cmdline = docker_run_cmdline(

cmdline,

image=docker_image,

cwd=cwd,

environ=environ,

docker_args=["--net=host", "--name=%s" % container_name],

)

if manual_cmd_log is not None:

if manual_cmd_log == []:

manual_cmd_log.append(

'echo "Testing ${docker_image:=%s}"' % docker_image

)

manual_cmd_log.append(manual_cmdline(cmdline, docker_image))

cwd = None

environ = None