import os
import pathlib
import stat
import subprocess
import sys
from collections import namedtuple
from contextlib import nullcontext
from decimal import Decimal
from fractions import Fraction
from functools import partial
from io import StringIO
from pathlib import Path
from tempfile import tempdir
from unittest.mock import patch

import pytest

import asyncclick as click
import asyncclick._termui_impl
import asyncclick.utils
from asyncclick._compat import WIN
from asyncclick._utils import UNSET


def test_unset_sentinel():
    value = UNSET

    assert value
    assert value is UNSET
    assert value == UNSET
    assert repr(value) == "Sentinel.UNSET"
    assert str(value) == "Sentinel.UNSET"
    assert bool(value) is True

    # Try all native Python values that can be falsy or truthy.
    # See: https://docs.python.org/3/library/stdtypes.html#truth-value-testing
    real_values = (
        None,
        True,
        False,
        0,
        1,
        0.0,
        1.0,
        0j,
        1j,
        Decimal(0),
        Decimal(1),
        Fraction(0, 1),
        Fraction(1, 1),
        "",
        "a",
        "UNSET",
        "Sentinel.UNSET",
        [1],
        (1),
        {1: "a"},
        set(),
        set([1]),
        frozenset(),
        frozenset([1]),
        range(0),
        range(1),
    )

    for real_value in real_values:
        assert value != real_value
        assert value is not real_value

    assert value not in real_values


def test_echo(runner):
    with runner.isolation() as outstreams:
        click.echo("\N{SNOWMAN}")
        click.echo(b"\x44\x44")
        click.echo(42, nl=False)
        click.echo(b"a", nl=False)
        click.echo("\x1b[31mx\x1b[39m", nl=False)
        bytes = outstreams[0].getvalue().replace(b"\r\n", b"\n")
        assert bytes == b"\xe2\x98\x83\nDD\n42ax"

    # if wrapped, we expect bytes to survive.
    @click.command()
    def cli():
        click.echo(b"\xf6")

    result = runner.invoke(cli, [])
    assert result.stdout_bytes == b"\xf6\n"

    # Ensure we do not strip for bytes.
    with runner.isolation() as outstreams:
        click.echo(bytearray(b"\x1b[31mx\x1b[39m"), nl=False)
        assert outstreams[0].getvalue() == b"\x1b[31mx\x1b[39m"


def test_echo_custom_file():
    f = StringIO()
    click.echo("hello", file=f)
    assert f.getvalue() == "hello\n"


def test_echo_no_streams(monkeypatch, runner):
    """echo should not fail when stdout and stderr are None with pythonw on Windows."""
    with runner.isolation():
        sys.stdout = None
        sys.stderr = None
        click.echo("test")
        click.echo("test", err=True)


