changelog.py

import itertools
import pathlib
import re
from datetime import datetime
from functools import cached_property
from collections.abc import Iterator

from packaging import version as _version

import abstracts

from aio.core.functional import async_property

from envoy.base import utils
from envoy.base.utils.abstract.project.changelog import (
    CHANGELOG_ENTRY_GLOB,
    ENTRY_SEPARATOR,
)
from envoy.code.check import abstract, interface


MAX_VERSION_FOR_CHANGES_SECTION = "1.16"
try:
    from envoy.base.utils.abstract.project.changelog import (
        CHANGELOG_AREAS_PATH,
    )
except ImportError:
    CHANGELOG_AREAS_PATH = "changelogs/areas.yaml"
VALID_CHANGELOG_AREA_RE = re.compile(r"^[a-z0-9_\-/]+$")
VALID_CHANGELOG_AREA_PATTERN = r"[a-z0-9_\-/]+"
CHANGELOG_AREAS_FILE = pathlib.Path(CHANGELOG_AREAS_PATH)


@abstracts.implementer(interface.IChangelogChangesChecker)
class AChangelogChangesChecker(metaclass=abstracts.Abstraction):
    error_message = (
        "{version}/{section}/{entry[area]}: "
        "{error}\n{entry[change]}")

    def __init__(
            self,
            sections: utils.typing.ChangelogSectionsDict,
            areas: "utils.typing.ChangelogAreasDict") -> None:
        self.sections = sections
        self.areas = areas

    @property  # type:ignore
    @abstracts.interfacemethod
    def change_checkers(self) -> tuple[interface.IRSTCheck, ...]:
        raise NotImplementedError

    @cached_property
    def max_version_for_changes_section(self) -> _version.Version:
        return _version.Version(MAX_VERSION_FOR_CHANGES_SECTION)

    def check_entry(
            self,
            version: _version.Version,
            section: str,
            entry: utils.typing.ChangeDict) -> tuple[str, ...]:
        change = entry["change"].strip()
        errors = [
            checker(change)
            for checker
            in self.change_checkers]
        return tuple(
            self.error_message.format(
                version=version,
                section=section,
                entry=entry,
                error=error)
            for error
            in errors
            if error)

    def check_section(
            self,
            version: _version.Version,
            section: str,
            data: utils.typing.ChangeList | None) -> tuple[str, ...]:
        name_error = self.check_section_name(version, section)
        return (
            *((name_error, )
              if name_error
              else ()),
            *itertools.chain.from_iterable(
                self.check_entry(version, section, entry)
                for entry
                in data or []))

    def check_sections(
            self,
            version: _version.Version,
            sections: utils.typing.ChangelogChangeSectionsDict) -> (
                tuple[str, ...]):
        return tuple(
            itertools.chain.from_iterable(
                self.check_section(version, section, data)  # type:ignore
                for section, data
                in sections.items()))

    def check_section_name(
            self,
            version: _version.Version,
            section: str) -> str | None:
        invalid_changes = (
            section == "changes"
            and version > self.max_version_for_changes_section)
        if invalid_changes:
            return (
                f"{version}/changes: Invalid `changes` section "
                "(this is no longer used)")

    def check_entry_filename(
            self,
            path: pathlib.Path) -> str | None:
        section = path.parent.name
        if section not in self.sections:
            return (
                f"{path}: Invalid section `{section}`. "
                f"Valid sections: {sorted(self.sections)}")
        if path.suffix != ".rst":
            return (
                f"{path}: Invalid file extension `{path.suffix}` "
                "(expected `.rst`)")
        if path.stem.count(ENTRY_SEPARATOR) != 1:
            return (
                f"{path}: Filename stem must contain exactly one "
                f"`{ENTRY_SEPARATOR}` separator "
                f"(expected `<area>{ENTRY_SEPARATOR}<slug>`)")
        area, slug = path.stem.split(ENTRY_SEPARATOR, 1)
        if not area:
            return f"{path}: Area part of filename is empty"
        if self.areas and area not in self.areas:
            return (
                f"{path}: Invalid area '{area}'. "
                f"Valid areas come from {CHANGELOG_AREAS_PATH}")
        if not slug:
            return f"{path}: Slug part of filename is empty"
        return None

    def check_areas_file(self) -> tuple[str, ...]:
        if not self.areas:
            return ()
        title_areas: dict[str, list[str]] = {}
        errors = []
        for area, area_data in self.areas.items():
            title = area_data["title"]
            title_areas.setdefault(title, []).append(area)
            if not VALID_CHANGELOG_AREA_RE.match(area):
                errors.append(
                    f"{CHANGELOG_AREAS_FILE}: "
                    f"Invalid area key '{area}' "
                    f"(must match {VALID_CHANGELOG_AREA_PATTERN})")
            if not VALID_CHANGELOG_AREA_RE.match(title):
                errors.append(
                    f"{CHANGELOG_AREAS_FILE}: "
                    f"Invalid title '{title}' for area '{area}' "
                    f"(must match {VALID_CHANGELOG_AREA_PATTERN})")
        for title, areas in sorted(title_areas.items()):
            if len(areas) < 2:
                continue
            errors.append(
                f"{CHANGELOG_AREAS_FILE}: "
                f"Duplicate title '{title}' used by areas: "
                f"{', '.join(sorted(areas))}")
        return tuple(errors)

    def check_entry_content(
            self,
            path: pathlib.Path) -> str | None:
        content = path.read_text()
        if not content.strip():
            return (
                f"{path}: Entry file is empty or contains only whitespace")
        return None

    def check_entry_files(
            self,
            paths: list[pathlib.Path]) -> tuple[str, ...]:
        errors = []
        for path in paths:
            if err := self.check_entry_filename(path):
                errors.append(err)
            if err := self.check_entry_content(path):
                errors.append(err)
        return tuple(errors)


