from __future__ import annotations

import json
import os
import platform
import shutil
import sys
import tarfile
import time
import zipfile
from datetime import datetime
from tempfile import TemporaryDirectory
from pathlib import Path

import pytest

import conda_package_handling
import conda_package_handling.tarball
from conda_package_handling import api, exceptions


this_dir = os.path.dirname(__file__)
data_dir = os.path.join(this_dir, "data")
version_file = Path(this_dir).parent / "src" / "conda_package_handling" / "__init__.py"
test_package_name = "mock-2.0.0-py37_1000"
test_package_name_2 = "cph_test_data-0.0.1-0"


def _write_package_dir(prefix: Path | str, files: dict[str, str]):
    """Write a package directory tree under ``prefix``.

    ``files`` maps archive-relative paths to file contents. Parent directories
    are created as needed.
    """
    root = Path(prefix)
    for relpath, content in files.items():
        path = root / relpath
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(content, encoding="utf-8")


def _component_member_paths(
    extract_dir: Path, conda_path: Path | str, component: str
) -> set[str]:
    """Return member paths from one .conda inner tarball.

    Extracts only ``component`` (``"info"`` or ``"pkg"``) from ``conda_path``
    into ``extract_dir`` and returns archive-relative paths of regular files.
    """
    api.extract(str(conda_path), extract_dir, components=component)
    return {
        path.relative_to(extract_dir).as_posix()
        for path in extract_dir.rglob("*")
        if path.is_file()
    }


@pytest.mark.skipif(
    bool(os.environ.get("GITHUB_ACTIONS", False)), reason="Fails on GitHub Actions"
)
@pytest.mark.skipif(not version_file.exists(), reason=f"Could not find {version_file}")
def test_correct_version():
    """
    Prevent accidentally running tests against a globally installed different version.
    """
    assert conda_package_handling.__version__ in version_file.read_text()


def test_api_extract_tarball_implicit_path(testing_workdir):
    tarfile = os.path.join(data_dir, test_package_name + ".tar.bz2")
    local_tarfile = os.path.join(testing_workdir, os.path.basename(tarfile))
    shutil.copy2(tarfile, local_tarfile)
    api.extract(local_tarfile)
    assert os.path.isfile(os.path.join(testing_workdir, test_package_name, "info", "index.json"))


def test_api_tarball_details(testing_workdir):
    tarfile = os.path.join(data_dir, test_package_name + ".tar.bz2")
    results = api.get_pkg_details(tarfile)
    assert results["size"] == 106576
    assert results["md5"] == "0f9cce120a73803a70abb14bd4d4900b"
    assert results["sha256"] == "34c659b0fdc53d28ae721fd5717446fb8abebb1016794bd61e25937853f4c29c"


def test_api_conda_v2_details(testing_workdir):
    condafile = os.path.join(data_dir, test_package_name + ".conda")
    results = api.get_pkg_details(condafile)
    assert results["size"] == 113421
    assert results["sha256"] == "181ec44eb7b06ebb833eae845bcc466ad96474be1f33ee55cab7ac1b0fdbbfa3"
    assert results["md5"] == "23c226430e35a3bd994db6c36b9ac8ae"


def test_api_extract_tarball_explicit_path(testing_workdir):
    tarfile = os.path.join(data_dir, test_package_name + ".tar.bz2")
    local_tarfile = os.path.join(testing_workdir, os.path.basename(tarfile))
    shutil.copy2(tarfile, local_tarfile)

    api.extract(local_tarfile, "manual_path")
    assert os.path.isfile(os.path.join(testing_workdir, "manual_path", "info", "index.json"))


def test_api_extract_conda_v2_implicit_path(testing_workdir):
    condafile = os.path.join(data_dir, test_package_name + ".conda")
    local_condafile = os.path.join(testing_workdir, os.path.basename(condafile))
    shutil.copy2(condafile, local_condafile)
    api.extract(local_condafile)
    assert os.path.isfile(os.path.join(testing_workdir, test_package_name, "info", "index.json"))


def test_api_extract_conda_v2_no_destdir_relative_path(testing_workdir):
    cwd = os.getcwd()
    os.chdir(testing_workdir)
    try:
        condafile = os.path.join(data_dir, test_package_name + ".conda")
        local_condafile = os.path.join(testing_workdir, os.path.basename(condafile))
        shutil.copy2(condafile, local_condafile)

        condafile = os.path.basename(local_condafile)
        assert os.path.exists(condafile)
        # cli passes dest=None, prefix=None
        api.extract(condafile, None, prefix=None)
    finally:
        os.chdir(cwd)


