import faulthandler
import io
import os
import pdb
import sys
from io import BytesIO

import pytest

import click
from click.exceptions import ClickException
from click.testing import CliRunner


def test_runner():
    @click.command()
    def test():
        i = click.get_binary_stream("stdin")
        o = click.get_binary_stream("stdout")
        while True:
            chunk = i.read(4096)
            if not chunk:
                break
            o.write(chunk)
            o.flush()

    runner = CliRunner()
    result = runner.invoke(test, input="Hello World!\n")
    assert not result.exception
    assert result.output == "Hello World!\n"


def test_echo_stdin_stream():
    @click.command()
    def test():
        i = click.get_binary_stream("stdin")
        o = click.get_binary_stream("stdout")
        while True:
            chunk = i.read(4096)
            if not chunk:
                break
            o.write(chunk)
            o.flush()

    runner = CliRunner(echo_stdin=True)
    result = runner.invoke(test, input="Hello World!\n")
    assert not result.exception
    assert result.output == "Hello World!\nHello World!\n"


def test_echo_stdin_prompts():
    @click.command()
    def test_python_input():
        foo = input("Foo: ")
        click.echo(f"foo={foo}")

    runner = CliRunner(echo_stdin=True)
    result = runner.invoke(test_python_input, input="bar bar\n")
    assert not result.exception
    assert result.output == "Foo: bar bar\nfoo=bar bar\n"

    @click.command()
    @click.option("--foo", prompt=True)
    def test_prompt(foo):
        click.echo(f"foo={foo}")

    result = runner.invoke(test_prompt, input="bar bar\n")
    assert not result.exception
    assert result.output == "Foo: bar bar\nfoo=bar bar\n"

    @click.command()
    @click.option("--foo", prompt=True, hide_input=True)
    def test_hidden_prompt(foo):
        click.echo(f"foo={foo}")

    result = runner.invoke(test_hidden_prompt, input="bar bar\n")
    assert not result.exception
    assert result.output == "Foo: \nfoo=bar bar\n"

    @click.command()
    @click.option("--foo", prompt=True)
    @click.option("--bar", prompt=True)
    def test_multiple_prompts(foo, bar):
        click.echo(f"foo={foo}, bar={bar}")

    result = runner.invoke(test_multiple_prompts, input="one\ntwo\n")
    assert not result.exception
    assert result.output == "Foo: one\nBar: two\nfoo=one, bar=two\n"


def test_runner_with_stream():
    @click.command()
    def test():
        i = click.get_binary_stream("stdin")
        o = click.get_binary_stream("stdout")
        while True:
            chunk = i.read(4096)
            if not chunk:
                break
            o.write(chunk)
            o.flush()

    runner = CliRunner()
    result = runner.invoke(test, input=BytesIO(b"Hello World!\n"))
    assert not result.exception
    assert result.output == "Hello World!\n"

    runner = CliRunner(echo_stdin=True)
    result = runner.invoke(test, input=BytesIO(b"Hello World!\n"))
    assert not result.exception
    assert result.output == "Hello World!\nHello World!\n"


def test_prompts():
    @click.command()
    @click.option("--foo", prompt=True)
    def test(foo):
        click.echo(f"foo={foo}")

    runner = CliRunner()
    result = runner.invoke(test, input="wau wau\n")
    assert not result.exception
    assert result.output == "Foo: wau wau\nfoo=wau wau\n"

    @click.command()
    @click.option("--foo", prompt=True, hide_input=True)
    def test(foo):
        click.echo(f"foo={foo}")

    runner = CliRunner()
    result = runner.invoke(test, input="wau wau\n")
    assert not result.exception
    assert result.output == "Foo: \nfoo=wau wau\n"


def test_getchar():
    @click.command()
    def continue_it():
        click.echo(click.getchar())

    runner = CliRunner()
    result = runner.invoke(continue_it, input="y")
    assert not result.exception
    assert result.output == "y\n"

    runner = CliRunner(echo_stdin=True)
    result = runner.invoke(continue_it, input="y")
    assert not result.exception
    assert result.output == "y\n"

    @click.command()
    def getchar_echo():
        click.echo(click.getchar(echo=True))

    runner = CliRunner()
    result = runner.invoke(getchar_echo, input="y")
    assert not result.exception
    assert result.output == "yy\n"

    runner = CliRunner(echo_stdin=True)
    result = runner.invoke(getchar_echo, input="y")
    assert not result.exception
    assert result.output == "yy\n"


