import os
import sys
from pathlib import Path

import pytest

from clarabot_nano.utility.path.chroot import (
    ChrootPath,
    ResolvedRootConsistencyError,
    CycledPathError,
    PartialNodeResolutionError,
)
from clarabot_nano.utility.path.error import coerce_platform_os_error
from clarabot_nano.utility.path.resolve import resolve_path
from clarabot_nano.utility.path.sensitive import SensitiveValidationError
from clarabot_nano.utility.path.validate import ValidationError, normalize_absolute_path
from clarabot_nano_tests.testing.file import temp_path
from clarabot_nano_tests.utility.path.utility import ComplexStruct

DUAL_BIN_PROD = [(True, True), (True, False), (False, True), (False, False)]


def test_utility_path_chroot_path_relative_start(tmpdir):
    _r = Path(normalize_absolute_path(temp_path(tmpdir)))
    _r.mkdir()
    sd = _r / "sd"
    sd.mkdir()
    cr = ChrootPath(_r)
    assert cr.path("") == _r
    assert cr.path(".") == _r
    assert cr.path("./") == _r
    assert cr.path("sd") == sd
    assert cr.path("./sd") == sd
    # test root acceptance and resolution with sensitivity
    roots = ["C:/", "C:\\", "c:/", "c:\\"] if sys.platform.startswith("win") else ["/"]
    lnr = _r / "ln2r"
    os.symlink(roots[0], lnr, target_is_directory=True)  # link to sensitive path
    roots.append(str(lnr))
    roots = set(roots)
    roots.update(normalize_absolute_path(r) for r in list(roots))
    for r in roots:
        r_sd = "Windows" if sys.platform.startswith("win") else "bin"
        r_sd_p = resolve_path(Path(normalize_absolute_path(r), r_sd))
        with pytest.raises(SensitiveValidationError):
            ChrootPath(r)
        cr = ChrootPath(r, restrict_sensitive=False)
        r = resolve_path(Path(normalize_absolute_path(r)))
        assert cr.path("") == r
        assert cr.path(".") == r
        assert cr.path("./") == r
        assert cr.path(r_sd) == r_sd_p
        assert cr.path(f"./{r_sd}") == r_sd_p


def test_utility_path_chroot_path_absolute_rel(tmpdir):
    _r = Path(normalize_absolute_path(temp_path(tmpdir)))
    _r.mkdir()
    cr = ChrootPath(_r)
    sps = (
        [
            "C:/",
            "C:\\",
            "/bin",
            "\\\\localhost",
            "\\\\?\\c:\\Win",
            "\\\\.\\Volume{b75e2c83-0000-0000-0000-602f00000000}\\Test\\Foo.txt",
            "\\\\?\\Volume{b75e2c83-0000-0000-0000-602f00000000}\\Test\\Foo.txt",
            "\\\\?\\server1\\e:\\utilities\\\\filecomparer\\",
            "\\Windows",
            "\\\\Server\\Share\\file",
            "\\\\system07\\C$\\",
        ]
        if sys.platform.startswith("win")
        else [
            "/",
            "/bin",
        ]
    )
    for sp in sps:
        with pytest.raises(ValidationError, match="Path cannot be absolute|Name contains reserved characters"):
            cr.path(sp)


@pytest.fixture
def root_struct(tmpdir):
    _r = Path(normalize_absolute_path(temp_path(tmpdir)))
    _r.mkdir()
    ds = [_r / f"d{i}" for i in range(5)]
    for d in ds:
        d.mkdir()
        (d / "mark").write_text(d.name)
    fs = [_r / f"f{i}" for i in range(5)]
    for f in fs:
        f.write_text(f.name)
    lna = _r / "lna"
    lnb = _r / "lnb"
    lnr = _r / "lnr"
    os.symlink(ds[0], lna, target_is_directory=True)  # link to dir0
    os.symlink(lna, lnb)  # link to lna
    os.symlink(_r, lnr, target_is_directory=True)  # link to main root
    return _r, ds, fs, lna, lnb, lnr


def test_utility_path_chroot_root_setup(root_struct):
    r, ds, fs, lna, lnb, lnr = root_struct
    # test root setup on real dirs
    for d in ds:
        cr = ChrootPath(d)
        assert cr.root == d
        assert cr.path("mark").read_text() == d.name
        assert len(ps := cr.list(".")) == 1 and ps[0].read_text() == d.name
    # test root setup on real files
    for f in fs:
        with pytest.raises(NotADirectoryError):
            ChrootPath(f)
    # test root setup on link dirs
    for ld in [lna, lnb]:
        d = resolve_path(ld, True)
        cr = ChrootPath(ld)
        assert cr.root == ds[0]
        assert cr.path("mark").read_text() == d.name
        assert len(ps := cr.list(".")) == 1 and ps[0].read_text() == d.name


