def run_step(step, catalog):  # noqa: PLR0912
    """Run an end-to-end query test step with a command and expected output."""
    command = step["command"]
    # Note that a process.returncode of -2 is the same as the shell returncode of 130
    # (canceled by KeyboardInterrupt)
    interrupt_exit_code = -2
    if sys.platform == "win32":
        # Windows has a different mechanism of creating a process group.
        popen_args = {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP}
        # This is STATUS_CONTROL_C_EXIT which is equivalent to 0xC000013A
        interrupt_exit_code = 3221225786
    else:
        popen_args = {"start_new_session": True}
    stdin_file = None
    if step.get("stdin_file"):
        # The "with" file open context manager cannot be used here without
        # additional code duplication, as a file is only opened if needed.
        stdin_file = open(step["stdin_file"])  # noqa: SIM115
    try:
        process = subprocess.Popen(  # noqa: S603
            command,
            shell=False,
            stdin=stdin_file,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            encoding="utf-8",
            env={
                **os.environ,
                "DATACHAIN__METASTORE": catalog.metastore.serialize(),
                "DATACHAIN__WAREHOUSE": catalog.warehouse.serialize(),
            },
            **popen_args,
        )
        interrupt_after = step.get("interrupt_after")
        if interrupt_after:
            stdout, stderr = communicate_and_interrupt_process(process, interrupt_after)
        else:
            stdout, stderr = process.communicate(timeout=E2E_STEP_TIMEOUT_SEC)
    finally:
        if stdin_file:
            stdin_file.close()

    if interrupt_after:
        if process.returncode not in (interrupt_exit_code, 11):
            print(f"Process stdout: {stdout}")
            print(f"Process stderr: {stderr}")
            raise RuntimeError(
                "Query script failed to interrupt correctly: "
                f"{process.returncode} Command: {command}"
            )
    elif process.returncode != 0:
        print(f"Process stdout: {stdout}")
        print(f"Process stderr: {stderr}")
        raise RuntimeError(
            "Query script failed with exit code: "
            f"{process.returncode} Command: {command}"
        )

    if step.get("sort_expected_lines"):
        assert sorted(stdout.split("\n")) == sorted(
            step["expected"].lstrip("\n").split("\n")
        )
    elif step.get("expected_in_stderr"):
        assert step["expected_in_stderr"] in stderr
        if step.get("expected_not_in_stderr"):
            assert step["expected_not_in_stderr"] not in stderr
    elif step.get("expected_in"):
        assert sorted(stdout.split("\n")) == sorted(
            step["expected_in"].lstrip("\n").split("\n")
        )
    elif step.get("expected_rows"):
        assert _comparable_row(stdout) == _comparable_row(step["expected_rows"])
    else:
        assert stdout == step["expected"].lstrip("\n")

    if step.get("listing"):
        assert "Listing" in stderr
    else:
        assert "Listing" not in stderr
Run an end-to-end query test step with a command and expected output.
tests/test_query_e2e.py