@pytest.mark.parametrize(
    ("styles", "ref"),
    [
        ({"fg": "black"}, "\x1b[30mx y\x1b[0m"),
        ({"fg": "red"}, "\x1b[31mx y\x1b[0m"),
        ({"fg": "green"}, "\x1b[32mx y\x1b[0m"),
        ({"fg": "yellow"}, "\x1b[33mx y\x1b[0m"),
        ({"fg": "blue"}, "\x1b[34mx y\x1b[0m"),
        ({"fg": "magenta"}, "\x1b[35mx y\x1b[0m"),
        ({"fg": "cyan"}, "\x1b[36mx y\x1b[0m"),
        ({"fg": "white"}, "\x1b[37mx y\x1b[0m"),
        ({"bg": "black"}, "\x1b[40mx y\x1b[0m"),
        ({"bg": "red"}, "\x1b[41mx y\x1b[0m"),
        ({"bg": "green"}, "\x1b[42mx y\x1b[0m"),
        ({"bg": "yellow"}, "\x1b[43mx y\x1b[0m"),
        ({"bg": "blue"}, "\x1b[44mx y\x1b[0m"),
        ({"bg": "magenta"}, "\x1b[45mx y\x1b[0m"),
        ({"bg": "cyan"}, "\x1b[46mx y\x1b[0m"),
        ({"bg": "white"}, "\x1b[47mx y\x1b[0m"),
        ({"bg": 91}, "\x1b[48;5;91mx y\x1b[0m"),
        ({"bg": (135, 0, 175)}, "\x1b[48;2;135;0;175mx y\x1b[0m"),
        ({"bold": True}, "\x1b[1mx y\x1b[0m"),
        ({"dim": True}, "\x1b[2mx y\x1b[0m"),
        ({"underline": True}, "\x1b[4mx y\x1b[0m"),
        ({"overline": True}, "\x1b[53mx y\x1b[0m"),
        ({"italic": True}, "\x1b[3mx y\x1b[0m"),
        ({"blink": True}, "\x1b[5mx y\x1b[0m"),
        ({"reverse": True}, "\x1b[7mx y\x1b[0m"),
        ({"strikethrough": True}, "\x1b[9mx y\x1b[0m"),
        ({"bold": False}, "\x1b[22mx y\x1b[0m"),
        ({"dim": False}, "\x1b[22mx y\x1b[0m"),
        ({"underline": False}, "\x1b[24mx y\x1b[0m"),
        ({"overline": False}, "\x1b[55mx y\x1b[0m"),
        ({"italic": False}, "\x1b[23mx y\x1b[0m"),
        ({"blink": False}, "\x1b[25mx y\x1b[0m"),
        ({"reverse": False}, "\x1b[27mx y\x1b[0m"),
        ({"strikethrough": False}, "\x1b[29mx y\x1b[0m"),
        ({"fg": "black", "reset": False}, "\x1b[30mx y"),
    ],
)
def test_styling(styles, ref):
    assert click.style("x y", **styles) == ref
    assert click.unstyle(ref) == "x y"


@pytest.mark.parametrize(("text", "expect"), [("\x1b[?25lx y\x1b[?25h", "x y")])
def test_unstyle_other_ansi(text, expect):
    assert click.unstyle(text) == expect


def test_filename_formatting():
    assert click.format_filename(b"foo.txt") == "foo.txt"
    assert click.format_filename(b"/x/foo.txt") == "/x/foo.txt"
    assert click.format_filename("/x/foo.txt") == "/x/foo.txt"
    assert click.format_filename("/x/foo.txt", shorten=True) == "foo.txt"
    assert click.format_filename("/x/\ufffd.txt", shorten=True) == "�.txt"


def test_prompts(runner):
    @click.command()
    def test():
        if click.confirm("Foo"):
            click.echo("yes!")
        else:
            click.echo("no :(")

    result = runner.invoke(test, input="y\n")
    assert not result.exception
    assert result.output == "Foo [y/N]: y\nyes!\n"

    result = runner.invoke(test, input="\n")
    assert not result.exception
    assert result.output == "Foo [y/N]: \nno :(\n"

    result = runner.invoke(test, input="n\n")
    assert not result.exception
    assert result.output == "Foo [y/N]: n\nno :(\n"

    @click.command()
    def test_no():
        if click.confirm("Foo", default=True):
            click.echo("yes!")
        else:
            click.echo("no :(")

    result = runner.invoke(test_no, input="y\n")
    assert not result.exception
    assert result.output == "Foo [Y/n]: y\nyes!\n"

    result = runner.invoke(test_no, input="\n")
    assert not result.exception
    assert result.output == "Foo [Y/n]: \nyes!\n"

    result = runner.invoke(test_no, input="n\n")
    assert not result.exception
    assert result.output == "Foo [Y/n]: n\nno :(\n"


def test_confirm_repeat(runner):
    cli = click.Command(
        "cli", params=[click.Option(["--a/--no-a"], default=None, prompt=True)]
    )
    result = runner.invoke(cli, input="\ny\n")
    assert result.output == "A [y/n]: \nError: invalid input\nA [y/n]: y\n"


