Merge branch '@invertase/feat-add-retry-for-storage-firestore-database' of https://github.com/firebase/firebase-functions-python into @invertase/feat-add-retry-for-storage-firestore-database

Author: IzaakGough

src/firebase_functions/options.py

@@ -254,10 +254,11 @@ class RuntimeOptions:
     to the value "gcf_gen1"
     """
 
-    vpc_connector: str | _util.Sentinel | None = None
+    vpc_connector: str | Expression[str] | _util.Sentinel | None = None
     """
     Connect function to specified VPC connector.
-    A value of ``RESET_VALUE`` removes the VPC connector.
+    Accepts a plain string, an ``Expression[str]`` (e.g. a ``StringParam`` or
+    ternary expression), or ``RESET_VALUE`` to remove the VPC connector.
     """
 
     vpc_connector_egress_settings: VpcEgressSetting | _util.Sentinel | None = None
@@ -1239,8 +1240,8 @@ def set_global_options(
     max_instances: int | Expression[int] | _util.Sentinel | None = None,
     concurrency: int | Expression[int] | _util.Sentinel | None = None,
     cpu: int | _typing.Literal["gcf_gen1"] | _util.Sentinel = "gcf_gen1",
-    vpc_connector: str | None = None,
-    vpc_connector_egress_settings: VpcEgressSetting | None = None,
+    vpc_connector: str | Expression[str] | _util.Sentinel | None = None,
+    vpc_connector_egress_settings: VpcEgressSetting | _util.Sentinel | None = None,
     service_account: str | _util.Sentinel | None = None,
     ingress: IngressSetting | _util.Sentinel | None = None,
     labels: dict[str, str] | None = None,

tests/conftest.py

@@ -0,0 +1,30 @@
+# Copyright 2026 Google Inc.
+#
+# 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.
+"""
+Pytest configuration and shared fixtures for the test suite.
+"""
+
+import pytest
+
+from firebase_functions import params
+
+# pylint: disable=protected-access
+
+
+@pytest.fixture(autouse=True)
+def _cleanup_params():
+    """Clear the global params registry so each test runs with a clean state."""
+    params._params.clear()
+    yield
+    params._params.clear()

tests/test_options.py

@@ -15,6 +15,7 @@
 Options unit tests.
 """
 
+import pytest
 from pytest import raises
 
 from firebase_functions import alerts_fn, https_fn, options, params
@@ -41,6 +42,15 @@ def asamplefunctionpreserved(_):
     return "hello world"
 
 
+@pytest.fixture(autouse=True)
+def _cleanup_global_options():
+    """Reset global options so each test runs with a clean state."""
+    original_options = options._GLOBAL_OPTIONS
+    options._GLOBAL_OPTIONS = options.RuntimeOptions()
+    yield
+    options._GLOBAL_OPTIONS = original_options
+
+
 def test_set_global_options():
     """
     Testing if setting a global option internally change the values.
@@ -64,6 +74,30 @@ def test_global_options_merged_with_provider_options():
     )
 
 
+@pytest.mark.parametrize(
+    "vpc_connector_expr_factory",
+    [
+        lambda: params.StringParam("VPC_CONNECTOR"),
+        lambda: params.BoolParam("USE_VPC").equals(True).then("my-vpc", ""),
+    ],
+)
+def test_set_global_options_accepts_vpc_connector_expression(vpc_connector_expr_factory):
+    vpc_connector_expr = vpc_connector_expr_factory()
+    options.set_global_options(vpc_connector=vpc_connector_expr)
+
+    assert options._GLOBAL_OPTIONS.vpc_connector == vpc_connector_expr, (
+        "global vpc_connector expression was not stored"
+    )
+
+    https_options = options.HttpsOptions()
+    endpoint = https_options._endpoint(func_name="test_vpc_global")
+
+    assert endpoint.vpc is not None, "vpc block was not set on endpoint"
+    assert endpoint.vpc["connector"] == str(vpc_connector_expr), (
+        "global vpc_connector Expression[str] was not applied to the endpoint"
+    )
+
+
 def test_https_options_removes_cors():
     """
     Testing _HttpsOptions strips out the 'cors' property when converted to a dict.
@@ -206,6 +240,48 @@ def test_invoker_with_no_element_throws():
         options.HttpsOptions(invoker=[])._endpoint(func_name="test")
 
 
+@pytest.mark.parametrize(
+    "vpc_connector_expr_factory",
+    [
+        lambda: params.StringParam("VPC_CONNECTOR"),
+        lambda: params.BoolParam("USE_VPC").equals(True).then("my-vpc", ""),
+    ],
+)
+def test_vpc_connector_accepts_expression(vpc_connector_expr_factory):
+    vpc_connector_expr = vpc_connector_expr_factory()
+    https_options = options.HttpsOptions(vpc_connector=vpc_connector_expr)
+    https_options_dict = https_options._asdict_with_global_options()
+
+    # The options dict should contain the CEL string representation for the expression.
+    assert https_options_dict["vpc_connector"] == str(vpc_connector_expr), (
+        "vpc_connector expression was not converted to CEL string"
+    )
+
+    # The generated endpoint should map the resolved vpc_connector into the vpc block.
+    endpoint = https_options._endpoint(func_name="test_vpc")
+    assert endpoint.vpc is not None, "vpc block was not set on endpoint"
+    assert endpoint.vpc["connector"] == str(vpc_connector_expr), (
+        "vpc connector was not set from vpc_connector Expression[str]"
+    )
+
+
+def test_vpc_connector_expression_with_egress_settings():
+    vpc_connector_expr = params.StringParam("VPC_CONNECTOR")
+    https_options = options.HttpsOptions(
+        vpc_connector=vpc_connector_expr,
+        vpc_connector_egress_settings=options.VpcEgressSetting.ALL_TRAFFIC,
+    )
+
+    endpoint = https_options._endpoint(func_name="test_vpc_egress")
+    assert endpoint.vpc is not None, "vpc block was not set on endpoint"
+    assert endpoint.vpc["connector"] == str(vpc_connector_expr), (
+        "vpc connector was not set from vpc_connector Expression[str]"
+    )
+    assert endpoint.vpc.get("egressSettings") == options.VpcEgressSetting.ALL_TRAFFIC.value, (
+        "egressSettings was not set alongside an Expression[str] vpc_connector"
+    )
+
+
 def _assert_alert_endpoint_options(endpoint, expected_alert_type, expect_app_id: str | None = None):
     assert endpoint.region == ["europe-west1"]
     assert endpoint.maxInstances == 1