def test_catch_exceptions():
    class CustomError(Exception):
        pass

    @click.command()
    def cli():
        raise CustomError(1)

    runner = CliRunner()

    result = runner.invoke(cli)
    assert isinstance(result.exception, CustomError)
    assert type(result.exc_info) is tuple
    assert len(result.exc_info) == 3

    with pytest.raises(CustomError):
        runner.invoke(cli, catch_exceptions=False)

    CustomError = SystemExit

    result = runner.invoke(cli)
    assert result.exit_code == 1


def test_catch_exceptions_cli_runner():
    """Test that invoke `catch_exceptions` takes the value from CliRunner if not set
    explicitly."""

    class CustomError(Exception):
        pass

    @click.command()
    def cli():
        raise CustomError(1)

    runner = CliRunner(catch_exceptions=False)

    result = runner.invoke(cli, catch_exceptions=True)
    assert isinstance(result.exception, CustomError)
    assert type(result.exc_info) is tuple
    assert len(result.exc_info) == 3

    with pytest.raises(CustomError):
        runner.invoke(cli)


def test_with_color():
    @click.command()
    def cli():
        click.secho("hello world", fg="blue")

    runner = CliRunner()

    result = runner.invoke(cli)
    assert result.output == "hello world\n"
    assert not result.exception

    result = runner.invoke(cli, color=True)
    assert result.output == f"{click.style('hello world', fg='blue')}\n"
    assert not result.exception


def test_with_color_errors():
    class CLIError(ClickException):
        def format_message(self) -> str:
            return click.style(self.message, fg="red")

    @click.command()
    def cli():
        raise CLIError("Red error")

    runner = CliRunner()

    result = runner.invoke(cli)
    assert result.output == "Error: Red error\n"
    assert result.exception

    result = runner.invoke(cli, color=True)
    assert result.output == f"Error: {click.style('Red error', fg='red')}\n"
    assert result.exception


def test_with_color_but_pause_not_blocking():
    @click.command()
    def cli():
        click.pause()

    runner = CliRunner()
    result = runner.invoke(cli, color=True)
    assert not result.exception
    assert result.output == ""


def test_with_echo_via_pager():
    @click.command()
    def cli():
        click.echo_via_pager("Hello, Click!")

    runner = CliRunner()
    result = runner.invoke(cli)
    assert not result.exception
    assert result.output == "Hello, Click!\n"


def test_exit_code_and_output_from_sys_exit():
    # See issue #362
    @click.command()
    def cli_string():
        click.echo("hello world")
        sys.exit("error")

    @click.command()
    @click.pass_context
    def cli_string_ctx_exit(ctx):
        click.echo("hello world")
        ctx.exit("error")

    @click.command()
    def cli_int():
        click.echo("hello world")
        sys.exit(1)

    @click.command()
    @click.pass_context
    def cli_int_ctx_exit(ctx):
        click.echo("hello world")
        ctx.exit(1)

    @click.command()
    def cli_float():
        click.echo("hello world")
        sys.exit(1.0)

    @click.command()
    @click.pass_context
    def cli_float_ctx_exit(ctx):
        click.echo("hello world")
        ctx.exit(1.0)

    @click.command()
    def cli_no_error():
        click.echo("hello world")

    runner = CliRunner()

    result = runner.invoke(cli_string)
    assert result.exit_code == 1
    assert result.output == "hello world\nerror\n"

    result = runner.invoke(cli_string_ctx_exit)
    assert result.exit_code == 1
    assert result.output == "hello world\nerror\n"

    result = runner.invoke(cli_int)
    assert result.exit_code == 1
    assert result.output == "hello world\n"

    result = runner.invoke(cli_int_ctx_exit)
    assert result.exit_code == 1
    assert result.output == "hello world\n"

    result = runner.invoke(cli_float)
    assert result.exit_code == 1
    assert result.output == "hello world\n1.0\n"

    result = runner.invoke(cli_float_ctx_exit)
    assert result.exit_code == 1
    assert result.output == "hello world\n1.0\n"

    result = runner.invoke(cli_no_error)
    assert result.exit_code == 0
    assert result.output == "hello world\n"


