The `replaced_task_nesting` ([docs](https://docs.celeryq.dev/en/stable/userguide/tasks.html#task-request) @ `New in version 5.2.` ) was missing from the `request` because it wasn't parsed correctly over the call stack. This made the whole feature invalid as you always lost the `replaced_task_nesting` value which made it reset to zero.

This PR fixes the handling of `replaced_task_nesting` and also adds integration tests for the feature that didn't exist before.
Fixed replaced_task_nesting. The `replaced_task_nesting` ([docs](https://docs.celeryq.dev/en/stable/userguide/tasks.html#task-request) @ `New in version 5.2.` ) was missing from the `request` because it wasn't parsed correctly over the call stack. This made the whole feature invalid as you always lost the `replaced_task_nesting` value which made it reset to zero.

This PR fixes the handling of `replaced_task_nesting` and also adds integration tests for the feature that didn't exist before.
"""Single-threaded execution pool."""
import os

from celery import signals

from .base import BasePool, apply_target

__all__ = ('TaskPool',)


class TaskPool(BasePool):
    """Solo task pool (blocking, inline, fast)."""

    body_can_be_buffer = True

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.on_apply = apply_target
        self.limit = 1
        signals.worker_process_init.send(sender=None)

    def _get_info(self):
        info = super()._get_info()
        info.update({
            'max-concurrency': 1,
            'processes': [os.getpid()],
            'max-tasks-per-child': None,
            'put-guarded-by-semaphore': True,
            'timeouts': (),
        })
        return info

"""Single-threaded execution pool."""
import os

from celery import signals

from .base import BasePool, apply_target

__all__ = ('TaskPool',)


class TaskPool(BasePool):
    """Solo task pool (blocking, inline, fast)."""

    body_can_be_buffer = True

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.on_apply = apply_target
        self.limit = 1
        signals.worker_process_init.send(sender=None)

    def _get_info(self):
        info = super()._get_info()
        info.update({
            'max-concurrency': 1,
            'processes': [os.getpid()],
            'max-tasks-per-child': None,
            'put-guarded-by-semaphore': True,
            'timeouts': (),
        })
        return info
