xref: /aosp_15_r20/external/bazelbuild-rules_android/test/bashunit/unittest_test.py (revision 9e965d6fece27a77de5377433c2f7e6999b8cc0b)

# Copyright 2020 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.

"""Tests for unittest.bash."""

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import os
import re
import shutil
import stat
import subprocess
import tempfile
import textwrap
import unittest

# The test setup for this external test is forwarded to the internal bash test.
# This allows the internal test to use the same runfiles to load unittest.bash.
_TEST_PREAMBLE = """
#!/bin/bash
# --- begin runfiles.bash initialization ---
if [[ -f "${RUNFILES_DIR:-/dev/null}/bazel_tools/tools/bash/runfiles/runfiles.bash" ]]; then
  source "${RUNFILES_DIR}/bazel_tools/tools/bash/runfiles/runfiles.bash"
else
  echo >&2 "ERROR: cannot find @bazel_tools//tools/bash/runfiles:runfiles.bash"
  exit 1
fi
# --- end runfiles.bash initialization ---

echo "Writing XML to ${XML_OUTPUT_FILE}"

source "$(rlocation "rules_android/test/bashunit/unittest.bash")" \
  || { echo "Could not source unittest.bash" >&2; exit 1; }
"""

ANSI_ESCAPE = re.compile(r"(\x9B|\x1B\[)[0-?]*[ -\/]*[@-~]")


def remove_ansi(line):
  """Remove ANSI-style escape sequences from the input."""
  return ANSI_ESCAPE.sub("", line)


class TestResult(object):
  """Save test results for easy checking."""

  def __init__(self, asserter, return_code, output, xmlfile):
    self._asserter = asserter
    self._return_code = return_code
    self._output = remove_ansi(output)

    # Read in the XML result file.
    if os.path.isfile(xmlfile):
      with open(xmlfile, "r") as f:
        self._xml = f.read()
    else:
      # Unable to read the file, errors will be reported later.
      self._xml = ""

  # Methods to assert on the state of the results.

  def assertLogMessage(self, message):
    self.assertExactlyOneMatch(self._output, message)

  def assertNotLogMessage(self, message):
    self._asserter.assertNotRegex(self._output, message)

  def assertXmlMessage(self, message):
    self.assertExactlyOneMatch(self._xml, message)

  def assertNotXmlMessage(self, message):
    self._asserter.assertNotRegex(self._xml, message)

  def assertSuccess(self, suite_name):
    self._asserter.assertEqual(0, self._return_code,
                               f"Script failed unexpectedly:\n{self._output}")
    self.assertLogMessage(suite_name)
    self.assertXmlMessage("<testsuites [^/]*failures=\"0\"")
    self.assertXmlMessage("<testsuites [^/]*errors=\"0\"")

  def assertNotSuccess(self, suite_name, failures=0, errors=0):
    self._asserter.assertNotEqual(0, self._return_code)
    self.assertLogMessage(suite_name)
    if failures:
      self.assertXmlMessage(f'<testsuites [^/]*failures="{failures}"')
    if errors:
      self.assertXmlMessage(f'<testsuites [^/]*errors="{errors}"')

  def assertTestPassed(self, test_name):
    self.assertLogMessage(f"PASSED: {test_name}")

  def assertTestFailed(self, test_name, message=""):
    self.assertLogMessage(f"{test_name} FAILED: {message}")

  def assertExactlyOneMatch(self, text, pattern):
    self._asserter.assertRegex(text, pattern)
    self._asserter.assertEqual(
        len(re.findall(pattern, text)),
        1,
        msg=f"Found more than 1 match of '{pattern}' in '{text}'")