def test_utility_path_chroot_root_moves_real(root_struct):
    r, ds, fs, lna, lnb, lnr = root_struct
    # test real dir/file moves
    cr = ChrootPath(ds[3])
    assert cr.path("mark").read_text() == ds[3].name
    assert cr.path("") == ds[3]
    assert cr.path(".") == ds[3]
    ds[3].rename(r / "dx")  # root dir is (re)moved
    for rel in ["mark", "", "."]:
        with pytest.raises(ResolvedRootConsistencyError):
            cr.path(rel)
    ds[4].rename(ds[3])  # root dir is replaced with a different with identical name
    assert cr.path("mark").read_text() == ds[4].name
    assert cr.path("") == ds[3]
    assert cr.path(".") == ds[3]
    ds[3].rename(r / "dx2")  # root dir is (re)moved (again)
    for rel in ["mark", "", "."]:
        with pytest.raises(ResolvedRootConsistencyError):
            cr.path(rel)
    fs[4].rename(ds[3])  # root dir is replaced with a file
    # could be improved to be a ResolvedRootConsistencyError
    with pytest.raises(FileNotFoundError if sys.platform.startswith("win") else NotADirectoryError):
        cr.path("mark")
    with pytest.raises(ResolvedRootConsistencyError):
        cr.path("")
    with pytest.raises(ResolvedRootConsistencyError):
        cr.path(".")


def test_utility_path_chroot_root_moves_link(root_struct):
    r, ds, fs, lna, lnb, lnr = root_struct
    # test link source moves (all paths go the the same dir0/mark file)
    crs_p = [
        (ChrootPath(lna), "mark"),
        (ChrootPath(lnb), "mark"),
        (ChrootPath(lnr), f"{ds[0].name}/mark"),
    ]
    assert all(cr.path(p).read_text() == ds[0].name for cr, p in crs_p)
    ds[0].rename(r / "dy")  # rename dir0
    for (cr, p), et in zip(crs_p, [ResolvedRootConsistencyError, ResolvedRootConsistencyError, FileNotFoundError]):
        with pytest.raises(et):
            cr.path(p)
    ds[1].rename(ds[0])  # dir0 path is replaced with a different with identical name
    assert all(cr.path(p).read_text() == ds[1].name for cr, p in crs_p)
    ds[0].rename(r / "dy2")  # rename new dir0
    for (cr, p), et in zip(crs_p, [ResolvedRootConsistencyError, ResolvedRootConsistencyError, FileNotFoundError]):
        with pytest.raises(et):
            cr.path(p)
    fs[3].rename(ds[0])  # dir0 path is replaced with a file
    for cr, p in crs_p:
        with pytest.raises(
            (FileNotFoundError, ResolvedRootConsistencyError) if sys.platform.startswith("win") else NotADirectoryError
        ):
            cr.path(p)
    ds[0].rename(r / "dy3")  # rename new dir0 (file)
    for (cr, p), et in zip(crs_p, [ResolvedRootConsistencyError, ResolvedRootConsistencyError, FileNotFoundError]):
        with pytest.raises(et):
            cr.path(p)
    ds[2].rename(ds[0])  # dir0 path is replaced with a different with identical name
    assert all(cr.path(p).read_text() == ds[2].name for cr, p in crs_p)
    # test link target moves
    cr = ChrootPath(lnb)
    assert cr.path("mark").read_text() == ds[2].name
    lna.rename(r / "dz")
    with pytest.raises(ResolvedRootConsistencyError):
        cr.path("mark")
    ds[0].rename(lna)
    with pytest.raises(ResolvedRootConsistencyError, match="Root path have changed since start"):
        cr.path("mark")


def test_utility_path_chroot_cycle(tmpdir):
    _r = Path(normalize_absolute_path(temp_path(tmpdir)))
    _r.mkdir()
    d = _r / "d"
    d.mkdir()
    f = d / "f"
    f.write_text("xy")
    os.symlink("..", d / "l0", target_is_directory=True)
    os.symlink(_r, d / "l1", target_is_directory=True)
    os.symlink(d, d / "l2", target_is_directory=True)
    # check cycle
    cr = ChrootPath(_r)
    for rel in ["", "d/l0", "d/l1"]:
        assert set(p.name for p in cr.list(rel)) == {"d"}
    for rel in ["d", "d/l2", "d/l0/d", "d/l1/d"]:
        assert set(p.name for p in cr.list(rel)) == {"l0", "l1", "l2", "f"}
    cr = ChrootPath(_r, allow_path_cycles=False)
    assert set(p.name for p in cr.list("")) == {"d"}
    assert set(p.name for p in cr.list("d")) == {"l0", "l1", "l2", "f"}
    for rel in ["d/l0", "d/l1", "d/l2", "d/l0/d", "d/l1/d"]:
        with pytest.raises(CycledPathError):
            cr.path(rel)


