diff --git a/packages/gapic-generator/gapic/generator/generator.py b/packages/gapic-generator/gapic/generator/generator.py
index 9fe56aa9de1d..c1932abbe426 100644
--- a/packages/gapic-generator/gapic/generator/generator.py
+++ b/packages/gapic-generator/gapic/generator/generator.py
@@ -120,7 +120,7 @@ def get_response(self, api_schema: api.API, opts: Options) -> CodeGeneratorRespo
         for template_name in client_templates:
             # Quick check: Skip "private" templates.
             filename = template_name.split("/")[-1]
-            if filename.startswith("_") and filename != "__init__.py.j2":
+            if filename.startswith("_") and filename not in ("__init__.py.j2", "_compat.py.j2"):
                 continue
 
             # Append to the output files dictionary.
diff --git a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2 b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2
new file mode 100644
index 000000000000..368b13450a76
--- /dev/null
+++ b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2
@@ -0,0 +1,270 @@
+# {% include '_license.j2' %}
+
+"""A compatibility module for older versions of google-api-core."""
+
+import functools
+import json
+import operator
+import os
+import re
+import uuid
+from typing import Any, Callable, Dict, List, Optional, Tuple, Union
+from google.auth.exceptions import MutualTLSChannelError
+import google.protobuf.message
+
+
+try:
+    from google.api_core.universe import (
+        get_default_mtls_endpoint,
+        get_api_endpoint,
+        get_universe_domain,
+    )
+except ImportError:
+    # TODO(https://github.com/googleapis/google-cloud-python/issues/17813): Remove these fallbacks when google-api-core >= 2.18.0 is the minimum required version.
+    def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
+        """Converts api endpoint to mTLS endpoint."""
+        if not api_endpoint:
+            return api_endpoint
+
+        mtls_endpoint_re = re.compile(
+            r"(?P<name>[^.]+)(?P<mtls>\.mtls)?(?P<sandbox>\.sandbox)?(?P<googledomain>\.googleapis\.com)?"
+        )
+
+        m = mtls_endpoint_re.match(api_endpoint)
+        if m is None:
+            # Could not parse api_endpoint; return as-is.
+            return api_endpoint
+
+        name, mtls, sandbox, googledomain = m.groups()
+        if mtls or not googledomain:
+            return api_endpoint
+
+        if sandbox:
+            return api_endpoint.replace(
+                "sandbox.googleapis.com", "mtls.sandbox.googleapis.com"
+            )
+
+        return api_endpoint.replace(".googleapis.com", ".mtls.googleapis.com")
+
+    def get_api_endpoint(
+        api_override: Optional[str],
+        client_cert_source: Optional[Callable[[], Tuple[bytes, bytes]]],
+        universe_domain: str,
+        use_mtls_endpoint: str,
+        default_universe: str,
+        default_mtls_endpoint: Optional[str],
+        default_endpoint_template: str,
+    ) -> Optional[str]:
+        """Return the API endpoint used by the client."""
+        if api_override is not None:
+            api_endpoint = api_override
+        elif use_mtls_endpoint == "always" or (use_mtls_endpoint == "auto" and client_cert_source):
+            if universe_domain != default_universe:
+                raise MutualTLSChannelError(
+                    f"mTLS is not supported in any universe other than {default_universe}."
+                )
+            api_endpoint = default_mtls_endpoint
+        else:
+            api_endpoint = default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
+        return api_endpoint
+
+    def get_universe_domain(
+        client_universe_domain: Optional[str],
+        universe_domain_env: Optional[str],
+        default_universe: str,
+    ) -> str:
+        """Return the universe domain used by the client."""
+        universe_domain = default_universe
+        if client_universe_domain is not None:
+            universe_domain = client_universe_domain
+        elif universe_domain_env is not None:
+            universe_domain = universe_domain_env
+        if len(universe_domain.strip()) == 0:
+            raise ValueError("Universe Domain cannot be an empty string.") 
+        return universe_domain
+
+
+try:
+    from google.api_core.gapic_v1.config import (
+        use_client_cert_effective,
+        get_client_cert_source,
+        read_environment_variables,
+    )
+except ImportError:
+    from google.auth.transport import mtls  # type: ignore
+
+    # TODO(https://github.com/googleapis/google-cloud-python/issues/17813): Remove these fallbacks when google-api-core >= 2.18.0 is the minimum required version.
+
+    def use_client_cert_effective() -> bool:
+        """Returns whether client certificate should be used for mTLS."""
+        if hasattr(mtls, "should_use_client_cert"):
+            return mtls.should_use_client_cert()
+        else:
+            use_client_cert_str = os.getenv("GOOGLE_API_USE_CLIENT_CERTIFICATE", "false").lower()
+            if use_client_cert_str not in ("true", "false"):
+                raise ValueError(
+                    "Environment variable `GOOGLE_API_USE_CLIENT_CERTIFICATE` must be"
+                    " either `true` or `false`"
+                )
+            return use_client_cert_str == "true"
+
+    def get_client_cert_source(
+        provided_cert_source: Optional[Callable[[], Tuple[bytes, bytes]]],
+        use_cert_flag: bool,
+    ) -> Optional[Callable[[], Tuple[bytes, bytes]]]:
+        """Return the client cert source to be used by the client."""
+        client_cert_source = None
+        if use_cert_flag:
+            if provided_cert_source:
+                client_cert_source = provided_cert_source
+            elif (
+                hasattr(mtls, "has_default_client_cert_source")
+                and mtls.has_default_client_cert_source()
+            ):
+                client_cert_source = mtls.default_client_cert_source()
+            else:
+                raise ValueError(
+                    "Client certificate is required for mTLS, but no client certificate source was provided or found."
+                )
+        return client_cert_source
+
+    def read_environment_variables() -> Tuple[bool, str, Optional[str]]:
+        """Returns the environment variables used by the client."""
+        use_client_cert = use_client_cert_effective()
+        use_mtls_endpoint = os.getenv("GOOGLE_API_USE_MTLS_ENDPOINT", "auto").lower()
+        universe_domain_env = os.getenv("GOOGLE_CLOUD_UNIVERSE_DOMAIN")
+        if use_mtls_endpoint not in ("auto", "never", "always"):
+            raise MutualTLSChannelError(
+                "Environment variable `GOOGLE_API_USE_MTLS_ENDPOINT` "
+                "must be `never`, `auto` or `always`"
+            )
+        return use_client_cert, use_mtls_endpoint, universe_domain_env
+
+
+try:
+    from google.api_core.gapic_v1.request import setup_request_id  # type: ignore
+except ImportError:
+    # TODO(https://github.com/googleapis/google-cloud-python/issues/17813): Remove this fallback when google-api-core >= 2.26.0 is the minimum required version.
+    def setup_request_id(request, field_name: str, is_proto3_optional: bool):
+        """Populate a UUID4 field in the request if it is not already set.
+
+        Args:
+            request (Union[google.protobuf.message.Message, dict]): The request object.
+            field_name (str): The name of the field to populate.
+            is_proto3_optional (bool): Whether the field is proto3 optional.
+        """
+        request_id_val = str(uuid.uuid4())
+        if request is None:
+            return
+
+        if isinstance(request, dict):
+            if is_proto3_optional:
+                if field_name not in request or request[field_name] is None:
+                    request[field_name] = request_id_val
+            elif not request.get(field_name):
+                request[field_name] = request_id_val
+            return
+
+        if is_proto3_optional:
+            try:
+                # Pure protobuf messages
+                if not request.HasField(field_name):
+                    setattr(request, field_name, request_id_val)
+            except (AttributeError, ValueError):
+                # Proto-plus messages or other objects
+                if getattr(request, field_name, None) is None:
+                    setattr(request, field_name, request_id_val)
+        else:
+            if not getattr(request, field_name, None):
+                setattr(request, field_name, request_id_val)
+
+
+try:
+    from google.api_core.rest_helpers import (
+        flatten_query_params,
+        transcode_request,
+    )
+except ImportError:  # pragma: NO COVER
+    # TODO: Remove these fallbacks when google-api-core >= 2.18.0 is the minimum required version.
+    from google.protobuf import json_format  # type: ignore
+    from google.api_core import path_template  # type: ignore
+
+    def flatten_query_params(obj, strict=False):  # pragma: NO COVER
+        if obj is not None and not isinstance(obj, dict):
+            raise TypeError("flatten_query_params must be called with dict object")
+        return _flatten(obj, key_path=[], strict=strict)
+
+    def _flatten(obj, key_path, strict=False):  # pragma: NO COVER
+        if obj is None:
+            return []
+        if isinstance(obj, dict):
+            return _flatten_dict(obj, key_path=key_path, strict=strict)
+        if isinstance(obj, list):
+            return _flatten_list(obj, key_path=key_path, strict=strict)
+        return _flatten_value(obj, key_path=key_path, strict=strict)
+
+    def _is_primitive_value(obj):  # pragma: NO COVER
+        if obj is None:
+            return False
+        if isinstance(obj, (list, dict)):
+            raise ValueError("query params may not contain repeated dicts or lists")
+        return True
+
+    def _flatten_value(obj, key_path, strict=False):  # pragma: NO COVER
+        return [(".".join(key_path), _canonicalize(obj, strict=strict))]
+
+    def _flatten_dict(obj, key_path, strict=False):  # pragma: NO COVER
+        items = (
+            _flatten(value, key_path=key_path + [key], strict=strict)
+            for key, value in obj.items()
+        )
+        return functools.reduce(operator.concat, items, [])
+
+    def _flatten_list(elems, key_path, strict=False):  # pragma: NO COVER
+        items = (
+            _flatten_value(elem, key_path=key_path, strict=strict)
+            for elem in elems
+            if _is_primitive_value(elem)
+        )
+        return functools.reduce(operator.concat, items, [])
+
+    def _canonicalize(obj, strict=False):  # pragma: NO COVER
+        if strict:
+            value = str(obj)
+            if isinstance(obj, bool):
+                value = value.lower()
+            return value
+        return obj
+
+    def transcode_request(  # pragma: NO COVER
+        http_options: List[Dict[str, str]],
+        request: Any,
+        required_fields_default_values: Optional[Dict[str, Any]] = None,
+        rest_numeric_enums: bool = False,
+    ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
+        pb_request = getattr(request, "_pb", request)
+        transcoded_request = path_template.transcode(http_options, pb_request)
+
+        body_json = None
+        if transcoded_request.get("body") is not None:
+            body_json = json_format.MessageToJson(
+                transcoded_request["body"],
+                use_integers_for_enums=rest_numeric_enums,
+            )
+
+        query_params_json = {}
+        if transcoded_request.get("query_params") is not None:
+            query_params_json = json.loads(json_format.MessageToJson(
+                transcoded_request["query_params"],
+                use_integers_for_enums=rest_numeric_enums,
+            ))
+
+        if required_fields_default_values:
+            for k, v in required_fields_default_values.items():
+                if k not in query_params_json:
+                    query_params_json[k] = v
+
+        if rest_numeric_enums:
+            query_params_json["$alt"] = "json;enum-encoding=int"
+
+        return transcoded_request, body_json, query_params_json
diff --git a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/_shared_macros.j2 b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/_shared_macros.j2
index 755e4530e7ba..6e58759112ca 100644
--- a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/_shared_macros.j2
+++ b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/_shared_macros.j2
@@ -198,22 +198,23 @@ def _get_http_options():
         service: The service.
         is_async (bool): Used to determine the code path i.e. whether for sync or async call.
         is_request_message_proto_plus_type (bool): Used to determine whether the request message is a proto-plus type. #}
-{% macro rest_call_method_common(body_spec, method_name, service, is_async=False, is_request_message_proto_plus_type=False) %}
+{% macro rest_call_method_common(body_spec, method_name, service, is_async=False, is_request_message_proto_plus_type=False, rest_numeric_enums=False) %}
 {% set service_name = service.name %}
 {% set await_prefix = "await " if is_async else "" %}
 {% set async_class_prefix = "Async" if is_async else "" %}
 
     http_options = _Base{{ service_name }}RestTransport._Base{{method_name}}._get_http_options()
-    {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2274): Add debug log before intercepting a request #} 
     request, metadata = {{ await_prefix }}self._interceptor.pre_{{ method_name|snake_case }}(request, metadata)
-    transcoded_request = _Base{{ service_name }}RestTransport._Base{{method_name}}._get_transcoded_request(http_options, request)
-
-    {% if body_spec %}
-    body = _Base{{ service_name }}RestTransport._Base{{method_name}}._get_request_body_json(transcoded_request)
-    {% endif %}{# body_spec #}
-
-    # Jsonify the query params
-    query_params = _Base{{ service_name }}RestTransport._Base{{method_name}}._get_query_params_json(transcoded_request)
+    transcoded_request, body, query_params = rest_helpers.transcode_request(
+        http_options,
+        request,
+        required_fields_default_values=getattr(
+            _Base{{ service_name }}RestTransport._Base{{method_name}},
+            "__REQUIRED_FIELDS_DEFAULT_VALUES",
+            None,
+        ),
+        rest_numeric_enums={{ rest_numeric_enums }},
+    )
 
     if CLIENT_LOGGING_SUPPORTED and _LOGGER.isEnabledFor(logging.DEBUG):  # pragma: NO COVER
         request_url = "{host}{uri}".format(host=self._host, uri=transcoded_request['uri'])
diff --git a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/_rest_mixins_base.py.j2 b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/_rest_mixins_base.py.j2
index 16cc77ea937c..f768500d5755 100644
--- a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/_rest_mixins_base.py.j2
+++ b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/_rest_mixins_base.py.j2
@@ -25,27 +25,7 @@
 
         {{ shared_macros.http_options_method(api.mixin_http_options["{}".format(name)])|indent(8)}}
 
-        @staticmethod
-        def _get_transcoded_request(http_options, request):
-            request_kwargs = json_format.MessageToDict(request)
-            transcoded_request = path_template.transcode(
-                http_options, **request_kwargs)
-            return transcoded_request
-
-        {% set body_spec = api.mixin_http_options["{}".format(name)][0].body %}
-        {%- if body_spec %}
-
-        @staticmethod
-        def _get_request_body_json(transcoded_request):
-            body = json.dumps(transcoded_request['body'])
-            return body
-
-        {%- endif %} {# body_spec #}
-
-        @staticmethod
-        def _get_query_params_json(transcoded_request):
-            query_params = json.loads(json.dumps(transcoded_request['query_params']))
-            return query_params
+        pass
 
     {% endfor %}
     {% endif %} {# rest in opts.transport #}
diff --git a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2 b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2
index 49c1374053b5..a3e08a447d06 100644
--- a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2
+++ b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2
@@ -10,7 +10,8 @@ from google.auth.transport.requests import AuthorizedSession  # type: ignore
 from google.auth import credentials as ga_credentials  # type: ignore
 from google.api_core import exceptions as core_exceptions
 from google.api_core import retry as retries
-from google.api_core import rest_helpers
+{% set package_path = api.naming.module_namespace|join('.') + "." + api.naming.versioned_module_name %}
+from {{package_path}} import _compat as rest_helpers
 from google.api_core import rest_streaming
 from google.api_core import gapic_v1
 import google.protobuf
@@ -245,7 +246,7 @@ class {{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
             {% endif %}
             """
 
-            {{ shared_macros.rest_call_method_common(body_spec, method.name, service, False, method.input.ident.is_proto_plus_type)|indent(8) }}
+            {{ shared_macros.rest_call_method_common(body_spec, method.name, service, False, method.input.ident.is_proto_plus_type, opts.rest_numeric_enums)|indent(8) }}
 
             {% if not method.void %}
             # Return the response
diff --git a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest_asyncio.py.j2 b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest_asyncio.py.j2
index 80980572c30a..83c8cd28b986 100644
--- a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest_asyncio.py.j2
+++ b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest_asyncio.py.j2
@@ -32,7 +32,8 @@ from google.iam.v1 import policy_pb2  # type: ignore
 from google.cloud.location import locations_pb2 # type: ignore
 {% endif %}
 from google.api_core import retry_async as retries
-from google.api_core import rest_helpers
+{% set package_path = api.naming.module_namespace|join('.') + "." + api.naming.versioned_module_name %}
+from {{package_path}} import _compat as rest_helpers
 from google.api_core import rest_streaming_async  # type: ignore
 import google.protobuf
 
@@ -203,7 +204,7 @@ class Async{{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
             {% endif %}
             """
 
-            {{ shared_macros.rest_call_method_common(body_spec, method.name, service, True, method.input.ident.is_proto_plus_type)|indent(8) }}
+            {{ shared_macros.rest_call_method_common(body_spec, method.name, service, True, method.input.ident.is_proto_plus_type, opts.rest_numeric_enums)|indent(8) }}
 
             {% if not method.void %}
             # Return the response
diff --git a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest_base.py.j2 b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest_base.py.j2
index b79785afc517..90c0e10d6403 100644
--- a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest_base.py.j2
+++ b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest_base.py.j2
@@ -120,51 +120,8 @@ class _Base{{ service.name }}RestTransport({{service.name}}Transport):
         def _get_unset_required_fields(cls, message_dict):
             return {k: v for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items() if k not in message_dict}
         {% endif %}{# required fields #}
-        
         {% set method_http_options = method.http_options %}
-        
         {{ shared_macros.http_options_method(method_http_options)|indent(8) }}
-        
-        @staticmethod
-        def _get_transcoded_request(http_options, request):
-            {% if method.input.ident.is_proto_plus_type %}
-            pb_request = {{method.input.ident}}.pb(request)
-            {% else %}
-            pb_request = request
-            {% endif %}
-            transcoded_request = path_template.transcode(http_options, pb_request)
-            return transcoded_request
-
-        {% set body_spec = method.http_options[0].body %}
-        {%- if body_spec %}
-
-        @staticmethod
-        def _get_request_body_json(transcoded_request):
-            # Jsonify the request body
-
-            body = json_format.MessageToJson(
-                transcoded_request['body'],
-                use_integers_for_enums={{ opts.rest_numeric_enums }}
-            )
-            return body
-
-        {%- endif %}{# body_spec #}
-
-        @staticmethod
-        def _get_query_params_json(transcoded_request):
-            query_params = json.loads(json_format.MessageToJson(
-                transcoded_request['query_params'],
-                use_integers_for_enums={{ opts.rest_numeric_enums }},
-            ))
-            {% if method.input.required_fields %}
-            query_params.update(_Base{{ service.name }}RestTransport._Base{{method.name}}._get_unset_required_fields(query_params))
-            {% endif %}{# required fields #}
-
-            {% if opts.rest_numeric_enums %}
-            query_params["$alt"] = "json;enum-encoding=int"
-            {% endif %}
-            return query_params
-
         {% endif %}{# method.http_options and not method.client_streaming #}
     {% endfor %}
 
diff --git a/packages/gapic-generator/noxfile.py b/packages/gapic-generator/noxfile.py
index 52209f41ff38..d19c9a895a9a 100644
--- a/packages/gapic-generator/noxfile.py
+++ b/packages/gapic-generator/noxfile.py
@@ -36,6 +36,18 @@
 
 showcase_version = os.environ.get("SHOWCASE_VERSION", "0.35.0")
 ADS_TEMPLATES = path.join(path.dirname(__file__), "gapic", "ads-templates")
+CURRENT_DIRECTORY = Path(__file__).parent.absolute()
+# Path to the centralized mypy configuration file at the repository root.
+# Search upwards to support running nox from both monorepo packages and integration test goldens.
+MYPY_CONFIG_FILE = next(
+    (
+        str(p / "mypy.ini")
+        for p in CURRENT_DIRECTORY.parents
+        if (p / "mypy.ini").exists()
+    ),
+    str(CURRENT_DIRECTORY.parent.parent / "mypy.ini"),
+)
+
 RUFF_VERSION = "ruff==0.14.14"
 LINT_PATHS = ["docs", "gapic", "tests", "test_utils", "noxfile.py", "setup.py"]
 # Ruff uses globs for excludes (different from Black's regex)
@@ -181,6 +193,7 @@ def fragment(session, use_ads_templates=False):
         "grpcio-tools",
     )
     session.install("-e", ".")
+    session.install("-e", "../google-api-core")
 
     # The specific failure is `Plugin output is unparseable`
     if session.python == "3.10":
@@ -245,6 +258,7 @@ def showcase_library(
 
     # Install gapic-generator-python
     session.install("-e", ".")
+    session.install("-e", "../google-api-core")
 
     # Install grpcio-tools for protoc
     session.install("grpcio-tools")
@@ -730,7 +744,7 @@ def mypy(session):
         "click==8.1.3",
     )
     session.install(".")
-    session.run("mypy", "-p", "gapic")
+    session.run("mypy", f"--config-file={MYPY_CONFIG_FILE}", "-p", "gapic")
 
 
 @nox.session(python=NEWEST_PYTHON)
@@ -742,7 +756,7 @@ def lint(session):
     """
 
     # TODO(https://github.com/googleapis/google-cloud-python/issues/16186):
-    # SKIP: This session was not enforced in the standalone (split) repo 
+    # SKIP: This session was not enforced in the standalone (split) repo
     # and is disabled here to ensure a "move-only" migration.
     session.skip(
         "Linting was not enforced in the split repo. "
@@ -772,9 +786,11 @@ def lint(session):
 @nox.session(python=NEWEST_PYTHON)
 def lint_setup_py(session):
     # TODO(https://github.com/googleapis/google-cloud-python/issues/16186):
-    # SKIP: This session was not enforced in the standalone (split) repo 
+    # SKIP: This session was not enforced in the standalone (split) repo
     # and is disabled here to ensure a "move-only" migration.
-    session.skip("Skipping now to avoid changing code during migration. See Issue #16186")
+    session.skip(
+        "Skipping now to avoid changing code during migration. See Issue #16186"
+    )
 
 
 @nox.session(python="3.10")
@@ -849,9 +865,11 @@ def prerelease_deps(session, protobuf_implementation):
     """
     Run all tests with pre-release versions of dependencies installed.
     """
-    # TODO(https://github.com/googleapis/google-cloud-python/issues/16184): 
+    # TODO(https://github.com/googleapis/google-cloud-python/issues/16184):
     # Implement pre-release dependency logic to test against upcoming runtime changes.
-    session.skip("prerelease_deps session is not yet implemented for gapic-generator-python.")
+    session.skip(
+        "prerelease_deps session is not yet implemented for gapic-generator-python."
+    )
 
 
 @nox.session(python=NEWEST_PYTHON)
@@ -861,6 +879,8 @@ def prerelease_deps(session, protobuf_implementation):
 )
 def core_deps_from_source(session, protobuf_implementation):
     """Run all tests with core dependencies installed from source."""
-    # TODO(https://github.com/googleapis/google-cloud-python/issues/16185): 
+    # TODO(https://github.com/googleapis/google-cloud-python/issues/16185):
     # Implement logic to install core packages directly from the mono-repo directories.
-    session.skip("core_deps_from_source session is not yet implemented for gapic-generator-python.")
\ No newline at end of file
+    session.skip(
+        "core_deps_from_source session is not yet implemented for gapic-generator-python."
+    )
diff --git a/packages/gapic-generator/tests/unit/generator/test_generator.py b/packages/gapic-generator/tests/unit/generator/test_generator.py
index 9d8545c4192f..f1566ee6e5a7 100644
--- a/packages/gapic-generator/tests/unit/generator/test_generator.py
+++ b/packages/gapic-generator/tests/unit/generator/test_generator.py
@@ -117,6 +117,8 @@ def test_get_response_ignores_private_files():
         list_templates.return_value = [
             "foo/bar/baz.py.j2",
             "foo/bar/_base.py.j2",
+            "foo/bar/__init__.py.j2",
+            "foo/bar/_compat.py.j2",
             "molluscs/squid/sample.py.j2",
         ]
         with mock.patch.object(jinja2.Environment, "get_template") as get_template:
@@ -128,12 +130,13 @@ def test_get_response_ignores_private_files():
             get_template.assert_has_calls(
                 [
                     mock.call("molluscs/squid/sample.py.j2"),
+                    mock.call("foo/bar/__init__.py.j2"),
+                    mock.call("foo/bar/_compat.py.j2"),
                     mock.call("foo/bar/baz.py.j2"),
-                ]
+                ],
+                any_order=True,
             )
-            assert len(cgr.file) == 1
-            assert cgr.file[0].name == "foo/bar/baz.py"
-            assert cgr.file[0].content == "I am a template result.\n"
+            assert len(cgr.file) == 3
 
 
 def test_get_response_fails_invalid_file_paths():