move low_level_address_tuple() to dns.inet; add some no-coverage comments

6 files changed, 26 insertions, 22 deletions

diff --git a/dns/_asyncbackend.py b/dns/_asyncbackend.py
index 093713d..705ce18 100644
--- a/dns/_asyncbackend.py
+++ b/dns/_asyncbackend.py
@@ -25,24 +25,10 @@ class NullContext:
         pass
 
 
-# This is handy, but should probably move somewhere else!
-
-def low_level_address_tuple(af, high_level_address_tuple):
-    address, port = high_level_address_tuple
-    if af == dns.inet.AF_INET:
-        return (address, port)
-    elif af == dns.inet.AF_INET6:
-        ai_flags = socket.AI_NUMERICHOST
-        ((*_, tup), *_) = socket.getaddrinfo(address, port, flags=ai_flags)
-        return tup
-    else:
-        raise NotImplementedError(f'unknown address family {af}')
-
-
 # These are declared here so backends can import them without creating
 # circular dependencies with dns.asyncbackend.
 
-class Socket:
+class Socket:  # pragma: no cover
     async def close(self):
         pass
 
@@ -53,7 +39,7 @@ class Socket:
         await self.close()
 
 
-class DatagramSocket(Socket):
+class DatagramSocket(Socket):  # pragma: no cover
     async def sendto(self, what, destination, timeout):
         pass
 
@@ -61,7 +47,7 @@ class DatagramSocket(Socket):
         pass
 
 
-class StreamSocket(Socket):
+class StreamSocket(Socket):  # pragma: no cover
     async def sendall(self, what, destination, timeout):
         pass
 
@@ -69,7 +55,7 @@ class StreamSocket(Socket):
         pass
 
 
-class Backend:
+class Backend:    # pragma: no cover
     def name(self):
         return 'unknown'
 
diff --git a/dns/_curio_backend.py b/dns/_curio_backend.py
index 699276d..836273b 100644
--- a/dns/_curio_backend.py
+++ b/dns/_curio_backend.py
@@ -8,6 +8,7 @@ import curio.socket  # type: ignore
 
 import dns._asyncbackend
 import dns.exception
+import dns.inet
 
 
 def _maybe_timeout(timeout):
@@ -18,7 +19,7 @@ def _maybe_timeout(timeout):
 
 
 # for brevity
-_lltuple = dns._asyncbackend.low_level_address_tuple
+_lltuple = dns.inet.low_level_address_tuple
 
 
 class DatagramSocket(dns._asyncbackend.DatagramSocket):
diff --git a/dns/_trio_backend.py b/dns/_trio_backend.py
index 0491511..418639c 100644
--- a/dns/_trio_backend.py
+++ b/dns/_trio_backend.py
@@ -8,6 +8,7 @@ import trio.socket  # type: ignore
 
 import dns._asyncbackend
 import dns.exception
+import dns.inet
 
 
 def _maybe_timeout(timeout):
@@ -18,7 +19,7 @@ def _maybe_timeout(timeout):
 
 
 # for brevity
-_lltuple = dns._asyncbackend.low_level_address_tuple
+_lltuple = dns.inet.low_level_address_tuple
 
 
 class DatagramSocket(dns._asyncbackend.DatagramSocket):
diff --git a/dns/asyncbackend.py b/dns/asyncbackend.py
index 1c9a102..26f2397 100644
--- a/dns/asyncbackend.py
+++ b/dns/asyncbackend.py
@@ -3,7 +3,7 @@
 import dns.exception
 
 from dns._asyncbackend import Socket, DatagramSocket, \
-    StreamSocket, Backend, low_level_address_tuple  # noqa:
+    StreamSocket, Backend  # noqa:
 
 
 _default_backend = None
diff --git a/dns/asyncquery.py b/dns/asyncquery.py
index 03e9fad..38141fe 100644
--- a/dns/asyncquery.py
+++ b/dns/asyncquery.py
@@ -35,7 +35,7 @@ from dns.query import _addresses_equal, _compute_times, UnexpectedSource, \
 
 
 # for brevity
-_lltuple = dns.asyncbackend.low_level_address_tuple
+_lltuple = dns.inet.low_level_address_tuple
 
 
 def _source_tuple(af, address, port):
diff --git a/dns/inet.py b/dns/inet.py
index f5b1fcb..7960e9f 100644
--- a/dns/inet.py
+++ b/dns/inet.py
@@ -139,3 +139,19 @@ def is_address(text):
             return True
         except Exception:
             return False
+
+
+def low_level_address_tuple(af, high_tuple):
+    """Given an address family and a "high-level" address tuple, i.e.
+    an (address, port) return the appropriate "low-level" address tuple
+    suitable for use in socket calls.
+    """
+    address, port = high_tuple
+    if af == dns.inet.AF_INET:
+        return (address, port)
+    elif af == dns.inet.AF_INET6:
+        ai_flags = socket.AI_NUMERICHOST
+        ((*_, tup), *_) = socket.getaddrinfo(address, port, flags=ai_flags)
+        return tup
+    else:
+        raise NotImplementedError(f'unknown address family {af}')