dataconnect: include relevant logcat snippets in github actions logs when androidTest tests fail

.github/workflows/dataconnect.yml

@@ -66,6 +66,12 @@ jobs:
         with:
           node-version: ${{ env.FDC_NODEJS_VERSION }}
 
+      - uses: actions/setup-python@42375524e23c412d93fb67b49958b491fce71c38 # v5.4.0
+        with:
+          python-version: ${{ env.FDC_PYTHON_VERSION }}
+
+      - run: pip install -r firebase-dataconnect/ci/requirements.txt
+
       - name: Install Firebase Tools ("firebase" command-line tool)
         run: |
           set -euo pipefail
@@ -229,6 +235,15 @@ jobs:
         if: steps.connectedCheck.outcome != 'success'
         run: |
           set -euo pipefail
+
+          if [[ ! -e logcat.log ]] ; then
+            echo "WARNING dsdta43sxk: logcat log file not found; skipping scanning for test failures" >&2
+          else
+            echo "Scanning logcat output for failure details"
+            python firebase-dataconnect/ci/logcat_error_report.py --logcat-file=logcat.log
+            echo
+          fi
+
           echo 'Failing because the outcome of the "Gradle connectedCheck" step ("${{ steps.connectedCheck.outcome }}") was not successful'
           exit 1
 

firebase-dataconnect/ci/README.md

@@ -18,5 +18,5 @@ pip install -r requirements.txt
 Then, run all of these presubmit checks by running the following command:
 
 ```
-ruff check && ruff format && pyright && pytest && echo 'SUCCESS!!!!!!!!!!!!!!!'
+ruff check --fix && ruff format && pyright && pytest && echo 'SUCCESS!!!!!!!!!!!!!!!'
 ```

firebase-dataconnect/ci/logcat_error_report.py

