More test fixes

salt/transport/base.py

@@ -128,6 +128,7 @@ def publish_client(opts, io_loop, host=None, port=None, path=None):
         ttype = opts["transport"]
     elif "transport" in opts.get("pillar", {}).get("master", {}):
         ttype = opts["pillar"]["master"]["transport"]
+
     # switch on available ttypes
     if ttype == "zeromq":
         import salt.transport.zeromq
@@ -147,6 +148,7 @@ def publish_client(opts, io_loop, host=None, port=None, path=None):
         return salt.transport.ws.PublishClient(
             opts, io_loop, host=host, port=port, path=path
         )
+
     raise Exception("Transport type not found: {}".format(ttype))
 
 

salt/transport/ipc.py

@@ -241,12 +241,8 @@ class IPCServer:
 
     # pylint: disable=W1701
     def __del__(self):
-        try:
-            self.close()
-        except TypeError:
-            # This is raised when Python's GC has collected objects which
-            # would be needed when calling self.close()
-            pass
+        if not self._closing:
+            log.warning("%r never closed")
 
     # pylint: enable=W1701
 
@@ -388,12 +384,8 @@ class IPCClient:
 
     # pylint: disable=W1701
     def __del__(self):
-        try:
-            self.close()
-        except TypeError:
-            # This is raised when Python's GC has collected objects which
-            # would be needed when calling self.close()
-            pass
+        if not self._closing:
+            log.warning("%r never closed", self)
 
     # pylint: enable=W1701
 

salt/transport/tcp.py

@@ -229,7 +229,6 @@ class TCPPubClient(salt.transport.base.PublishClient):
     def __init__(self, opts, io_loop, **kwargs):  # pylint: disable=W0231
         self.opts = opts
         self.io_loop = io_loop
-        self.message_client = None
         self.unpacker = salt.utils.msgpack.Unpacker()
         self.connected = False
         self._closing = False
@@ -260,9 +259,9 @@ class TCPPubClient(salt.transport.base.PublishClient):
         if self._closing:
             return
         self._closing = True
-        if self.message_client is not None:
-            self.message_client.close()
-            self.message_client = None
+        self._stream.close()
+        self._stream = None
+        self._closed = True
 
     # pylint: disable=W1701
     def __del__(self):
@@ -345,7 +344,7 @@ class TCPPubClient(salt.transport.base.PublishClient):
         return body
 
     async def send(self, msg):
-        await self.message_client.send(msg, reply=False)
+        await self._stream.send(msg)
 
     async def recv(self, timeout=None):
         try:

salt/transport/zeromq.py

@@ -1099,7 +1099,7 @@ class RequestClient(salt.transport.base.RequestClient):
         if self.socket:
             self.socket.close()
             self.socket = None
-        if self.context.closed is False:
+        if self.context and self.context.closed is False:
             # This hangs if closing the stream causes an import error
             self.context.term()
             self.context = None

tests/pytests/unit/channel/test_request_channel.py

@@ -330,7 +330,7 @@ def test_clear_req_channel_master_uri_override(temp_salt_minion, temp_salt_maste
         master_ip="localhost", master_port=opts["master_port"]
     )
     with salt.channel.client.ReqChannel.factory(opts, master_uri=master_uri) as channel:
-        assert "127.0.0.1" in channel.transport.message_client.addr
+        assert "127.0.0.1" in channel.transport.master_uri
 
 
 def run_loop_in_thread(loop, evt):