@pytest.mark.skipif(WIN, reason="Different behavior on windows.")
@pytest.mark.anyio
async def test_prompts_abort(monkeypatch, capsys):
    def f(_):
        raise KeyboardInterrupt()

    monkeypatch.setattr("asyncclick.termui.hidden_prompt_func", f)

    try:
        await click.prompt("Password", hide_input=True)
    except click.Abort:
        click.echo("interrupted")

    out, err = capsys.readouterr()
    assert out == "Password:\ninterrupted\n"


def test_prompts_eof(runner):
    """If too few lines of input are given, prompt should exit, not hang."""

    @click.command
    async def echo():
        for _ in range(3):
            click.echo(await click.prompt("", type=int))

    # only provide two lines of input for three prompts
    result = runner.invoke(echo, input="1\n2\n")
    assert result.exit_code == 1


def _test_gen_func():
    yield "a"
    yield "b"
    yield "c"
    yield "abc"


def _test_gen_func_fails():
    yield "test"
    raise RuntimeError("This is a test.")


def _test_gen_func_echo(file=None):
    yield "test"
    click.echo("hello", file=file)
    yield "test"


def _test_simulate_keyboard_interrupt(file=None):
    yield "output_before_keyboard_interrupt"
    raise KeyboardInterrupt()


EchoViaPagerTest = namedtuple(
    "EchoViaPagerTest",
    (
        "description",
        "test_input",
        "expected_pager",
        "expected_stdout",
        "expected_stderr",
        "expected_error",
    ),
)


@pytest.mark.skipif(WIN, reason="Different behavior on windows.")
@pytest.mark.parametrize(
    "pager_cmd", ["cat", "cat ", " cat ", "less", " less", " less "]
)
@pytest.mark.parametrize(
    "test",
    [
        # We need to pass a parameter function instead of a plain param
        # as pytest.mark.parametrize will reuse the parameters causing the
        # generators to be used up so they will not yield anymore
        EchoViaPagerTest(
            description="Plain string argument",
            test_input=lambda: "just text",
            expected_pager="just text\n",
            expected_stdout="",
            expected_stderr="",
            expected_error=None,
        ),
        EchoViaPagerTest(
            description="Iterable argument",
            test_input=lambda: ["itera", "ble"],
            expected_pager="iterable\n",
            expected_stdout="",
            expected_stderr="",
            expected_error=None,
        ),
        EchoViaPagerTest(
            description="Generator function argument",
            test_input=lambda: _test_gen_func,
            expected_pager="abcabc\n",
            expected_stdout="",
            expected_stderr="",
            expected_error=None,
        ),
        EchoViaPagerTest(
            description="String generator argument",
            test_input=lambda: _test_gen_func(),
            expected_pager="abcabc\n",
            expected_stdout="",
            expected_stderr="",
            expected_error=None,
        ),
        EchoViaPagerTest(
            description="Number generator expression argument",
            test_input=lambda: (c for c in range(6)),
            expected_pager="012345\n",
            expected_stdout="",
            expected_stderr="",
            expected_error=None,
        ),
        EchoViaPagerTest(
            description="Exception in generator function argument",
            test_input=lambda: _test_gen_func_fails,
            # Because generator throws early on, the pager did not have
            # a chance yet to write the file.
            expected_pager="",
            expected_stdout="",
            expected_stderr="",
            expected_error=RuntimeError,
        ),
        EchoViaPagerTest(
            description="Exception in generator argument",
            test_input=lambda: _test_gen_func_fails,
            # Because generator throws early on, the pager did not have a
            # chance yet to write the file.
            expected_pager="",
            expected_stdout="",
            expected_stderr="",
            expected_error=RuntimeError,
        ),
        EchoViaPagerTest(
            description="Keyboard interrupt should not terminate the pager",
            test_input=lambda: _test_simulate_keyboard_interrupt(),
            # Due to the keyboard interrupt during pager execution, click program
            # should abort, but the pager should stay open.
            # This allows users to cancel the program and search in the pager
            # output, before they decide to terminate the pager.
            expected_pager="output_before_keyboard_interrupt",
            expected_stdout="",
            expected_stderr="",
            expected_error=KeyboardInterrupt,
        ),
        EchoViaPagerTest(
            description="Writing to stdout during generator execution",
            test_input=lambda: _test_gen_func_echo(),
            expected_pager="testtest\n",
            expected_stdout="hello\n",
            expected_stderr="",
            expected_error=None,
        ),
        EchoViaPagerTest(
            description="Writing to stderr during generator execution",
            test_input=lambda: _test_gen_func_echo(file=sys.stderr),
            expected_pager="testtest\n",
            expected_stdout="",
            expected_stderr="hello\n",
            expected_error=None,
        ),
    ],
)
def test_echo_via_pager(monkeypatch, capfd, pager_cmd, test):
    monkeypatch.setitem(os.environ, "PAGER", pager_cmd)
    monkeypatch.setattr(asyncclick._termui_impl, "isatty", lambda x: True)

    test_input = test.test_input()
    expected_pager = test.expected_pager
    expected_stdout = test.expected_stdout
    expected_stderr = test.expected_stderr
    expected_error = test.expected_error

    check_raise = pytest.raises(expected_error) if expected_error else nullcontext()

    pager_out_tmp = Path(tempdir) / "pager_out.txt"
    pager_out_tmp.unlink(missing_ok=True)
    with pager_out_tmp.open("w") as f:
        force_subprocess_stdout = patch.object(
            subprocess,
            "Popen",
            partial(subprocess.Popen, stdout=f),
        )
        with force_subprocess_stdout:
            with check_raise:
                click.echo_via_pager(test_input)

    out, err = capfd.readouterr()

    pager = pager_out_tmp.read_text()

    assert pager == expected_pager, (
        f"Unexpected pager output in test case '{test.description}'"
    )
    assert out == expected_stdout, (
        f"Unexpected stdout in test case '{test.description}'"
    )
    assert err == expected_stderr, (
        f"Unexpected stderr in test case '{test.description}'"
    )


