src/test/py/bazel/test_base.py - bazel - Git at Google

 # Copyright 2017 The Bazel Authors. All rights reserved.
 #
 # 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.

 import os
 import subprocess
 import sys
 import unittest


 class Error(Exception):
   """Base class for errors in this module."""
   pass


 class ArgumentError(Error):
   """A function received a bad argument."""
   pass


 class EnvVarUndefinedError(Error):
   """An expected environment variable is not defined."""

   def __init__(self, name):
     Error.__init__(self, 'Environment variable "%s" is not defined' % name)


 class TestBase(unittest.TestCase):

   _stderr = None
   _stdout = None
   _runfiles = None
   _temp = None
   _tests_root = None

   def setUp(self):
     unittest.TestCase.setUp(self)
     if self._runfiles is None:
       self._runfiles = TestBase._LoadRunfiles()
     test_tmpdir = TestBase.GetEnv('TEST_TMPDIR')
     self._stdout = os.path.join(test_tmpdir, 'bazel.stdout')
     self._stderr = os.path.join(test_tmpdir, 'bazel.stderr')
     self._temp = os.path.join(test_tmpdir, 'tmp')
     self._tests_root = os.path.join(test_tmpdir, 'tests_root')
     os.mkdir(self._tests_root)
     os.chdir(self._tests_root)

   def AssertExitCode(self, actual_exit_code, expected_exit_code, stderr_lines):
     """Assert that `actual_exit_code` == `expected_exit_code`."""
     if actual_exit_code != expected_exit_code:
       self.fail('\n'.join([
           'Bazel exited with %d (expected %d), stderr:' % (actual_exit_code,
                                                            expected_exit_code),
           '(start stderr)----------------------------------------',
       ] + (stderr_lines or []) + [
           '(end stderr)------------------------------------------',
       ]))

   @staticmethod
   def GetEnv(name, default=None):
     """Returns environment variable `name`.

     Args:
       name: string; name of the environment variable
       default: anything; return this value if the envvar is not defined
     Returns:
       string, the envvar's value if defined, or `default` if the envvar is not
       defined but `default` is
     Raises:
       EnvVarUndefinedError: if `name` is not a defined envvar and `default` is
         None
     """
     value = os.getenv(name, '__undefined_envvar__')
     if value == '__undefined_envvar__':
       if default:
         return default
       raise EnvVarUndefinedError(name)
     return value

   @staticmethod
   def IsWindows():
     """Returns true if the current platform is Windows."""
     return os.name == 'nt'

   def Path(self, path):
     """Returns the absolute path of `path` relative to the scratch directory.

     Args:
       path: string; a path, relative to the test's scratch directory,
         e.g. "foo/bar/BUILD"
     Returns:
       an absolute path
     Raises:
       ArgumentError: if `path` is absolute or contains uplevel references
     """
     if os.path.isabs(path) or '..' in path:
       raise ArgumentError(('path="%s" may not be absolute and may not contain '
                            'uplevel references') % path)
     return os.path.join(self._tests_root, path)

   def Rlocation(self, runfile):
     """Returns the absolute path to a runfile."""
     if TestBase.IsWindows():
       return self._runfiles.get(runfile)
     else:
       return os.path.join(self._runfiles, runfile)

   def ScratchDir(self, path):
     """Creates directories under the test's scratch directory.

     Args:
       path: string; a path, relative to the test's scratch directory,
         e.g. "foo/bar"
     Raises:
       ArgumentError: if `path` is absolute or contains uplevel references
       IOError: if an I/O error occurs
     """
     if not path:
       return
     abspath = self.Path(path)
     if os.path.exists(abspath):
       if os.path.isdir(abspath):
         return
       raise IOError('"%s" (%s) exists and is not a directory' % (path, abspath))
     os.makedirs(abspath)

   def ScratchFile(self, path, lines=None):
     """Creates a file under the test's scratch directory.

     Args:
       path: string; a path, relative to the test's scratch directory,
         e.g. "foo/bar/BUILD"
       lines: [string]; the contents of the file (newlines are added
         automatically)
     Raises:
       ArgumentError: if `path` is absolute or contains uplevel references
       IOError: if an I/O error occurs
     """
     if not path:
       return
     abspath = self.Path(path)
     if os.path.exists(abspath) and not os.path.isfile(abspath):
       raise IOError('"%s" (%s) exists and is not a file' % (path, abspath))
     self.ScratchDir(os.path.dirname(path))
     with open(abspath, 'w') as f:
       if lines:
         for l in lines:
           f.write(l)
           f.write('\n')

   def RunBazel(self, args, env_remove=None):
     """Runs "bazel <args>", waits for it to exit.

     Args:
       args: [string]; flags to pass to bazel (e.g. ['--batch', 'build', '//x'])
       env_remove: set(string); optional; environment variables to NOT pass to
         Bazel
     Returns:
       (int, [string], [string]) tuple: exit code, stdout lines, stderr lines
     """
     with open(self._stdout, 'w') as stdout:
       with open(self._stderr, 'w') as stderr:
         proc = subprocess.Popen(
             [
                 self.Rlocation('io_bazel/src/bazel'), '--bazelrc=/dev/null',
                 '--nomaster_bazelrc'
             ] + args,
             stdout=stdout,
             stderr=stderr,
             cwd=self._tests_root,
             env=self._BazelEnvMap(env_remove))
         exit_code = proc.wait()

     with open(self._stdout, 'r') as f:
       stdout = [l.strip() for l in f.readlines()]
     with open(self._stderr, 'r') as f:
       stderr = [l.strip() for l in f.readlines()]
     return exit_code, stdout, stderr

   def _BazelEnvMap(self, env_remove=None):
     """Returns the environment variable map to run Bazel."""
     if TestBase.IsWindows():
       result = []
       if sys.version_info.major == 3:
         # Python 3.2 has os.listdir
         result = [
             n for n in os.listdir('c:\\program files\\java')
             if n.startswith('jdk')
         ]
       else:
         # Python 2.7 has os.path.walk
         def _Visit(result, _, names):
           result.extend(n for n in names if n.startswith('jdk'))
           while names:
             names.pop()

         os.path.walk('c:\\program files\\java\\', _Visit, result)

       env = {
           'SYSTEMROOT': TestBase.GetEnv('SYSTEMROOT'),
           # TODO(laszlocsomor): Let Bazel pass BAZEL_SH and JAVA_HOME to tests
           # and use those here instead of hardcoding paths.
           'JAVA_HOME': 'c:\\program files\\java\\' + sorted(result)[-1],
           'BAZEL_SH': 'c:\\tools\\msys64\\usr\\bin\\bash.exe',
       }
     else:
       env = {'HOME': os.path.join(self._temp, 'home')}

     env['PATH'] = TestBase.GetEnv('PATH')
     # The inner Bazel must know that it's running as part of a test (so that it
     # uses --max_idle_secs=15 by default instead of 3 hours, etc.), and it knows
     # that by checking for TEST_TMPDIR.
     env['TEST_TMPDIR'] = TestBase.GetEnv('TEST_TMPDIR')
     env['TMP'] = self._temp
     if env_remove:
       for e in env_remove:
         del env[e]
     return env

   @staticmethod
   def _LoadRunfiles():
     """Loads the runfiles manifest from ${TEST_SRCDIR}/MANIFEST.

     Only necessary to use on Windows, where runfiles are not symlinked in to the
     runfiles directory, but are written to a MANIFEST file instead.

     Returns:
       on Windows: {string: string} dictionary, keys are runfiles-relative paths,
         values are absolute paths that the runfiles entry is mapped to;
       on other platforms: string; value of $TEST_SRCDIR
     """
     test_srcdir = TestBase.GetEnv('TEST_SRCDIR')
     if not TestBase.IsWindows():
       return test_srcdir

     result = {}
     with open(os.path.join(test_srcdir, 'MANIFEST'), 'r') as f:
       for l in f:
         tokens = l.strip().split(' ')
         if len(tokens) == 2:
           result[tokens[0]] = tokens[1]
     return result
	# Copyright 2017 The Bazel Authors. All rights reserved.
	#
	# 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.

	import os
	import subprocess
	import sys
	import unittest


	class Error(Exception):
	"""Base class for errors in this module."""
	pass


	class ArgumentError(Error):
	"""A function received a bad argument."""
	pass


	class EnvVarUndefinedError(Error):
	"""An expected environment variable is not defined."""

	def __init__(self, name):
	Error.__init__(self, 'Environment variable "%s" is not defined' % name)


	class TestBase(unittest.TestCase):

	_stderr = None
	_stdout = None
	_runfiles = None
	_temp = None
	_tests_root = None

	def setUp(self):
	unittest.TestCase.setUp(self)
	if self._runfiles is None:
	self._runfiles = TestBase._LoadRunfiles()
	test_tmpdir = TestBase.GetEnv('TEST_TMPDIR')
	self._stdout = os.path.join(test_tmpdir, 'bazel.stdout')
	self._stderr = os.path.join(test_tmpdir, 'bazel.stderr')
	self._temp = os.path.join(test_tmpdir, 'tmp')
	self._tests_root = os.path.join(test_tmpdir, 'tests_root')
	os.mkdir(self._tests_root)
	os.chdir(self._tests_root)

	def AssertExitCode(self, actual_exit_code, expected_exit_code, stderr_lines):
	"""Assert that `actual_exit_code` == `expected_exit_code`."""
	if actual_exit_code != expected_exit_code:
	self.fail('\n'.join([
	'Bazel exited with %d (expected %d), stderr:' % (actual_exit_code,
	expected_exit_code),
	'(start stderr)----------------------------------------',
	] + (stderr_lines or []) + [
	'(end stderr)------------------------------------------',
	]))

	@staticmethod
	def GetEnv(name, default=None):
	"""Returns environment variable `name`.

	Args:
	name: string; name of the environment variable
	default: anything; return this value if the envvar is not defined
	Returns:
	string, the envvar's value if defined, or `default` if the envvar is not
	defined but `default` is
	Raises:
	EnvVarUndefinedError: if `name` is not a defined envvar and `default` is
	None
	"""
	value = os.getenv(name, '__undefined_envvar__')
	if value == '__undefined_envvar__':
	if default:
	return default
	raise EnvVarUndefinedError(name)
	return value

	@staticmethod
	def IsWindows():
	"""Returns true if the current platform is Windows."""
	return os.name == 'nt'

	def Path(self, path):
	"""Returns the absolute path of `path` relative to the scratch directory.

	Args:
	path: string; a path, relative to the test's scratch directory,
	e.g. "foo/bar/BUILD"
	Returns:
	an absolute path
	Raises:
	ArgumentError: if `path` is absolute or contains uplevel references
	"""
	if os.path.isabs(path) or '..' in path:
	raise ArgumentError(('path="%s" may not be absolute and may not contain '
	'uplevel references') % path)
	return os.path.join(self._tests_root, path)

	def Rlocation(self, runfile):
	"""Returns the absolute path to a runfile."""
	if TestBase.IsWindows():
	return self._runfiles.get(runfile)
	else:
	return os.path.join(self._runfiles, runfile)

	def ScratchDir(self, path):
	"""Creates directories under the test's scratch directory.

	Args:
	path: string; a path, relative to the test's scratch directory,
	e.g. "foo/bar"
	Raises:
	ArgumentError: if `path` is absolute or contains uplevel references
	IOError: if an I/O error occurs
	"""
	if not path:
	return
	abspath = self.Path(path)
	if os.path.exists(abspath):
	if os.path.isdir(abspath):
	return
	raise IOError('"%s" (%s) exists and is not a directory' % (path, abspath))
	os.makedirs(abspath)

	def ScratchFile(self, path, lines=None):
	"""Creates a file under the test's scratch directory.

	Args:
	path: string; a path, relative to the test's scratch directory,
	e.g. "foo/bar/BUILD"
	lines: [string]; the contents of the file (newlines are added
	automatically)
	Raises:
	ArgumentError: if `path` is absolute or contains uplevel references
	IOError: if an I/O error occurs
	"""
	if not path:
	return
	abspath = self.Path(path)
	if os.path.exists(abspath) and not os.path.isfile(abspath):
	raise IOError('"%s" (%s) exists and is not a file' % (path, abspath))
	self.ScratchDir(os.path.dirname(path))
	with open(abspath, 'w') as f:
	if lines:
	for l in lines:
	f.write(l)
	f.write('\n')

	def RunBazel(self, args, env_remove=None):
	"""Runs "bazel <args>", waits for it to exit.

	Args:
	args: [string]; flags to pass to bazel (e.g. ['--batch', 'build', '//x'])
	env_remove: set(string); optional; environment variables to NOT pass to
	Bazel
	Returns:
	(int, [string], [string]) tuple: exit code, stdout lines, stderr lines
	"""
	with open(self._stdout, 'w') as stdout:
	with open(self._stderr, 'w') as stderr:
	proc = subprocess.Popen(
	[
	self.Rlocation('io_bazel/src/bazel'), '--bazelrc=/dev/null',
	'--nomaster_bazelrc'
	] + args,
	stdout=stdout,
	stderr=stderr,
	cwd=self._tests_root,
	env=self._BazelEnvMap(env_remove))
	exit_code = proc.wait()

	with open(self._stdout, 'r') as f:
	stdout = [l.strip() for l in f.readlines()]
	with open(self._stderr, 'r') as f:
	stderr = [l.strip() for l in f.readlines()]
	return exit_code, stdout, stderr

	def _BazelEnvMap(self, env_remove=None):
	"""Returns the environment variable map to run Bazel."""
	if TestBase.IsWindows():
	result = []
	if sys.version_info.major == 3:
	# Python 3.2 has os.listdir
	result = [
	n for n in os.listdir('c:\\program files\\java')
	if n.startswith('jdk')
	]
	else:
	# Python 2.7 has os.path.walk
	def _Visit(result, _, names):
	result.extend(n for n in names if n.startswith('jdk'))
	while names:
	names.pop()

	os.path.walk('c:\\program files\\java\\', _Visit, result)

	env = {
	'SYSTEMROOT': TestBase.GetEnv('SYSTEMROOT'),
	# TODO(laszlocsomor): Let Bazel pass BAZEL_SH and JAVA_HOME to tests
	# and use those here instead of hardcoding paths.
	'JAVA_HOME': 'c:\\program files\\java\\' + sorted(result)[-1],
	'BAZEL_SH': 'c:\\tools\\msys64\\usr\\bin\\bash.exe',
	}
	else:
	env = {'HOME': os.path.join(self._temp, 'home')}

	env['PATH'] = TestBase.GetEnv('PATH')
	# The inner Bazel must know that it's running as part of a test (so that it
	# uses --max_idle_secs=15 by default instead of 3 hours, etc.), and it knows
	# that by checking for TEST_TMPDIR.
	env['TEST_TMPDIR'] = TestBase.GetEnv('TEST_TMPDIR')
	env['TMP'] = self._temp
	if env_remove:
	for e in env_remove:
	del env[e]
	return env

	@staticmethod
	def _LoadRunfiles():
	"""Loads the runfiles manifest from ${TEST_SRCDIR}/MANIFEST.

	Only necessary to use on Windows, where runfiles are not symlinked in to the
	runfiles directory, but are written to a MANIFEST file instead.

	Returns:
	on Windows: {string: string} dictionary, keys are runfiles-relative paths,
	values are absolute paths that the runfiles entry is mapped to;
	on other platforms: string; value of $TEST_SRCDIR
	"""
	test_srcdir = TestBase.GetEnv('TEST_SRCDIR')
	if not TestBase.IsWindows():
	return test_srcdir

	result = {}
	with open(os.path.join(test_srcdir, 'MANIFEST'), 'r') as f:
	for l in f:
	tokens = l.strip().split(' ')
	if len(tokens) == 2:
	result[tokens[0]] = tokens[1]
	return result