chore: migrate off of bazelisk query command (#14214)

This PR migrates off of the `bazelisk query` command and instead uses a
regex to read the bazel rule from the `BUILD.bazel` file directly.

Author: ohmayr

.generator/cli.py

@@ -16,6 +16,7 @@
 import json
 import logging
 import os
+import re
 import subprocess
 import sys
 import subprocess
@@ -66,37 +67,43 @@ def handle_configure():
 
 
 def _determine_bazel_rule(api_path: str, source: str) -> str:
-    """Executes a `bazelisk query` to find a Bazel rule.
+    """Finds a Bazel rule by parsing the BUILD.bazel file directly.
 
     Args:
-        api_path(str): The API path to query for.
+        api_path (str): The API path, e.g., 'google/cloud/language/v1'.
         source(str): The path to the root of the Bazel workspace.
 
     Returns:
-        str: The discovered Bazel rule.
+        str: The discovered Bazel rule, e.g., '//google/cloud/language/v1:language-v1-py'.
 
     Raises:
-        ValueError: If the subprocess call fails or returns an empty result.
+        ValueError: If the file can't be processed or no matching rule is found.
     """
-    logger.info(f"Determining Bazel rule for api_path: '{api_path}'")
+    logger.info(f"Determining Bazel rule for api_path: '{api_path}' by parsing file.")
     try:
-        query = f'filter("-py$", kind("rule", //{api_path}/...:*))'
-        command = ["bazelisk", "query", query]
-        result = subprocess.run(
-            command,
-            cwd=source,
-            capture_output=True,
-            text=True,
-            check=True,
-        )
-        bazel_rule = result.stdout.strip()
-        if not bazel_rule:
-            raise ValueError(f"Bazelisk query `{query}` returned an empty bazel rule.")
+        build_file_path = os.path.join(source, api_path, "BUILD.bazel")
+
+        with open(build_file_path, "r") as f:
+            content = f.read()
+
+        match = re.search(r'name\s*=\s*"([^"]+-py)"', content)
+
+        # This check is for a logical failure (no match), not a runtime exception.
+        # It's good to keep it for clear error messaging.
+        if not match:
+            raise ValueError(
+                f"No Bazel rule with a name ending in '-py' found in {build_file_path}"
+            )
+
+        rule_name = match.group(1)
+        bazel_rule = f"//{api_path}:{rule_name}"
 
         logger.info(f"Found Bazel rule: {bazel_rule}")
         return bazel_rule
     except Exception as e:
-        raise ValueError(f"Bazelisk query `{query}` failed") from e
+        raise ValueError(
+            f"Failed to determine Bazel rule for '{api_path}' by parsing."
+        ) from e
 
 
 def _get_library_id(request_data: Dict) -> str:
@@ -227,11 +234,11 @@ def handle_generate(
     Args:
         librarian(str): Path to the directory in the container which contains
             the librarian configuration.
-        source(str): Path to the directory in the container which contains 
+        source(str): Path to the directory in the container which contains
             API protos.
-        output(str): Path to the directory in the container where code 
+        output(str): Path to the directory in the container where code
             should be generated.
-        input(str): The path path to the directory in the container 
+        input(str): The path path to the directory in the container
             which contains additional generator input.
 
     Raises:

.generator/test_cli.py

@@ -110,10 +110,8 @@ def test_determine_bazel_rule_success(mocker, caplog):
     Tests the happy path of _determine_bazel_rule.
     """
     caplog.set_level(logging.INFO)
-    mock_result = MagicMock(
-        stdout="//google/cloud/language/v1:google-cloud-language-v1-py\n"
-    )
-    mocker.patch("cli.subprocess.run", return_value=mock_result)
+    mock_content = 'name = "google-cloud-language-v1-py",\n'
+    mocker.patch("cli.open", mock_open(read_data=mock_content))
 
     rule = _determine_bazel_rule("google/cloud/language/v1", "source")
 
@@ -131,15 +129,35 @@ def test_build_bazel_target_success(mocker, caplog):
     assert "Bazel build for mock/bazel:rule rule completed successfully" in caplog.text
 
 
+def test_build_bazel_target_fails_to_find_rule_match(mocker, caplog):
+    """
+    Tests that ValueError is raised if the subprocess command fails.
+    """
+    caplog.set_level(logging.ERROR)
+    mock_content = '"google-cloud-language-v1-py",\n'
+    mocker.patch("cli.open", mock_open(read_data=mock_content))
+
+    with pytest.raises(ValueError):
+        _build_bazel_target("mock/bazel:rule", "source")
+
+
+def test_build_bazel_target_fails_to_determine_rule(caplog):
+    """
+    Tests that ValueError is raised if the subprocess command fails.
+    """
+    caplog.set_level(logging.ERROR)
+    with pytest.raises(ValueError):
+        _build_bazel_target("mock/bazel:rule", "source")
+
+
 def test_build_bazel_target_fails(mocker, caplog):
     """
     Tests that ValueError is raised if the subprocess command fails.
     """
     caplog.set_level(logging.ERROR)
-    mocker.patch(
-        "cli.subprocess.run",
-        side_effect=subprocess.CalledProcessError(1, "cmd", stderr="Build failed"),
-    )
+    mock_content = '"google-cloud-language-v1-py",\n'
+    mocker.patch("cli.open", mock_open(read_data=mock_content))
+
     with pytest.raises(ValueError):
         _build_bazel_target("mock/bazel:rule", "source")
 
@@ -174,7 +192,7 @@ def test_locate_and_extract_artifact_success(mocker, caplog):
         "google-cloud-language",
         "source",
         "output",
-        "google/cloud/language/v1"
+        "google/cloud/language/v1",
     )
     assert (
         "Found artifact at: /path/to/bazel-bin/google/cloud/language/v1/rule-py.tar.gz"