def test_echo_color_flag(monkeypatch, capfd):
    isatty = True
    monkeypatch.setattr(click._compat, "isatty", lambda x: isatty)

    text = "foo"
    styled_text = click.style(text, fg="red")
    assert styled_text == "\x1b[31mfoo\x1b[0m"

    click.echo(styled_text, color=False)
    out, err = capfd.readouterr()
    assert out == f"{text}\n"

    click.echo(styled_text, color=True)
    out, err = capfd.readouterr()
    assert out == f"{styled_text}\n"

    isatty = True
    click.echo(styled_text)
    out, err = capfd.readouterr()
    assert out == f"{styled_text}\n"

    isatty = False
    # Faking isatty() is not enough on Windows;
    # the implementation caches the colorama wrapped stream
    # so we have to use a new stream for each test
    stream = StringIO()
    click.echo(styled_text, file=stream)
    assert stream.getvalue() == f"{text}\n"

    stream = StringIO()
    click.echo(styled_text, file=stream, color=True)
    assert stream.getvalue() == f"{styled_text}\n"


@pytest.mark.anyio
async def test_prompt_cast_default(capfd, monkeypatch):
    monkeypatch.setattr(sys, "stdin", StringIO("\n"))
    value = await click.prompt("value", default="100", type=int)
    capfd.readouterr()
    assert isinstance(value, int)


