import errno
import os
import re
import signal
import sys
import tempfile
from unittest.mock import Mock, call, patch

import pytest

import t.skip
from celery import _find_option_with_arg, platforms
from celery.exceptions import SecurityError, SecurityWarning
from celery.platforms import (ASSUMING_ROOT, ROOT_DISALLOWED, ROOT_DISCOURAGED, DaemonContext, LockFailed, Pidfile,
                              _setgroups_hack, check_privileges, close_open_fds, create_pidlock, detached,
                              fd_by_path, get_fdmax, ignore_errno, initgroups, isatty, maybe_drop_privileges,
                              parse_gid, parse_uid, set_mp_process_title, set_pdeathsig, set_process_title, setgid,
                              setgroups, setuid, signals)
from celery.utils.text import WhateverIO
from t.unit import conftest

try:
    import resource
except ImportError:
    resource = None


def test_isatty():
    fh = Mock(name='fh')
    assert isatty(fh) is fh.isatty()
    fh.isatty.side_effect = AttributeError()
    assert not isatty(fh)


class test_find_option_with_arg:

    def test_long_opt(self):
        assert _find_option_with_arg(
            ['--foo=bar'], long_opts=['--foo']) == 'bar'

    def test_short_opt(self):
        assert _find_option_with_arg(
            ['-f', 'bar'], short_opts=['-f']) == 'bar'


@t.skip.if_win32
def test_fd_by_path():
    test_file = tempfile.NamedTemporaryFile()
    try:
        keep = fd_by_path([test_file.name])
        assert keep == [test_file.file.fileno()]
        with patch('os.open') as _open:
            _open.side_effect = OSError()
            assert not fd_by_path([test_file.name])
    finally:
        test_file.close()


def test_close_open_fds(patching):
    _close = patching('os.close')
    fdmax = patching('billiard.compat.get_fdmax')
    with patch('os.closerange', create=True) as closerange:
        fdmax.return_value = 3
        close_open_fds()
        if not closerange.called:
            _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:

    def test_raises_EBADF(self):
        with ignore_errno('EBADF'):
            exc = OSError()
            exc.errno = errno.EBADF
            raise exc

    def test_otherwise(self):
        with pytest.raises(OSError):
            with ignore_errno('EBADF'):
                exc = OSError()
                exc.errno = errno.ENOENT
                raise exc


class test_set_process_title:

    def test_no_setps(self):
        prev, platforms._setproctitle = platforms._setproctitle, None
        try:
            set_process_title('foo')
        finally:
            platforms._setproctitle = prev

    @patch('celery.platforms.set_process_title')
    @patch('celery.platforms.current_process')
    def test_mp_no_hostname(self, current_process, set_process_title):
        current_process().name = 'Foo'
        set_mp_process_title('foo', info='hello')
        set_process_title.assert_called_with('foo:Foo', info='hello')

    @patch('celery.platforms.set_process_title')
    @patch('celery.platforms.current_process')
    def test_mp_hostname(self, current_process, set_process_title):
        current_process().name = 'Foo'
        set_mp_process_title('foo', hostname='a@q.com', info='hello')
        set_process_title.assert_called_with('foo: a@q.com:Foo', info='hello')


class test_Signals:

    @patch('signal.getsignal')
    def test_getitem(self, getsignal):
        signals['SIGINT']
        getsignal.assert_called_with(signal.SIGINT)

    def test_supported(self):
        assert signals.supported('INT')
        assert not signals.supported('SIGIMAGINARY')

    @t.skip.if_win32
    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)
                seti.assert_called()

    def test_signum(self):
        assert signals.signum(13) == 13
        assert signals.signum('INT') == signal.SIGINT
        assert signals.signum('SIGINT') == signal.SIGINT
        with pytest.raises(TypeError):
            signals.signum('int')
            signals.signum(object())

    @patch('signal.signal')
    def test_ignore(self, set):
        signals.ignore('SIGINT')
        set.assert_called_with(signals.signum('INT'), signals.ignored)
        signals.ignore('SIGTERM')
        set.assert_called_with(signals.signum('TERM'), signals.ignored)

    @patch('signal.signal')
    def test_reset(self, set):
        signals.reset('SIGINT')
        set.assert_called_with(signals.signum('INT'), signals.default)

    @patch('signal.signal')
    def test_setitem(self, set):
        def handle(*args):
            return args

        signals['INT'] = handle
        set.assert_called_with(signal.SIGINT, handle)

    @patch('signal.signal')
    def test_setitem_raises(self, set):
        set.side_effect = ValueError()
        signals['INT'] = lambda *a: a


