reno reporter & linter that is far faster and leverages filtered git logs to handle topology for you

reno-report.py

#!/usr/bin/python
import argparse
import contextlib
import os.path
import re
import shutil
import subprocess
import logging
import time
import sys
from collections import defaultdict

import pytest
import yaml.representer

yaml.add_representer(defaultdict, yaml.representer.Representer.represent_dict)

DEFAULT_CONFIG = {"release_tag_re": r"^v?((?:[\d.ab]|rc)+)"}

log = logging.getLogger()


def normalize(git_dir):
    return git_dir.replace("\\", "/").replace("./", "")


class Runner:  # pylint: disable=too-many-instance-attributes
    def __init__(self, args):
        self.args = args
        try:
            self.cfg = yaml.safe_load(open("./reno.yaml"))
        except FileNotFoundError:
            self.cfg = DEFAULT_CONFIG.copy()

        self.earliest = self.cfg.get("earliest_version")
        self.version_regex = (
            args.version_regex or self.cfg.get("release_tag_re") or DEFAULT_CONFIG.get("release_tag_re")
        )
        self.tags = []
        self.logs = []
        self.notes = {}
        self.report = ""
        self.ver_start = self.args.previous
        self.ver_end = self.args.version or "HEAD"
        self.notes_dir = normalize(self.args.rel_notes_dir)

        log.debug("notes_dir: %s", self.notes_dir)
        if not os.path.exists(self.notes_dir):
            raise FileNotFoundError("expected folder: %s" % self.notes_dir)

        self.sections = dict(self.cfg.get("sections", {}))
        self.valid_sections = {"release_summary", *self.sections.keys()}

        self.__git = shutil.which("git")

    def git(self, *args):
        log.debug("+ git %s", " ".join(args))
        cmd = [self.__git] + list(args)
        ret = subprocess.run(cmd, check=True, stdout=subprocess.PIPE, encoding="utf8")
        return ret.stdout

    def get_tags(self):
        self.tags = []

        for tag in self.git("log", self.ver_end, "--tags", "--pretty=%D").split("\n"):
            tag = tag.strip()
            if not tag:
                continue
            head = re.match(r"HEAD, tag:", tag)
            tag = re.search(r"\btag: ([^\s,]+)", tag)
            if not tag:
                continue
            tag = tag[1]
            if re.match(self.version_regex, tag):
                self.tags.append(tag)
                if head:
                    self.ver_end = tag
            if tag == self.earliest:
                break

        self.tags = list(reversed(self.tags))

        log.debug("tags: %s", self.tags)

    def get_start_from_end(self):
        if not self.ver_start:
            if self.ver_end == "HEAD":
                self.ver_start = self.tags[-1] if self.tags else "HEAD"

            prev = None
            for t in self.tags:
                if self.ver_end == t:
                    self.ver_start = prev
                prev = t

        log.debug("prev: %s, cur: %s", self.ver_start, self.ver_end)

    def get_logs(self):
        cur_tag = self.ver_end
        ct = 0
        cname = ""
        hsh = ""
        vers = self.ver_start + ".." + self.ver_end
        for ent in self.git("log", vers, "--name-only", "--format=%D^%ct^%cn^%h", "--diff-filter=A").split("\n"):
            ent = ent.strip()
            info = ent.split("^")
            if len(info) > 1:
                tag, ct, cname, hsh = info
                tag = re.match(r"^tag: ([\S,]+)", ent)
                if tag:
                    cur_tag = tag[1]
            if ent.startswith(self.notes_dir):
                self.logs.append((cur_tag, ct, cname, hsh, ent))

    def load_note(self, tag, file, ct, cname, hsh, notes):
        try:
            with open(file) as f:
                note = yaml.safe_load(f)
                for k, v in note.items():
                    assert k in self.valid_sections, "%s: %s is not a valid section" % (file, k)
                    if type(v) is str:
                        v = [v]
                    assert type(v) is list, "%s: '%s' : list of entries or single string" % (file, k)
                    for line in v:
                        assert type(line) is str, "%s: '%s' : must be a simple string" % (file, line)
                        line = {"time": int(ct), "name": cname, "hash": hsh, "note": line}
                        notes[tag][k].append(line)
        except Exception as e:
            print("Error reading file %s: %s" % (file, repr(e)))
            raise

    def get_notes(self):
        seen = {}
        notes = defaultdict(lambda: defaultdict(lambda: []))
        for tag, ct, cname, hsh, file in self.logs:
            if seen.get(file):
                continue
            seen[file] = True
            try:
                self.load_note(tag, file, ct, cname, hsh, notes)
            except FileNotFoundError:
                pass

        cname = self.git("config", "user.name").strip()

        for file in self.git("diff", "--name-only").split("\n"):
            path = file.strip()
            self._load_uncommitted(seen, notes, path, cname)

        if self.args.lint:
            # every file, not just diffs
            for file in os.listdir(self.notes_dir):
                path = normalize(os.path.join(self.notes_dir, file))
                self._load_uncommitted(seen, notes, path, cname)

        self.notes = notes

    def _load_uncommitted(self, seen, notes, path, cname):
        if seen.get(path):
            return
        if not os.path.isfile(path):
            return
        if not path.endswith(".yaml"):
            return
        self.load_note("Uncommitted", path, os.stat(path).st_mtime, cname, None, notes)

    def get_report(self):
        num = 0
        for tag, sections in self.notes.items():
            if tag == "HEAD":
                tag = "Current Branch"
            if num > 0:
                print("")
            num += 1
            print(tag)
            print("=" * len(tag))

            ents = sections.get("release_summary", {})
            for ent in sorted(ents, key=lambda ent: ent["time"], reverse=True):
                note = ent["note"].strip()
                print(note, "\n")

            for sec, title in self.sections.items():
                ents = sections.get(sec, {})
                if not ents:
                    continue
                print()
                print(title)
                print("-" * len(title))
                for ent in sorted(ents, key=lambda ent: ent["time"], reverse=True):
                    note = ent["note"]
                    if self.args.blame:
                        epoch = ent["time"]
                        name = ent["name"]
                        hsh = ent["hash"]
                        hsh = "`" + hsh + "`" if hsh else ""
                        print("-", note, hsh, "(" + name + ")", time.strftime("%y-%m-%d", time.localtime(epoch)))
                    else:
                        print("-", note)

    def get_branch(self):
        return self.git("rev-parse", "--abbrev-ref", "HEAD").strip()

    def switch_branch(self, branch):
        self.git("-c", "advice.detachedHead=false", "checkout", branch)

    def run(self):
        orig = None
        if self.ver_end != "HEAD":
            orig = self.get_branch()
            self.switch_branch(self.ver_end)
        try:
            self.get_tags()
            self.get_start_from_end()
            self.get_logs()
            if orig:
                self.switch_branch(orig)
                orig = None
            self.get_notes()
            if self.args.lint:
                return
            if self.args.yaml:
                print(yaml.dump(self.notes))
                return
            self.get_report()

            print(self.report)
        finally:
            if orig:
                self.switch_branch(orig)