@pytest.mark.skipif(WIN, reason="Test too complex to make work windows.")
@pytest.mark.anyio
async def test_echo_writing_to_standard_error(capfd, monkeypatch):
    def emulate_input(text):
        """Emulate keyboard input."""
        monkeypatch.setattr(sys, "stdin", StringIO(text))

    click.echo("Echo to standard output")
    out, err = capfd.readouterr()
    assert out == "Echo to standard output\n"
    assert err == ""

    click.echo("Echo to standard error", err=True)
    out, err = capfd.readouterr()
    assert out == ""
    assert err == "Echo to standard error\n"

    emulate_input("asdlkj\n")
    await click.prompt("Prompt to stdin")
    out, err = capfd.readouterr()
    assert out == "Prompt to stdin: "
    assert err == ""

    emulate_input("asdlkj\n")
    await click.prompt("Prompt to stderr", err=True)
    out, err = capfd.readouterr()
    assert out == " "
    assert err == "Prompt to stderr:"

    emulate_input("y\n")
    click.confirm("Prompt to stdin")
    out, err = capfd.readouterr()
    assert out == "Prompt to stdin [y/N]: "
    assert err == ""

    emulate_input("y\n")
    click.confirm("Prompt to stderr", err=True)
    out, err = capfd.readouterr()
    assert out == " "
    assert err == "Prompt to stderr [y/N]:"

    monkeypatch.setattr(click.termui, "isatty", lambda x: True)
    monkeypatch.setattr(click.termui, "getchar", lambda: " ")

    click.pause("Pause to stdout")
    out, err = capfd.readouterr()
    assert out == "Pause to stdout\n"
    assert err == ""

    click.pause("Pause to stderr", err=True)
    out, err = capfd.readouterr()
    assert out == ""
    assert err == "Pause to stderr\n"


def test_echo_with_capsys(capsys):
    click.echo("Capture me.")
    out, err = capsys.readouterr()
    assert out == "Capture me.\n"


def test_open_file(runner):
    @click.command()
    @click.argument("filename")
    def cli(filename):
        with click.open_file(filename) as f:
            click.echo(f.read())

        click.echo("meep")

    with runner.isolated_filesystem():
        with open("hello.txt", "w") as f:
            f.write("Cool stuff")

        result = runner.invoke(cli, ["hello.txt"])
        assert result.exception is None
        assert result.output == "Cool stuff\nmeep\n"

        result = runner.invoke(cli, ["-"], input="foobar")
        assert result.exception is None
        assert result.output == "foobar\nmeep\n"


def test_open_file_pathlib_dash(runner):
    @click.command()
    @click.argument(
        "filename", type=click.Path(allow_dash=True, path_type=pathlib.Path)
    )
    def cli(filename):
        click.echo(str(type(filename)))

        with click.open_file(filename) as f:
            click.echo(f.read())

        result = runner.invoke(cli, ["-"], input="value")
        assert result.exception is None
        assert result.output == "pathlib.Path\nvalue\n"


def test_open_file_ignore_errors_stdin(runner):
    @click.command()
    @click.argument("filename")
    def cli(filename):
        with click.open_file(filename, errors="ignore") as f:
            click.echo(f.read())

    result = runner.invoke(cli, ["-"], input=os.urandom(16))
    assert result.exception is None


def test_open_file_respects_ignore(runner):
    with runner.isolated_filesystem():
        with open("test.txt", "w") as f:
            f.write("Hello world!")

        with click.open_file("test.txt", encoding="utf8", errors="ignore") as f:
            assert f.errors == "ignore"


def test_open_file_ignore_invalid_utf8(runner):
    with runner.isolated_filesystem():
        with open("test.txt", "wb") as f:
            f.write(b"\xe2\x28\xa1")

        with click.open_file("test.txt", encoding="utf8", errors="ignore") as f:
            f.read()


def test_open_file_ignore_no_encoding(runner):
    with runner.isolated_filesystem():
        with open("test.bin", "wb") as f:
            f.write(os.urandom(16))

        with click.open_file("test.bin", errors="ignore") as f:
            f.read()


@pytest.mark.skipif(WIN, reason="os.chmod() is not fully supported on Windows.")
@pytest.mark.parametrize("permissions", [0o400, 0o444, 0o600, 0o644])
def test_open_file_atomic_permissions_existing_file(runner, permissions):
    with runner.isolated_filesystem():
        with open("existing.txt", "w") as f:
            f.write("content")
        os.chmod("existing.txt", permissions)

        @click.command()
        @click.argument("filename")
        def cli(filename):
            click.open_file(filename, "w", atomic=True).close()

        result = runner.invoke(cli, ["existing.txt"])
        assert result.exception is None
        assert stat.S_IMODE(os.stat("existing.txt").st_mode) == permissions


