Last active
November 3, 2021 15:11
-
-
Save earonesty/eb38cecabf7c629cd8ed7d04e6aa1bf1 to your computer and use it in GitHub Desktop.
reno reporter & linter that is far faster and leverages filtered git logs to handle topology for you
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment