Function _invoke_lambda

python/vercel-runtime/tests/test_runtime.py:133–199

Run vc_init.py in legacy mode and call vc_handler. Uses an extra pipe (fd 3) so vc_init.py's own print() calls don't pollute the JSON result.

(
    *,
    entrypoint_abs: pathlib.Path,
    entrypoint_rel: str,
    module_name: str,
    event: dict[str, Any],
    variable_name: str = "app",
)

async def _invoke_lambda(
    *,
    entrypoint_abs: pathlib.Path,
    entrypoint_rel: str,
    module_name: str,
    event: dict[str, Any],
    variable_name: str = "app",
) -> dict[str, Any]:
    """Run vc_init.py in legacy mode and call vc_handler.

    Uses an extra pipe (fd 3) so vc_init.py's own print()
    calls don't pollute the JSON result.
    """
    env = {
        **_base_env(),
        "__VC_HANDLER_ENTRYPOINT": entrypoint_rel,
        "__VC_HANDLER_ENTRYPOINT_ABS": str(entrypoint_abs),
        "__VC_HANDLER_MODULE_NAME": module_name,
        "__VC_HANDLER_VARIABLE_NAME": variable_name,
    }
    env.pop("VERCEL_IPC_PATH", None)

    result_r, result_w = os.pipe()
    env["_RESULT_FD"] = str(result_w)

    cmd = [sys.executable]
    if _coverage_active():
        cmd.append(str(_COV_WRAPPER))
    cmd.extend([str(_LAMBDA_INVOKER), str(_VC_INIT)])

    proc = await asyncio.create_subprocess_exec(
        *cmd,
        env=env,
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        pass_fds=(result_w,),
    )
    os.close(result_w)

    try:
        assert proc.stdin is not None
        proc.stdin.write(json.dumps(event).encode())
        proc.stdin.close()

        loop = asyncio.get_running_loop()
        result_data, _ = await asyncio.wait_for(
            asyncio.gather(
                loop.run_in_executor(
                    None,
                    lambda: os.read(result_r, 1_000_000),
                ),
                proc.wait(),
            ),
            timeout=10.0,
        )
    finally:
        os.close(result_r)