@pytest.mark.skipif(WIN, reason="os.stat() is not fully supported on Windows.")
def test_open_file_atomic_permissions_new_file(runner):
    with runner.isolated_filesystem():

        @click.command()
        @click.argument("filename")
        def cli(filename):
            click.open_file(filename, "w", atomic=True).close()

        # Create a test file to get the expected permissions for new files
        # according to the current umask.
        with open("test.txt", "w"):
            pass
        permissions = stat.S_IMODE(os.stat("test.txt").st_mode)

        result = runner.invoke(cli, ["new.txt"])
        assert result.exception is None
        assert stat.S_IMODE(os.stat("new.txt").st_mode) == permissions


def test_iter_keepopenfile(tmpdir):
    expected = list(map(str, range(10)))
    p = tmpdir.mkdir("testdir").join("testfile")
    p.write("\n".join(expected))
    with p.open() as f:
        for e_line, a_line in zip(
            expected, asyncclick.utils.KeepOpenFile(f), strict=False
        ):
            assert e_line == a_line.strip()


def test_iter_lazyfile(tmpdir):
    expected = list(map(str, range(10)))
    p = tmpdir.mkdir("testdir").join("testfile")
    p.write("\n".join(expected))
    with p.open() as f:
        with asyncclick.utils.LazyFile(f.name) as lf:
            for e_line, a_line in zip(expected, lf, strict=False):
                assert e_line == a_line.strip()


class MockMain:
    __slots__ = "__package__"

    def __init__(self, package_name):
        self.__package__ = package_name


@pytest.mark.parametrize(
    ("path", "main", "expected"),
    [
        ("example.py", None, "example.py"),
        (str(pathlib.Path("/foo/bar/example.py")), None, "example.py"),
        ("example", None, "example"),
        (str(pathlib.Path("example/__main__.py")), "example", "python -m example"),
        (str(pathlib.Path("example/cli.py")), "example", "python -m example.cli"),
        (str(pathlib.Path("./example")), "", "example"),
    ],
)
def test_detect_program_name(path, main, expected):
    assert asyncclick.utils._detect_program_name(path, _main=MockMain(main)) == expected


def test_expand_args(monkeypatch):
    user = os.path.expanduser("~")
    assert user in asyncclick.utils._expand_args(["~"])
    monkeypatch.setenv("CLICK_TEST", "hello")
    assert "hello" in asyncclick.utils._expand_args(["$CLICK_TEST"])
    assert "pyproject.toml" in asyncclick.utils._expand_args(["*.toml"])
    assert os.path.join("tests", "conftest.py") in asyncclick.utils._expand_args(
        ["**/conftest.py"]
    )
    assert "*.not-found" in asyncclick.utils._expand_args(["*.not-found"])
    # a bad glob pattern, such as a pytest identifier, should return itself
    assert (
        asyncclick.utils._expand_args(["test.py::test_bad"])[0] == "test.py::test_bad"
    )


@pytest.mark.parametrize(
    ("value", "max_length", "expect"),
    [
        pytest.param("", 10, "", id="empty"),
        pytest.param("123 567 90", 10, "123 567 90", id="equal length, no dot"),
        pytest.param("123 567 9. aaaa bbb", 10, "123 567 9.", id="sentence < max"),
        pytest.param("123 567\n\n 9. aaaa bbb", 10, "123 567", id="paragraph < max"),
        pytest.param("123 567 90123.", 10, "123 567...", id="truncate"),
        pytest.param("123 5678 xxxxxx", 10, "123...", id="length includes suffix"),
        pytest.param(
            "token in ~/.netrc ciao ciao",
            20,
            "token in ~/.netrc...",
            id="ignore dot in word",
        ),
    ],
)
@pytest.mark.parametrize(
    "alter",
    [
        pytest.param(None, id=""),
        pytest.param(
            lambda text: "\n\b\n" + "  ".join(text.split(" ")) + "\n", id="no-wrap mark"
        ),
    ],
)
def test_make_default_short_help(value, max_length, alter, expect):
    assert len(expect) <= max_length

    if alter:
        value = alter(value)

    out = asyncclick.utils.make_default_short_help(value, max_length)
    assert out == expect