Method test_with_uid

t/unit/utils/test_platforms.py:220–279

(self, initgroups, setuid, setgid,
                      getpwuid, parse_gid, parse_uid, getuid, geteuid,
                      getgid, getegid)

    @patch('celery.platforms.setuid')
    @patch('celery.platforms.initgroups')
    def test_with_uid(self, initgroups, setuid, setgid,
                      getpwuid, parse_gid, parse_uid, getuid, geteuid,
                      getgid, getegid):
        geteuid.return_value = 10
        getuid.return_value = 10

        class pw_struct:
            pw_gid = 50001

        def raise_on_second_call(*args, **kwargs):
            setuid.side_effect = OSError()
            setuid.side_effect.errno = errno.EPERM

        setuid.side_effect = raise_on_second_call
        getpwuid.return_value = pw_struct()
        parse_uid.return_value = 5001
        parse_gid.return_value = 5001
        maybe_drop_privileges(uid='user')
        parse_uid.assert_called_with('user')
        getpwuid.assert_called_with(5001)
        setgid.assert_called_with(50001)
        initgroups.assert_called_with(5001, 50001)
        setuid.assert_has_calls([call(5001), call(0)])

        setuid.side_effect = raise_on_second_call

        def to_root_on_second_call(mock, first):
            return_value = [first]

            def on_first_call(*args, **kwargs):
                ret, return_value[0] = return_value[0], 0
                return ret

            mock.side_effect = on_first_call

        to_root_on_second_call(geteuid, 10)
        to_root_on_second_call(getuid, 10)
        with pytest.raises(SecurityError):
            maybe_drop_privileges(uid='user')

        getuid.return_value = getuid.side_effect = None
        geteuid.return_value = geteuid.side_effect = None
        getegid.return_value = 0
        getgid.return_value = 0
        setuid.side_effect = raise_on_second_call
        with pytest.raises(SecurityError):
            maybe_drop_privileges(gid='group')

        getuid.reset_mock()
        geteuid.reset_mock()
        setuid.reset_mock()
        getuid.side_effect = geteuid.side_effect = None

        def raise_on_second_call(*args, **kwargs):
            setuid.side_effect = OSError()
            setuid.side_effect.errno = errno.ENOENT

        setuid.side_effect = raise_on_second_call