def test_utility_path_chroot_path_chroot(tmpdir):
    _r = Path(normalize_absolute_path(temp_path(tmpdir)))
    _r.mkdir()
    os.symlink("..", _r / "l0", target_is_directory=True)
    os.symlink(_r.parent, _r / "l1", target_is_directory=True)
    cr = ChrootPath(_r)
    # test no escape
    with pytest.raises(ValidationError, match="Names of relative-directives are reserved"):
        cr.path("..")
    with pytest.raises(ValidationError, match="Combined path is not contained by root"):
        cr.path("l0")
    with pytest.raises(ValidationError, match="Combined path is not contained by root"):
        cr.path("l1")
    with pytest.raises(ValidationError, match="Path cannot be absolute"):
        cr.path(_r.parent)
    # test no escape with new_path
    with pytest.raises(ValidationError, match="Names of relative-directives are reserved"):
        cr.new_path("../xy")
    with pytest.raises(ValidationError, match="Combined path is not contained by root"):
        cr.new_path("l0/xy")
    with pytest.raises(ValidationError, match="Combined path is not contained by root"):
        cr.new_path("l1/xy")
    with pytest.raises(ValidationError, match="Path cannot be absolute"):
        cr.new_path(_r.parent / "xy")


def test_utility_path_chroot_new_path(tmpdir):
    out = Path(normalize_absolute_path(temp_path(tmpdir)))
    out.write_text("asdf")
    _r = Path(normalize_absolute_path(temp_path(tmpdir)))
    _r.mkdir()
    (_r / "f0").write_text("f0")
    (_r / "d0").mkdir()
    (_r / "d0" / "d1").mkdir()
    (_r / "d0" / "f1").write_text("f1")
    (_r / "d2").mkdir()
    (_r / "d2" / "f2").write_text("f2")
    (_r / "l_i_f").symlink_to(_r / "f0")
    (_r / "l_i_d").symlink_to(_r / "d0")
    (_r / "l_o_f").symlink_to(out)
    cr = ChrootPath(_r)
    for p in ["/", ".", "./."]:
        with pytest.raises(ValidationError, match="Name is empty"):
            cr.new_path(p)
    with pytest.raises(ValidationError, match="Invalid path"):
        cr.new_path("")
    for p in ["..", "./..", "../.."]:
        with pytest.raises(ValidationError, match="Names of relative-directives are reserved"):
            cr.new_path(p)
    for p in ["f0", "d0", "d0/d1", "d0/f1", "l_i_f", "l_i_d"]:
        cr.new_path(p).exists()
    with pytest.raises(ValidationError, match="Final path is not contained by root"):
        cr.new_path("l_o_f")
    for p in ["f0/x", "d0/f1/x"]:
        with pytest.raises(ValidationError, match="Path must be inside an existing directory"):
            cr.new_path(p)
    for p in ["d9/x", "d0/d9/x"]:
        with pytest.raises(FileNotFoundError):
            cr.new_path(p)
    assert sorted(x.name for x in cr.iter(".")) == ["d0", "d2", "f0", "l_i_d", "l_i_f"]
    # disallowed symlinks
    cr = ChrootPath(_r, allow_symlinks=False)
    with pytest.raises(ValidationError, match="Final part may not be a symlink"):
        cr.new_path("l_i_f")
    assert sorted(x.name for x in cr.iter(".")) == ["d0", "d2", "f0"]
    # free symlinks
    cr = ChrootPath(_r, contain_symlinks=False)
    assert sorted(x.name for x in cr.iter(".")) == ["d0", "d2", "f0", "l_i_d", "l_i_f", "l_o_f"]