@abstracts.implementer(interface.IChangelogStatus)
class AChangelogStatus(metaclass=abstracts.Abstraction):

    def __init__(
            self,
            check: interface.IChangelogCheck,
            changelog: utils.interface.IChangelog) -> None:
        self._check = check
        self.changelog = changelog

    @property
    def checker(self) -> interface.IChangelogChangesChecker:
        return self._check.changes_checker

    @async_property
    async def data(self) -> utils.typing.ChangelogDict:
        return await self.changelog.data

    @async_property
    async def date(self) -> str:
        return (await self.data)["date"]

    @cached_property
    def date_format(self) -> str:
        return self.project.changelogs.date_format.replace("-", "")

    @async_property
    async def dev_not_pending(self) -> bool:
        return (
            self.is_current
            and self.project.is_dev
            and not await self.is_pending)

    @property
    def duplicate_current(self) -> bool:
        return (
            self.is_current
            and self.project.changelogs.changelog_path(
                self.version).exists())

    @property
    def entry_dir(self) -> pathlib.Path | None:
        if not self.is_current:
            return None
        return (
            self.project.changelogs
                        .changelog_path(self.version)
                        .with_suffix(""))

    @async_property(cache=True)
    async def errors(self) -> tuple[str, ...]:
        areas_errors = await self.check_areas_file()
        entry_errors = await self.check_entry_files()
        try:
            return (
                *self.check_version(),
                *await self.check_date(),
                *await self.check_sections(),
                *areas_errors,
                *entry_errors)
        except utils.exceptions.ChangelogParseError as e:
            return (*areas_errors, *entry_errors, f"{self.version}: {e}")

    @async_property
    async def invalid_date(self) -> str | None:
        if await self.is_pending:
            return None
        date = await self.date
        try:
            datetime.strptime(date, self.date_format)
        except ValueError:
            return date

    @cached_property
    def is_current(self) -> bool:
        return self.project.is_current(self.version)

    @async_property
    async def is_pending(self) -> bool:
        return (await self.date) == "Pending"

    @async_property
    async def pending_not_dev(self) -> bool:
        return (
            (not self.is_current
             or not self.project.is_dev)
            and await self.is_pending)

    @property
    def project(self) -> utils.interface.IProject:
        return self._check.project

    @async_property
    async def sections(self) -> utils.typing.ChangelogChangeSectionsDict:
        return utils.typed(
            utils.typing.ChangelogChangeSectionsDict,
            {k: v
             for k, v
             in (await self.data).items()
             if k != "date"})

    @property
    def version(self) -> _version.Version:
        return self.changelog.version

    @property
    def version_higher_than_current(self) -> bool:
        return (
            self.version
            > _version.Version(self.project.version.base_version))

    async def check_date(self) -> tuple[str, ...]:
        errors = []
        if invalid_date := await self.invalid_date:
            errors.append(f"Format not recognized \"{invalid_date}\"")
        if await self.dev_not_pending:
            errors.append("Should be set to `Pending`")
        elif await self.pending_not_dev:
            errors.append("Should not be set to `Pending`")
        return tuple(
            f"{self.version}/date: {e}"
            for e
            in errors)

    async def check_sections(self) -> tuple[str, ...]:
        # Runs checker in executor, uncomment following line for debugging
        # return self.checker.check_sections(self.version, await self.sections)
        return await self.project.execute(
            self.checker.check_sections,
            self.version,
            await self.sections)

    async def check_entry_files(self) -> tuple[str, ...]:
        entry_dir = self.entry_dir
        if entry_dir is None or not entry_dir.exists():
            return ()
        paths = sorted(entry_dir.glob(CHANGELOG_ENTRY_GLOB))
        if not paths:
            return ()
        return await self.project.execute(
            self.checker.check_entry_files,
            paths)

    async def check_areas_file(self) -> tuple[str, ...]:
        areas = self.project.changelogs.areas
        if not self.is_current or not areas:
            return ()
        return await self.project.execute(self.checker.check_areas_file)

    def check_version(self) -> tuple[str, ...]:
        errors = []
        if self.duplicate_current:
            errors.append(
                "Duplicate current version file. "
                "Only `current.yaml` should exist for the current version "
                f"({self.project.version.base_version})")
        elif self.version_higher_than_current:
            errors.append(
                "Changelog version is higher than "
                f"the current version ({self.project.version.base_version})")
        return tuple(
            f"{self.version}: {e}"
            for e
            in errors)


@abstracts.implementer(interface.IChangelogCheck)
class AChangelogCheck(
        abstract.AProjectCodeCheck,
        metaclass=abstracts.Abstraction):
    """Changelog check."""

    def __iter__(self) -> Iterator[interface.IChangelogStatus]:
        for changelog in self.changelogs:
            yield changelog

    @property  # type:ignore
    @abstracts.interfacemethod
    def changes_checker_class(
            self) -> type[interface.IChangelogChangesChecker]:
        raise NotImplementedError

    @cached_property
    def changes_checker(self) -> interface.IChangelogChangesChecker:
        return self.changes_checker_class(
            self.project.changelogs.sections,
            self.project.changelogs.areas)

    @property  # type:ignore
    @abstracts.interfacemethod
    def changelog_status_class(self) -> type[interface.IChangelogStatus]:
        raise NotImplementedError

    @cached_property
    def changelogs(self) -> tuple[interface.IChangelogStatus, ...]:
        return tuple(
            self.changelog_status_class(self, changelog)
            for changelog
            in self.project.changelogs.values())