class test_set_pdeathsig:

    def test_call(self):
        set_pdeathsig('SIGKILL')

    @t.skip.if_win32
    def test_call_with_correct_parameter(self):
        with patch('celery.platforms._set_pdeathsig') as _set_pdeathsig:
            set_pdeathsig('SIGKILL')
            _set_pdeathsig.assert_called_once_with(signal.SIGKILL)


@t.skip.if_win32
class test_get_fdmax:

    @patch('resource.getrlimit')
    def test_when_infinity(self, getrlimit):
        with patch('os.sysconf') as sysconfig:
            sysconfig.side_effect = KeyError()
            getrlimit.return_value = [None, resource.RLIM_INFINITY]
            default = object()
            assert get_fdmax(default) is default

    @patch('resource.getrlimit')
    def test_when_actual(self, getrlimit):
        with patch('os.sysconf') as sysconfig:
            sysconfig.side_effect = KeyError()
            getrlimit.return_value = [None, 13]
            assert get_fdmax(None) == 13


@t.skip.if_win32
class test_maybe_drop_privileges:

    def test_on_windows(self):
        prev, sys.platform = sys.platform, 'win32'
        try:
            maybe_drop_privileges()
        finally:
            sys.platform = prev

    @patch('os.getegid')
    @patch('os.getgid')
    @patch('os.geteuid')
    @patch('os.getuid')
    @patch('celery.platforms.parse_uid')
    @patch('celery.platforms.parse_gid')
    @patch('pwd.getpwuid')
    @patch('celery.platforms.setgid')
    @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
        with pytest.raises(OSError):
            maybe_drop_privileges(uid='user')

    @patch('celery.platforms.parse_uid')
    @patch('celery.platforms.parse_gid')
    @patch('celery.platforms.setgid')
    @patch('celery.platforms.setuid')
    @patch('celery.platforms.initgroups')
    def test_with_guid(self, initgroups, setuid, setgid,
                       parse_gid, parse_uid):

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

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

        setuid.side_effect = None
        with pytest.raises(SecurityError):
            maybe_drop_privileges(uid='user', gid='group')
        setuid.side_effect = OSError()
        setuid.side_effect.errno = errno.EINVAL
        with pytest.raises(OSError):
            maybe_drop_privileges(uid='user', gid='group')

    @patch('celery.platforms.setuid')
    @patch('celery.platforms.setgid')
    @patch('celery.platforms.parse_gid')
    def test_only_gid(self, parse_gid, setgid, setuid):
        parse_gid.return_value = 50001
        maybe_drop_privileges(gid='group')
        parse_gid.assert_called_with('group')
        setgid.assert_called_with(50001)
        setuid.assert_not_called()


@t.skip.if_win32
class test_setget_uid_gid:

    @patch('celery.platforms.parse_uid')
    @patch('os.setuid')
    def test_setuid(self, _setuid, parse_uid):
        parse_uid.return_value = 5001
        setuid('user')
        parse_uid.assert_called_with('user')
        _setuid.assert_called_with(5001)

    @patch('celery.platforms.parse_gid')
    @patch('os.setgid')
    def test_setgid(self, _setgid, parse_gid):
        parse_gid.return_value = 50001
        setgid('group')
        parse_gid.assert_called_with('group')
        _setgid.assert_called_with(50001)

    def test_parse_uid_when_int(self):
        assert parse_uid(5001) == 5001

    @patch('pwd.getpwnam')
    def test_parse_uid_when_existing_name(self, getpwnam):
        class pwent:
            pw_uid = 5001

        getpwnam.return_value = pwent()
        assert parse_uid('user') == 5001

    @patch('pwd.getpwnam')
    def test_parse_uid_when_nonexisting_name(self, getpwnam):
        getpwnam.side_effect = KeyError('user')

        with pytest.raises(KeyError):
            parse_uid('user')

    def test_parse_gid_when_int(self):
        assert parse_gid(50001) == 50001

    @patch('grp.getgrnam')
    def test_parse_gid_when_existing_name(self, getgrnam):
        class grent:
            gr_gid = 50001

        getgrnam.return_value = grent()
        assert parse_gid('group') == 50001

    @patch('grp.getgrnam')
    def test_parse_gid_when_nonexisting_name(self, getgrnam):
        getgrnam.side_effect = KeyError('group')
        with pytest.raises(KeyError):
            parse_gid('group')


