98% coverage for celery.platforms

celery/platforms.py

@@ -52,13 +52,15 @@ Seems we're already running? (pid: {1})"""
 try:
     from io import UnsupportedOperation
     FILENO_ERRORS = (AttributeError, UnsupportedOperation)
-except ImportError:  # Py2
+except ImportError:  # pragma: no cover
+    # Py2
     FILENO_ERRORS = (AttributeError, )  # noqa
 
 try:
     from io import UnsupportedOperation
     FILENO_ERRORS = (AttributeError, UnsupportedOperation)
-except ImportError:  # Py2
+except ImportError:  # pragma: no cover
+    # Py2
     FILENO_ERRORS = (AttributeError, )  # noqa
 
 
@@ -522,9 +524,8 @@ def maybe_drop_privileges(uid=None, gid=None):
                 raise
             pass  # Good: cannot restore privileges.
         else:
-            if uid:
-                raise RuntimeError(
-                    'non-root user able to restore privileges after setuid.')
+            raise RuntimeError(
+                'non-root user able to restore privileges after setuid.')
 
     else:
         gid and setgid(gid)
@@ -574,14 +575,14 @@ class Signals(object):
 
         def arm_alarm(self, seconds):
             _signal.setitimer(_signal.ITIMER_REAL, seconds)
-    else:
+    else:  # pragma: no cover
         try:
             from itimer import alarm as _itimer_alarm  # noqa
         except ImportError:
 
             def arm_alarm(self, seconds):  # noqa
                 _signal.alarm(math.ceil(seconds))
-        else:
+        else:  # pragma: no cover
 
             def arm_alarm(self, seconds):      # noqa
                 return _itimer_alarm(seconds)  # noqa

celery/tests/utilities/test_platforms.py

@@ -26,12 +26,40 @@ from celery.platforms import (
     Pidfile,
     LockFailed,
     setgroups,
-    _setgroups_hack
+    _setgroups_hack,
+    _find_option_with_arg,
+    close_open_fds,
 )
 
 from celery.tests.utils import Case, WhateverIO, override_stdouts, mock_open
 
 
+class test_find_option_with_arg(Case):
+
+    def test_long_opt(self):
+        self.assertEqual(
+            _find_option_with_arg(['--foo=bar'], long_opts=['--foo']),
+            'bar'
+        )
+
+    def test_short_opt(self):
+        self.assertEqual(
+            _find_option_with_arg(['-f', 'bar'], short_opts=['-f']),
+            'bar'
+        )
+
+class test_close_open_fds(Case):
+
+    def test_closes(self):
+        with patch('os.close') as _close:
+            with patch('celery.platforms.get_fdmax') as fdmax:
+                fdmax.return_value = 3
+                close_open_fds()
+                _close.assert_has_calls([call(2), call(1), call(0)])
+                _close.side_effect = OSError()
+                _close.side_effect.errno = errno.EBADF
+                close_open_fds()
+
 class test_ignore_errno(Case):
 
     def test_raises_EBADF(self):
@@ -69,6 +97,17 @@ class test_Signals(Case):
         self.assertTrue(signals.supported('INT'))
         self.assertFalse(signals.supported('SIGIMAGINARY'))
 
+    def test_reset_alarm(self):
+        with patch('signal.alarm') as _alarm:
+            signals.reset_alarm()
+            _alarm.assert_called_with(0)
+
+    def test_arm_alarm(self):
+        if hasattr(signal, 'setitimer'):
+            with patch('signal.setitimer', create=True) as seti:
+                signals.arm_alarm(30)
+                self.assertTrue(seti.called)
+
     def test_signum(self):
         self.assertEqual(signals.signum(13), 13)
         self.assertEqual(signals.signum('INT'), signal.SIGINT)
@@ -346,6 +385,11 @@ if not platforms.IS_WINDOWS:
                 pass
             self.assertFalse(x._detach.called)
 
+            x.after_chdir = Mock()
+            with x:
+                pass
+            x.after_chdir.assert_called_with()
+
     class test_Pidfile(Case):
 
         @patch('celery.platforms.Pidfile')