Function unpickle_in_subprocess

lib/sqlalchemy/testing/util.py:68–103

pickle ``obj`` to a file, then unpickle it in a new interpreter. ``code`` is Python source run by that interpreter, which receives the name of the pickle file as ``sys.argv[1]``. The new interpreter has the current ``sys.path``, so that the SQLAlchemy under test, as well as the ``t

(obj, code)

def unpickle_in_subprocess(obj, code):
    """pickle ``obj`` to a file, then unpickle it in a new interpreter.

    ``code`` is Python source run by that interpreter, which receives the
    name of the pickle file as ``sys.argv[1]``.  The new interpreter has
    the current ``sys.path``, so that the SQLAlchemy under test, as well
    as the ``test`` package, are importable.

    Returns the stripped stdout of the subprocess; a non-zero exit status
    fails the test, reporting its stderr.

    """

    fd, filename = mkstemp("pkl")
    try:
        with os.fdopen(fd, "wb") as file_:
            pickle.dump(obj, file_)

        parts = list(sys.path)
        if os.environ.get("PYTHONPATH"):
            parts.append(os.environ["PYTHONPATH"])

        proc = subprocess.run(
            [sys.executable, "-c", code, filename.replace(os.sep, "/")],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            env={**os.environ, "PYTHONPATH": os.pathsep.join(parts)},
        )
    finally:
        os.unlink(filename)

    if proc.returncode != 0:
        raise AssertionError(
            "subprocess failed: %s" % proc.stderr.decode(errors="replace")
        )
    return proc.stdout.strip()


def random_choices(population, k=1):