Function setup_background_callback_app

tests/async_tests/utils.py:77–155

(manager_name, app_name)

@contextmanager
def setup_background_callback_app(manager_name, app_name):
    from dash.testing.application_runners import import_app

    if manager_name == "celery":
        os.environ["LONG_CALLBACK_MANAGER"] = "celery"
        redis_url = os.environ["REDIS_URL"].rstrip("/")
        os.environ["CELERY_BROKER"] = f"{redis_url}/0"
        os.environ["CELERY_BACKEND"] = f"{redis_url}/1"

        # Clear redis of cached values
        redis_conn = redis.Redis(host="localhost", port=6379, db=1)
        cache_keys = redis_conn.keys()
        if cache_keys:
            redis_conn.delete(*cache_keys)

        worker = subprocess.Popen(
            [
                "celery",
                "-A",
                f"tests.async_tests.{app_name}:handle",
                "worker",
                "-P",
                "prefork",
                "--concurrency",
                "2",
                "--loglevel=info",
            ],
            encoding="utf8",
            preexec_fn=os.setpgrp,
            stderr=subprocess.PIPE,
        )
        # Wait for the worker to be ready, if you cancel before it is ready, the job
        # will still be queued.
        lines = []
        for line in iter(worker.stderr.readline, ""):
            if "ready" in line:
                break
            lines.append(line)
        else:
            error = "\n".join(lines)
            raise RuntimeError(f"celery failed to start: {error}")

        try:
            yield import_app(f"tests.async_tests.{app_name}")
        finally:
            # Interval may run one more time after settling on final app state
            # Sleep for 1 interval of time
            time.sleep(0.5)
            os.environ.pop("LONG_CALLBACK_MANAGER")
            os.environ.pop("CELERY_BROKER")
            os.environ.pop("CELERY_BACKEND")
            kill(worker.pid)
            from dash import page_registry

            page_registry.clear()

    elif manager_name == "diskcache":
        os.environ["LONG_CALLBACK_MANAGER"] = "diskcache"