def parse_args(args):
    parser = argparse.ArgumentParser(description="Sane reno reporter")
    parser.add_argument("--version", help="Version to report on (default: current branch)")
    parser.add_argument("--previous", help="Previous version, (default: ordinal previous tag)")
    parser.add_argument("--version-regex", help="Regex to use when parsing (default: from reno.yaml)")
    parser.add_argument("--rel-notes-dir", help="Release notes folder", default="./releasenotes/notes")
    parser.add_argument("--debug", help="Debug mode", action="store_true")
    parser.add_argument("--yaml", help="Dump yaml", action="store_true")
    parser.add_argument("--lint", help="Lint notes for valid markdown", action="store_true")
    parser.add_argument("--blame", help="Show more commit info in the report", action="store_true")
    return parser.parse_args(args)


def main():
    logging.basicConfig(level=logging.INFO, stream=sys.stderr)
    args = parse_args(sys.argv[1:])
    if args.debug:
        log.setLevel(logging.DEBUG)
        log.debug("args: %s", args)
    r = Runner(args)
    try:
        r.run()
    except (subprocess.CalledProcessError, AssertionError) as e:
        print("ERROR:", str(e), file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()


def test_lint():
    args = parse_args(["--lint", "--debug"])
    r = Runner(args)
    r.run()


def test_report(capsys):
    args = parse_args([])
    r = Runner(args)
    r.run()
    captured = capsys.readouterr()
    assert "Current Branch" in captured.out


def test_yaml(capsys):
    args = parse_args(["--yaml"])
    r = Runner(args)
    r.run()
    captured = capsys.readouterr()
    res = yaml.safe_load(captured.out)
    assert res["HEAD"]


@contextlib.contextmanager
def mock_git(runner, regex, result):
    func = runner.git

    def new_git(*args):
        cmd = " ".join(args)
        if re.match(regex, cmd):
            return result
        return func(*args)

    runner.git = new_git
    yield
    runner.git = func


def test_diff(capsys, tmp_path):
    args = parse_args(["--yaml", "--rel-notes-dir", str(tmp_path)])
    r = Runner(args)
    with open(tmp_path / "rel.yaml", "w") as f:
        f.write("release_summary: rel")

    norm_path = normalize(str(tmp_path)) + "/rel.yaml"
    with mock_git(r, r"diff --name-only", f"{norm_path}\n"), mock_git(r, r"log.*", ""):
        r.run()

    captured = capsys.readouterr()
    res = yaml.safe_load(captured.out)
    assert res["Uncommitted"]["release_summary"][0]["note"] == "rel"


def test_lint_all(tmp_path):
    assert os.path.exists(tmp_path)
    args = parse_args(["--lint", "--rel-notes-dir", str(tmp_path)])
    r = Runner(args)
    with open(tmp_path / "rel.yaml", "w") as f:
        f.write("releaxxxxx: rel")

    with pytest.raises(AssertionError, match=".*is not a valid section.*"):
        r.run()


def test_bad_section(tmp_path):
    assert os.path.exists(tmp_path)
    args = parse_args(["--yaml", "--rel-notes-dir", str(tmp_path)])
    r = Runner(args)
    with open(tmp_path / "rel.yaml", "w") as f:
        f.write("releaxxxxx: rel")

    norm_path = normalize(str(tmp_path)) + "/rel.yaml"
    with mock_git(r, r"diff --name-only", f"{norm_path}\n"), mock_git(r, r"log.*", ""):
        with pytest.raises(AssertionError, match=".*is not a valid section.*"):
            r.run()


def test_bad_entry(tmp_path):
    args = parse_args(["--rel-notes-dir", str(tmp_path)])
    r = Runner(args)
    with open(tmp_path / "rel.yaml", "w") as f:
        f.write("release_summary: [{'bad': 'summary'}]")

    norm_path = normalize(str(tmp_path)) + "/rel.yaml"
    with mock_git(r, r"diff --name-only", f"{norm_path}\n"), mock_git(r, r"log.*", ""):
        with pytest.raises(AssertionError, match=".*simple string.*"):
            r.run()


def test_bad_entry2(tmp_path):
    args = parse_args(["--rel-notes-dir", str(tmp_path)])
    r = Runner(args)
    with open(tmp_path / "rel.yaml", "w") as f:
        f.write("release_summary: {'bad': 'summary'}")

    norm_path = normalize(str(tmp_path)) + "/rel.yaml"
    with mock_git(r, r"diff --name-only", f"{norm_path}\n"), mock_git(r, r"log.*", ""):
        with pytest.raises(AssertionError, match=".*of entries.*"):
            r.run()


def test_main(tmp_path):
    sys.argv = ("whatever", "--lint", "--debug")
    main()

    sys.argv = ("whatever", "--rel-notes-dir=notexist")
    with pytest.raises(FileNotFoundError):
        main()

    with open(tmp_path / "rel.yaml", "w") as f:
        f.write("releaxxxxx: rel")

    sys.argv = ("whatever", "--rel-notes-dir", str(tmp_path), "--lint")
    with pytest.raises(SystemExit):
        main()


def test_args():
    args = parse_args(["--version", "4.5.6", "--debug", "--previous", "4.5.1", "--lint"])
    assert args.version == "4.5.6"
    assert args.previous == "4.5.1"
    assert args.lint
    assert args.debug
    args = parse_args(["--version", "4.5.6", "--debug", "--previous", "4.5.1", "--yaml"])
    assert args.yaml
    args = parse_args(["--version", "4.5.6", "--debug", "--previous", "4.5.1", "--blame"])
    assert args.blame