Add tests for closing poller in b263dde

Author: mnaberez

supervisor/tests/base.py

@@ -1169,6 +1169,7 @@ def __str__(self):
 class DummyPoller:
     def __init__(self, options):
         self.result = [], []
+        self.closed = False
 
     def register_readable(self, fd):
         pass
@@ -1179,6 +1180,9 @@ def register_writable(self, fd):
     def poll(self, timeout):
         return self.result
 
+    def close(self):
+        self.closed = True
+
 def dummy_handler(event, result):
     pass
 

supervisor/tests/test_options.py

@@ -19,6 +19,7 @@
 from supervisor.tests.base import DummySupervisor
 from supervisor.tests.base import DummyLogger
 from supervisor.tests.base import DummyOptions
+from supervisor.tests.base import DummyPoller
 from supervisor.tests.base import DummyPConfig
 from supervisor.tests.base import DummyProcess
 from supervisor.tests.base import DummySocketConfig
@@ -1371,6 +1372,25 @@ def test_cleanup_pidfile_ignores_oserror_enoent(self):
         instance.pidfile = notfound
         instance.cleanup() # shouldn't raise
 
+    def test_cleanup_closes_poller(self):
+        pidfile = tempfile.mktemp()
+        try:
+            with open(pidfile, 'w') as f:
+                f.write('2')
+            instance = self._makeOne()
+            instance.pidfile = pidfile
+
+            poller = DummyPoller({})
+            instance.poller = poller
+            self.assertFalse(poller.closed)
+            instance.cleanup()
+            self.assertTrue(poller.closed)
+        finally:
+            try:
+                os.unlink(pidfile)
+            except OSError:
+                pass
+
     def test_cleanup_fds_closes_5_upto_minfds_ignores_oserror(self):
         instance = self._makeOne()
         instance.minfds = 10

supervisor/tests/test_poller.py

@@ -223,6 +223,15 @@ def test_after_daemonize_restores_kqueue(self):
         poller.register_readable.assert_called_with(1)
         poller.register_writable.assert_called_with(3)
 
+    def test_close_closes_kqueue(self):
+        mock_kqueue = Mock()
+        options = DummyOptions()
+        poller = self._makeOne(options)
+        poller._kqueue = mock_kqueue
+        poller.close()
+        mock_kqueue.close.assert_called_once_with()
+        self.assertEqual(poller._kqueue, None)
+
     def assertReadEventAdded(self, kevent, fd):
         self.assertKevent(kevent, fd, select.KQ_FILTER_READ, select.KQ_EV_ADD)