def test_env():
    @click.command()
    def cli_env():
        click.echo(f"ENV={os.environ['TEST_CLICK_ENV']}")

    runner = CliRunner()

    env_orig = dict(os.environ)
    env = dict(env_orig)
    assert "TEST_CLICK_ENV" not in env
    env["TEST_CLICK_ENV"] = "some_value"
    result = runner.invoke(cli_env, env=env)
    assert result.exit_code == 0
    assert result.output == "ENV=some_value\n"

    assert os.environ == env_orig


def test_stderr():
    @click.command()
    def cli_stderr():
        click.echo("1 - stdout")
        click.echo("2 - stderr", err=True)
        click.echo("3 - stdout")
        click.echo("4 - stderr", err=True)

    runner_mix = CliRunner()
    result_mix = runner_mix.invoke(cli_stderr)

    assert result_mix.output == "1 - stdout\n2 - stderr\n3 - stdout\n4 - stderr\n"
    assert result_mix.stdout == "1 - stdout\n3 - stdout\n"
    assert result_mix.stderr == "2 - stderr\n4 - stderr\n"

    @click.command()
    def cli_empty_stderr():
        click.echo("stdout")

    runner = CliRunner()
    result = runner.invoke(cli_empty_stderr)

    assert result.output == "stdout\n"
    assert result.stdout == "stdout\n"
    assert result.stderr == ""


@pytest.mark.parametrize(
    "args, expected_output",
    [
        (None, "bar\n"),
        ([], "bar\n"),
        ("", "bar\n"),
        (["--foo", "one two"], "one two\n"),
        ('--foo "one two"', "one two\n"),
    ],
)
def test_args(args, expected_output):
    @click.command()
    @click.option("--foo", default="bar")
    def cli_args(foo):
        click.echo(foo)

    runner = CliRunner()
    result = runner.invoke(cli_args, args=args)
    assert result.exit_code == 0
    assert result.output == expected_output


def test_setting_prog_name_in_extra():
    @click.command()
    def cli():
        click.echo("ok")

    runner = CliRunner()
    result = runner.invoke(cli, prog_name="foobar")
    assert not result.exception
    assert result.output == "ok\n"


def test_command_standalone_mode_returns_value():
    @click.command()
    def cli():
        click.echo("ok")
        return "Hello, World!"

    runner = CliRunner()
    result = runner.invoke(cli, standalone_mode=False)
    assert result.output == "ok\n"
    assert result.return_value == "Hello, World!"
    assert result.exit_code == 0


def test_file_stdin_attrs(runner):
    @click.command()
    @click.argument("f", type=click.File())
    def cli(f):
        click.echo(f.name)
        click.echo(f.mode, nl=False)

    result = runner.invoke(cli, ["-"])
    assert result.output == "<stdin>\nr"


def test_isolated_runner(runner):
    with runner.isolated_filesystem() as d:
        assert os.path.exists(d)

    assert not os.path.exists(d)


def test_isolated_runner_custom_tempdir(runner, tmp_path):
    with runner.isolated_filesystem(temp_dir=tmp_path) as d:
        assert os.path.exists(d)

    assert os.path.exists(d)
    os.rmdir(d)


def test_isolation_stderr_errors():
    """Writing to stderr should escape invalid characters instead of
    raising a UnicodeEncodeError.
    """
    runner = CliRunner()

    with runner.isolation() as (_, err, _):
        click.echo("\udce2", err=True, nl=False)
        assert err.getvalue() == b"\\udce2"


def test_isolation_flushes_unflushed_stderr():
    """An un-flushed write to stderr, as with `print(..., file=sys.stderr)`, will end up
    flushed by the runner at end of invocation.
    """
    runner = CliRunner()

    with runner.isolation() as (_, err, _):
        click.echo("\udce2", err=True, nl=False)
        assert err.getvalue() == b"\\udce2"

    @click.command()
    def cli():
        # set end="", flush=False so that it's totally clear that we won't get any
        # auto-flush behaviors
        print("gyarados gyarados gyarados", file=sys.stderr, end="", flush=False)

    result = runner.invoke(cli)
    assert result.stderr == "gyarados gyarados gyarados"