def test_api_extract_conda_v2_explicit_path(testing_workdir):
    condafile = os.path.join(data_dir, test_package_name + ".conda")
    local_condafile = os.path.join(testing_workdir, os.path.basename(condafile))
    shutil.copy2(condafile, local_condafile)

    api.extract(condafile, "manual_path")
    assert os.path.isfile(os.path.join(testing_workdir, "manual_path", "info", "index.json"))


def test_api_extract_conda_v2_explicit_path_prefix(testing_workdir):
    tarfile = os.path.join(data_dir, test_package_name + ".conda")
    api.extract(tarfile, prefix=os.path.join(testing_workdir, "folder"))
    assert os.path.isfile(
        os.path.join(testing_workdir, "folder", test_package_name, "info", "index.json")
    )

    api.extract(tarfile, dest_dir="steve", prefix=os.path.join(testing_workdir, "folder"))
    assert os.path.isfile(os.path.join(testing_workdir, "folder", "steve", "info", "index.json"))


def test_api_extract_dest_dir_and_prefix_both_abs_raises():
    tarfile = os.path.join(data_dir, test_package_name + ".conda")
    with pytest.raises(ValueError):
        api.extract(tarfile, prefix=os.path.dirname(tarfile), dest_dir=os.path.dirname(tarfile))


def test_api_extract_info_conda_v2(testing_workdir):
    condafile = os.path.join(data_dir, test_package_name + ".conda")
    local_condafile = os.path.join(testing_workdir, os.path.basename(condafile))
    shutil.copy2(condafile, local_condafile)
    api.extract(local_condafile, "manual_path", components="info")
    assert os.path.isfile(os.path.join(testing_workdir, "manual_path", "info", "index.json"))
    assert not os.path.isdir(os.path.join(testing_workdir, "manual_path", "lib"))


def check_conda_v2_metadata(condafile):
    with zipfile.ZipFile(condafile) as zf:
        d = json.loads(zf.read("metadata.json"))
    assert d["conda_pkg_format_version"] == 2


def test_api_transmute_tarball_to_conda_v2(testing_workdir):
    tarfile = os.path.join(data_dir, test_package_name + ".tar.bz2")
    # lower compress level makes the test run much faster, even 15 is much
    # better than 22
    errors = api.transmute(tarfile, ".conda", testing_workdir, zstd_compress_level=3)
    assert not errors
    condafile = os.path.join(testing_workdir, test_package_name + ".conda")
    assert os.path.isfile(condafile)
    check_conda_v2_metadata(condafile)


def test_api_transmute_tarball_info_sorts_first(testing_workdir):
    test_packages = [test_package_name]
    test_packages_with_symlinks = [test_package_name_2]
    if sys.platform != "win32":
        test_packages += test_packages_with_symlinks
    for test_package in test_packages:
        test_file = os.path.join(data_dir, test_package + ".tar.bz2")

        # transmute/convert doesn't re-sort files; extract to folder.
        api.extract(test_file, testing_workdir)
        out_fn = os.path.join(testing_workdir, test_package + ".tar.bz2")
        out = api.create(testing_workdir, None, out_fn)
        assert out == out_fn

        # info must be first
        with tarfile.open(out_fn, "r:bz2") as repacked:
            info_seen = False
            not_info_seen = False
            for member in repacked:
                if member.name.startswith("info"):
                    assert not_info_seen is False, (
                        f"{test_package} package info/ must sort first, "
                    )
                    f"but {[m.name for m in repacked.getmembers()]}"
                    info_seen = True
                else:
                    not_info_seen = True
            assert info_seen, "package had no info/ files"


