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

diff --git a/dns/_asyncbackend.py b/dns/_asyncbackend.py
index 093713d1baf39b340ff4ae7267d445ad92f11b7e..705ce180c2790c9e565a7a759c8b13d421bb1fcd 100644 (file)
--- 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 699276d344bf0827d340b1e41e87561f73cf3e99..836273b382fa938ab4010a94ab28c4c5f11bb082 100644 (file)
--- 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 049151117532923bc33713189863acf45d394b4a..418639cb7786556ecd8e0f45ae6c4f43453133df 100644 (file)
--- 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 1c9a102462a19cf86999966eaf7205952a8b92ce..26f239769d2d7e919cf41240b6dd5984adbe1576 100644 (file)
--- 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 03e9fadafdc9ce4cbe89c121b482a19b550c4062..38141feb8d383eec0c2df293a9fd3867283d2f5a 100644 (file)
--- 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 f5b1fcb3eea65207fa6033148254d9727a3d6985..7960e9f7dbc92ef8c1241c7e5dfcd810f8eee622 100644 (file)
--- 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}')