ronantakizawa/codereview-bench huggingface dataset:
Repository: celery/celery
File: celery/contrib/testing/worker.py (Python)

Code before:
    name='test_worker_starting',
    providing_args={},
)
test_worker_started = Signal(
    name='test_worker_started',
    providing_args={'worker', 'consumer'},
)
test_worker_stopped = Signal(
    name='test_worker_stopped',
    providing_args={'worker'},
)


class TestWorkController(worker.WorkController):
    """Worker that can synchronize on being fully started."""

    logger_queue = None

    def __init__(self, *args, **kwargs):
        # type: (*Any, **Any) -> None
        self._on_started = threading.Event()

        super().__init__(*args, **kwargs)

        if self.pool_cls.__module__.split('.')[-1] == 'prefork':
            from multiprocessing import Queue
            self.logger_queue = Queue()
            self.pid = os.getpid()

            try:
                from tblib import pickling_support
                pickling_support.install()
            except ImportError:
                pass

            # collect logs from forked process.
            # XXX: those logs will appear twice in the live log
            self.queue_listener = logging.handlers.QueueListener(self.logger_queue, logging.getLogger())
            self.queue_listener.start()

    class QueueHandler(logging.handlers.QueueHandler):
        def prepare(self, record):
            record.from_queue = True
            # Keep origin record.
            return record

        def handleError(self, record):
            if logging.raiseExceptions:
                raise

    def start(self):

Review comment (suggestion):
```suggestion
            from billiard import Queue
```

Code after:
    name='test_worker_starting',
    providing_args={},
)
test_worker_started = Signal(
    name='test_worker_started',
    providing_args={'worker', 'consumer'},
)
test_worker_stopped = Signal(
    name='test_worker_stopped',
    providing_args={'worker'},
)


class TestWorkController(worker.WorkController):
    """Worker that can synchronize on being fully started."""

    logger_queue = None

    def __init__(self, *args, **kwargs):
        # type: (*Any, **Any) -> None
        self._on_started = threading.Event()

        super().__init__(*args, **kwargs)

        if self.pool_cls.__module__.split('.')[-1] == 'prefork':
            from billiard import Queue
            self.logger_queue = Queue()
            self.pid = os.getpid()

            try:
                from tblib import pickling_support
                pickling_support.install()
            except ImportError:
                pass

            # collect logs from forked process.
            # XXX: those logs will appear twice in the live log
            self.queue_listener = logging.handlers.QueueListener(self.logger_queue, logging.getLogger())
            self.queue_listener.start()

    class QueueHandler(logging.handlers.QueueHandler):
        def prepare(self, record):
            record.from_queue = True
            # Keep origin record.
            return record

        def handleError(self, record):
            if logging.raiseExceptions:
                raise

    def start(self):
Repository: celery/celery
File: t/unit/app/test_preload_cli.py (Python)

Code before:
    )

    assert "No such option: --ini" in res_without_preload.stdout
    assert res_without_preload.exit_code == 2

    res_with_preload = isolated_cli_runner.invoke(
        celery,
        [
            "-A",
            "t.unit.bin.proj.pyramid_celery_app",
            "purge",
            "-f",
            "--ini",
            "some_ini.ini",
        ],
        catch_exceptions=True,
    )

    assert res_with_preload.exit_code == 0

    res_with_preload = isolated_cli_runner.invoke(
        celery,
        [
            "-A",
            "t.unit.bin.proj.pyramid_celery_app",
            "shell",
            "--ini",
            "some_ini.ini",
        ],
        catch_exceptions=True,
    )
    assert res_with_preload.exit_code == 0


Review comment (suggestion):
Is there a recommended way to fully clean/reset the applications called by the isolated_cli_runner? I noticed that if I run the `t.unit.bin.proj.pyramid_celery_app`  cases first then the --ini option seems to be defined even when running the `t.unit.bin.proj.app`. I'm guessing that Celery has a global or some other persistence that I am not aware of that needs to get reset or added to the tear-down of  the tests. I've also tried running these as separate tests and even in separate test files and there still seems to be some persistence or leakage of the user_options. Basically I want two separate and independent Celery processes to test and configure here.