@@ -0,0 +1,175 @@
+# Copyright 2025 Google LLC
+#
+# 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.
+
+from __future__ import annotations
+
+import argparse
+import dataclasses
+import logging
+import pathlib
+import re
+import tempfile
+import typing
+
+if typing.TYPE_CHECKING:
+  from _typeshed import SupportsWrite
+
+TEST_STARTED_TOKEN = "TestRunner: started:"  # noqa: S105
+TEST_STARTED_PATTERN = r"(\W|^)" + re.escape(TEST_STARTED_TOKEN) + r"\s+(?P<name>.*\S)"
+TEST_FAILED_TOKEN = "TestRunner: failed:"  # noqa: S105
+TEST_FAILED_PATTERN = r"(\W|^)" + re.escape(TEST_FAILED_TOKEN) + r"\s+(?P<name>.*\S)"
+TEST_FINISHED_TOKEN = "TestRunner: finished:"  # noqa: S105
+TEST_FINISHED_PATTERN = r"(\W|^)" + re.escape(TEST_FINISHED_TOKEN) + r"\s+(?P<name>.*\S)"
+
+
+@dataclasses.dataclass
+class TestResult:
+  test_name: str
+  output_file: pathlib.Path
+  passed: bool
+
+
+def main() -> None:
+  args = parse_args()
+  logging.basicConfig(format="%(message)s", level=args.log_level)
+
+  if args.work_dir is None:
+    work_temp_dir = tempfile.TemporaryDirectory("dd9rh9apdf")
+    work_dir = pathlib.Path(work_temp_dir.name)
+    logging.debug("Using temporary directory as work directory: %s", work_dir)
+  else:
+    work_temp_dir = None
+    work_dir = args.work_dir
+    logging.debug("Using specified directory as work directory: %s", work_dir)
+    work_dir.mkdir(parents=True, exist_ok=True)
+
+  logging.info("Extracting test failures from %s", args.logcat_file)
+  test_results: list[TestResult] = []
+  cur_test_result: TestResult | None = None
+  cur_test_result_output_file: SupportsWrite[str] | None = None
+
+  with args.logcat_file.open("rt", encoding="utf8", errors="ignore") as logcat_file_handle:
+    for line in logcat_file_handle:
+      test_started_match = TEST_STARTED_TOKEN in line and re.search(TEST_STARTED_PATTERN, line)
+      if test_started_match:
+        test_name = test_started_match.group("name")
+        logging.debug('Found "Test Started" logcat line for test: %s', test_name)
+        if cur_test_result_output_file is not None:
+          cur_test_result_output_file.close()
+        test_output_file = work_dir / f"{len(test_results)}.txt"
+        cur_test_result = TestResult(test_name=test_name, output_file=test_output_file, passed=True)
+        test_results.append(cur_test_result)
+        cur_test_result_output_file = test_output_file.open("wt", encoding="utf8", errors="replace")
+
+      if cur_test_result_output_file is not None:
+        cur_test_result_output_file.write(line)
+
+      test_failed_match = TEST_FAILED_TOKEN in line and re.search(TEST_FAILED_PATTERN, line)
+      if test_failed_match:
+        test_name = test_failed_match.group("name")
+        logging.warning("FAILED TEST: %s", test_name)
+        if cur_test_result is None:
+          logging.warning(
+            "WARNING: failed test reported without matching test started: %s", test_name
+          )
+        else:
+          cur_test_result.passed = False
+
+      test_finished_match = TEST_FINISHED_TOKEN in line and re.search(TEST_FINISHED_PATTERN, line)
+      if test_finished_match:
+        test_name = test_finished_match.group("name")
+        logging.debug('Found "Test Finished" logcat line for test: %s', test_name)
+        if cur_test_result_output_file is not None:
+          cur_test_result_output_file.close()
+        cur_test_result_output_file = None
+        cur_test_result = None
+
+  if cur_test_result_output_file is not None:
+    cur_test_result_output_file.close()
+  del cur_test_result_output_file
+
+  passed_tests = [test_result for test_result in test_results if test_result.passed]
+  failed_tests = [test_result for test_result in test_results if not test_result.passed]
+  print_line(
+    f"Found results for {len(test_results)} tests: "
+    f"{len(passed_tests)} passed, {len(failed_tests)} failed"
+  )
+
+  if len(failed_tests) > 0:
+    fail_number = 0
+    for failed_test_result in failed_tests:
+      fail_number += 1
+      print_line("")
+      print_line(f"Failure {fail_number}/{len(failed_tests)}: {failed_test_result.test_name}:")
+      try:
+        with failed_test_result.output_file.open(
+          "rt", encoding="utf8", errors="ignore"
+        ) as test_output_file:
+          for line in test_output_file:
+            print_line(line.rstrip())
+      except OSError:
+        logging.warning("WARNING: reading file failed: %s", failed_test_result.output_file)
+        continue
+
+  if work_temp_dir is not None:
+    logging.debug("Cleaning up temporary directory: %s", work_dir)
+  del work_dir
+  del work_temp_dir
+
+
+def print_line(line: str) -> None:
+  print(line)  # noqa: T201
+
+
+class ParsedArgs(typing.Protocol):
+  logcat_file: pathlib.Path
+  log_level: int
+  work_dir: pathlib.Path | None
+
+
+def parse_args() -> ParsedArgs:
+  arg_parser = argparse.ArgumentParser()
+  arg_parser.add_argument(
+    "--logcat-file",
+    required=True,
+    help="The text file containing the logcat logs to scan.",
+  )
+  arg_parser.add_argument(
+    "--work-dir",
+    default=None,
+    help="The directory into which to write temporary files; "
+    "if not specified, use a temporary directory that is deleted "
+    "when this script completes; this is primarily intended for "
+    "developers of this script to use in testing and debugging",
+  )
+  arg_parser.add_argument(
+    "--verbose",
+    action="store_const",
+    dest="log_level",
+    default=logging.INFO,
+    const=logging.DEBUG,
+    help="Include debug logging output",
+  )
+
+  parse_result = arg_parser.parse_args()
+
+  parse_result.logcat_file = pathlib.Path(parse_result.logcat_file)
+  parse_result.work_dir = (
+    None if parse_result.work_dir is None else pathlib.Path(parse_result.work_dir)
+  )
+  return typing.cast("ParsedArgs", parse_result)
+
+
+if __name__ == "__main__":
+  main()

firebase-dataconnect/ci/logcat_error_report_test.py