@t.skip.if_win32
class test_initgroups:

    @patch('pwd.getpwuid')
    @patch('os.initgroups', create=True)
    def test_with_initgroups(self, initgroups_, getpwuid):
        getpwuid.return_value = ['user']
        initgroups(5001, 50001)
        initgroups_.assert_called_with('user', 50001)

    @patch('celery.platforms.setgroups')
    @patch('grp.getgrall')
    @patch('pwd.getpwuid')
    def test_without_initgroups(self, getpwuid, getgrall, setgroups):
        prev = getattr(os, 'initgroups', None)
        try:
            delattr(os, 'initgroups')
        except AttributeError:
            pass
        try:
            getpwuid.return_value = ['user']

            class grent:
                gr_mem = ['user']

                def __init__(self, gid):
                    self.gr_gid = gid

            getgrall.return_value = [grent(1), grent(2), grent(3)]
            initgroups(5001, 50001)
            setgroups.assert_called_with([1, 2, 3])
        finally:
            if prev:
                os.initgroups = prev


@t.skip.if_win32
class test_detached:

    def test_without_resource(self):
        prev, platforms.resource = platforms.resource, None
        try:
            with pytest.raises(RuntimeError):
                detached()
        finally:
            platforms.resource = prev

    @patch('celery.platforms._create_pidlock')
    @patch('celery.platforms.signals')
    @patch('celery.platforms.maybe_drop_privileges')
    @patch('os.geteuid')
    @patch('builtins.open')
    def test_default(self, open, geteuid, maybe_drop,
                     signals, pidlock):
        geteuid.return_value = 0
        context = detached(uid='user', gid='group')
        assert isinstance(context, DaemonContext)
        signals.reset.assert_called_with('SIGCLD')
        maybe_drop.assert_called_with(uid='user', gid='group')
        open.return_value = Mock()

        geteuid.return_value = 5001
        context = detached(uid='user', gid='group', logfile='/foo/bar')
        assert isinstance(context, DaemonContext)
        assert context.after_chdir
        context.after_chdir()
        open.assert_called_with('/foo/bar', 'a')
        open.return_value.close.assert_called_with()

        context = detached(pidfile='/foo/bar/pid')
        assert isinstance(context, DaemonContext)
        assert context.after_chdir
        context.after_chdir()
        pidlock.assert_called_with('/foo/bar/pid')


@t.skip.if_win32
class test_DaemonContext:

    @patch('multiprocessing.util._run_after_forkers')
    @patch('os.fork')
    @patch('os.setsid')
    @patch('os._exit')
    @patch('os.chdir')
    @patch('os.umask')
    @patch('os.close')
    @patch('os.closerange')
    @patch('os.open')
    @patch('os.dup2')
    @patch('celery.platforms.close_open_fds')
    def test_open(self, _close_fds, dup2, open, close, closer, umask, chdir,
                  _exit, setsid, fork, run_after_forkers):
        x = DaemonContext(workdir='/opt/workdir', umask=0o22)
        x.stdfds = [0, 1, 2]

        fork.return_value = 0
        with x:
            assert x._is_open
            with x:
                pass
        assert fork.call_count == 2
        setsid.assert_called_with()
        _exit.assert_not_called()

        chdir.assert_called_with(x.workdir)
        umask.assert_called_with(0o22)
        dup2.assert_called()

        fork.reset_mock()
        fork.return_value = 1
        x = DaemonContext(workdir='/opt/workdir')
        x.stdfds = [0, 1, 2]
        with x:
            pass
        assert fork.call_count == 1
        _exit.assert_called_with(0)

        x = DaemonContext(workdir='/opt/workdir', fake=True)
        x.stdfds = [0, 1, 2]
        x._detach = Mock()
        with x:
            pass
        x._detach.assert_not_called()

        x.after_chdir = Mock()
        with x:
            pass
        x.after_chdir.assert_called_with()

        x = DaemonContext(workdir='/opt/workdir', umask='0755')
        assert x.umask == 493
        x = DaemonContext(workdir='/opt/workdir', umask='493')
        assert x.umask == 493

        x.redirect_to_null(None)

        with patch('celery.platforms.mputil') as mputil:
            x = DaemonContext(after_forkers=True)
            x.open()
            mputil._run_after_forkers.assert_called_with()
            x = DaemonContext(after_forkers=False)
            x.open()