Code after:

    assert "No such option: --ini" in res_without_preload.stdout
    assert res_without_preload.exit_code == 2

    res_without_preload = isolated_cli_runner.invoke(
        celery,
        ["-A", "t.unit.bin.proj.app", "shell", "--ini", "some_ini.ini"],
        catch_exceptions=True,
    )

    assert "No such option: --ini" in res_without_preload.stdout
    assert res_without_preload.exit_code == 2

    res_with_preload = isolated_cli_runner.invoke(
        celery,
        [
            "-A",
            "t.unit.bin.proj.pyramid_celery_app",
            "purge",
            "-f",
            "--ini",
            "some_ini.ini",
        ],
        catch_exceptions=True,
    )

    assert res_with_preload.exit_code == 0

    res_with_preload = isolated_cli_runner.invoke(
        celery,
        [
            "-A",
            "t.unit.bin.proj.pyramid_celery_app",
            "shell",
            "--ini",
            "some_ini.ini",
        ],
        catch_exceptions=True,
    )
    assert res_with_preload.exit_code == 0
Repository: celery/celery
File: t/unit/bin/proj/pyramid_celery_app.py (Python)

Code before:
# The Pyramid-Celery project https://github.com/sontek/pyramid_celery
# assumes that you want to configure Celery via an ini settings file.
# The .ini files are the standard configuration file for Pyramid
# applications.
# See https://docs.pylonsproject.org/projects/pyramid/en/latest/quick_tutorial/ini.html
#

app = Celery(set_as_current=False)
app.config_from_object("t.integration.test_worker_config")

class PurgeMock:
    def queue_purge(self, queue):
        return 0

class ConnMock:
    default_channel = PurgeMock()
    channel_errors = KeyError

mock = Mock()
mock.__enter__ = Mock(return_value=ConnMock())
mock.__exit__ = Mock(return_value=False)

app.connection_for_write = MagicMock(return_value=mock)

# Below are taken from pyramid-celery's __init__.py
# Ref: https://github.com/sontek/pyramid_celery/blob/cf8aa80980e42f7235ad361874d3c35e19963b60/pyramid_celery/__init__.py#L25-L36
ini_option = Option(
    ('--ini', '-i',),
    help='Paste ini configuration file.'
)

ini_var_option = Option(
    ('--ini-var',),
    help='Comma separated list of key=value to pass to ini.'
)

app.user_options['preload'].add(
        ini_option
)
app.user_options['preload'].add(
    ini_var_option
)



Review comment (suggestion):
I use this Mock to prevent hanging of execution and get immediate return during the test. If there is a more standard way to mock the purge functionality in the context of tests let me know.

Code after:
# The Pyramid-Celery project https://github.com/sontek/pyramid_celery
# assumes that you want to configure Celery via an ini settings file.
# The .ini files are the standard configuration file for Pyramid
# applications.
# See https://docs.pylonsproject.org/projects/pyramid/en/latest/quick_tutorial/ini.html
#

app = Celery(set_as_current=False)
app.config_from_object("t.integration.test_worker_config")


class PurgeMock:
    def queue_purge(self, queue):
        return 0


class ConnMock:
    default_channel = PurgeMock()
    channel_errors = KeyError


mock = Mock()
mock.__enter__ = Mock(return_value=ConnMock())
mock.__exit__ = Mock(return_value=False)

app.connection_for_write = MagicMock(return_value=mock)

# Below are taken from pyramid-celery's __init__.py
# Ref: https://github.com/sontek/pyramid_celery/blob/cf8aa80980e42f7235ad361874d3c35e19963b60/pyramid_celery/__init__.py#L25-L36 # noqa: E501
ini_option = Option(
    (
        "--ini",
        "-i",
    ),
    help="Paste ini configuration file.",
)

ini_var_option = Option(
    ("--ini-var",), help="Comma separated list of key=value to pass to ini."
)

app.user_options["preload"].add(ini_option)
app.user_options["preload"].add(ini_var_option)
Repository: celery/celery
File: t/unit/tasks/test_result.py (Python)

Code before:
import copy
import datetime
import platform
import traceback
from contextlib import contextmanager
from unittest import mock
from unittest.mock import Mock, call, patch
from zoneinfo import ZoneInfo

import pytest

from celery import states, uuid
from celery.app.task import Context
from celery.backends.base import Backend, SyncBackendMixin
from celery.exceptions import ImproperlyConfigured, IncompleteStream, TimeoutError
from celery.result import AsyncResult, EagerResult, GroupResult, ResultSet, assert_will_not_block, result_from_tuple
from celery.utils.serialization import pickle

