import os
import re
import subprocess
import sys
from os.path import dirname, normcase
from pathlib import Path
from shlex import quote
from subprocess import Popen

import pytest

from virtualenv.run import cli_run


class ActivationTester:
    def __init__(self, of_class, session, cmd, activate_script, extension):
        self.of_class = of_class
        self._creator = session.creator
        self._version_cmd = [cmd, "--version"]
        self._invoke_script = [cmd]
        self.activate_script = activate_script
        self.extension = extension
        self.activate_cmd = "source"
        self.deactivate = "deactivate"
        self.pydoc_call = "pydoc -w pydoc_test"
        self.script_encoding = "utf-8"
        self._version = None
        self.unix_line_ending = True

    def get_version(self, raise_on_fail):
        if self._version is None:
            # locally we disable, so that contributors don't need to have everything setup
            try:
                process = Popen(
                    self._version_cmd,
                    universal_newlines=True,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                )
                out, err = process.communicate()
                result = out if out else err
                self._version = result
                return result
            except Exception as exception:
                self._version = exception
                if raise_on_fail:
                    raise
                return RuntimeError(f"{self} is not available due {exception}")
        return self._version

    def __repr__(self):
        return (
            f"{self.__class__.__name__}(\nversion={self._version!r},\ncreator={self._creator},\n"
            f"interpreter={self._creator.interpreter})"
        )

    def __call__(self, monkeypatch, tmp_path):
        activate_script = self._creator.bin_dir / self.activate_script

        # check line endings are correct type
        script_content = activate_script.read_bytes()
        for line in script_content.split(b"\n")[:-1]:
            if self.unix_line_ending:
                assert line == b"" or line[-1] != 13, script_content.decode("utf-8")
            else:
                assert line[-1] == 13, script_content.decode("utf-8")

        test_script = self._generate_test_script(activate_script, tmp_path)
        monkeypatch.chdir(tmp_path)

        monkeypatch.delenv("VIRTUAL_ENV", raising=False)
        invoke, env = self._invoke_script + [str(test_script)], self.env(tmp_path)

        try:
            process = Popen(invoke, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=env)
            raw_, _ = process.communicate()
            raw = raw_.decode()
        except subprocess.CalledProcessError as exception:
            output = exception.output + exception.stderr
            assert not exception.returncode, output  # noqa: PT017
            return

        out = re.sub(r"pydev debugger: process \d+ is connecting\n\n", "", raw, re.M).strip().splitlines()
        self.assert_output(out, raw, tmp_path)
        return env, activate_script

    def non_source_activate(self, activate_script):
        return self._invoke_script + [str(activate_script)]

    # noinspection PyMethodMayBeStatic
    def env(self, tmp_path):  # noqa: U100
        env = os.environ.copy()
        # add the current python executable folder to the path so we already have another python on the path
        # also keep the path so the shells (fish, bash, etc can be discovered)
        env["PYTHONIOENCODING"] = "utf-8"
        env["PATH"] = os.pathsep.join([dirname(sys.executable)] + env.get("PATH", "").split(os.pathsep))
        # clear up some environment variables so they don't affect the tests
        for key in [k for k in env.keys() if k.startswith("_OLD") or k.startswith("VIRTUALENV_")]:
            del env[key]
        return env

    def _generate_test_script(self, activate_script, tmp_path):
        commands = self._get_test_lines(activate_script)
        script = os.linesep.join(commands)
        test_script = tmp_path / f"script.{self.extension}"
        with test_script.open("wb") as file_handler:
            file_handler.write(script.encode(self.script_encoding))
        return test_script

    def _get_test_lines(self, activate_script):
        commands = [
            self.print_python_exe(),
            self.print_os_env_var("VIRTUAL_ENV"),
            self.activate_call(activate_script),
            self.print_python_exe(),
            self.print_os_env_var("VIRTUAL_ENV"),
            self.print_prompt(),
            # \\ loads documentation from the virtualenv site packages
            self.pydoc_call,
            self.deactivate,
            self.print_python_exe(),
            self.print_os_env_var("VIRTUAL_ENV"),
            "",  # just finish with an empty new line
        ]
        return commands

    def assert_output(self, out, raw, tmp_path):
        # pre-activation
        assert out[0], raw
        assert out[1] == "None", raw
        # post-activation
        expected = self._creator.exe.parent / os.path.basename(sys.executable)
        assert self.norm_path(out[2]) == self.norm_path(expected), raw
        assert self.norm_path(out[3]) == self.norm_path(self._creator.dest).replace("\\\\", "\\"), raw
        # Some attempts to test the prompt output print more than 1 line.
        # So we need to check if the prompt exists on any of them.
        prompt_text = f"({self._creator.env_name}) "
        assert any(prompt_text in line for line in out[4:-3]), raw

        assert out[-3] == "wrote pydoc_test.html", raw
        content = tmp_path / "pydoc_test.html"
        assert content.exists(), raw
        # post deactivation, same as before
        assert out[-2] == out[0], raw
        assert out[-1] == "None", raw

    def quote(self, s):
        return quote(s)

    def python_cmd(self, cmd):
        return f"{os.path.basename(sys.executable)} -c {self.quote(cmd)}"

    def print_python_exe(self):
        return self.python_cmd("import sys; print(sys.executable)")

    def print_os_env_var(self, var):
        val = f'"{var}"'
        return self.python_cmd(f"import os; import sys; v = os.environ.get({val}); print(v)")

    def print_prompt(self):
        return NotImplemented

    def activate_call(self, script):
        cmd = self.quote(str(self.activate_cmd))
        scr = self.quote(str(script))
        return f"{cmd} {scr}".strip()

    @staticmethod
    def norm_path(path):
        # python may return Windows short paths, normalize
        if not isinstance(path, Path):
            path = Path(path)
        path = str(path.resolve())
        if sys.platform != "win32":
            result = path
        else:
            from ctypes import create_unicode_buffer, windll

            buffer_cont = create_unicode_buffer(256)
            get_long_path_name = windll.kernel32.GetLongPathNameW
            get_long_path_name(str(path), buffer_cont, 256)
            result = buffer_cont.value or path
        return normcase(result)


