More test fixes

salt/__init__.py

@@ -2,11 +2,15 @@
 Salt package
 """
 
+import asyncio
 import importlib
 import os
 import sys
 import warnings
 
+if sys.platform.startswith("win"):
+    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+
 if sys.version_info < (3,):
     sys.stderr.write(
         "\n\nAfter the Sodium release, 3001, Salt no longer supports Python 2. Exiting.\n\n"
@@ -90,6 +94,9 @@ warnings.filterwarnings(
 def __define_global_system_encoding_variable__():
     import sys
 
+    print("define global system encoding")
+    sys.stdout.flush()
+
     # This is the most trustworthy source of the system encoding, though, if
     # salt is being imported after being daemonized, this information is lost
     # and reset to None

salt/loader/lazy.py

@@ -739,9 +739,9 @@ class LazyLoader(salt.utils.lazy.LazyDict):
                     # loading using exec_module has been causing odd things
                     # with the magic dunders we pack into the loaded
                     # modules, most notably with salt-ssh's __opts__.
-                    mod = spec.loader.load_module()
-                    # mod = importlib.util.module_from_spec(spec)
-                    # spec.loader.exec_module(mod)
+                    # mod = spec.loader.load_module()
+                    mod = importlib.util.module_from_spec(spec)
+                    spec.loader.exec_module(mod)
                     # pylint: enable=no-member
                     sys.modules[mod_namespace] = mod
                     # reload all submodules if necessary
@@ -760,9 +760,9 @@ class LazyLoader(salt.utils.lazy.LazyDict):
                     # loading using exec_module has been causing odd things
                     # with the magic dunders we pack into the loaded
                     # modules, most notably with salt-ssh's __opts__.
-                    mod = self.run(spec.loader.load_module)
-                    # mod = importlib.util.module_from_spec(spec)
-                    # spec.loader.exec_module(mod)
+                    # mod = self.run(spec.loader.load_module)
+                    mod = importlib.util.module_from_spec(spec)
+                    spec.loader.exec_module(mod)
                     # pylint: enable=no-member
                     sys.modules[mod_namespace] = mod
         except OSError:

salt/transport/zeromq.py

@@ -239,7 +239,8 @@ class PublishClient(salt.transport.base.PublishClient):
             for cb in self.callbacks:
                 running, task = self.callbacks[cb]
                 try:
-                    task.cancel()
+                    if task:
+                        task.cancel()
                 except RuntimeError:
                     log.warning("Tasks loop already closed")
 
@@ -643,6 +644,7 @@ class AsyncReqMessageClient:
 
         self._closing = False
         self.socket = None
+        self.sending = False
 
     async def connect(self):
         if self.socket is None:
@@ -708,11 +710,16 @@ class AsyncReqMessageClient:
         """
         Return a future which will be completed when the message has a response
         """
-
-        response = await asyncio.wait_for(self._send_recv(message), timeout=timeout)
-        if callback:
-            callback(response)
-        return response
+        while self.sending:
+            await asyncio.sleep(0.03)
+        self.sending = True
+        try:
+            response = await asyncio.wait_for(self._send_recv(message), timeout=timeout)
+            if callback:
+                callback(response)
+            return response
+        finally:
+            self.sending = False
 
 
 class ZeroMQSocketMonitor:

salt/utils/network.py

@@ -2246,12 +2246,12 @@ def _test_addrs(addrinfo, port):
             s = socket.socket(ip_family, socket.SOCK_STREAM)
             s.settimeout(2)
             s.connect((ip_addr, port))
-            s.close()
-
             ip_addrs = [ip_addr]
             break
         except OSError:
             pass
+        finally:
+            s.close()
     return ip_addrs
 
 

tests/pytests/conftest.py

@@ -610,12 +610,15 @@ def pytest_pyfunc_call(pyfuncitem):
     # loop.run_sync(
     #    CoroTestFunction(pyfuncitem.obj, testargs), timeout=get_test_timeout(pyfuncitem)
     # )
+    # try:
     loop.asyncio_loop.run_until_complete(
         asyncio.wait_for(
             CoroTestFunction(pyfuncitem.obj, testargs)(),
             timeout=get_test_timeout(pyfuncitem),
         )
     )
+    # except RuntimeError as exc:
+    #    log.warning("WTFSON %r", dir(exc))
     return True
 
 
@@ -624,13 +627,13 @@ def io_loop():
     """
     Create new io loop for each test, and tear it down after.
     """
-    loop = asyncio.new_event_loop()
-    asyncio.set_event_loop(loop)
+    asyncio_loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(asyncio_loop)
     loop = tornado.ioloop.IOLoop.current()
     try:
         yield loop
     finally:
-        loop.close(all_fds=True)
+        loop.close(all_fds=True)  # Also closes asyncio_loop
         asyncio.set_event_loop(None)
 
 

tests/pytests/unit/test_minion.py

@@ -28,8 +28,11 @@ def test_minion_load_grains_false():
     opts = {"random_startup_delay": 0, "grains": {"foo": "bar"}}
     with patch("salt.loader.grains") as grainsfunc:
         minion = salt.minion.Minion(opts, load_grains=False)
-        assert minion.opts["grains"] == opts["grains"]
-        grainsfunc.assert_not_called()
+        try:
+            assert minion.opts["grains"] == opts["grains"]
+            grainsfunc.assert_not_called()
+        finally:
+            minion.destroy()
 
 
 def test_minion_load_grains_true():
@@ -39,8 +42,11 @@ def test_minion_load_grains_true():
     opts = {"random_startup_delay": 0, "grains": {}}
     with patch("salt.loader.grains") as grainsfunc:
         minion = salt.minion.Minion(opts, load_grains=True)
-        assert minion.opts["grains"] != {}
-        grainsfunc.assert_called()
+        try:
+            assert minion.opts["grains"] != {}
+            grainsfunc.assert_called()
+        finally:
+            minion.destroy()
 
 
 def test_minion_load_grains_default():
@@ -50,8 +56,11 @@ def test_minion_load_grains_default():
     opts = {"random_startup_delay": 0, "grains": {}}
     with patch("salt.loader.grains") as grainsfunc:
         minion = salt.minion.Minion(opts)
-        assert minion.opts["grains"] != {}
-        grainsfunc.assert_called()
+        try:
+            assert minion.opts["grains"] != {}
+            grainsfunc.assert_called()
+        finally:
+            minion.destroy()
 
 
 @pytest.mark.parametrize(
@@ -80,40 +89,47 @@ def test_send_req_fires_completion_event(event, minion_opts):
         with patch("salt.loader.grains"):
             minion = salt.minion.Minion(minion_opts)
 
-            load = {"load": "value"}
-            timeout = 60
+            try:
+                load = {"load": "value"}
+                timeout = 60
 
-            # XXX This is buggy because "async" in event[0] will never evaluate
-            # to True and if it *did* evaluate to true the test would fail
-            # because you Mock isn't a co-routine.
-            if "async" in event[0]:
-                rtn = minion._send_req_async(load, timeout).result()
-            else:
-                rtn = minion._send_req_sync(load, timeout)
+                # XXX This is buggy because "async" in event[0] will never evaluate
+                # to True and if it *did* evaluate to true the test would fail
+                # because you Mock isn't a co-routine.
+                if "async" in event[0]:
+                    rtn = minion._send_req_async(load, timeout).result()
+                else:
+                    rtn = minion._send_req_sync(load, timeout)
 
-            # get the
-            for idx, call in enumerate(event.mock_calls, 1):
-                if "fire_event" in call[0]:
-                    condition_event_tag = (
-                        len(call.args) > 1
-                        and call.args[1] == "__master_req_channel_payload"
-                    )
-                    condition_event_tag_error = "{} != {}; Call(number={}): {}".format(
-                        idx, call, call.args[1], "__master_req_channel_payload"
-                    )
-                    condition_timeout = (
-                        len(call.kwargs) == 1 and call.kwargs["timeout"] == timeout
-                    )
-                    condition_timeout_error = "{} != {}; Call(number={}): {}".format(
-                        idx, call, call.kwargs["timeout"], timeout
-                    )
+                # get the
+                for idx, call in enumerate(event.mock_calls, 1):
+                    if "fire_event" in call[0]:
+                        condition_event_tag = (
+                            len(call.args) > 1
+                            and call.args[1] == "__master_req_channel_payload"
+                        )
+                        condition_event_tag_error = (
+                            "{} != {}; Call(number={}): {}".format(
+                                idx, call, call.args[1], "__master_req_channel_payload"
+                            )
+                        )
+                        condition_timeout = (
+                            len(call.kwargs) == 1 and call.kwargs["timeout"] == timeout
+                        )
+                        condition_timeout_error = (
+                            "{} != {}; Call(number={}): {}".format(
+                                idx, call, call.kwargs["timeout"], timeout
+                            )
+                        )
 
-                    fire_event_called = True
-                    assert condition_event_tag, condition_event_tag_error
-                    assert condition_timeout, condition_timeout_error
+                        fire_event_called = True
+                        assert condition_event_tag, condition_event_tag_error
+                        assert condition_timeout, condition_timeout_error
 
-            assert fire_event_called
-            assert rtn
+                assert fire_event_called
+                assert rtn
+            finally:
+                minion.destroy()
 
 
 async def test_send_req_async_regression_62453(minion_opts):
@@ -956,6 +972,8 @@ async def test_master_type_failover_no_masters(minion_opts):
     ):
         with pytest.raises(SaltClientError):
             minion = salt.minion.Minion(minion_opts)
+            # Mock the io_loop so calls to stop/close won't happen.
+            minion.io_loop = MagicMock()
             await minion.connect_master()
 
 
@@ -1075,7 +1093,7 @@ async def test_master_type_disable(minion_opts):
 
         try:
             minion_man = salt.minion.MinionManager(minion_opts)
-            minion_man._connect_minion(minion)
+            await minion_man._connect_minion(minion)
         except RuntimeError:
             pytest.fail("_connect_minion(minion) threw an error, This was not expected")
 
@@ -1084,4 +1102,6 @@ async def test_master_type_disable(minion_opts):
         assert "schedule" in minion.periodic_callbacks
         assert minion.connected is False
     finally:
+        # Mock the io_loop so calls to stop/close won't happen.
+        minion.io_loop = MagicMock()
         minion.destroy()

tests/pytests/unit/transport/test_tcp.py

@@ -29,7 +29,7 @@ def fake_crypto():
 
 
 @pytest.fixture
-def fake_authd():
+def fake_authd(io_loop):
     @tornado.gen.coroutine
     def return_nothing():
         raise tornado.gen.Return()