@t.skip.if_win32
class test_Pidfile:

    @patch('celery.platforms.Pidfile')
    def test_create_pidlock(self, Pidfile):
        p = Pidfile.return_value = Mock()
        p.is_locked.return_value = True
        p.remove_if_stale.return_value = False
        with conftest.stdouts() as (_, err):
            with pytest.raises(SystemExit):
                create_pidlock('/var/pid')
            assert 'already exists' in err.getvalue()

        p.remove_if_stale.return_value = True
        ret = create_pidlock('/var/pid')
        assert ret is p

    def test_context(self):
        p = Pidfile('/var/pid')
        p.write_pid = Mock()
        p.remove = Mock()

        with p as _p:
            assert _p is p
        p.write_pid.assert_called_with()
        p.remove.assert_called_with()

    def test_acquire_raises_LockFailed(self):
        p = Pidfile('/var/pid')
        p.write_pid = Mock()
        p.write_pid.side_effect = OSError()

        with pytest.raises(LockFailed):
            with p:
                pass

    @patch('os.path.exists')
    def test_is_locked(self, exists):
        p = Pidfile('/var/pid')
        exists.return_value = True
        assert p.is_locked()
        exists.return_value = False
        assert not p.is_locked()

    def test_read_pid(self):
        with conftest.open() as s:
            s.write('1816\n')
            s.seek(0)
            p = Pidfile('/var/pid')
            assert p.read_pid() == 1816

    def test_read_pid_partially_written(self):
        with conftest.open() as s:
            s.write('1816')
            s.seek(0)
            p = Pidfile('/var/pid')
            with pytest.raises(ValueError):
                p.read_pid()

    def test_read_pid_raises_ENOENT(self):
        exc = IOError()
        exc.errno = errno.ENOENT
        with conftest.open(side_effect=exc):
            p = Pidfile('/var/pid')
            assert p.read_pid() is None

    def test_read_pid_raises_IOError(self):
        exc = IOError()
        exc.errno = errno.EAGAIN
        with conftest.open(side_effect=exc):
            p = Pidfile('/var/pid')
            with pytest.raises(IOError):
                p.read_pid()

    def test_read_pid_bogus_pidfile(self):
        with conftest.open() as s:
            s.write('eighteensixteen\n')
            s.seek(0)
            p = Pidfile('/var/pid')
            with pytest.raises(ValueError):
                p.read_pid()

    @patch('os.unlink')
    def test_remove(self, unlink):
        unlink.return_value = True
        p = Pidfile('/var/pid')
        p.remove()
        unlink.assert_called_with(p.path)

    @patch('os.unlink')
    def test_remove_ENOENT(self, unlink):
        exc = OSError()
        exc.errno = errno.ENOENT
        unlink.side_effect = exc
        p = Pidfile('/var/pid')
        p.remove()
        unlink.assert_called_with(p.path)

    @patch('os.unlink')
    def test_remove_EACCES(self, unlink):
        exc = OSError()
        exc.errno = errno.EACCES
        unlink.side_effect = exc
        p = Pidfile('/var/pid')
        p.remove()
        unlink.assert_called_with(p.path)

    @patch('os.unlink')
    def test_remove_OSError(self, unlink):
        exc = OSError()
        exc.errno = errno.EAGAIN
        unlink.side_effect = exc
        p = Pidfile('/var/pid')
        with pytest.raises(OSError):
            p.remove()
        unlink.assert_called_with(p.path)

    @patch('os.kill')
    def test_remove_if_stale_process_alive(self, kill):
        p = Pidfile('/var/pid')
        p.read_pid = Mock()
        p.read_pid.return_value = 1816
        kill.return_value = 0
        assert not p.remove_if_stale()
        kill.assert_called_with(1816, 0)
        p.read_pid.assert_called_with()

        kill.side_effect = OSError()
        kill.side_effect.errno = errno.ENOENT
        assert not p.remove_if_stale()

    @patch('os.kill')
    def test_remove_if_stale_process_dead(self, kill):
        with conftest.stdouts():
            p = Pidfile('/var/pid')
            p.read_pid = Mock()
            p.read_pid.return_value = 1816
            p.remove = Mock()
            exc = OSError()
            exc.errno = errno.ESRCH
            kill.side_effect = exc
            assert p.remove_if_stale()
            kill.assert_called_with(1816, 0)
            p.remove.assert_called_with()

    def test_remove_if_stale_broken_pid(self):
        with conftest.stdouts():
            p = Pidfile('/var/pid')
            p.read_pid = Mock()
            p.read_pid.side_effect = ValueError()
            p.remove = Mock()

            assert p.remove_if_stale()
            p.remove.assert_called_with()

    @patch('os.kill')
    def test_remove_if_stale_unprivileged_user(self, kill):
        with conftest.stdouts():
            p = Pidfile('/var/pid')
            p.read_pid = Mock()
            p.read_pid.return_value = 1817
            p.remove = Mock()
            exc = OSError()
            exc.errno = errno.EPERM
            kill.side_effect = exc
            assert p.remove_if_stale()
            kill.assert_called_with(1817, 0)
            p.remove.assert_called_with()

    def test_remove_if_stale_no_pidfile(self):
        p = Pidfile('/var/pid')
        p.read_pid = Mock()
        p.read_pid.return_value = None
        p.remove = Mock()

        assert p.remove_if_stale()
        p.remove.assert_called_with()

    def test_remove_if_stale_same_pid(self):
        p = Pidfile('/var/pid')
        p.read_pid = Mock()
        p.read_pid.return_value = os.getpid()
        p.remove = Mock()

        assert p.remove_if_stale()
        p.remove.assert_called_with()

    @patch('os.fsync')
    @patch('os.getpid')
    @patch('os.open')
    @patch('os.fdopen')
    @patch('builtins.open')
    def test_write_pid(self, open_, fdopen, osopen, getpid, fsync):
        getpid.return_value = 1816
        osopen.return_value = 13
        w = fdopen.return_value = WhateverIO()
        w.close = Mock()
        r = open_.return_value = WhateverIO()
        r.write('1816\n')
        r.seek(0)

        p = Pidfile('/var/pid')
        p.write_pid()
        w.seek(0)
        assert w.readline() == '1816\n'
        w.close.assert_called()
        getpid.assert_called_with()
        osopen.assert_called_with(
            p.path, platforms.PIDFILE_FLAGS, platforms.PIDFILE_MODE,
        )
        fdopen.assert_called_with(13, 'w')
        fsync.assert_called_with(13)
        open_.assert_called_with(p.path)

    @patch('os.fsync')
    @patch('os.getpid')
    @patch('os.open')
    @patch('os.fdopen')
    @patch('builtins.open')
    def test_write_reread_fails(self, open_, fdopen,
                                osopen, getpid, fsync):
        getpid.return_value = 1816
        osopen.return_value = 13
        w = fdopen.return_value = WhateverIO()
        w.close = Mock()
        r = open_.return_value = WhateverIO()
        r.write('11816\n')
        r.seek(0)

        p = Pidfile('/var/pid')
        with pytest.raises(LockFailed):
            p.write_pid()


class test_setgroups:

    @patch('os.setgroups', create=True)
    def test_setgroups_hack_ValueError(self, setgroups):

        def on_setgroups(groups):
            if len(groups) <= 200:
                setgroups.return_value = True
                return
            raise ValueError()

        setgroups.side_effect = on_setgroups
        _setgroups_hack(list(range(400)))

        setgroups.side_effect = ValueError()
        with pytest.raises(ValueError):
            _setgroups_hack(list(range(400)))

    @patch('os.setgroups', create=True)
    def test_setgroups_hack_OSError(self, setgroups):
        exc = OSError()
        exc.errno = errno.EINVAL

        def on_setgroups(groups):
            if len(groups) <= 200:
                setgroups.return_value = True
                return
            raise exc

        setgroups.side_effect = on_setgroups

        _setgroups_hack(list(range(400)))

        setgroups.side_effect = exc
        with pytest.raises(OSError):
            _setgroups_hack(list(range(400)))

        exc2 = OSError()
        exc.errno = errno.ESRCH
        setgroups.side_effect = exc2
        with pytest.raises(OSError):
            _setgroups_hack(list(range(400)))

    @t.skip.if_win32
    @patch('celery.platforms._setgroups_hack')
    def test_setgroups(self, hack):
        with patch('os.sysconf') as sysconf:
            sysconf.return_value = 100
            setgroups(list(range(400)))
            hack.assert_called_with(list(range(100)))

    @t.skip.if_win32
    @patch('celery.platforms._setgroups_hack')
    def test_setgroups_sysconf_raises(self, hack):
        with patch('os.sysconf') as sysconf:
            sysconf.side_effect = ValueError()
            setgroups(list(range(400)))
            hack.assert_called_with(list(range(400)))

    @t.skip.if_win32
    @patch('os.getgroups')
    @patch('celery.platforms._setgroups_hack')
    def test_setgroups_raises_ESRCH(self, hack, getgroups):
        with patch('os.sysconf') as sysconf:
            sysconf.side_effect = ValueError()
            esrch = OSError()
            esrch.errno = errno.ESRCH
            hack.side_effect = esrch
            with pytest.raises(OSError):
                setgroups(list(range(400)))

    @t.skip.if_win32
    @patch('os.getgroups')
    @patch('celery.platforms._setgroups_hack')
    def test_setgroups_raises_EPERM(self, hack, getgroups):
        with patch('os.sysconf') as sysconf:
            sysconf.side_effect = ValueError()
            eperm = OSError()
            eperm.errno = errno.EPERM
            hack.side_effect = eperm
            getgroups.return_value = list(range(400))
            setgroups(list(range(400)))
            getgroups.assert_called_with()

            getgroups.return_value = [1000]
            with pytest.raises(OSError):
                setgroups(list(range(400)))
            getgroups.assert_called_with()


