#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
# tools/validate -- applies the asset format (format 1) to a publisher repository, or to one set directory.
#
# ```bash
# python3 tools/validate [--root DIR] [--profile source|release] [--set NAME]...
# python3 tools/validate --set-directory DIR [--publisher NAME] [--licenses ID,...] [--license-texts DIR]... \
#                        [--profile source|release]
# python3 tools/validate --list-rules [--markdown]
# ```
#
# Over a repository (the current directory, or `--root`), it reads `publisher.conf`, validates every set under `sets/`
# (or the ones `--set` names), and once every set passed runs `check-publisher` and holds the committed plugin manifests
# and the marketplace to what `sync-manifests` writes; every declared licence is held to the SPDX list in force
# (`publisher.conf` `licenses=`, or the default). Over one set directory -- a conformance fixture, or a set built by
# `build-set` under `--profile release` -- it applies the set-level rules alone, with `--publisher` where the set-name
# rules need one and `--licenses` where the list is not the default. Each finding is one line,
# `validate: <path>: <rule-id>: <what>`, with `warning:` after the command name where the rule warns; the exit status
# is 1 when a refusing finding was made, 2 on a usage error. The rule ids and what each requires are `--list-rules`, and
# `format/FORMAT.md` states them for a publisher.
#
# The root is the operator's argument and is opened by path; `sets/`, each set and every file inside them are opened from
# that descriptor without following a symbolic link (`tools/lib/safe_read.py`), so a linked `sets/`, a linked set
# directory or `--set-directory` naming a link is refused before a byte of the target is read. Reading a set once is
# the validator's property (`tools/lib/set_validation.py`): every set-level rule judges the records one walk kept. The
# repository-level checks that follow -- `check-publisher` in its own process, and the marketplace comparison -- are
# descriptor reads of their own over `set.conf`, the manifests, `publisher.conf` and the marketplace, made after every
# set has passed, so a file changed between the walk and those reads is read in its new form by them alone; they follow
# no link either. A set is read as data and no file in it is executed. Standard library only.
from __future__ import annotations

import argparse
import os
import pathlib
import subprocess
import sys

TOOLS = pathlib.Path(__file__).resolve().parent
sys.path.insert(0, str(TOOLS / "lib"))

import asset_format as fmt  # noqa: E402
from findings import FindingCollector, refuse_unconverted_errors  # noqa: E402
from key_value_config import parse_key_value_text  # noqa: E402
from license_policy import resolve_license_allowlist  # noqa: E402
from license_texts import check_license_text_directory, read_declared_digests  # noqa: E402
from manifests import MARKETPLACE_PATH, render_manifest_files, find_stale_manifest_files  # noqa: E402
from safe_read import RefusedRead, open_directory, open_root, read_text_under  # noqa: E402
from set_validation import (PROFILE_SOURCE, PROFILES, ValidationOptions, dynamic_injection_setting,  # noqa: E402
                            read_reserved_words, validate_set_directory)

RESERVED_WORDS_FILE = TOOLS / "reserved-words.txt"
CHECK_PUBLISHER = TOOLS / "check-publisher"


def list_rules(markdown: bool) -> int:
    print(rules_listing(markdown), end="")
    return 0


def rules_listing(markdown: bool) -> str:
    """Every rule id, whether it refuses or warns, and what it requires; as a Markdown table with `--markdown`."""
    if markdown:
        lines = ["| Rule | Outcome | Requires |", "|---|---|---|"]
        for rule_id, statement in fmt.RULES.items():
            outcome = "warns" if rule_id in fmt.RULES_WARNING else "refuses"
            lines.append(f"| `{rule_id}` | {outcome} | {statement} |")
        return "\n".join(lines) + "\n"
    width = max(len(rule_id) for rule_id in fmt.RULES)
    return "".join(f"{rule_id:<{width}}  {'warns  ' if rule_id in fmt.RULES_WARNING else 'refuses'}  {statement}\n"
                   for rule_id, statement in fmt.RULES.items())


LAYOUT_MESSAGE = "a publisher repository holds publisher.conf and sets/ at its root; name it with --root, or validate one set with --set-directory"


def validate_repository(root: pathlib.Path, profile: str, only_sets: list, collector: FindingCollector) -> None:
    try:
        root_fd = open_root(root)
    except OSError:
        collector.refuse(str(root), "repo.layout", LAYOUT_MESSAGE)
        return
    try:
        validate_repository_under(root_fd, root, profile, only_sets, collector)
    finally:
        os.close(root_fd)