def test_utility_path_chroot_node(tmpdir):
    _rx = str(temp_path(tmpdir))
    _r = Path(normalize_absolute_path(_rx))
    _r.mkdir()
    d = _r / "d"
    d.mkdir()
    df = d / "df"
    df.write_text("xyd")
    f = _r / "f"
    f.write_text("xyf")
    ds = _r / "ds"
    ds.symlink_to(d, target_is_directory=True)
    fs = _r / "fs"
    fs.symlink_to(f)
    d1 = d / "d1"
    d1.mkdir()
    d1f = d1 / "d1f"
    d1f.write_text("xyd1")

    def _check(res, p, s: os.stat_result, ls_or_parent):
        assert p == res[0]
        # NOTE: stat can fail if access-timestamp is updated so just compare some attributes
        for a in ["st_dev", "st_ino", "st_mode", "st_size", "st_mtime_ns"]:
            assert getattr(s, a) == getattr(res[1], a)
        if not isinstance(ls_or_parent, list):
            assert res[2] == ls_or_parent
        else:
            assert isinstance(res[2], list)
            assert set(ls_or_parent) == set(res[2])

    cr = ChrootPath(_rx)
    _check(cr.node(""), [(_r, "")], _r.stat(), [(d, None), (ds, d), (f, None), (fs, f)])
    _check(cr.node("d"), [(_r, ""), (d, "d")], d.stat(), [(d1, None), (df, None)])
    _check(cr.node("d/d1"), [(_r, ""), (d, "d"), (d1, "d/d1")], d1.stat(), [(d1f, None)])
    _check(cr.node("d/d1/d1f"), [(_r, ""), (d, "d"), (d1, "d/d1"), (d1f, "d/d1/d1f")], d1f.stat(), None)
    _check(cr.node("ds"), [(_r, ""), (d, "ds")], d.stat(), [(d1, None), (df, None)])
    _check(cr.node("ds/d1"), [(_r, ""), (d, "ds"), (d1, "ds/d1")], d1.stat(), [(d1f, None)])
    _check(cr.node("f"), [(_r, ""), (f, "f")], f.stat(), None)
    _check(cr.node("d/df"), [(_r, ""), (d, "d"), (df, "d/df")], df.stat(), None)

    if sys.platform.startswith("win"):
        return  # TODO simulate permission error with win32security

    try:
        d.chmod(0)
        try:
            cr.node("d")
        except PartialNodeResolutionError as e:
            _check((e.args[0], e.args[1], None), [(_r, ""), (d, "d")], d.stat(), None)
            assert isinstance(e.args[2], PermissionError)
        else:
            assert False
    finally:
        d.chmod(0o755)

    try:
        df.chmod(0)
        d1.chmod(0)
        _check(cr.node("d"), [(_r, ""), (d, "d")], d.stat(), [(d1, None), (df, None)])
        df.stat()
        d1.stat()
    finally:
        df.chmod(0o755)
        d1.chmod(0o755)


def test_utility_path_chroot_alternates(tmpdir):
    _r = Path(normalize_absolute_path(temp_path(tmpdir)))
    _r.mkdir()
    _s = Path(normalize_absolute_path(temp_path(tmpdir)))
    _s.symlink_to(_r, target_is_directory=True)
    assert _r != _s
    assert _r == resolve_path(_s)

    cr = ChrootPath(_r)
    assert cr.path("") == _r
    cr.add_alternate_root_path(_s)
    assert cr.path("") == _r

    _s.unlink()
    with pytest.raises(ResolvedRootConsistencyError):
        cr.path("")


def test_utility_path_chroot_path_complex_struct(tmpdir):
    with ComplexStruct.create(tmpdir) as cs:
        valid_roots, invalid_roots = cs.split_roots_by_validity()
        for r in valid_roots:
            for allow_sym, contain_sym in DUAL_BIN_PROD:
                chr_p = ChrootPath(r, allow_symlinks=allow_sym, contain_symlinks=contain_sym)
                allowed_names = cs.get_allowed_items(r, allow_sym, contain_sym)
                for name, p in cs.all_by_name_iter():
                    try:
                        rel = p.relative_to(r)
                        inside = True
                    except ValueError:
                        rel = Path(*[".." for _ in r.relative_to(cs.root).parts], *p.relative_to(cs.root).parts)
                        inside = False
                    if name in allowed_names and inside:
                        # assert path direct access
                        try:
                            with coerce_platform_os_error():
                                fp = chr_p.path(rel)
                        except FileNotFoundError:
                            assert cs.missing_by_name[name] == p
                        else:
                            assert name not in cs.missing_by_name
                            if p.is_symlink():
                                assert fp == resolve_path(p, True)
                            else:
                                assert fp == p
                            # assert listing access of a path
                            if fp.is_dir():
                                lps = chr_p.list(rel)
                                for lp in lps:
                                    if lp.name in allowed_names:
                                        pass
                                    else:
                                        assert (
                                            False
                                        ), f"Test root: {cs.root}\nRoo: {r}\nQue: {p}\nRel: {rel}\n{fp}\n{lp}\n{lps}"
                                    if lp.is_symlink():
                                        try:
                                            assert resolve_path(lp, True).name in allowed_names
                                        except FileNotFoundError:
                                            pass
                            else:
                                try:
                                    chr_p.list(rel)
                                except NotADirectoryError:
                                    pass
                                else:
                                    assert False, f"{cs.root}\n{r}\n{p}\n{rel}"
                    else:
                        # assert path direct access
                        try:
                            with coerce_platform_os_error():
                                chr_p.path(rel)
                        except (ValidationError, FileNotFoundError):
                            pass
                        else:
                            assert False, f"{cs.root}\n{r}\n{p}\n{rel}"
                        # assert listing access of a path
                        try:
                            with coerce_platform_os_error():
                                chr_p.list(rel)
                        except (ValidationError, FileNotFoundError):
                            pass
                        else:
                            assert False, f"{cs.root}\n{r}\n{p}\n{rel}"