@@ -0,0 +1,149 @@
+# Copyright 2025 Google LLC
+#
+# 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.
+
+from __future__ import annotations
+
+import re
+
+import pytest
+
+import logcat_error_report as sut
+
+
+class TestRegularExpressionPatterns:
+  @pytest.mark.parametrize(
+    "string",
+    [
+      "",
+      "XTestRunner: started: fooTest1234",
+      "TestRunner: started:fooTest1234",
+      pytest.param(
+        "TestRunner: started: fooTest1234",
+        marks=pytest.mark.xfail(
+          reason="make sure that the test would otherwise pass on match",
+          strict=True,
+        ),
+      ),
+    ],
+  )
+  def test_test_started_pattern_no_match(self, string: str) -> None:
+    assert re.search(sut.TEST_STARTED_PATTERN, string) is None
+
+  @pytest.mark.parametrize(
+    ("string", "expected_name"),
+    [
+      ("TestRunner: started: fooTest1234", "fooTest1234"),
+      ("  TestRunner: started: fooTest1234", "fooTest1234"),
+      ("TestRunner: started:   fooTest1234", "fooTest1234"),
+      ("TestRunner: started: fooTest1234  ", "fooTest1234"),
+      ("TestRunner: started: fooTest1234(abc.123)", "fooTest1234(abc.123)"),
+      ("TestRunner: started: a $ 2 ^ %% .  ", "a $ 2 ^ %% ."),
+      pytest.param(
+        "i do not match the pattern",
+        None,
+        marks=pytest.mark.xfail(
+          reason="make sure that the test would otherwise pass on match",
+          strict=True,
+        ),
+      ),
+    ],
+  )
+  def test_test_started_pattern_match(self, string: str, expected_name: str) -> None:
+    match = re.search(sut.TEST_STARTED_PATTERN, string)
+    assert match is not None
+    assert match.group("name") == expected_name
+
+  @pytest.mark.parametrize(
+    "string",
+    [
+      "",
+      "XTestRunner: finished: fooTest1234",
+      "TestRunner: finished:fooTest1234",
+      pytest.param(
+        "TestRunner: finished: fooTest1234",
+        marks=pytest.mark.xfail(
+          reason="make sure that the test would otherwise pass on match",
+          strict=True,
+        ),
+      ),
+    ],
+  )
+  def test_test_finished_pattern_no_match(self, string: str) -> None:
+    assert re.search(sut.TEST_FINISHED_PATTERN, string) is None
+
+  @pytest.mark.parametrize(
+    ("string", "expected_name"),
+    [
+      ("TestRunner: finished: fooTest1234", "fooTest1234"),
+      ("  TestRunner: finished: fooTest1234", "fooTest1234"),
+      ("TestRunner: finished:   fooTest1234", "fooTest1234"),
+      ("TestRunner: finished: fooTest1234  ", "fooTest1234"),
+      ("TestRunner: finished: fooTest1234(abc.123)", "fooTest1234(abc.123)"),
+      ("TestRunner: finished: a $ 2 ^ %% .  ", "a $ 2 ^ %% ."),
+      pytest.param(
+        "i do not match the pattern",
+        None,
+        marks=pytest.mark.xfail(
+          reason="make sure that the test would otherwise pass on match",
+          strict=True,
+        ),
+      ),
+    ],
+  )
+  def test_test_finished_pattern_match(self, string: str, expected_name: str) -> None:
+    match = re.search(sut.TEST_FINISHED_PATTERN, string)
+    assert match is not None
+    assert match.group("name") == expected_name
+
+  @pytest.mark.parametrize(
+    "string",
+    [
+      "",
+      "XTestRunner: failed: fooTest1234",
+      "TestRunner: failed:fooTest1234",
+      pytest.param(
+        "TestRunner: failed: fooTest1234",
+        marks=pytest.mark.xfail(
+          reason="make sure that the test would otherwise pass on match",
+          strict=True,
+        ),
+      ),
+    ],
+  )
+  def test_test_failed_pattern_no_match(self, string: str) -> None:
+    assert re.search(sut.TEST_FAILED_PATTERN, string) is None
+
+  @pytest.mark.parametrize(
+    ("string", "expected_name"),
+    [
+      ("TestRunner: failed: fooTest1234", "fooTest1234"),
+      ("  TestRunner: failed: fooTest1234", "fooTest1234"),
+      ("TestRunner: failed:   fooTest1234", "fooTest1234"),
+      ("TestRunner: failed: fooTest1234  ", "fooTest1234"),
+      ("TestRunner: failed: fooTest1234(abc.123)", "fooTest1234(abc.123)"),
+      ("TestRunner: failed: a $ 2 ^ %% .  ", "a $ 2 ^ %% ."),
+      pytest.param(
+        "i do not match the pattern",
+        None,
+        marks=pytest.mark.xfail(
+          reason="make sure that the test would otherwise pass on match",
+          strict=True,
+        ),
+      ),
+    ],
+  )
+  def test_test_failed_pattern_match(self, string: str, expected_name: str) -> None:
+    match = re.search(sut.TEST_FAILED_PATTERN, string)
+    assert match is not None
+    assert match.group("name") == expected_name

firebase-dataconnect/ci/pyproject.toml

@@ -16,6 +16,7 @@ indent-width = 2
 [tool.ruff.lint]
 select = ["ALL"]
 ignore = [
+  "C901", # function is too complex
   "COM812", # missing-trailing-comma
   "D100", # Missing docstring in public module
   "D101", # Missing docstring in public class
@@ -29,6 +30,8 @@ ignore = [
   "E501", # Line too long (will be fixed by the formatter)
   "EM101", # Exception must not use a string literal, assign to variable first
   "LOG015", # root-logger-call
+  "PLR0912", # Too many branches
+  "PLR0915", # Too many statements
   "TRY003", # Avoid specifying long messages outside the exception class
 ]