Add unit tests for StaticConnectionInfoCache

Author: rhatgadkar-goog

tests/unit/mocks.py

@@ -149,7 +149,7 @@ def __init__(
         cluster: str = "test-cluster",
         name: str = "test-instance",
         ip_addrs: dict = {
-            "PRIVATE": "127.0.0.1",
+            "PRIVATE": "127.0.0.1",  # "private" IP is localhost in testing
             "PUBLIC": "0.0.0.0",
             "PSC": "x.y.alloydb.goog",
         },
@@ -443,8 +443,8 @@ def write_static_info(i: FakeInstance) -> io.StringIO:
     static[i.uri()] = {
         "pemCertificateChain": chain,
         "caCert": ca_cert,
-        "ipAddress": "127.0.0.1",  # "private" IP is localhost in testing
-        "publicIpAddress": "",
-        "pscInstanceConfig": {"pscDnsName": ""},
+        "ipAddress": i.ip_addrs["PRIVATE"],
+        "publicIpAddress": i.ip_addrs["PUBLIC"],
+        "pscInstanceConfig": {"pscDnsName": i.ip_addrs["PSC"]},
     }
     return io.StringIO(json.dumps(static))

tests/unit/test_static.py

@@ -0,0 +1,95 @@
+# 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 mocks import FakeInstance, write_static_info
+
+from google.cloud.alloydb.connector.connection_info import ConnectionInfo
+from google.cloud.alloydb.connector.static import StaticConnectionInfoCache
+
+
+def test_StaticConnectionInfoCache_init() -> None:
+    """
+    Test that StaticConnectionInfoCache.__init__ populates its ConnectionInfo
+    object.
+    """
+    i = FakeInstance()
+    static_info = write_static_info(i)
+    cache = StaticConnectionInfoCache(i.uri(), static_info)
+    assert len(cache._info.cert_chain) == 3
+    assert cache._info.ca_cert
+    assert cache._info.key
+    assert cache._info.ip_addrs == {
+        "PRIVATE": i.ip_addrs["PRIVATE"],
+        "PUBLIC": i.ip_addrs["PUBLIC"],
+        "PSC": i.ip_addrs["PSC"],
+    }
+    assert cache._info.expiration
+
+def test_StaticConnectionInfoCache_init_trailing_dot_dns() -> None:
+    """
+    Test that StaticConnectionInfoCache.__init__ populates its ConnectionInfo
+    object correctly when its PSC DNS name contains a trailing dot.
+    """
+    i = FakeInstance()
+    no_trailing_dot_dns = i.ip_addrs["PSC"]
+    i.ip_addrs["PSC"] += "."
+    static_info = write_static_info(i)
+    cache = StaticConnectionInfoCache(i.uri(), static_info)
+    assert len(cache._info.cert_chain) == 3
+    assert cache._info.ca_cert
+    assert cache._info.key
+    assert cache._info.ip_addrs == {
+        "PRIVATE": i.ip_addrs["PRIVATE"],
+        "PUBLIC": i.ip_addrs["PUBLIC"],
+        "PSC": no_trailing_dot_dns,
+    }
+    assert cache._info.expiration
+
+async def test_StaticConnectionInfoCache_force_refresh() -> None:
+    """
+    Test that StaticConnectionInfoCache.force_refresh is a no-op.
+    """
+    i = FakeInstance()
+    static_info = write_static_info(i)
+    cache = StaticConnectionInfoCache(i.uri(), static_info)
+    conn_info = cache._info
+    await cache.force_refresh()
+    conn_info2 = cache._info
+    assert conn_info2 == conn_info
+
+async def test_StaticConnectionInfoCache_connect_info() -> None:
+    """
+    Test that StaticConnectionInfoCache.connect_info works as expected.
+    """
+    i = FakeInstance()
+    static_info = write_static_info(i)
+    cache = StaticConnectionInfoCache(i.uri(), static_info)
+    # check that cached connection info is now set
+    assert isinstance(cache._info, ConnectionInfo)
+    conn_info = await cache.connect_info()
+    # check that calling connect_info uses cached info
+    conn_info2 = await cache.connect_info()
+    assert conn_info2 == conn_info
+
+async def test_StaticConnectionInfoCache_close() -> None:
+    """
+    Test that StaticConnectionInfoCache.close is a no-op.
+    """
+    i = FakeInstance()
+    static_info = write_static_info(i)
+    cache = StaticConnectionInfoCache(i.uri(), static_info)
+    conn_info = cache._info
+    await cache.close()
+    conn_info2 = cache._info
+    assert conn_info2 == conn_info