@pytest.mark.skipif(sys.platform == "win32", reason="windows and symlinks are not great")
def test_api_transmute_to_conda_v2_contents(testing_workdir):
    def _walk(path):
        for entry in os.scandir(path):
            if entry.is_dir(follow_symlinks=False):
                yield from _walk(entry.path)
                continue
            yield entry

    tar_path = os.path.join(data_dir, test_package_name_2 + ".tar.bz2")
    conda_path = os.path.join(testing_workdir, test_package_name_2 + ".conda")
    api.transmute(tar_path, ".conda", testing_workdir, zstd_compress_level=3)

    # Verify original contents were all put in the right place
    pkg_tarbz2 = tarfile.open(tar_path, mode="r:bz2")
    info_items = [item for item in pkg_tarbz2.getmembers() if item.path.startswith("info/")]
    pkg_items = [item for item in pkg_tarbz2.getmembers() if not item.path.startswith("info/")]

    errors = []
    for component, expected in (("info", info_items), ("pkg", pkg_items)):
        with TemporaryDirectory() as root:
            api.extract(conda_path, root, components=component)

            contents = {
                os.path.relpath(entry.path, root): {
                    "is_symlink": entry.is_symlink(),
                    "target": os.readlink(entry.path) if entry.is_symlink() else None,
                }
                for entry in _walk(root)
            }

            for item in expected:
                if item.path not in contents:
                    errors.append(f"'{item.path}' not found in {component} contents")
                    continue

                ct = contents.pop(item.path)
                if item.issym():
                    if not ct["is_symlink"] or ct["target"] != item.linkname:
                        errors.append(
                            f"{item.name} -> {item.linkname} incorrect in {component} contents"
                        )
                elif not item.isfile():
                    # Raise an exception rather than appending to `errors`
                    # because getting to this point is an indication that our
                    # test data (i.e., .tar.bz2 package) is corrupt, rather
                    # than the `.transmute` function having problems (which is
                    # what `errors` is meant to track).  For context, conda
                    # packages should only contain regular files and symlinks.
                    raise ValueError(f"unexpected item '{item.path}' in test .tar.bz2")
            if contents:
                errors.append(f"extra files [{', '.join(contents)}] in {component} contents")
    assert not errors


@pytest.mark.parametrize("via", ["create", "transmute"])
def test_api_cep35_info_placement(tmp_path: Path, via):
    package_name = "cep35-test-1.0.0-0"
    prefix = tmp_path / "prefix"
    files = {
        "info/licenses/LICENSE": "MIT License\n",
        "bin/cep35-test": "#!/bin/sh\n",
        "info/index.json": json.dumps(
            {
                "arch": "x86_64",
                "build": "0",
                "build_number": 0,
                "name": "cep35-test",
                "platform": "linux",
                "version": "1.0.0",
            }
        ),
    }
    _write_package_dir(prefix, files)

    if via == "create":
        conda_path = tmp_path / f"{package_name}.conda"
        api.create(str(prefix), None, f"{package_name}.conda", out_folder=str(tmp_path))
    else:
        tar_path = tmp_path / f"{package_name}.tar.bz2"
        api.create(str(prefix), None, f"{package_name}.tar.bz2", out_folder=str(tmp_path))
        errors = api.transmute(str(tar_path), ".conda", str(tmp_path), zstd_compress_level=3)
        assert not errors
        conda_path = tmp_path / f"{package_name}.conda"

    info_paths = _component_member_paths(tmp_path / "info", conda_path, "info")
    pkg_paths = _component_member_paths(tmp_path / "pkg", conda_path, "pkg")

    errors = []
    for path in files:
        if path.startswith("info/"):
            if path in pkg_paths:
                errors.append(f"{path} incorrectly placed in pkg-*")
            elif path not in info_paths:
                errors.append(f"{path} missing from info-*")
        elif path in info_paths:
            errors.append(f"{path} incorrectly placed in info-*")
        elif path not in pkg_paths:
            errors.append(f"{path} missing from pkg-*")

    assert not errors, "CEP 35 component split failed:\n" + "\n".join(errors)


def test_api_transmute_conda_v2_to_tarball(testing_workdir):
    condafile = os.path.join(data_dir, test_package_name + ".conda")
    outfile = Path(testing_workdir, test_package_name + ".tar.bz2")
    # one quiet=True in the test suite for coverage
    api.transmute(condafile, ".tar.bz2", testing_workdir, quiet=True)
    assert outfile.is_file()

    # test that no-force keeps file, and force overwrites file
    for force in False, True:
        mtime = outfile.stat().st_mtime
        time.sleep(2 if platform.platform() == "Windows" else 0)
        api.transmute(condafile, ".tar.bz2", testing_workdir, force=force)
        mtime2 = outfile.stat().st_mtime
        assert (mtime2 == mtime) != force


def test_warning_when_bundling_no_metadata(testing_workdir):
    pass