class RaiseOnNonSourceCall(ActivationTester):
    def __init__(self, of_class, session, cmd, activate_script, extension, non_source_fail_message):
        super().__init__(of_class, session, cmd, activate_script, extension)
        self.non_source_fail_message = non_source_fail_message

    def __call__(self, monkeypatch, tmp_path):
        env, activate_script = super().__call__(monkeypatch, tmp_path)
        process = Popen(
            self.non_source_activate(activate_script),
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            env=env,
        )
        out, _err = process.communicate()
        err = _err.decode("utf-8")
        assert process.returncode
        assert self.non_source_fail_message in err


@pytest.fixture(scope="session")
def activation_tester_class():
    return ActivationTester


@pytest.fixture(scope="session")
def raise_on_non_source_class():
    return RaiseOnNonSourceCall


@pytest.fixture(scope="session", params=[True, False], ids=["with_prompt", "no_prompt"])
def activation_python(request, tmp_path_factory, special_char_name, current_fastest):
    dest = os.path.join(str(tmp_path_factory.mktemp("activation-tester-env")), special_char_name)
    cmd = ["--without-pip", dest, "--creator", current_fastest, "-vv", "--no-periodic-update"]
    if request.param:
        cmd += ["--prompt", special_char_name]
    session = cli_run(cmd)
    pydoc_test = session.creator.purelib / "pydoc_test.py"
    pydoc_test.write_text('"""This is pydoc_test.py"""')
    return session


@pytest.fixture()
def activation_tester(activation_python, monkeypatch, tmp_path, is_inside_ci):
    def _tester(tester_class):
        tester = tester_class(activation_python)
        if not tester.of_class.supports(activation_python.creator.interpreter):
            pytest.skip(f"{tester.of_class.__name__} not supported")
        version = tester.get_version(raise_on_fail=is_inside_ci)
        if not isinstance(version, str):
            pytest.skip(msg=str(version))
        return tester(monkeypatch, tmp_path)

    return _tester