def test_pdb_uses_real_streams():
    """``pdb.Pdb()`` inside ``CliRunner`` defaults to real terminal streams
    so that interactive debuggers work instead of reading from the
    captured ``BytesIO`` stdin.
    """

    @click.command()
    def cli():
        debugger = pdb.Pdb()
        assert debugger.stdin is sys.__stdin__
        assert debugger.stdout is sys.__stdout__
        click.echo("after debugger")

    runner = CliRunner()
    result = runner.invoke(cli, catch_exceptions=False)
    assert result.output == "after debugger\n"


def test_pdb_explicit_streams_honored():
    """Explicit ``stdin``/``stdout`` arguments to ``pdb.Pdb()`` are not
    overridden by the ``CliRunner`` patch.
    """

    @click.command()
    def cli():
        custom_in = sys.stdin
        custom_out = sys.stdout
        debugger = pdb.Pdb(stdin=custom_in, stdout=custom_out)
        assert debugger.stdin is custom_in
        assert debugger.stdout is custom_out

    runner = CliRunner()
    runner.invoke(cli, catch_exceptions=False)


def test_pdb_init_restored_after_invoke():
    """``pdb.Pdb.__init__`` is restored to its original after invoke."""
    original = pdb.Pdb.__init__

    @click.command()
    def cli():
        pass

    runner = CliRunner()
    runner.invoke(cli)

    assert pdb.Pdb.__init__ is original


needs_fd_capture = pytest.mark.skipif(
    sys.platform == "win32",
    reason="fd capture not supported on Windows",
)


def test_capture_invalid_mode():
    """Invalid capture mode raises ValueError."""
    with pytest.raises(ValueError, match="capture"):
        CliRunner(capture="invalid")  # type: ignore[arg-type]


@pytest.mark.skipif(sys.platform != "win32", reason="Windows-only test")
def test_capture_fd_windows_error():
    """fd capture raises ValueError on Windows."""
    with pytest.raises(ValueError, match="not supported on Windows"):
        CliRunner(capture="fd")


@needs_fd_capture
def test_capture_fd_os_write():
    """capture='fd' captures writes to fd 1/2 that bypass sys.stdout."""

    @click.command()
    def cli():
        click.echo("python stdout")
        os.write(1, b"fd stdout\n")
        os.write(2, b"fd stderr\n")

    runner = CliRunner(capture="fd")
    result = runner.invoke(cli)
    assert "python stdout" in result.stdout
    assert "fd stdout" in result.stdout
    assert "fd stderr" in result.stderr


@needs_fd_capture
def test_capture_fd_stale_reference():
    """capture='fd' captures writes from stale stdout references (issue #2874).

    Simulates ``from sys import stdout`` at import time, which grabs
    the real stdout connected to fd 1.  The stale object's underlying
    FileIO uses fd 1, so redirecting fd 1 captures its writes.
    """
    # open(1, ..., closefd=False) creates a writer whose underlying
    # FileIO uses fd 1 directly. This mirrors the real scenario:
    # the original sys.stdout is a TextIOWrapper -> BufferedWriter ->
    # FileIO(fd=1).
    stale_stdout = open(1, "w", closefd=False)  # noqa: SIM115

    @click.command()
    def cli():
        stale_stdout.write("stale write\n")
        stale_stdout.flush()
        click.echo("normal write")

    runner = CliRunner(capture="fd")
    result = runner.invoke(cli)
    assert "normal write" in result.stdout
    assert "stale write" in result.stdout


@needs_fd_capture
def test_capture_fd_logging_handler(tmp_path):
    """capture='fd' captures logging output from a handler holding a stale
    stderr reference (issue #2827).

    stdlib logging.StreamHandler grabs sys.stderr at configuration time.
    Under normal CliRunner (sys-level capture), the handler still writes
    to the original stream object and output is lost. fd-level capture
    redirects the underlying file descriptor, so the writes are captured.
    """
    import logging

    # Create a writer backed by the real fd 2, simulating a handler
    # configured at import time before pytest or CliRunner replaced
    # sys.stderr. open(2, closefd=False) mirrors the real scenario:
    # the original sys.stderr is a TextIOWrapper -> BufferedWriter ->
    # FileIO(fd=2).
    stale_stderr = open(2, "w", closefd=False)  # noqa: SIM115
    handler = logging.StreamHandler(stale_stderr)
    handler.setFormatter(logging.Formatter("%(message)s"))

    logger = logging.getLogger(f"click_test_{tmp_path.name}")
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    logger.propagate = False

    @click.command()
    def cli():
        logger.info("log from stale handler")
        click.echo("normal echo")

    # sys-level capture misses the log line (it bypasses sys.stderr).
    runner_sys = CliRunner(capture="sys")
    result_sys = runner_sys.invoke(cli)
    assert "normal echo" in result_sys.output
    assert "log from stale handler" not in result_sys.output

    # fd-level capture catches it by redirecting fd 2.
    runner_fd = CliRunner(capture="fd")
    result_fd = runner_fd.invoke(cli)
    assert "normal echo" in result_fd.output
    assert "log from stale handler" in result_fd.output

    logger.removeHandler(handler)