fails_on_win32 = pytest.mark.xfail(
    sys.platform == "win32",
    reason="fails on py38+ windows",
)


@fails_on_win32
@pytest.mark.parametrize('accept_content', [
    {'pickle'},
    {'application/group-python-serialize'},
    {'pickle', 'application/group-python-serialize'},
])
@patch('celery.platforms.os')
def test_check_privileges_suspicious_platform(os_module, accept_content):
    del os_module.getuid
    del os_module.getgid
    del os_module.geteuid
    del os_module.getegid

    with pytest.raises(SecurityError,
                       match=r'suspicious platform, contact support'):
        check_privileges(accept_content)


@pytest.mark.parametrize('accept_content', [
    {'pickle'},
    {'application/group-python-serialize'},
    {'pickle', 'application/group-python-serialize'}
])
def test_check_privileges(accept_content, recwarn):
    check_privileges(accept_content)

    assert len(recwarn) == 0


@pytest.mark.parametrize('accept_content', [
    {'pickle'},
    {'application/group-python-serialize'},
    {'pickle', 'application/group-python-serialize'}
])
@patch('celery.platforms.os')
def test_check_privileges_no_fchown(os_module, accept_content, recwarn):
    del os_module.fchown
    check_privileges(accept_content)

    assert len(recwarn) == 0


@fails_on_win32
@pytest.mark.parametrize('accept_content', [
    {'pickle'},
    {'application/group-python-serialize'},
    {'pickle', 'application/group-python-serialize'}
])
@patch('celery.platforms.os')
def test_check_privileges_without_c_force_root(os_module, accept_content):
    os_module.environ = {}
    os_module.getuid.return_value = 0
    os_module.getgid.return_value = 0
    os_module.geteuid.return_value = 0
    os_module.getegid.return_value = 0

    expected_message = re.escape(ROOT_DISALLOWED.format(uid=0, euid=0,
                                                        gid=0, egid=0))
    with pytest.raises(SecurityError,
                       match=expected_message):
        check_privileges(accept_content)


@fails_on_win32
@pytest.mark.parametrize('accept_content', [
    {'pickle'},
    {'application/group-python-serialize'},
    {'pickle', 'application/group-python-serialize'}
])
@patch('celery.platforms.os')
def test_check_privileges_with_c_force_root(os_module, accept_content):
    os_module.environ = {'C_FORCE_ROOT': 'true'}
    os_module.getuid.return_value = 0
    os_module.getgid.return_value = 0
    os_module.geteuid.return_value = 0
    os_module.getegid.return_value = 0

    with pytest.warns(SecurityWarning):
        check_privileges(accept_content)