PYTRACEBACK = """\
Traceback (most recent call last):
  File "foo.py", line 2, in foofunc
    don't matter
  File "bar.py", line 3, in barfunc
    don't matter
Doesn't matter: really!\
"""


def mock_task(name, state, result, traceback=None):
    return {
        'id': uuid(), 'name': name, 'state': state,
        'result': result, 'traceback': traceback,
    }

Review comment (suggestion):
https://docs.python.org/3/library/zoneinfo.html was added to the Standard Library in Python 3.9.

https://pypi.org/project/backports.zoneinfo could be installed ___conditionally___ on Python < v3.9.

Code after:
import copy
import datetime
import platform
import traceback
from contextlib import contextmanager
from unittest.mock import Mock, call, patch

import pytest

from celery import states, uuid
from celery.app.task import Context
from celery.backends.base import Backend, SyncBackendMixin
from celery.exceptions import ImproperlyConfigured, IncompleteStream, TimeoutError
from celery.result import AsyncResult, EagerResult, GroupResult, ResultSet, assert_will_not_block, result_from_tuple
from celery.utils.serialization import pickle

PYTRACEBACK = """\
Traceback (most recent call last):
  File "foo.py", line 2, in foofunc
    don't matter
  File "bar.py", line 3, in barfunc
    don't matter
Doesn't matter: really!\
"""


def mock_task(name, state, result, traceback=None):
    return {
        'id': uuid(), 'name': name, 'state': state,
        'result': result, 'traceback': traceback,
    }
Repository: celery/celery
File: celery/contrib/django/task.py (Python)

Code before:
import functools

try:
    from django.db import transaction
except ImportError:
    transaction = None

from celery.app.task import Task as BaseTask


class DjangoTask(BaseTask):
    """
    Extend the base :class:`~celery.app.task.Task` for Django.

    Provide a nicer API to trigger tasks at the end of the DB transaction.
    """

    def delay_on_commit(self, *args, **kwargs):
        """Call :meth:`~celery.app.task.Task.delay` with Django's ``on_commit()``."""
        return transaction.on_commit(functools.partial(self.delay, *args, **kwargs))

    def apply_async_on_commit(self, *args, **kwargs):
        """Call :meth:`~celery.app.task.Task.apply_async` with Django's ``on_commit()``."""
        return transaction.on_commit(functools.partial(self.apply_async, *args, **kwargs))


Review comment (suggestion):
```suggestion
from celery.app.task import Task
```

Code after:
import functools

from django.db import transaction

from celery.app.task import Task


class DjangoTask(Task):
    """
    Extend the base :class:`~celery.app.task.Task` for Django.

    Provide a nicer API to trigger tasks at the end of the DB transaction.
    """

    def delay_on_commit(self, *args, **kwargs):
        """Call :meth:`~celery.app.task.Task.delay` with Django's ``on_commit()``."""
        return transaction.on_commit(functools.partial(self.delay, *args, **kwargs))

    def apply_async_on_commit(self, *args, **kwargs):
        """Call :meth:`~celery.app.task.Task.apply_async` with Django's ``on_commit()``."""
        return transaction.on_commit(functools.partial(self.apply_async, *args, **kwargs))
Repository: celery/celery
File: celery/utils/nodenames.py (Python)

Code before:
    return parts


def default_nodename(hostname: str) -> str:
    """Return the default nodename for this process."""
    name, host = nodesplit(hostname or '')
    return nodename(name or NODENAME_DEFAULT, host or gethostname())


def node_format(s: str, name: str, **extra: Any) -> str:
    """Format worker node name (name@host.com)."""
    shortname, host = nodesplit(name)
    return host_format(s, host, shortname or NODENAME_DEFAULT, p=name, **extra)


def _fmt_process_index(prefix: str = '', default: str = '0') -> str:
    from .log import current_process_index

    index = current_process_index()
    return f'{prefix}{index}' if index else default


_fmt_process_index_with_prefix = partial(_fmt_process_index, '-', '')


def host_format(s: str, host: str | None = None, name: str | None = None, **extra: Any) -> str:
    """Format host %x abbreviations."""
    host = host or gethostname()
    hname, _, domain = host.partition('.')
    name = name or hname
    keys = dict(
        {
            'h': host,
            'n': name,
            'd': domain,
            'i': _fmt_process_index,
            'I': _fmt_process_index_with_prefix,
        },
        **extra,
    )
    return simple_format(s, keys)


Review comment (suggestion):
```suggestion
def host_format(s: str, host: str | None = None, name: str | None = None, **extra: dict) -> str:
```