def validate_repository_under(root_fd: int, root: pathlib.Path, profile: str, only_sets: list, collector: FindingCollector) -> None:
    try:
        publisher_conf = parse_key_value_text(read_text_under(root_fd, pathlib.PurePath("publisher.conf"), fmt.FILE_MAX_BYTES))
    except FileNotFoundError:
        collector.refuse(str(root), "repo.layout", LAYOUT_MESSAGE)
        return
    except RefusedRead as refusal:
        collector.refuse("publisher.conf", "repo.publisher-conf", refusal.message)
        return
    try:
        sets_fd = open_directory("sets", root_fd)
    except FileNotFoundError:
        collector.refuse(str(root), "repo.layout", LAYOUT_MESSAGE)
        return
    except NotADirectoryError:
        collector.refuse("sets", "repo.layout", "is not a directory; " + LAYOUT_MESSAGE)
        return
    except RefusedRead as refusal:
        collector.refuse("sets", "repo.layout", refusal.message)
        return
    for message in publisher_conf.syntax_errors():
        collector.refuse("publisher.conf", "repo.publisher-conf", message)
    missing = [key for key in fmt.PUBLISHER_CONF_REQUIRED if not publisher_conf.get(key).strip()]
    if missing:
        collector.refuse("publisher.conf", "repo.publisher-conf", "missing or empty: " + ", ".join(f"`{key}=`" for key in missing))
    # The same unknown-key rule as set.conf's, so a misspelt `licenses=` or `allow_dynamic_injection=` does not read
    # as the default; a publisher's own key is `x_<key>`.
    for key in publisher_conf.values:
        if key not in fmt.PUBLISHER_CONF_REQUIRED and key not in fmt.PUBLISHER_CONF_OPTIONAL and not fmt.EXTENSION_KEY_PATTERN.match(key):
            collector.refuse("publisher.conf", "repo.publisher-conf", f"`{key}` is not a key this format reads; a publisher's own key is `x_<key>`")
    _, setting_error = dynamic_injection_setting(publisher_conf)
    if setting_error:
        collector.refuse("publisher.conf", "repo.publisher-conf", setting_error)
    allowlist, allowlist_error = resolve_license_allowlist(publisher_conf)
    if allowlist_error:
        collector.refuse("publisher.conf", "license.allowlist", allowlist_error)
    declared_digests, declared_error = read_declared_digests(publisher_conf)
    if declared_error:
        collector.refuse("publisher.conf", "repo.publisher-conf", declared_error)
    # The repository's own texts are the ones build-set copies into a set that carries none.
    check_license_text_directory(collector, root_fd, declared_digests)

    options = ValidationOptions(publisher=publisher_conf.get("publisher") or None, profile=profile,
                                reserved_words=read_reserved_words(RESERVED_WORDS_FILE), license_allowlist=allowlist,
                                license_text_directories=[root], display_root=root, publisher_conf=publisher_conf)
    validated = []
    try:
        with os.scandir(sets_fd) as listing:
            entries = sorted((entry for entry in listing if not entry.name.startswith(".")), key=lambda entry: entry.name)
        if only_sets:
            # A selector is a name, matched against the sets discovered under sets/; it is not composed into a path.
            for name in sorted(set(only_sets)):
                if not fmt.is_valid_name(name):
                    collector.refuse(name, "name.grammar", "--set takes a set name: 1-64 characters of a-z, 0-9 and single hyphens")
                elif name not in {entry.name for entry in entries}:
                    collector.refuse(f"sets/{name}", "set.conf.missing", "no such set directory")
            entries = [entry for entry in entries if entry.name in only_sets]
        for entry in entries:
            if entry.is_symlink():
                collector.refuse(f"sets/{entry.name}", "file.symlink", "is a symbolic link; a set is a directory under sets/")
                continue
            if not entry.is_dir(follow_symlinks=False):
                collector.refuse(f"sets/{entry.name}", "set.conf.missing", "is not a directory")
                continue
            summary = validate_set_directory(root / "sets" / entry.name, options, collector, parent_fd=sets_fd)
            if summary.set_conf is not None:
                validated.append((entry.name, summary.set_conf))
    finally:
        os.close(sets_fd)
    if collector.has_refusals():
        return  # the repository-level checks read manifests the set walk has judged
    run_check_publisher(root, collector)
    if not only_sets:
        # Each set's manifests were compared inside its validation (the options carry publisher.conf); the marketplace
        # is the repository's one generated file left.
        expected = {MARKETPLACE_PATH: render_manifest_files(publisher_conf, validated)[MARKETPLACE_PATH]}
        for relative_path, reason in find_stale_manifest_files(root_fd, expected):
            collector.refuse(str(relative_path), "repo.marketplace", f"{reason}; run `python3 tools/sync-manifests`")


def run_check_publisher(root: pathlib.Path, collector: FindingCollector) -> None:
    completed = subprocess.run([sys.executable, str(CHECK_PUBLISHER), "--root", str(root)], capture_output=True, text=True)
    for line in completed.stderr.splitlines():
        print(line, file=sys.stderr)
    if completed.returncode == 1:
        collector.refuse("publisher.conf", "repo.publisher-conf", "check-publisher found the names and contacts above")
    elif completed.returncode != 0:
        collector.refuse("publisher.conf", "repo.publisher-conf", f"check-publisher exited {completed.returncode}: {completed.stderr.strip()}")