@pytest.mark.skipif(sys.platform == "win32", reason="windows and symlinks are not great")
def test_create_package_with_uncommon_conditions_captures_all_content(testing_workdir):
    os.makedirs("src/a_folder")
    os.makedirs("src/empty_folder")
    os.makedirs("src/symlink_stuff")
    with open("src/a_folder/text_file", "w") as f:
        f.write("weee")
    open("src/empty_file", "w").close()
    os.link("src/a_folder/text_file", "src/a_folder/hardlink_to_text_file")
    os.symlink("../a_folder", "src/symlink_stuff/symlink_to_a")
    os.symlink("../empty_file", "src/symlink_stuff/symlink_to_empty_file")
    os.symlink("../a_folder/text_file", "src/symlink_stuff/symlink_to_text_file")

    with tarfile.open("pinkie.tar.bz2", "w:bz2") as tf:

        def add(source, target):
            tf.add(source, target, recursive=False)

        add("src/empty_folder", "empty_folder")
        add("src/empty_file", "empty_file")
        add("src/a_folder", "a_folder")
        add("src/a_folder/text_file", "a_folder/text_file")
        add("src/a_folder/hardlink_to_text_file", "a_folder/hardlink_to_text_file")
        add("src/symlink_stuff/symlink_to_a", "symlink_stuff/symlink_to_a")
        add(
            "src/symlink_stuff/symlink_to_empty_file",
            "symlink_stuff/symlink_to_empty_file",
        )
        add(
            "src/symlink_stuff/symlink_to_text_file",
            "symlink_stuff/symlink_to_text_file",
        )

    api.create("src", None, "thebrain.tar.bz2")

    # test against both archives created manually and those created by cph.
    # They should be equal in all ways.
    for fn in ("pinkie.tar.bz2", "thebrain.tar.bz2"):
        api.extract(fn)
        target_dir = fn[:-8]
        flist = [
            "empty_folder",
            "empty_file",
            "a_folder/text_file",
            "a_folder/hardlink_to_text_file",
            "symlink_stuff/symlink_to_a",
            "symlink_stuff/symlink_to_text_file",
            "symlink_stuff/symlink_to_empty_file",
        ]

        # no symlinks on windows
        if sys.platform != "win32":
            # not directly included but checked symlink
            flist.append("symlink_stuff/symlink_to_a/text_file")

        missing_content = []
        for f in flist:
            path_that_should_be_there = os.path.join(testing_workdir, target_dir, f)
            if not (
                os.path.exists(path_that_should_be_there)
                or os.path.lexists(path_that_should_be_there)  # noqa
            ):
                missing_content.append(f)
        if missing_content:
            print("missing files in output package")
            print(missing_content)
            sys.exit(1)

        # hardlinks should be preserved, but they're currently not with libarchive
        # hardlinked_file = os.path.join(testing_workdir, target_dir, 'a_folder/text_file')
        # stat = os.stat(hardlinked_file)
        # assert stat.st_nlink == 2

        hardlinked_file = os.path.join(testing_workdir, target_dir, "empty_file")
        stat = os.stat(hardlinked_file)
        if sys.platform != "win32":
            assert stat.st_nlink == 1


@pytest.mark.skipif(
    datetime.now() <= datetime(2020, 12, 1),
    reason="Don't understand why this doesn't behave.  Punt.",
)
def test_secure_refusal_to_extract_abs_paths(testing_workdir):
    with tarfile.open("pinkie.tar.bz2", "w:bz2") as tf:
        open("thebrain", "w").close()
        tf.add(os.path.join(testing_workdir, "thebrain"), "/naughty/abs_path")
        try:
            tf.getmember("/naughty/abs_path")
        except KeyError:
            pytest.skip("Tar implementation does not generate unsafe paths in archive.")

    with pytest.raises(api.InvalidArchiveError):
        api.extract("pinkie.tar.bz2")


def tests_secure_refusal_to_extract_dotdot(testing_workdir):
    with tarfile.open("pinkie.tar.bz2", "w:bz2") as tf:
        open("thebrain", "w").close()
        tf.add(os.path.join(testing_workdir, "thebrain"), "../naughty/abs_path")

    with pytest.raises(api.InvalidArchiveError):
        api.extract("pinkie.tar.bz2")


def test_api_bad_filename(testing_workdir):
    with pytest.raises(ValueError):
        api.extract("pinkie.rar", testing_workdir)


def test_details_bad_extension():
    with pytest.raises(ValueError):
        # TODO this function should not exist
        api.get_pkg_details("pinkie.rar")


