Miscellaneous flake8-bugbear issues (#6814)

distributed/client.py

@@ -836,8 +836,8 @@ def __init__(
         elif isinstance(getattr(address, "scheduler_address", None), str):
             # It's a LocalCluster or LocalCluster-compatible object
             self.cluster = address
-            status = getattr(self.cluster, "status")
-            if status and status in [Status.closed, Status.closing]:
+            status = self.cluster.status
+            if status in (Status.closed, Status.closing):
                 raise RuntimeError(
                     f"Trying to connect to an already closed or closing Cluster {self.cluster}."
                 )

distributed/comm/addressing.py

@@ -272,7 +272,10 @@ def address_from_user_args(  # type: ignore[no-untyped-def]
     if security and security.require_encryption and not protocol:
         protocol = "tls"
 
-    if protocol and protocol.rstrip("://") == "inplace":
+    if protocol and protocol.endswith("://"):
+        protocol, _, _ = protocol.rpartition("://")
+
+    if protocol == "inplace":
         if host or port or interface:
             raise ValueError(
                 "Can not specify inproc protocol and host or port or interface"
@@ -287,14 +290,14 @@ def address_from_user_args(  # type: ignore[no-untyped-def]
             host = get_ip_interface(interface)
 
     if protocol and host and "://" not in host:
-        host = protocol.rstrip("://") + "://" + host
+        host = protocol + "://" + host
 
     if host or port:
         addr = uri_from_host_port(host, port, default_port)
     else:
         addr = ""
 
     if protocol:
-        addr = protocol.rstrip("://") + "://" + addr.split("://")[-1]
+        addr = protocol + "://" + addr.split("://")[-1]
 
     return addr

distributed/tests/test_client.py

@@ -3866,8 +3866,8 @@ def test_get_versions_sync(c):
     assert v["scheduler"] is not None
     assert v["client"] is not None
     assert len(v["workers"]) == 2
-    for v in v["workers"].values():
-        assert v is not None
+    for wv in v["workers"].values():
+        assert wv is not None
 
     c.get_versions(check=True)
     # smoke test for versions

distributed/utils.py

@@ -1227,7 +1227,7 @@ def command_has_keyword(cmd, k):
             except ImportError:
                 raise ImportError("Module for command %s is not available" % cmd)
 
-        if isinstance(getattr(cmd, "main"), click.core.Command):
+        if isinstance(cmd.main, click.core.Command):
             cmd = cmd.main
         if isinstance(cmd, click.core.Command):
             cmd_params = {

distributed/worker.py

@@ -212,7 +212,7 @@ async def _force_close(self):
         await asyncio.wait_for(self.close(nanny=False, executor_wait=False), 30)
     except (KeyboardInterrupt, SystemExit):  # pragma: nocover
         raise
-    except (Exception, BaseException):  # pragma: nocover
+    except BaseException:  # pragma: nocover
         # Worker is in a very broken state if closing fails. We need to shut down
         # immediately, to ensure things don't get even worse and this worker potentially
         # deadlocks the cluster.