]> git.ipfire.org Git - thirdparty/knot-resolver.git/commitdiff
tests/pytests: ruff linter fixes
authorAleš Mrázek <ales.mrazek@nic.cz>
Tue, 15 Oct 2024 15:58:00 +0000 (17:58 +0200)
committerAleš Mrázek <ales.mrazek@nic.cz>
Thu, 17 Oct 2024 12:39:47 +0000 (14:39 +0200)
tests/pytests/kresd.py
tests/pytests/proxy.py
tests/pytests/utils.py

index 21d964555dfb1ef8753cb335dac9a7976ca0910a..6373e63aa82c67d1046f87dd7ba9c22678eac386 100644 (file)
@@ -48,7 +48,7 @@ Forward = namedtuple('Forward', ['proto', 'ip', 'port', 'hostname', 'ca_file'])
 
 
 class Kresd(ContextDecorator):
-    def __init__(
+    def __init__(  # noqa: PLR0913
             self, workdir, port=None, tls_port=None, ip=None, ip6=None, certname=None,
             verbose=True, hints=None, forward=None, policy_test_pass=False, rr=False,
             valgrind=False):
@@ -179,7 +179,7 @@ class Kresd(ContextDecorator):
         port = self.tls_port if tls else self.port
         if family == socket.AF_INET:
             return self.ip, port
-        elif family == socket.AF_INET6:
+        if family == socket.AF_INET6:
             return self.ip6, port, 0, 0
         raise RuntimeError("Unsupported socket family: {}".format(family))
 
@@ -237,13 +237,12 @@ class Kresd(ContextDecorator):
             for line in log:
                 if past_startup:
                     partial_log += line
-                else:  # find real start of test log (after initial alive-pings)
-                    if not past_startup_msgid:
-                        if re.match(KRESD_LOG_STARTUP_MSGID, line) is not None:
-                            past_startup_msgid = True
-                    else:
-                        if re.match(KRESD_LOG_IO_CLOSE, line) is not None:
-                            past_startup = True
+                # find real start of test log (after initial alive-pings)
+                elif not past_startup_msgid:
+                    if re.match(KRESD_LOG_STARTUP_MSGID, line) is not None:
+                        past_startup_msgid = True
+                elif re.match(KRESD_LOG_IO_CLOSE, line) is not None:
+                    past_startup = True
         return partial_log
 
 
@@ -263,8 +262,7 @@ def is_port_free(port, ip=None, ip6=None):
     except OSError as exc:
         if exc.errno == 98:  # address already in use
             return False
-        else:
-            raise
+        raise
     return True
 
 
index b8a53cd8051f1c1192cce9d5d14bd0ab07c57f5f..57bb6e961e0f0d2c6cffe18d99a7d75e4e0bd67c 100644 (file)
@@ -82,7 +82,7 @@ class Proxy(ContextDecorator):
 class TLSProxy(Proxy):
     EXECUTABLE = 'tlsproxy'
 
-    def __init__(
+    def __init__(  # noqa: PLR0913
                 self,
                 local_ip: str = '127.0.0.1',
                 local_port: Optional[int] = None,
@@ -130,9 +130,9 @@ def kresd_tls_client(
             kresd_fwd_target_kwargs: Optional[Dict[Any, Any]] = None
         ) -> Kresd:
     """kresd_tls_client --(tls)--> tlsproxy --(tcp)--> kresd_fwd_target"""
-    ALLOWED_IPS = {'127.0.0.1', '::1'}
-    assert proxy.local_ip in ALLOWED_IPS, "only localhost IPs supported for proxy"
-    assert proxy.upstream_ip in ALLOWED_IPS, "only localhost IPs are supported for proxy"
+    allowed_ips = {'127.0.0.1', '::1'}
+    assert proxy.local_ip in allowed_ips, "only localhost IPs supported for proxy"
+    assert proxy.upstream_ip in allowed_ips, "only localhost IPs are supported for proxy"
 
     if kresd_tls_client_kwargs is None:
         kresd_tls_client_kwargs = {}
index 8af71aadfb926b2798c7cc4ae8dca4878c43da6d..1290c86d6a4d21721c7d6ea512f9ae0463121794 100644 (file)
@@ -40,8 +40,7 @@ def receive_parse_answer(sock):
     if data_answer is None:
         raise BrokenPipeError("kresd closed connection")
 
-    msg_answer = dns.message.from_wire(data_answer, one_rr_per_rrset=True)
-    return msg_answer
+    return dns.message.from_wire(data_answer, one_rr_per_rrset=True)
 
 
 def prepare_wire(