@needs_fd_capture
def test_capture_fd_faulthandler():
    """faulthandler.enable() works with capture='fd' (issue #2865)."""

    @click.command()
    def cli():
        faulthandler.enable()
        click.echo("after faulthandler")

    runner = CliRunner(capture="fd")
    result = runner.invoke(cli)
    assert result.exit_code == 0
    assert "after faulthandler" in result.output


def test_capture_sys_fileno_raises():
    """capture='sys' leaves fileno() raising UnsupportedOperation, so user
    code that does ``os.dup2(w, sys.stdout.fileno())`` cannot mutate the host
    runner's stdout (issue #3384).
    """

    @click.command()
    def cli():
        with pytest.raises(io.UnsupportedOperation):
            sys.stdout.fileno()
        with pytest.raises(io.UnsupportedOperation):
            sys.stderr.fileno()
        click.echo("ok")

    runner = CliRunner(capture="sys")
    result = runner.invoke(cli)
    assert result.exit_code == 0, result.output
    assert "ok" in result.stdout


@needs_fd_capture
def test_capture_fd_stderr_separation():
    """capture='fd' properly separates fd-level stdout and stderr."""

    @click.command()
    def cli():
        click.echo("py-out")
        click.echo("py-err", err=True)
        os.write(1, b"fd-out\n")
        os.write(2, b"fd-err\n")

    runner = CliRunner(capture="fd")
    result = runner.invoke(cli)
    assert "py-out" in result.stdout
    assert "fd-out" in result.stdout
    assert "py-err" in result.stderr
    assert "fd-err" in result.stderr
    # Mixed output has all of them
    assert "py-out" in result.output
    assert "py-err" in result.output
    assert "fd-out" in result.output
    assert "fd-err" in result.output


@needs_fd_capture
def test_capture_fd_nesting():
    """Nested CliRunner.invoke() with fd capture works correctly."""

    @click.command("inner")
    def inner_cli():
        click.echo("inner")

    @click.command("outer")
    def outer_cli():
        click.echo("outer")
        os.write(1, b"outer fd\n")
        inner_runner = CliRunner(capture="fd")
        inner_result = inner_runner.invoke(inner_cli)
        click.echo(f"inner captured: {inner_result.stdout.strip()}")

    runner = CliRunner(capture="fd")
    result = runner.invoke(outer_cli)
    assert "outer" in result.stdout
    assert "outer fd" in result.stdout
    assert "inner captured: inner" in result.stdout


@needs_fd_capture
@pytest.mark.parametrize("runner_capture", ["sys", "fd"])
@pytest.mark.parametrize("pytest_fixture", ["capsys", "capfd"])
def test_capture_pytest_matrix(
    request: pytest.FixtureRequest,
    runner_capture: str,
    pytest_fixture: str,
) -> None:
    """Matrix of ``CliRunner`` capture modes and Pytest capture fixtures."""
    pytest_cap = request.getfixturevalue(pytest_fixture)

    @click.command()
    def cli() -> None:
        click.echo("inside-stdout")
        click.echo("inside-stderr", err=True)

    runner = CliRunner(capture=runner_capture)
    result = runner.invoke(cli)

    # CliRunner sees Click's output.
    assert result.exit_code == 0
    assert "inside-stdout" in result.stdout
    assert "inside-stderr" in result.stderr

    # Pytest capture machinery is still working after invoke().
    print("post-invoke-stdout")
    print("post-invoke-stderr", file=sys.stderr)
    captured = pytest_cap.readouterr()
    assert "post-invoke-stdout" in captured.out
    assert "post-invoke-stderr" in captured.err

    # Click output is not leaking into Pytest.
    assert "inside-stdout" not in captured.out
    assert "inside-stderr" not in captured.err