Code after:
    """Return the default nodename for this process."""
    name, host = nodesplit(hostname or '')
    return nodename(name or NODENAME_DEFAULT, host or gethostname())


def node_format(s: str, name: str, **extra: dict) -> str:
    """Format worker node name (name@host.com)."""
    shortname, host = nodesplit(name)
    return host_format(s, host, shortname or NODENAME_DEFAULT, p=name, **extra)


def _fmt_process_index(prefix: str = '', default: str = '0') -> str:
    from .log import current_process_index

    index = current_process_index()
    return f'{prefix}{index}' if index else default


_fmt_process_index_with_prefix = partial(_fmt_process_index, '-', '')


def host_format(s: str, host: str | None = None, name: str | None = None, **extra: dict) -> str:
    """Format host %x abbreviations."""
    host = host or gethostname()
    hname, _, domain = host.partition('.')
    name = name or hname
    keys = dict(
        {
            'h': host,
            'n': name,
            'd': domain,
            'i': _fmt_process_index,
            'I': _fmt_process_index_with_prefix,
        },
        **extra,
    )
    return simple_format(s, keys)
Repository: celery/celery
File: celery/backends/asynchronous.py (Python)

Code before:

    def stop(self):
        self._stopped.set()
        self._shutdown.wait(THREAD_TIMEOUT_MAX)

    def wait_for(self, p, wait, timeout=None):
        self.start()
        if not p.ready:
            self._drain_complete_event.wait(timeout=timeout)

            self._ensure_not_shut_down()

    def _ensure_not_shut_down(self):
        """Currently used to ensure the drainer has not run to completion.

        Raises if the shutdown event has been signaled (either due to an exception
        or stop() being called).

        The _shutdown event acts as synchronization to ensure _exc is properly
        set before it is read from, avoiding need for locks.
        """
        if self._shutdown.is_set():
            if self._exc is not None:
                raise self._exc
            else:
                raise CeleryRestartRequired(E_CELERY_RESTART_REQUIRED)


@register_drainer('eventlet')
class eventletDrainer(greenletDrainer):

    def spawn(self, func):
        from eventlet import sleep, spawn
        g = spawn(func)
        sleep(0)
        return g

    def _event(self):
        return EventletAdaptedEvent()


@register_drainer('gevent')
class geventDrainer(greenletDrainer):

    def spawn(self, func):
        import gevent
        g = gevent.spawn(func)
        gevent.sleep(0)
        return g

    def _event(self):

Review comment (suggestion):
```suggestion
                raise Exception(E_CELERY_RESTART_REQUIRED)
```

Code after:
            self._started.wait()

    def stop(self):
        self._stopped.set()
        self._shutdown.wait(THREAD_TIMEOUT_MAX)

    def wait_for(self, p, wait, timeout=None):
        self.start()
        if not p.ready:
            self._drain_complete_event.wait(timeout=timeout)

            self._ensure_not_shut_down()

    def _ensure_not_shut_down(self):
        """Currently used to ensure the drainer has not run to completion.

        Raises if the shutdown event has been signaled (either due to an exception
        or stop() being called).

        The _shutdown event acts as synchronization to ensure _exc is properly
        set before it is read from, avoiding need for locks.
        """
        if self._shutdown.is_set():
            if self._exc is not None:
                raise self._exc
            else:
                raise Exception(E_CELERY_RESTART_REQUIRED)


@register_drainer('eventlet')
class eventletDrainer(greenletDrainer):

    def spawn(self, func):
        from eventlet import sleep, spawn
        g = spawn(func)
        sleep(0)
        return g

    def _event(self):
        return EventletAdaptedEvent()


@register_drainer('gevent')
class geventDrainer(greenletDrainer):

    def spawn(self, func):
        import gevent
        g = gevent.spawn(func)
        gevent.sleep(0)
        return g
Repository: celery/celery
File: celery/utils/term.py (Python)

Code before:
        return self.node(s, fg(40 + GREEN))

    def iyellow(self, *s: object) -> colored:
        return self.node(s, fg(40 + YELLOW))

    def iblue(self, *s: colored) -> colored:
        return self.node(s, fg(40 + BLUE))

    def imagenta(self, *s: object) -> colored:
        return self.node(s, fg(40 + MAGENTA))

    def icyan(self, *s: object) -> colored:
        return self.node(s, fg(40 + CYAN))

    def iwhite(self, *s: object) -> colored:
        return self.node(s, fg(40 + WHITE))

    def reset(self, *s: object) -> colored:
        return self.node(s or ('',), RESET_SEQ)

    def __add__(self, other: object) -> str:
        return f"{self}{other}"