class UnittestTest(unittest.TestCase):

  def setUp(self):
    """Create a working directory under our temp dir."""
    super(UnittestTest, self).setUp()
    self.work_dir = tempfile.mkdtemp(dir=os.environ["TEST_TMPDIR"])

  def tearDown(self):
    """Clean up the working directory."""
    super(UnittestTest, self).tearDown()
    shutil.rmtree(self.work_dir)

  def write_file(self, filename, contents=""):
    """Write the contents to a file in the workdir."""

    filepath = os.path.join(self.work_dir, filename)
    with open(filepath, "w") as f:
      f.write(_TEST_PREAMBLE.strip())
      f.write(contents)
    os.chmod(filepath, stat.S_IEXEC | stat.S_IWRITE | stat.S_IREAD)

  def find_runfiles(self):
    if "RUNFILES_DIR" in os.environ:
      return os.environ["RUNFILES_DIR"]

    # Fall back to being based on the srcdir.
    if "TEST_SRCDIR" in os.environ:
      return os.environ["TEST_SRCDIR"]

    # Base on the current dir
    return f"{os.getcwd()}/.."

  def execute_test(self, filename, env=None, args=()):
    """Executes the file and stores the results."""

    filepath = os.path.join(self.work_dir, filename)
    xmlfile = os.path.join(self.work_dir, "dummy-testlog.xml")
    test_env = {
        "TEST_TMPDIR": self.work_dir,
        "RUNFILES_DIR": self.find_runfiles(),
        "TEST_SRCDIR": os.environ["TEST_SRCDIR"],
        "XML_OUTPUT_FILE": xmlfile,
    }
    # Add in env, forcing everything to be a string.
    if env:
      for k, v in env.items():
        test_env[k] = str(v)
    completed = subprocess.run(
        [filepath, *args],
        env=test_env,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
    )
    return TestResult(self, completed.returncode,
                      completed.stdout.decode("utf-8"), xmlfile)

  # Actual test cases.

  def test_success(self):
    self.write_file(
        "thing.sh", """
function test_success() {
  echo foo >&${TEST_log} || fail "expected echo to succeed"
  expect_log "foo"
}

run_suite "success tests"
""")

    result = self.execute_test("thing.sh")
    result.assertSuccess("success tests")
    result.assertTestPassed("test_success")

  def test_timestamp(self):
    self.write_file(
        "thing.sh", """
function test_timestamp() {
  local ts=$(timestamp)
  [[ $ts =~ ^[0-9]{13}$ ]] || fail "timestamp wan't valid: $ts"

  local time_diff=$(get_run_time 100000 223456)
  assert_equals $time_diff 123.456
}

run_suite "timestamp tests"
""")

    result = self.execute_test("thing.sh")
    result.assertSuccess("timestamp tests")
    result.assertTestPassed("test_timestamp")

  def test_failure(self):
    self.write_file(
        "thing.sh", """
function test_failure() {
  fail "I'm a failure with <>&\\" escaped symbols"
}

run_suite "failure tests"
""")

    result = self.execute_test("thing.sh")
    result.assertNotSuccess("failure tests", failures=0, errors=1)
    result.assertTestFailed("test_failure")
    result.assertXmlMessage(
        "message=\"I'm a failure with &lt;&gt;&amp;&quot; escaped symbols\"")
    result.assertXmlMessage("I'm a failure with <>&\" escaped symbols")

  def test_set_bash_errexit_prints_stack_trace(self):
    self.write_file(
        "thing.sh", """
set -euo pipefail

function helper() {
  echo before
  false
  echo after
}

function test_failure_in_helper() {
  helper
}

run_suite "bash errexit tests"
""")

    result = self.execute_test("thing.sh")
    result.assertNotSuccess("bash errexit tests")
    result.assertTestFailed("test_failure_in_helper")
    result.assertLogMessage(r"./thing.sh:\d*: in call to helper")
    result.assertLogMessage(
        r"./thing.sh:\d*: in call to test_failure_in_helper")

  def test_set_bash_errexit_runs_tear_down(self):
    self.write_file(
        "thing.sh", """
set -euo pipefail

function tear_down() {
  echo "Running tear_down"
}

function testenv_tear_down() {
  echo "Running testenv_tear_down"
}

function test_failure_in_helper() {
  wrong_command
}

run_suite "bash errexit tests"
""")

    result = self.execute_test("thing.sh")
    result.assertNotSuccess("bash errexit tests")
    result.assertTestFailed("test_failure_in_helper")
    result.assertLogMessage("Running tear_down")
    result.assertLogMessage("Running testenv_tear_down")

  def test_set_bash_errexit_pipefail_propagates_failure_through_pipe(self):
    self.write_file(
        "thing.sh", """
set -euo pipefail

function test_pipefail() {
  wrong_command | cat
  echo after
}

run_suite "bash errexit tests"
""")

    result = self.execute_test("thing.sh")
    result.assertNotSuccess("bash errexit tests")
    result.assertTestFailed("test_pipefail")
    result.assertLogMessage("wrong_command: command not found")
    result.assertNotLogMessage("after")

  def test_set_bash_errexit_no_pipefail_ignores_failure_before_pipe(self):
    self.write_file(
        "thing.sh", """
set -eu
set +o pipefail

function test_nopipefail() {
  wrong_command | cat
  echo after
}

run_suite "bash errexit tests"
""")

    result = self.execute_test("thing.sh")
    result.assertSuccess("bash errexit tests")
    result.assertTestPassed("test_nopipefail")
    result.assertLogMessage("wrong_command: command not found")
    result.assertLogMessage("after")

  def test_set_bash_errexit_pipefail_long_testname_succeeds(self):
    test_name = "x" * 1000
    self.write_file(
        "thing.sh", """
set -euo pipefail

function test_%s() {
  :
}

run_suite "bash errexit tests"
""" % test_name)

    result = self.execute_test("thing.sh")
    result.assertSuccess("bash errexit tests")

  def test_empty_test_fails(self):
    self.write_file("thing.sh", """
# No tests present.

run_suite "empty test suite"
""")

    result = self.execute_test("thing.sh")
    result.assertNotSuccess("empty test suite")
    result.assertLogMessage("No tests found.")

  def test_empty_test_succeeds_sharding(self):
    self.write_file(
        "thing.sh", """
# Only one test.
function test_thing() {
  echo
}

run_suite "empty test suite"
""")

    # First shard.
    result = self.execute_test(
        "thing.sh", env={
            "TEST_TOTAL_SHARDS": 2,
            "TEST_SHARD_INDEX": 0,
        })
    result.assertSuccess("empty test suite")
    result.assertLogMessage("No tests executed due to sharding")

    # Second shard.
    result = self.execute_test(
        "thing.sh", env={
            "TEST_TOTAL_SHARDS": 2,
            "TEST_SHARD_INDEX": 1,
        })
    result.assertSuccess("empty test suite")
    result.assertNotLogMessage("No tests")

  def test_filter_runs_only_matching_test(self):
    self.write_file(
        "thing.sh",
        textwrap.dedent("""
        function test_abc() {
          :
        }

        function test_def() {
          echo "running def"
        }

        run_suite "tests to filter"
        """))

    result = self.execute_test(
        "thing.sh", env={"TESTBRIDGE_TEST_ONLY": "test_a*"})

    result.assertSuccess("tests to filter")
    result.assertTestPassed("test_abc")
    result.assertNotLogMessage("running def")

  def test_filter_prefix_match_only_skips_test(self):
    self.write_file(
        "thing.sh",
        textwrap.dedent("""
        function test_abc() {
          echo "running abc"
        }

        run_suite "tests to filter"
        """))

    result = self.execute_test(
        "thing.sh", env={"TESTBRIDGE_TEST_ONLY": "test_a"})

    result.assertNotSuccess("tests to filter")
    result.assertLogMessage("No tests found.")

  def test_filter_multiple_globs_runs_tests_matching_any(self):
    self.write_file(
        "thing.sh",
        textwrap.dedent("""
        function test_abc() {
          echo "running abc"
        }

        function test_def() {
          echo "running def"
        }

        run_suite "tests to filter"
        """))

    result = self.execute_test(
        "thing.sh", env={"TESTBRIDGE_TEST_ONLY": "donotmatch:*a*"})

    result.assertSuccess("tests to filter")
    result.assertTestPassed("test_abc")
    result.assertNotLogMessage("running def")

  def test_filter_character_group_runs_only_matching_tests(self):
    self.write_file(
        "thing.sh",
        textwrap.dedent("""
        function test_aaa() {
          :
        }

        function test_daa() {
          :
        }

        function test_zaa() {
          echo "running zaa"
        }

        run_suite "tests to filter"
        """))

    result = self.execute_test(
        "thing.sh", env={"TESTBRIDGE_TEST_ONLY": "test_[a-f]aa"})

    result.assertSuccess("tests to filter")
    result.assertTestPassed("test_aaa")
    result.assertTestPassed("test_daa")
    result.assertNotLogMessage("running zaa")

  def test_filter_sharded_runs_subset_of_filtered_tests(self):
    for index in range(2):
      with self.subTest(index=index):
        self.__filter_sharded_runs_subset_of_filtered_tests(index)

  def __filter_sharded_runs_subset_of_filtered_tests(self, index):
    self.write_file(
        "thing.sh",
        textwrap.dedent("""
        function test_a0() {
          echo "running a0"
        }

        function test_a1() {
          echo "running a1"
        }

        function test_bb() {
          echo "running bb"
        }

        run_suite "tests to filter"
        """))

    result = self.execute_test(
        "thing.sh",
        env={
            "TESTBRIDGE_TEST_ONLY": "test_a*",
            "TEST_TOTAL_SHARDS": 2,
            "TEST_SHARD_INDEX": index
        })

    result.assertSuccess("tests to filter")
    # The sharding logic is shifted by 1, starts with 2nd shard.
    result.assertTestPassed("test_a" + str(index ^ 1))
    result.assertLogMessage("running a" + str(index ^ 1))
    result.assertNotLogMessage("running a" + str(index))
    result.assertNotLogMessage("running bb")

  def test_arg_runs_only_matching_test_and_issues_warning(self):
    self.write_file(
        "thing.sh",
        textwrap.dedent("""
        function test_abc() {
          :
        }

        function test_def() {
          echo "running def"
        }

        run_suite "tests to filter"
        """))

    result = self.execute_test("thing.sh", args=["test_abc"])

    result.assertSuccess("tests to filter")
    result.assertTestPassed("test_abc")
    result.assertNotLogMessage("running def")
    result.assertLogMessage(
        r"WARNING: Passing test names in arguments \(--test_arg\) is "
        "deprecated, please use --test_filter='test_abc' instead.")

  def test_arg_multiple_tests_issues_warning_with_test_filter_command(self):
    self.write_file(
        "thing.sh",
        textwrap.dedent("""
        function test_abc() {
          :
        }

        function test_def() {
          :
        }

        run_suite "tests to filter"
        """))

    result = self.execute_test("thing.sh", args=["test_abc", "test_def"])

    result.assertSuccess("tests to filter")
    result.assertTestPassed("test_abc")
    result.assertTestPassed("test_def")
    result.assertLogMessage(
        r"WARNING: Passing test names in arguments \(--test_arg\) is "
        "deprecated, please use --test_filter='test_abc:test_def' instead.")

  def test_arg_and_filter_ignores_arg(self):
    self.write_file(
        "thing.sh",
        textwrap.dedent("""
        function test_abc() {
          :
        }

        function test_def() {
          echo "running def"
        }

        run_suite "tests to filter"
        """))

    result = self.execute_test(
        "thing.sh", args=["test_def"], env={"TESTBRIDGE_TEST_ONLY": "test_a*"})

    result.assertSuccess("tests to filter")
    result.assertTestPassed("test_abc")
    result.assertNotLogMessage("running def")
    result.assertLogMessage(
        "WARNING: Both --test_arg and --test_filter specified, ignoring --test_arg"
    )

  def test_custom_ifs_variable_finds_and_runs_test(self):
    for sharded in (False, True):
      for ifs in (r"\t", "t"):
        with self.subTest(ifs=ifs, sharded=sharded):
          self.__custom_ifs_variable_finds_and_runs_test(ifs, sharded)

  def __custom_ifs_variable_finds_and_runs_test(self, ifs, sharded):
    self.write_file(
        "thing.sh",
        textwrap.dedent(r"""
        set -euo pipefail
        IFS=$'%s'
        function test_foo() {
          :
        }

        run_suite "custom IFS test"
        """ % ifs))

    result = self.execute_test(
        "thing.sh",
        env={} if not sharded else {
            "TEST_TOTAL_SHARDS": 2,
            "TEST_SHARD_INDEX": 1
        })

    result.assertSuccess("custom IFS test")
    result.assertTestPassed("test_foo")

  def test_fail_in_teardown_reports_failure(self):
    self.write_file(
        "thing.sh",
        textwrap.dedent(r"""
        function tear_down() {
          echo "tear_down log" >"${TEST_log}"
          fail "tear_down failure"
        }

        function test_foo() {
          :
        }

        run_suite "Failure in tear_down test"
        """))

    result = self.execute_test("thing.sh")

    result.assertNotSuccess("Failure in tear_down test", errors=1)
    result.assertTestFailed("test_foo", "tear_down failure")
    result.assertXmlMessage('message="tear_down failure"')
    result.assertLogMessage("tear_down log")

  def test_fail_in_teardown_after_test_failure_reports_both_failures(self):
    self.write_file(
        "thing.sh",
        textwrap.dedent(r"""
        function tear_down() {
          echo "tear_down log" >"${TEST_log}"
          fail "tear_down failure"
        }

        function test_foo() {
          echo "test_foo log" >"${TEST_log}"
          fail "Test failure"
        }

        run_suite "Failure in tear_down test"
        """))

    result = self.execute_test("thing.sh")

    result.assertNotSuccess("Failure in tear_down test", errors=1)
    result.assertTestFailed("test_foo", "Test failure")
    result.assertTestFailed("test_foo", "tear_down failure")
    result.assertXmlMessage('message="Test failure"')
    result.assertNotXmlMessage('message="tear_down failure"')
    result.assertXmlMessage("test_foo log")
    result.assertXmlMessage("tear_down log")
    result.assertLogMessage("Test failure")
    result.assertLogMessage("tear_down failure")
    result.assertLogMessage("test_foo log")
    result.assertLogMessage("tear_down log")

  def test_errexit_in_teardown_reports_failure(self):
    self.write_file(
        "thing.sh",
        textwrap.dedent(r"""
        set -euo pipefail

        function tear_down() {
          invalid_command
        }

        function test_foo() {
          :
        }

        run_suite "errexit in tear_down test"
        """))

    result = self.execute_test("thing.sh")

    result.assertNotSuccess("errexit in tear_down test")
    result.assertLogMessage("invalid_command: command not found")
    result.assertXmlMessage('message="No failure message"')
    result.assertXmlMessage("invalid_command: command not found")

  def test_fail_in_tear_down_after_errexit_reports_both_failures(self):
    self.write_file(
        "thing.sh",
        textwrap.dedent(r"""
        set -euo pipefail

        function tear_down() {
          echo "tear_down log" >"${TEST_log}"
          fail "tear_down failure"
        }

        function test_foo() {
          invalid_command
        }

        run_suite "fail after failure"
        """))

    result = self.execute_test("thing.sh")

    result.assertNotSuccess("fail after failure")
    result.assertTestFailed(
        "test_foo",
        "terminated because this command returned a non-zero status")
    result.assertTestFailed("test_foo", "tear_down failure")
    result.assertLogMessage("invalid_command: command not found")
    result.assertLogMessage("tear_down log")
    result.assertXmlMessage('message="No failure message"')
    result.assertXmlMessage("invalid_command: command not found")

  def test_errexit_in_tear_down_after_errexit_reports_both_failures(self):
    self.write_file(
        "thing.sh",
        textwrap.dedent(r"""
        set -euo pipefail

        function tear_down() {
          invalid_command_tear_down
        }

        function test_foo() {
          invalid_command_test
        }

        run_suite "fail after failure"
        """))

    result = self.execute_test("thing.sh")

    result.assertNotSuccess("fail after failure")
    result.assertTestFailed(
        "test_foo",
        "terminated because this command returned a non-zero status")
    result.assertLogMessage("invalid_command_test: command not found")
    result.assertLogMessage("invalid_command_tear_down: command not found")
    result.assertXmlMessage('message="No failure message"')
    result.assertXmlMessage("invalid_command_test: command not found")
    result.assertXmlMessage("invalid_command_tear_down: command not found")


if __name__ == "__main__":
  unittest.main()