def test_convert_bad_extension(testing_workdir):
    api._convert("pinkie.rar", ".conda", testing_workdir)


def test_convert_keyerror(tmpdir, mocker):
    tarfile = os.path.join(data_dir, test_package_name + ".tar.bz2")

    mocker.patch(
        "conda_package_streaming.transmute.transmute",
        side_effect=KeyboardInterrupt(),
    )

    # interrupted before ".conda" was created
    with pytest.raises(KeyboardInterrupt):
        api._convert(tarfile, ".conda", tmpdir)

    def create_file_and_raise(*args, **kwargs):
        out_fn = Path(tmpdir, Path(tarfile[: -len(".tar.bz2")] + ".conda").name)
        print("out fn", out_fn)
        out_fn.write_text("")
        raise KeyboardInterrupt()

    mocker.patch("conda_package_streaming.transmute.transmute", side_effect=create_file_and_raise)

    # interrupted after ".conda" was created
    with pytest.raises(KeyboardInterrupt):
        api._convert(tarfile, ".conda", tmpdir)


def test_create_filelist(tmpdir, mocker):
    # another bad API, tested for coverage
    filelist = Path(tmpdir, "filelist.txt")
    filelist.write_text("\n".join(["filelist.txt", "anotherfile"]))

    # when looking for filelist-not-found.txt
    with pytest.raises(FileNotFoundError):
        api.create(str(tmpdir), "filelist-not-found.txt", str(tmpdir / "newconda.conda"))

    # when adding anotherfile
    with pytest.raises(FileNotFoundError):
        api.create(str(tmpdir), str(filelist), str(tmpdir / "newconda.conda"))

    # unrecognized target extension
    with pytest.raises(ValueError):
        api.create(str(tmpdir), str(filelist), str(tmpdir / "newpackage.rar"))

    def create_file_and_raise(prefix, file_list, out_fn, *args, **kwargs):
        Path(prefix, out_fn).write_text("")
        raise KeyboardInterrupt()

    mocker.patch(
        "conda_package_handling.conda_fmt.CondaFormat_v2.create",
        side_effect=create_file_and_raise,
    )

    # failure inside inner create()
    with pytest.raises(KeyboardInterrupt):
        api.create(str(tmpdir), str(filelist), str(tmpdir / "newpackage.conda"))


def test_api_transmute_fail_validation(tmpdir, mocker):
    package = os.path.join(data_dir, test_package_name + ".conda")

    # this code is only called for .conda -> .tar.bz2; a streaming validate for
    # .tar.bz2 -> .conda would be a good idea.
    mocker.patch(
        "conda_package_handling.validate.validate_converted_files_match_streaming",
        return_value=(str(package), {"missing-file.txt"}, {"mismatched-size.txt"}),
    )

    errors = api.transmute(package, ".tar.bz2", tmpdir)
    assert errors


def test_api_transmute_fail_validation_to_conda(tmpdir, mocker):
    package = os.path.join(data_dir, test_package_name + ".tar.bz2")

    mocker.patch(
        "conda_package_handling.validate.validate_converted_files_match_streaming",
        return_value=(str(package), {"missing-file.txt"}, {"mismatched-size.txt"}),
    )

    errors = api.transmute(package, ".conda", tmpdir, zstd_compress_level=3)
    assert errors


def test_api_transmute_fail_validation_2(tmpdir, mocker):
    package = os.path.join(data_dir, test_package_name + ".conda")
    tmptarfile = tmpdir / Path(package).name
    shutil.copy(package, tmptarfile)

    mocker.patch(
        "conda_package_handling.validate.validate_converted_files_match_streaming",
        side_effect=Exception("not today"),
    )

    # run with out_folder=None
    errors = api.transmute(str(tmptarfile), ".tar.bz2")
    assert errors


def test_api_translates_exception(mocker, tmpdir):
    from conda_package_streaming.extract import exceptions as cps_exceptions

    tarfile = os.path.join(data_dir, test_package_name + ".tar.bz2")

    # translates their exception to our exception of the same name
    mocker.patch(
        "conda_package_streaming.package_streaming.stream_conda_component",
        side_effect=cps_exceptions.CaseInsensitiveFileSystemError(),
    )

    # should this be exported from the api or inherit from InvalidArchiveError?
    with pytest.raises(exceptions.CaseInsensitiveFileSystemError):
        api.extract(tarfile, tmpdir)