chore: remove need dns fixture, use mocks

Author: jackwotherspoon

requirements-test.txt

@@ -10,5 +10,4 @@ pg8000==1.31.2
 asyncpg==0.30.0
 python-tds==1.16.0
 aioresponses==0.7.7
-pytest-aiohttp==1.0.5
-dnserver==0.4.0
+pytest-aiohttp==1.0.5

tests/conftest.py

@@ -21,7 +21,6 @@
 from typing import Any, AsyncGenerator, Generator
 
 from aiohttp import web
-from dnserver import DNSServer
 import pytest  # noqa F401 Needed to run the tests
 from unit.mocks import FakeCredentials  # type: ignore
 from unit.mocks import FakeCSQLInstance  # type: ignore
@@ -152,13 +151,3 @@ async def cache(fake_client: CloudSQLClient) -> AsyncGenerator[RefreshAheadCache
     )
     yield cache
     await cache.close()
-
-
-@pytest.fixture
-def dns_server() -> Generator:
-    """Setup local DNS server for tests with TXT records."""
-    server = DNSServer.from_toml("tests/test_zones.toml", port=5053, upstream=None)
-    server.start()
-    assert server.is_running
-    yield server
-    server.stop()

tests/test_zones.toml

@@ -12,15 +12,20 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import dns.message
+import dns.rdataclass
+import dns.rdatatype
+import dns.resolver
+from mock import patch
 import pytest
 
 from google.cloud.sql.connector.connection_name import ConnectionName
 from google.cloud.sql.connector.exceptions import DnsResolutionError
 from google.cloud.sql.connector.resolver import DefaultResolver
 from google.cloud.sql.connector.resolver import DnsResolver
 
-conn_str = "test-project:test-region:test-instance"
-conn_name = ConnectionName("test-project", "test-region", "test-instance")
+conn_str = "my-project:my-region:my-instance"
+conn_name = ConnectionName("my-project", "my-region", "my-instance")
 
 
 async def test_DefaultResolver() -> None:
@@ -37,39 +42,87 @@ async def test_DnsResolver_with_conn_str() -> None:
     assert result == conn_name
 
 
-@pytest.mark.usefixtures("dns_server")
+query_text = """id 1234
+opcode QUERY
+rcode NOERROR
+flags QR AA RD RA
+;QUESTION
+db.example.com. IN TXT
+;ANSWER
+db.example.com. 0 IN TXT "test-project:test-region:test-instance"
+db.example.com. 0 IN TXT "my-project:my-region:my-instance"
+;AUTHORITY
+;ADDITIONAL
+"""
+
+
 async def test_DnsResolver_with_dns_name() -> None:
-    """Test DnsResolver resolves TXT record into proper instance connection name."""
-    resolver = DnsResolver()
-    resolver.port = 5053
-    result = await resolver.resolve("db.example.com")
-    assert result == conn_name
+    """Test DnsResolver resolves TXT record into proper instance connection name.
+
+    Should sort valid TXT records alphabetically and take first one.
+    """
+    # Patch DNS resolution with valid TXT records
+    with patch("dns.asyncresolver.Resolver.resolve") as mock_connect:
+        answer = dns.resolver.Answer(
+            "db.example.com",
+            dns.rdatatype.TXT,
+            dns.rdataclass.IN,
+            dns.message.from_text(query_text),
+        )
+        mock_connect.return_value = answer
+        resolver = DnsResolver()
+        resolver.port = 5053
+        # Resolution should return first value sorted alphabetically
+        result = await resolver.resolve("db.example.com")
+        assert result == conn_name
+
+
+query_text_malformed = """id 1234
+opcode QUERY
+rcode NOERROR
+flags QR AA RD RA
+;QUESTION
+bad.example.com. IN TXT
+;ANSWER
+bad.example.com. 0 IN TXT "malformed-instance-name"
+;AUTHORITY
+;ADDITIONAL
+"""
 
 
-@pytest.mark.usefixtures("dns_server")
 async def test_DnsResolver_with_malformed_txt() -> None:
     """Test DnsResolver with TXT record that holds malformed instance connection name.
 
     Should throw DnsResolutionError
     """
-    resolver = DnsResolver()
-    resolver.port = 5053
-    with pytest.raises(DnsResolutionError) as exc_info:
-        await resolver.resolve("bad.example.com")
-        assert (
-            exc_info.value.args[0]
-            == "Unable to parse TXT record for `bad.example.com` -> `bad-instance-name`"
+    # patch DNS resolution with malformed TXT record
+    with patch("dns.asyncresolver.Resolver.resolve") as mock_connect:
+        answer = dns.resolver.Answer(
+            "bad.example.com",
+            dns.rdatatype.TXT,
+            dns.rdataclass.IN,
+            dns.message.from_text(query_text_malformed),
         )
+        mock_connect.return_value = answer
+        resolver = DnsResolver()
+        resolver.port = 5053
+        with pytest.raises(DnsResolutionError) as exc_info:
+            await resolver.resolve("bad.example.com")
+            assert (
+                exc_info.value.args[0]
+                == "Unable to parse TXT record for `bad.example.com` -> `malformed-instance-name`"
+            )
 
 
-@pytest.mark.usefixtures("dns_server")
 async def test_DnsResolver_with_bad_dns_name() -> None:
     """Test DnsResolver with bad dns name.
 
     Should throw DnsResolutionError
     """
     resolver = DnsResolver()
     resolver.port = 5053
+    # set lifetime to 1 second for shorter timeout
+    resolver.lifetime = 1
     with pytest.raises(DnsResolutionError) as exc_info:
         await resolver.resolve("bad.dns.com")
     assert exc_info.value.args[0] == "Unable to resolve TXT record for `bad.dns.com`"