def supports_images() -> bool:
    return isatty(sys.stdin) and ITERM_PROFILE is not None


def _read_as_base64(path: str) -> str:
    with open(path, mode='rb') as fh:
        encoded = base64.b64encode(fh.read())
        return encoded.decode('ascii')


def imgcat(path: str, inline: int = 1, preserve_aspect_ratio: int = 0, **kwargs: Any) -> str:
    return '\n%s1337;File=inline=%d;preserveAspectRatio=%d:%s%s' % (
        _IMG_PRE, inline, preserve_aspect_ratio,
        _read_as_base64(path), _IMG_POST)


Review comment (suggestion):
```suggestion
    
    try:
        return sys.stdin.isatty() and ITERM_PROFILE
    except AttributeError:
        pass
```

Code after:
    def iyellow(self, *s: object) -> colored:
        return self.node(s, fg(40 + YELLOW))

    def iblue(self, *s: colored) -> colored:
        return self.node(s, fg(40 + BLUE))

    def imagenta(self, *s: object) -> colored:
        return self.node(s, fg(40 + MAGENTA))

    def icyan(self, *s: object) -> colored:
        return self.node(s, fg(40 + CYAN))

    def iwhite(self, *s: object) -> colored:
        return self.node(s, fg(40 + WHITE))

    def reset(self, *s: object) -> colored:
        return self.node(s or ('',), RESET_SEQ)

    def __add__(self, other: object) -> str:
        return f"{self}{other}"


def supports_images() -> bool:

    try:
        return sys.stdin.isatty() and bool(os.environ.get('ITERM_PROFILE'))
    except AttributeError:
        return False


def _read_as_base64(path: str) -> str:
    with open(path, mode='rb') as fh:
        encoded = base64.b64encode(fh.read())
        return encoded.decode('ascii')


def imgcat(path: str, inline: int = 1, preserve_aspect_ratio: int = 0, **kwargs: Any) -> str:
    return '\n%s1337;File=inline=%d;preserveAspectRatio=%d:%s%s' % (
        _IMG_PRE, inline, preserve_aspect_ratio,
        _read_as_base64(path), _IMG_POST)
Repository: celery/celery
File: t/integration/test_rabbitmq_default_queue_type_fallback.py (Python)

Code before:
import socket
import time

import pytest
from kombu import Connection

from celery import Celery


def wait_for_port(host, port, timeout=30.0):
    """Wait for a port to become available."""
    start = time.time()
    while time.time() - start < timeout:
        try:
            with socket.create_connection((host, port), timeout=2):
                return
        except OSError:
            time.sleep(1)
    raise TimeoutError(f"Timed out waiting for {host}:{port}")


@pytest.fixture()
def redis():
    """Fixture to provide Redis hostname and port."""
    return {"hostname": "redis", "port": 6379}


@pytest.fixture()
def app(rabbitmq, redis):
    wait_for_port(rabbitmq.hostname, rabbitmq.ports[5672])
    wait_for_port(redis["hostname"], redis["port"])

    return Celery(
        "test_app",
        broker=f"pyamqp://guest:guest@{rabbitmq.hostname}:{rabbitmq.ports[5672]}/",

Review comment (suggestion):
```suggestion
def wait_for_port(host, port, timeout=60.0):
```

Code after:
import socket
import time

import pytest
from kombu import Connection

from celery import Celery


def wait_for_port(host, port, timeout=60.0):
    """Wait for a port to become available."""
    start = time.time()
    while time.time() - start < timeout:
        try:
            with socket.create_connection((host, port), timeout=2):
                return
        except OSError:
            time.sleep(1)
    raise TimeoutError(f"Timed out waiting for {host}:{port}")


@pytest.fixture()
def redis():
    """Fixture to provide Redis hostname and port."""
    return {"hostname": "redis", "port": 6379}


@pytest.fixture()
def app(rabbitmq, redis):
    wait_for_port(rabbitmq.hostname, rabbitmq.ports[5672])
    wait_for_port(redis["hostname"], redis["port"])

    return Celery(
        "test_app",
        broker=f"pyamqp://guest:guest@{rabbitmq.hostname}:{rabbitmq.ports[5672]}/",