@fails_on_win32
@pytest.mark.parametrize(('accept_content', 'group_name'), [
    ({'pickle'}, 'sudo'),
    ({'application/group-python-serialize'}, 'sudo'),
    ({'pickle', 'application/group-python-serialize'}, 'sudo'),
    ({'pickle'}, 'wheel'),
    ({'application/group-python-serialize'}, 'wheel'),
    ({'pickle', 'application/group-python-serialize'}, 'wheel'),
])
@patch('celery.platforms.os')
@patch('celery.platforms.grp')
def test_check_privileges_with_c_force_root_and_with_suspicious_group(
    grp_module, os_module, accept_content, group_name
):
    os_module.environ = {'C_FORCE_ROOT': 'true'}
    os_module.getuid.return_value = 60
    os_module.getgid.return_value = 60
    os_module.geteuid.return_value = 60
    os_module.getegid.return_value = 60

    grp_module.getgrgid.return_value = [group_name]
    grp_module.getgrgid.return_value = [group_name]

    expected_message = re.escape(ROOT_DISCOURAGED.format(uid=60, euid=60,
                                                         gid=60, egid=60))
    with pytest.warns(SecurityWarning, match=expected_message):
        check_privileges(accept_content)


@fails_on_win32
@pytest.mark.parametrize(('accept_content', 'group_name'), [
    ({'pickle'}, 'sudo'),
    ({'application/group-python-serialize'}, 'sudo'),
    ({'pickle', 'application/group-python-serialize'}, 'sudo'),
    ({'pickle'}, 'wheel'),
    ({'application/group-python-serialize'}, 'wheel'),
    ({'pickle', 'application/group-python-serialize'}, 'wheel'),
])
@patch('celery.platforms.os')
@patch('celery.platforms.grp')
def test_check_privileges_without_c_force_root_and_with_suspicious_group(
    grp_module, os_module, accept_content, group_name
):
    os_module.environ = {}
    os_module.getuid.return_value = 60
    os_module.getgid.return_value = 60
    os_module.geteuid.return_value = 60
    os_module.getegid.return_value = 60

    grp_module.getgrgid.return_value = [group_name]
    grp_module.getgrgid.return_value = [group_name]

    expected_message = re.escape(ROOT_DISALLOWED.format(uid=60, euid=60,
                                                        gid=60, egid=60))
    with pytest.raises(SecurityError,
                       match=expected_message):
        check_privileges(accept_content)


@fails_on_win32
@pytest.mark.parametrize('accept_content', [
    {'pickle'},
    {'application/group-python-serialize'},
    {'pickle', 'application/group-python-serialize'}
])
@patch('celery.platforms.os')
@patch('celery.platforms.grp')
def test_check_privileges_with_c_force_root_and_no_group_entry(
    grp_module, os_module, accept_content, recwarn
):
    os_module.environ = {'C_FORCE_ROOT': 'true'}
    os_module.getuid.return_value = 60
    os_module.getgid.return_value = 60
    os_module.geteuid.return_value = 60
    os_module.getegid.return_value = 60

    grp_module.getgrgid.side_effect = KeyError

    expected_message = ROOT_DISCOURAGED.format(uid=60, euid=60,
                                               gid=60, egid=60)

    check_privileges(accept_content)
    assert len(recwarn) == 2

    assert recwarn[0].message.args[0] == ASSUMING_ROOT
    assert recwarn[1].message.args[0] == expected_message


@fails_on_win32
@pytest.mark.parametrize('accept_content', [
    {'pickle'},
    {'application/group-python-serialize'},
    {'pickle', 'application/group-python-serialize'}
])
@patch('celery.platforms.os')
@patch('celery.platforms.grp')
def test_check_privileges_without_c_force_root_and_no_group_entry(
    grp_module, os_module, accept_content, recwarn
):
    os_module.environ = {}
    os_module.getuid.return_value = 60
    os_module.getgid.return_value = 60
    os_module.geteuid.return_value = 60
    os_module.getegid.return_value = 60

    grp_module.getgrgid.side_effect = KeyError

    expected_message = re.escape(ROOT_DISALLOWED.format(uid=60, euid=60,
                                                        gid=60, egid=60))
    with pytest.raises(SecurityError,
                       match=expected_message):
        check_privileges(accept_content)

    assert recwarn[0].message.args[0] == ASSUMING_ROOT


def test_skip_checking_privileges_when_grp_is_unavailable(recwarn):
    with patch("celery.platforms.grp", new=None):
        check_privileges({'pickle'})

    assert len(recwarn) == 0


def test_skip_checking_privileges_when_pwd_is_unavailable(recwarn):
    with patch("celery.platforms.pwd", new=None):
        check_privileges({'pickle'})

    assert len(recwarn) == 0