def validate_one_set(set_directory: pathlib.Path, publisher: str, licenses: str, license_texts: list, profile: str,
                     publisher_conf_path: pathlib.Path, collector: FindingCollector) -> None:
    allowlist = tuple(item for item in licenses.replace(",", " ").split() if item) if licenses is not None else fmt.DEFAULT_LICENSE_ALLOWLIST
    publisher_conf = None
    if publisher_conf_path is not None:
        publisher_conf = read_publisher_conf(publisher_conf_path, collector)
        if publisher_conf is None:
            return
        _, setting_error = dynamic_injection_setting(publisher_conf)
        if setting_error:
            collector.refuse(str(publisher_conf_path), "repo.publisher-conf", setting_error)
    options = ValidationOptions(publisher=publisher, profile=profile, reserved_words=read_reserved_words(RESERVED_WORDS_FILE),
                                license_allowlist=allowlist, license_text_directories=[path.resolve() for path in license_texts],
                                display_root=set_directory.parent, publisher_conf=publisher_conf)
    validate_set_directory(set_directory, options, collector)


def read_publisher_conf(path: pathlib.Path, collector: FindingCollector):
    """The publisher.conf at `path`, read from its directory's descriptor, or None after a `repo.publisher-conf` finding."""
    try:
        directory_fd = open_root(path.parent)
    except OSError:
        collector.refuse(str(path), "repo.publisher-conf", "cannot be read: its directory does not exist")
        return None
    try:
        return parse_key_value_text(read_text_under(directory_fd, pathlib.PurePath(path.name), fmt.FILE_MAX_BYTES))
    except FileNotFoundError:
        collector.refuse(str(path), "repo.publisher-conf", "does not exist")
    except RefusedRead as refusal:
        collector.refuse(str(path), "repo.publisher-conf", refusal.message)
    finally:
        os.close(directory_fd)
    return None


def default_publisher_conf(set_directory: pathlib.Path) -> pathlib.Path:
    """`<set-directory>/../../publisher.conf` -- the repository root's, for a set under `sets/` -- where it exists."""
    candidate = set_directory.parent.parent / "publisher.conf"
    return candidate if candidate.is_file() else None


def main() -> int:
    parser = argparse.ArgumentParser(description="Apply the asset format to a publisher repository or one set.")
    parser.add_argument("--root", type=pathlib.Path, default=pathlib.Path("."), help="the publisher repository (default: the current directory)")
    parser.add_argument("--profile", choices=PROFILES, default=PROFILE_SOURCE, help="source (committed tree) or release (a built set with SHA256SUMS)")
    parser.add_argument("--set", action="append", default=[], metavar="NAME", help="validate this set alone; repeatable")
    parser.add_argument("--set-directory", type=pathlib.Path, metavar="DIR", help="validate one set directory instead of a repository")
    parser.add_argument("--publisher", metavar="NAME", help="with --set-directory: the publisher the set-name rules read")
    parser.add_argument("--licenses", metavar="ID,...", help="with --set-directory: the SPDX allowlist in force")
    parser.add_argument("--license-texts", action="append", default=[], type=pathlib.Path, metavar="DIR",
                        help="with --set-directory: a directory whose LICENSES/ holds the texts the set does not carry; repeatable")
    parser.add_argument("--publisher-conf", type=pathlib.Path, metavar="PATH",
                        help="with --set-directory: the publisher.conf the manifests are rendered from, compared whole "
                             "(default: <set-directory>/../../publisher.conf where it exists)")
    parser.add_argument("--list-rules", action="store_true", help="print every rule id with what it requires")
    parser.add_argument("--markdown", action="store_true", help="with --list-rules: as the table format/FORMAT.md embeds")
    args = parser.parse_args()
    if args.list_rules:
        return list_rules(args.markdown)
    collector = FindingCollector("validate")
    return refuse_unconverted_errors(collector, "repo.layout", args.root, lambda: validate(args, collector))


def validate(args: argparse.Namespace, collector: FindingCollector) -> int:
    if args.set_directory is not None:
        # The path is made absolute without resolving its last component, which the validator opens without following
        # a link; its parent is the operator's argument.
        set_directory = pathlib.Path(os.path.abspath(args.set_directory))
        publisher_conf_path = args.publisher_conf.resolve() if args.publisher_conf is not None else default_publisher_conf(set_directory)
        validate_one_set(set_directory, args.publisher, args.licenses, args.license_texts, args.profile, publisher_conf_path, collector)
    else:
        validate_repository(args.root.resolve(), args.profile, args.set, collector)
    return collector.report()


if __name__ == "__main__":
    sys.exit(main())
