diff --git a/scripts/review_pr.py b/scripts/review_pr.py new file mode 100644 index 0000000..47ab10e --- /dev/null +++ b/scripts/review_pr.py @@ -0,0 +1,631 @@ +#!/usr/bin/env python3 +"""Automated PR reviewer for awesome-quant README contributions.""" + +from __future__ import annotations + +import argparse +import difflib +import http.client +import ipaddress +import os +import re +import socket +import ssl +import sys +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Callable +from urllib.parse import urlsplit, urlunsplit + +from github import Github, GithubException + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from scripts.readme_entries import ( + ENTRY_RE, + GITHUB_LINK_RE, + MARKDOWN_URL_RE, + VALID_SECTIONS, + extract_languages, +) + + +NO_TAG_SECTIONS = { + "Commercial & Proprietary Services", + "Cross-Language Frameworks", + "Reproducing Works, Training & Books", + "Related Lists", +} + +RECENT_CLOSED_PULL_DAYS = 365 +CHECK_ORDER = ( + "description", + "files", + "entry-count", + "content", + "format", + "placement", + "tags", + "period", + "url", + "github-link", + "github", + "activity", + "documentation", + "reachability", + "duplicates", +) + + +@dataclass(frozen=True) +class Finding: + check: str + detail: str + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Review one awesome-quant pull request." + ) + parser.add_argument("--pr-number", type=int, default=None) + return parser.parse_args() + + +def env(name: str) -> str | None: + value = os.environ.get(name, "").strip() + return value or None + + +def fail(message: str) -> int: + print(f"ERROR {message}", file=sys.stderr) + return 2 + + +def normalize(text: str) -> str: + return " ".join(text.casefold().split()) + + +def canonicalize_url(url: str) -> str: + parsed = urlsplit(url.strip()) + scheme = parsed.scheme.casefold() + hostname = (parsed.hostname or "").casefold() + port = parsed.port + if port and not ( + (scheme == "https" and port == 443) + or (scheme == "http" and port == 80) + ): + netloc = f"{hostname}:{port}" + else: + netloc = hostname + path = parsed.path.rstrip("/") or "/" + return urlunsplit((scheme, netloc, path, parsed.query, "")) + + +def parse_github_repository_url(url: str) -> tuple[str, str] | None: + parsed = urlsplit(url.strip()) + if ( + parsed.scheme.casefold() != "https" + or parsed.hostname != "github.com" + or parsed.port is not None + or parsed.query + or parsed.fragment + ): + return None + path_parts = [part for part in parsed.path.split("/") if part] + if len(path_parts) != 2: + return None + return path_parts[0], path_parts[1] + + +def description_ends_with_period(description: str) -> bool: + github_match = GITHUB_LINK_RE.search(description) + text = description[: github_match.start()].rstrip() if github_match else description + previous = None + while previous != text: + previous = text + text = re.sub(r"\s*\[[^\]]+\]\([^)]+\)\s*$", "", text).rstrip() + text = re.sub( + r"\s*\(\[[^\]]+\]\([^)]+\)\)\s*$", + "", + text, + ).rstrip() + return bool(text) and text.endswith(".") + + +def parse_patch(patch: str | None) -> list[tuple[str, str]]: + if not patch: + return [] + + result: list[tuple[str, str]] = [] + section = "" + for raw_line in patch.splitlines(): + if raw_line.startswith("@@"): + continue + if raw_line.startswith("+"): + line = raw_line[1:] + stripped = line.strip() + if stripped.startswith("## "): + section = stripped[3:].strip() + result.append((section, line)) + elif raw_line.startswith(" "): + line = raw_line[1:] + stripped = line.strip() + if stripped.startswith("## "): + section = stripped[3:].strip() + return result + + +def extract_entry_line(added_lines: list[tuple[str, str]]) -> tuple[str, str] | None: + entry_lines = [item for item in added_lines if item[1].strip().startswith("- ")] + if len(entry_lines) != 1: + return None + return entry_lines[0] + + +def read_readme(repository: Any, ref: str) -> str: + content = repository.get_contents("README.md", ref=ref) + if isinstance(content, list): + raise RuntimeError(f"README.md at {ref} did not resolve to a file") + return content.decoded_content.decode("utf-8") + + +def analyze_readme_change( + base_readme: str, + head_readme: str, +) -> tuple[str | None, list[Finding]]: + added_lines, removed_lines = readme_changed_lines(base_readme, head_readme) + substantive_added = [line for line in added_lines if line.strip()] + substantive_removed = [line for line in removed_lines if line.strip()] + entry_lines = [ + line for line in substantive_added if line.strip().startswith("- ") + ] + findings: list[Finding] = [] + if len(entry_lines) != 1: + findings.append( + Finding( + "entry-count", + "expected exactly one added README entry line", + ) + ) + return None, findings + + unauthorized_additions = [ + line for line in substantive_added if line != entry_lines[0] + ] + if substantive_removed or unauthorized_additions: + findings.append( + Finding( + "content", + "README changes must add one entry without other substantive edits", + ) + ) + return entry_lines[0], findings + + +def readme_changed_lines( + base_readme: str, + head_readme: str, +) -> tuple[list[str], list[str]]: + base_lines = base_readme.splitlines() + head_lines = head_readme.splitlines() + added_lines: list[str] = [] + removed_lines: list[str] = [] + matcher = difflib.SequenceMatcher( + a=base_lines, + b=head_lines, + autojunk=False, + ) + for operation, base_start, base_end, head_start, head_end in matcher.get_opcodes(): + if operation in {"replace", "delete"}: + removed_lines.extend(base_lines[base_start:base_end]) + if operation in {"replace", "insert"}: + added_lines.extend(head_lines[head_start:head_end]) + return added_lines, removed_lines + + +def find_entry_section(readme_text: str, entry_line: str) -> str: + current_section = "" + matches = 0 + matched_section = "" + for line in readme_text.splitlines(): + stripped = line.strip() + if stripped.startswith("## ") and not stripped.startswith("### "): + current_section = stripped[3:].strip() + if line == entry_line: + matches += 1 + matched_section = current_section + if matches != 1: + raise RuntimeError( + "added README entry could not be located uniquely in the PR head" + ) + return matched_section + + +class PinnedHTTPSConnection(http.client.HTTPSConnection): + """HTTPS connection pinned to an address that was checked as public.""" + + def __init__(self, hostname: str, ip_address: str, port: int) -> None: + context = ssl.create_default_context() + super().__init__( + hostname, + port=port, + timeout=10, + context=context, + ) + self.ip_address = ip_address + self.ssl_context = context + + def connect(self) -> None: + sock = socket.create_connection( + (self.ip_address, self.port), + self.timeout, + ) + self.sock = self.ssl_context.wrap_socket(sock, server_hostname=self.host) + + +def request_url_status( + hostname: str, + ip_address: str, + port: int, + target: str, + method: str, +) -> int: + connection = PinnedHTTPSConnection(hostname, ip_address, port) + try: + connection.request( + method, + target, + headers={ + "Accept": "*/*", + "User-Agent": "awesome-quant-pr-review", + }, + ) + return connection.getresponse().status + finally: + connection.close() + + +def url_reachable( + url: str, + *, + resolver: Callable[..., list[tuple[Any, ...]]] = socket.getaddrinfo, + requester: Callable[[str, str, int, str, str], int] = request_url_status, +) -> bool: + try: + parsed = urlsplit(url) + if ( + parsed.scheme.casefold() != "https" + or not parsed.hostname + or parsed.username is not None + or parsed.password is not None + ): + return False + parsed_port = parsed.port + if parsed_port == 0: + return False + port = parsed_port or 443 + addresses = { + result[4][0] + for result in resolver( + parsed.hostname, + port, + type=socket.SOCK_STREAM, + ) + } + address_objects = [ipaddress.ip_address(address) for address in addresses] + if not address_objects or any( + not address.is_global + or address.is_multicast + or address.is_loopback + or address.is_link_local + or address.is_private + or address.is_reserved + or address.is_unspecified + for address in address_objects + ): + return False + + target = parsed.path or "/" + if parsed.query: + target = f"{target}?{parsed.query}" + for method in ("HEAD", "GET"): + status = requester( + parsed.hostname, + sorted(addresses)[0], + port, + target, + method, + ) + if status not in {405, 501}: + return 200 <= status < 400 + return False + except (OSError, ValueError, ssl.SSLError): + return False + + +def readme_has_duplicate(readme_text: str, name: str, urls: list[str]) -> bool: + name_key = normalize(name) + url_keys = {canonicalize_url(url) for url in urls if url} + for line in readme_text.splitlines(): + match = ENTRY_RE.match(line) + if match and normalize(match.group(1)) == name_key: + return True + if match: + existing_urls = { + canonicalize_url(url) + for url in MARKDOWN_URL_RE.findall(line) + } + if url_keys & existing_urls: + return True + return False + + +def entry_line_has_duplicate(line: str, name: str, urls: list[str]) -> bool: + match = ENTRY_RE.match(line) + if not match: + return False + if normalize(match.group(1)) == normalize(name): + return True + target_urls = {canonicalize_url(url) for url in urls if url} + entry_urls = { + canonicalize_url(url) + for url in MARKDOWN_URL_RE.findall(line) + } + return bool(target_urls & entry_urls) + + +def pull_request_has_duplicate( + repository: Any, + pull_request: Any, + name: str, + urls: list[str], +) -> bool: + base_readme = read_readme(repository, pull_request.base.sha) + head_readme = read_readme(repository, pull_request.head.sha) + added_lines, _removed_lines = readme_changed_lines(base_readme, head_readme) + return any( + entry_line_has_duplicate(line, name, urls) + for line in added_lines + if line.strip().startswith("- ") + ) + + +def repository_has_pull_request_duplicate( + repository: Any, + current_pr_number: int, + name: str, + urls: list[str], + *, + now: datetime, +) -> bool: + cutoff = now - timedelta(days=RECENT_CLOSED_PULL_DAYS) + for state in ("open", "closed"): + pulls = repository.get_pulls( + state=state, + sort="updated", + direction="desc", + ) + for pull_request in pulls: + if pull_request.number == current_pr_number: + continue + if state == "closed": + if pull_request.updated_at < cutoff: + break + closed_at = pull_request.closed_at + if closed_at is None or closed_at < cutoff: + continue + if pull_request_has_duplicate(repository, pull_request, name, urls): + return True + return False + + +def review_pr( + repository_name: str, + pr_number: int, + client: Github, + *, + now: datetime | None = None, +) -> tuple[list[Finding], str]: + current_time = now or datetime.now(timezone.utc) + + repository = client.get_repo(repository_name) + pull_request = repository.get_pull(pr_number) + findings: list[Finding] = [] + + if not (pull_request.body or "").strip(): + findings.append(Finding("description", "PR body is empty")) + + files = list(pull_request.get_files()) + if len(files) != 1 or files[0].filename != "README.md": + changed = ", ".join(file.filename for file in files) or "none" + findings.append( + Finding("files", f"only README.md may change, found: {changed}") + ) + return findings, pull_request.title + + base_readme = read_readme(repository, pull_request.base.sha) + head_readme = read_readme(repository, pull_request.head.sha) + entry_line, change_findings = analyze_readme_change(base_readme, head_readme) + findings.extend(change_findings) + if entry_line is None: + return findings, pull_request.title + + line = entry_line + match = ENTRY_RE.match(line) + if not match: + findings.append( + Finding("format", "added README bullet does not match the entry regex") + ) + return findings, pull_request.title + + section = find_entry_section(head_readme, line) + name = match.group(1).strip() + url = match.group(2).strip() + tail = match.group(3).strip() + tags, clean_description = extract_languages(tail) + + if section not in VALID_SECTIONS: + findings.append( + Finding("placement", f"entry is under unknown section {section!r}") + ) + + if section not in NO_TAG_SECTIONS and not tags: + findings.append(Finding("tags", "missing required backtick tag prefix")) + + github_label_count = clean_description.count("[GitHub](") + github_marker = clean_description.rfind("[GitHub](") + github_suffix = ( + clean_description[github_marker:] + if github_marker >= 0 + else "" + ) + if github_label_count and ( + github_label_count != 1 + or GITHUB_LINK_RE.fullmatch(github_suffix) is None + ): + findings.append( + Finding( + "github-link", + "optional GitHub link must use " + "[GitHub](https://github.com/owner/repo)", + ) + ) + + if not description_ends_with_period(clean_description): + findings.append( + Finding( + "period", + "description must end with a period before the optional GitHub link", + ) + ) + + for markdown_url in MARKDOWN_URL_RE.findall(line): + if not markdown_url.startswith("https://"): + findings.append( + Finding("url", f"URL must use https://: {markdown_url}") + ) + + github_urls: list[str] = [] + primary_github = parse_github_repository_url(url) + if primary_github: + github_urls.append(url) + elif (urlsplit(url).hostname or "").casefold() == "github.com": + findings.append( + Finding("github", f"invalid GitHub repository URL: {url}") + ) + github_urls.extend(GITHUB_LINK_RE.findall(line)) + github_urls = list(dict.fromkeys(github_urls)) + if not github_urls: + findings.append( + Finding("github", "no GitHub repository URL found; cannot verify activity") + ) + else: + repository_parts = parse_github_repository_url(github_urls[0]) + if not repository_parts: + findings.append( + Finding( + "github", f"unable to parse GitHub repository URL: {github_urls[0]}" + ) + ) + else: + owner, repo_name = repository_parts + github_repo = client.get_repo(f"{owner}/{repo_name}") + if github_repo.archived: + findings.append(Finding("activity", "repository is archived")) + pushed_at = github_repo.pushed_at + if ( + pushed_at is None + or pushed_at < current_time - timedelta(days=365) + ): + findings.append( + Finding( + "activity", + "repository has not been updated within the last 365 days", + ) + ) + try: + github_repo.get_readme() + except GithubException as exc: + if exc.status != 404: + raise + findings.append( + Finding( + "documentation", + "repository does not have a README", + ) + ) + + if not url_reachable(url): + findings.append(Finding("reachability", f"primary URL is not reachable: {url}")) + + if readme_has_duplicate(base_readme, name, [url, *github_urls]): + findings.append( + Finding("duplicates", "project name or URL already exists in README.md") + ) + elif repository_has_pull_request_duplicate( + repository, + pr_number, + name, + [url, *github_urls], + now=current_time, + ): + findings.append( + Finding( + "duplicates", + "project name or URL already exists in an open or recently closed PR", + ) + ) + + return findings, pull_request.title + + +def main() -> int: + args = parse_args() + if args.pr_number is not None: + pr_number = args.pr_number + else: + raw_pr_number = env("PR_NUMBER") + if raw_pr_number is None: + return fail("PR number is required via --pr-number or PR_NUMBER") + try: + pr_number = int(raw_pr_number) + except ValueError: + return fail("PR_NUMBER must be an integer") + if pr_number <= 0: + return fail("PR number is required via --pr-number or PR_NUMBER") + + token = env("GITHUB_TOKEN") or env("GITHUB_ACCESS_TOKEN") + if not token: + return fail("GITHUB_TOKEN or GITHUB_ACCESS_TOKEN is required") + + try: + repository_name = env("GITHUB_REPOSITORY") + if not repository_name: + return fail("GITHUB_REPOSITORY is required") + client = Github(token) + findings, title = review_pr(repository_name, pr_number, client) + except Exception as exc: + return fail(str(exc)) + + print(f"PR #{pr_number}: {title}") + print("Entries reviewed: 1") + if not findings: + for check in CHECK_ORDER: + print(f"- {check}: pass") + print("Verdict: APPROVE") + print("Recommended action: merge") + return 0 + + print("Findings:") + for finding in findings: + print(f"- {finding.check}: fail - {finding.detail}") + print("Verdict: NEEDS CHANGES") + print("Recommended action: no action") + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..d226956 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Tests for awesome-quant maintenance tooling.""" diff --git a/tests/test_review_pr.py b/tests/test_review_pr.py new file mode 100644 index 0000000..7321406 --- /dev/null +++ b/tests/test_review_pr.py @@ -0,0 +1,844 @@ +import io +import unittest +from contextlib import redirect_stderr, redirect_stdout +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace +from unittest.mock import patch + +from github import GithubException + +from scripts.review_pr import ( + Finding, + main, + readme_has_duplicate, + review_pr, + url_reachable, +) + + +NOW = datetime(2026, 8, 10, tzinfo=timezone.utc) +ENTRY_URL = "https://github.com/example/fresh" +VALID_PATCH = """@@ -1,1 +1,2 @@ + ## Trading & Backtesting ++- [Fresh](https://github.com/example/fresh) - `Python` - Fresh project. +""" + + +class FakePull: + def __init__( + self, + number, + *, + body="A useful contribution.", + files=None, + state="open", + closed_at=None, + title="Add Fresh", + base_sha=None, + head_sha=None, + base_readme=None, + head_readme=None, + content_error=False, + ): + self.number = number + self.body = body + self.state = state + self.closed_at = closed_at + self.updated_at = closed_at or NOW + self.title = title + self.base = SimpleNamespace( + sha=base_sha or ("base-sha" if number == 10 else f"base-{number}") + ) + self.head = SimpleNamespace( + sha=head_sha or ("head-sha" if number == 10 else f"head-{number}") + ) + self._files = files if files is not None else [ + SimpleNamespace(filename="README.md", patch=VALID_PATCH) + ] + self.base_readme = base_readme or ( + "# awesome-quant\n\n" + "## Trading & Backtesting\n" + ) + added_lines = [ + line[1:] + for changed_file in self._files + if changed_file.filename == "README.md" and changed_file.patch + for line in changed_file.patch.splitlines() + if line.startswith("+- ") + ] + self.head_readme = head_readme or ( + self.base_readme.rstrip() + "\n" + "\n".join(added_lines) + "\n" + ) + self.content_error = content_error + + def get_files(self): + return list(self._files) + + +class FakeProjectRepository: + def __init__(self): + self.archived = False + self.pushed_at = NOW + self.has_root_readme = True + self.has_readme = True + self.readme_error = None + + def get_contents(self, path): + if path != "README.md": + raise AssertionError(f"unexpected project path: {path}") + if not self.has_root_readme: + raise RuntimeError("README not found") + return SimpleNamespace(decoded_content=b"# Fresh") + + def get_readme(self): + if self.readme_error: + raise self.readme_error + if not self.has_readme: + raise GithubException(404, {"message": "Not Found"}) + return SimpleNamespace(decoded_content=b"# Fresh") + + +class FakeBaseRepository: + default_branch = "main" + + def __init__( + self, + pull, + *, + other_pulls=(), + base_readme=( + "# awesome-quant\n\n" + "## Trading & Backtesting\n" + ), + head_readme=None, + ): + self.pull = pull + self.other_pulls = list(other_pulls) + self.base_readme = base_readme + self.head_readme = head_readme or ( + base_readme.rstrip() + + "\n" + + VALID_PATCH.splitlines()[-1][1:] + + "\n" + ) + self.pull_query_error = None + self.content_refs = [] + + def get_pull(self, number): + if number != self.pull.number: + raise AssertionError(f"unexpected PR number: {number}") + return self.pull + + def get_contents(self, path, ref=None): + if path != "README.md": + raise AssertionError(f"unexpected base content request: {path}, {ref}") + self.content_refs.append(ref) + if ref in {self.default_branch, "base-sha"}: + content = self.base_readme + elif ref == "head-sha": + content = self.head_readme + else: + matching_pull = next( + ( + pull + for pull in self.other_pulls + if ref in {pull.base.sha, pull.head.sha} + ), + None, + ) + if matching_pull is None: + raise AssertionError(f"unexpected README ref: {ref}") + if matching_pull.content_error: + raise RuntimeError("candidate README content should not be fetched") + content = ( + matching_pull.base_readme + if ref == matching_pull.base.sha + else matching_pull.head_readme + ) + return SimpleNamespace(decoded_content=content.encode()) + + def get_pulls(self, **kwargs): + if self.pull_query_error: + raise self.pull_query_error + state = kwargs.get("state") + return [pull for pull in self.other_pulls if pull.state == state] + + +class FakeClient: + def __init__(self, repository): + self.repository = repository + self.project_repository = FakeProjectRepository() + self.requested_repositories = [] + + def get_repo(self, name): + self.requested_repositories.append(name) + if name == "owner/list": + return self.repository + if name == "example/fresh": + return self.project_repository + raise AssertionError(f"unexpected repository: {name}") + + +class ReadmeDuplicateTests(unittest.TestCase): + def test_rejects_exact_name_when_url_differs(self): + readme = ( + "## Trading & Backtesting\n\n" + "- [Example](https://github.com/example/old) - `Python` - Existing project.\n" + ) + + self.assertTrue( + readme_has_duplicate( + readme, + "Example", + ["https://github.com/example/new"], + ) + ) + + def test_does_not_match_url_prefix(self): + readme = ( + "## Trading & Backtesting\n\n" + "- [Other](https://github.com/example/freshness) - " + "`Python` - Existing project.\n" + ) + + self.assertFalse( + readme_has_duplicate( + readme, + "Fresh", + ["https://github.com/example/fresh"], + ) + ) + + def test_matches_canonical_trailing_slash(self): + readme = ( + "## Trading & Backtesting\n\n" + "- [Fresh](https://github.com/example/fresh/) - " + "`Python` - Existing project.\n" + ) + + self.assertTrue( + readme_has_duplicate( + readme, + "Other Name", + ["https://github.com/example/fresh"], + ) + ) + + + +class UrlReachabilityTests(unittest.TestCase): + @staticmethod + def address(ip_address): + return [(2, 1, 6, "", (ip_address, 443))] + + def test_rejects_non_https_url_without_network_request(self): + requester = unittest.mock.Mock() + + self.assertFalse( + url_reachable( + "http://example.com/project", + resolver=lambda *_args, **_kwargs: self.address("93.184.216.34"), + requester=requester, + ) + ) + requester.assert_not_called() + + def test_rejects_private_address_without_network_request(self): + requester = unittest.mock.Mock() + + self.assertFalse( + url_reachable( + "https://localhost/project", + resolver=lambda *_args, **_kwargs: self.address("127.0.0.1"), + requester=requester, + ) + ) + requester.assert_not_called() + + def test_rejects_mixed_public_and_private_dns_answers(self): + requester = unittest.mock.Mock() + + self.assertFalse( + url_reachable( + "https://example.com/project", + resolver=lambda *_args, **_kwargs: ( + self.address("93.184.216.34") + self.address("10.0.0.1") + ), + requester=requester, + ) + ) + requester.assert_not_called() + + def test_rejects_multicast_address_without_network_request(self): + requester = unittest.mock.Mock() + + self.assertFalse( + url_reachable( + "https://example.com/project", + resolver=lambda *_args, **_kwargs: self.address("224.0.0.1"), + requester=requester, + ) + ) + requester.assert_not_called() + + def test_rejects_port_zero_without_network_request(self): + requester = unittest.mock.Mock() + + self.assertFalse( + url_reachable( + "https://example.com:0/project", + resolver=lambda *_args, **_kwargs: self.address("93.184.216.34"), + requester=requester, + ) + ) + requester.assert_not_called() + + def test_accepts_public_url_without_following_redirect(self): + requester = unittest.mock.Mock(return_value=302) + + self.assertTrue( + url_reachable( + "https://example.com/project?source=test", + resolver=lambda *_args, **_kwargs: self.address("93.184.216.34"), + requester=requester, + ) + ) + requester.assert_called_once_with( + "example.com", + "93.184.216.34", + 443, + "/project?source=test", + "HEAD", + ) + + def test_retries_with_get_when_head_is_not_supported(self): + requester = unittest.mock.Mock(side_effect=[405, 200]) + + self.assertTrue( + url_reachable( + "https://example.com/project", + resolver=lambda *_args, **_kwargs: self.address("93.184.216.34"), + requester=requester, + ) + ) + self.assertEqual( + [call.args[-1] for call in requester.call_args_list], + ["HEAD", "GET"], + ) + + +class PullRequestDuplicateTests(unittest.TestCase): + def review(self, *, other_pulls=()): + repository = FakeBaseRepository( + FakePull(10), + other_pulls=other_pulls, + ) + client = FakeClient(repository) + with patch("scripts.review_pr.url_reachable", return_value=True): + findings, title = review_pr( + "owner/list", + 10, + client, + now=NOW, + ) + return findings, title, client + + def test_repository_name_is_an_explicit_input(self): + findings, title, client = self.review() + + self.assertEqual(findings, []) + self.assertEqual(title, "Add Fresh") + self.assertEqual(client.requested_repositories[0], "owner/list") + + def test_rejects_duplicate_in_open_pull_request(self): + duplicate = FakePull( + 9, + body=f"Previously proposed {ENTRY_URL}", + ) + + findings, _title, _client = self.review(other_pulls=[duplicate]) + + self.assertIn("duplicates", {finding.check for finding in findings}) + + def test_rejects_duplicate_in_recently_closed_pull_request(self): + duplicate = FakePull( + 8, + body="Previously proposed Fresh", + state="closed", + closed_at=NOW - timedelta(days=30), + ) + + findings, _title, _client = self.review(other_pulls=[duplicate]) + + self.assertIn("duplicates", {finding.check for finding in findings}) + + def test_ignores_old_closed_pull_request(self): + old_duplicate = FakePull( + 7, + body=f"Previously proposed {ENTRY_URL}", + state="closed", + closed_at=NOW - timedelta(days=366), + content_error=True, + ) + + findings, _title, _client = self.review(other_pulls=[old_duplicate]) + + self.assertEqual(findings, []) + + def test_uses_full_readmes_when_candidate_patch_is_truncated(self): + duplicate = FakePull( + 4, + files=[ + SimpleNamespace( + filename="README.md", + patch="@@ -500,0 +501,1 @@\n context only", + ) + ], + head_readme=( + "# awesome-quant\n\n" + "## Trading & Backtesting\n" + "- [Fresh](https://github.com/example/fresh) - " + "`Python` - Fresh project.\n" + ), + ) + + findings, _title, _client = self.review(other_pulls=[duplicate]) + + self.assertIn("duplicates", {finding.check for finding in findings}) + + def test_does_not_match_project_name_inside_unrelated_word(self): + unrelated_patch = """@@ -1,1 +1,2 @@ + ## Trading & Backtesting ++- [Other](https://github.com/example/other) - `Python` - Other project. +""" + unrelated = FakePull( + 6, + title="Maintenance", + body="Refresh metadata for the list.", + files=[ + SimpleNamespace( + filename="README.md", + patch=unrelated_patch, + ) + ], + ) + + findings, _title, _client = self.review(other_pulls=[unrelated]) + + self.assertEqual(findings, []) + + def test_requires_matching_entry_not_matching_title(self): + unrelated_patch = """@@ -1,1 +1,2 @@ + ## Trading & Backtesting ++- [Other](https://github.com/example/other) - `Python` - Other project. +""" + unrelated = FakePull( + 5, + title="Fresh ideas for the list", + body="A maintenance proposal.", + files=[ + SimpleNamespace( + filename="README.md", + patch=unrelated_patch, + ) + ], + ) + + findings, _title, _client = self.review(other_pulls=[unrelated]) + + self.assertEqual(findings, []) + + def test_pull_request_search_errors_fail_closed(self): + repository = FakeBaseRepository(FakePull(10)) + repository.pull_query_error = RuntimeError("pull search failed") + client = FakeClient(repository) + + with ( + patch("scripts.review_pr.url_reachable", return_value=True), + self.assertRaisesRegex(RuntimeError, "pull search failed"), + ): + review_pr("owner/list", 10, client, now=NOW) + + +class ValidationPipelineTests(unittest.TestCase): + def review( + self, + *, + patch_text=VALID_PATCH, + body="A useful contribution.", + files=None, + base_readme=( + "# awesome-quant\n\n" + "## Trading & Backtesting\n" + ), + head_readme=None, + reachable=True, + configure_project=None, + ): + changed_files = files or [ + SimpleNamespace(filename="README.md", patch=patch_text) + ] + if head_readme is None: + section = "Trading & Backtesting" + added_lines = [] + for raw_line in patch_text.splitlines(): + if raw_line.startswith(" ## "): + section = raw_line[4:] + elif raw_line.startswith("+- "): + added_lines.append(raw_line[1:]) + heading = ( + "" + if f"## {section}" in base_readme + else f"\n## {section}\n" + ) + head_readme = ( + base_readme.rstrip() + + heading + + "\n" + + "\n".join(added_lines) + + "\n" + ) + repository = FakeBaseRepository( + FakePull(10, body=body, files=changed_files), + base_readme=base_readme, + head_readme=head_readme, + ) + client = FakeClient(repository) + if configure_project: + configure_project(client.project_repository) + with patch( + "scripts.review_pr.url_reachable", + return_value=reachable, + ): + findings, _title = review_pr( + "owner/list", + 10, + client, + now=NOW, + ) + return {finding.check for finding in findings} + + def test_uses_pinned_base_and_head_readmes(self): + repository = FakeBaseRepository(FakePull(10)) + client = FakeClient(repository) + + with patch("scripts.review_pr.url_reachable", return_value=True): + review_pr("owner/list", 10, client, now=NOW) + + self.assertIn("base-sha", repository.content_refs) + self.assertIn("head-sha", repository.content_refs) + self.assertNotIn("main", repository.content_refs) + + def test_accepts_entry_when_heading_is_outside_patch_context(self): + patch_text = """@@ -20,0 +21,1 @@ ++- [Fresh](https://github.com/example/fresh) - `Python` - Fresh project. +""" + head_readme = ( + "# awesome-quant\n\n" + "## Trading & Backtesting\n\n" + "- [Existing](https://github.com/example/existing) - " + "`Python` - Existing project.\n" + "- [Fresh](https://github.com/example/fresh) - " + "`Python` - Fresh project.\n" + ) + base_readme = head_readme.replace( + "- [Fresh](https://github.com/example/fresh) - " + "`Python` - Fresh project.\n", + "", + ) + + self.assertNotIn( + "placement", + self.review( + patch_text=patch_text, + base_readme=base_readme, + head_readme=head_readme, + ), + ) + + def test_rejects_deleting_an_existing_entry(self): + base_readme = ( + "# awesome-quant\n\n" + "## Trading & Backtesting\n\n" + "- [Existing](https://github.com/example/existing) - " + "`Python` - Existing project.\n" + ) + head_readme = ( + "# awesome-quant\n\n" + "## Trading & Backtesting\n\n" + "- [Fresh](https://github.com/example/fresh) - " + "`Python` - Fresh project.\n" + ) + + self.assertIn( + "content", + self.review( + base_readme=base_readme, + head_readme=head_readme, + ), + ) + + def test_accepts_valid_entry(self): + self.assertEqual(self.review(), set()) + + def test_rejects_empty_pr_description(self): + self.assertIn("description", self.review(body=" ")) + + def test_rejects_changes_outside_readme(self): + files = [ + SimpleNamespace(filename="README.md", patch=VALID_PATCH), + SimpleNamespace(filename="code.py", patch="+print(1)"), + ] + self.assertIn("files", self.review(files=files)) + + def test_rejects_multiple_entries(self): + patch_text = VALID_PATCH + ( + "+- [Other](https://github.com/example/other) - " + "`Python` - Other project.\n" + ) + self.assertIn("entry-count", self.review(patch_text=patch_text)) + + def test_rejects_malformed_entry(self): + patch_text = """@@ -1,1 +1,2 @@ + ## Trading & Backtesting ++- Fresh project without Markdown links +""" + self.assertIn("format", self.review(patch_text=patch_text)) + + def test_rejects_unknown_section(self): + self.assertIn( + "placement", + self.review( + patch_text=VALID_PATCH.replace( + "Trading & Backtesting", + "Unknown Section", + ) + ), + ) + + def test_rejects_missing_tags(self): + self.assertIn( + "tags", + self.review( + patch_text=VALID_PATCH.replace( + "`Python` - ", + "", + ) + ), + ) + + def test_rejects_tags_without_required_separator(self): + self.assertIn( + "tags", + self.review( + patch_text=VALID_PATCH.replace( + "`Python` - Fresh", + "`Python` Fresh", + ) + ), + ) + + def test_rejects_description_without_period(self): + self.assertIn( + "period", + self.review( + patch_text=VALID_PATCH.replace( + "Fresh project.", + "Fresh project", + ) + ), + ) + + def test_rejects_insecure_primary_url(self): + self.assertIn( + "url", + self.review( + patch_text=VALID_PATCH.replace( + "https://github.com/example/fresh", + "http://github.com/example/fresh", + ) + ), + ) + + def test_rejects_insecure_trailing_url(self): + patch_text = VALID_PATCH.replace( + "Fresh project.", + "Fresh project. [Website](http://example.com)", + ) + self.assertIn("url", self.review(patch_text=patch_text)) + + def test_rejects_malformed_github_suffix(self): + patch_text = VALID_PATCH.replace( + "Fresh project.", + "Fresh project. [GitHub](http://github.com/example/fresh)", + ) + self.assertIn("github-link", self.review(patch_text=patch_text)) + + def test_rejects_github_link_that_is_not_the_suffix(self): + patch_text = VALID_PATCH.replace( + "Fresh project.", + ( + "Fresh project. " + "[GitHub](https://github.com/example/fresh) trailing text" + ), + ) + self.assertIn("github-link", self.review(patch_text=patch_text)) + + def test_rejects_entry_without_github_repository(self): + patch_text = VALID_PATCH.replace( + "https://github.com/example/fresh", + "https://example.com/fresh", + ) + self.assertIn("github", self.review(patch_text=patch_text)) + + def test_rejects_non_repository_github_path(self): + patch_text = VALID_PATCH.replace( + "https://github.com/example/fresh", + "https://github.com/example/fresh/issues", + ) + self.assertIn("github", self.review(patch_text=patch_text)) + + def test_rejects_archived_repository(self): + def archive(repository): + repository.archived = True + + self.assertIn( + "activity", + self.review(configure_project=archive), + ) + + def test_rejects_stale_repository(self): + def make_stale(repository): + repository.pushed_at = NOW - timedelta(days=366) + + self.assertIn( + "activity", + self.review(configure_project=make_stale), + ) + + def test_rejects_repository_without_readme(self): + def remove_readme(repository): + repository.has_root_readme = False + repository.has_readme = False + + self.assertIn( + "documentation", + self.review(configure_project=remove_readme), + ) + + def test_accepts_alternate_readme_name(self): + def use_alternate_readme(repository): + repository.has_root_readme = False + repository.has_readme = True + + self.assertNotIn( + "documentation", + self.review(configure_project=use_alternate_readme), + ) + + def test_readme_api_error_fails_closed(self): + def fail_readme_lookup(repository): + repository.readme_error = GithubException( + 500, + {"message": "Server Error"}, + ) + + with self.assertRaises(GithubException): + self.review(configure_project=fail_readme_lookup) + + def test_rejects_unreachable_primary_url(self): + self.assertIn("reachability", self.review(reachable=False)) + + def test_rejects_duplicate_in_base_readme(self): + base_readme = ( + "## Trading & Backtesting\n" + "- [Fresh](https://github.com/example/old) - " + "`Python` - Existing project.\n" + ) + self.assertIn( + "duplicates", + self.review(base_readme=base_readme), + ) + + +class MainTests(unittest.TestCase): + def run_main(self, review_result): + stdout = io.StringIO() + stderr = io.StringIO() + environment = { + "GITHUB_TOKEN": "token", + "GITHUB_REPOSITORY": "owner/list", + "PR_NUMBER": "10", + } + with ( + patch.dict("os.environ", environment, clear=True), + patch("sys.argv", ["review_pr.py"]), + patch("scripts.review_pr.Github"), + patch("scripts.review_pr.review_pr", return_value=review_result), + redirect_stdout(stdout), + redirect_stderr(stderr), + ): + result = main() + return result, stdout.getvalue(), stderr.getvalue() + + def test_success_reports_passed_checks(self): + result, stdout, _stderr = self.run_main(([], "Add Fresh")) + + self.assertEqual(result, 0) + self.assertIn("description: pass", stdout) + self.assertIn("duplicates: pass", stdout) + + def test_failure_reports_failed_check_and_nonzero_status(self): + finding = Finding("description", "PR body is empty") + + result, stdout, _stderr = self.run_main(([finding], "Add Fresh")) + + self.assertEqual(result, 1) + self.assertIn("description: fail - PR body is empty", stdout) + + def test_api_error_fails_closed(self): + stdout = io.StringIO() + stderr = io.StringIO() + environment = { + "GITHUB_TOKEN": "token", + "GITHUB_REPOSITORY": "owner/list", + "PR_NUMBER": "10", + } + with ( + patch.dict("os.environ", environment, clear=True), + patch("sys.argv", ["review_pr.py"]), + patch("scripts.review_pr.Github"), + patch( + "scripts.review_pr.review_pr", + side_effect=RuntimeError("API failed"), + ), + redirect_stdout(stdout), + redirect_stderr(stderr), + ): + result = main() + + self.assertEqual(result, 2) + self.assertIn("ERROR API failed", stderr.getvalue()) + + def test_invalid_pr_number_fails_closed(self): + stdout = io.StringIO() + stderr = io.StringIO() + environment = { + "GITHUB_TOKEN": "token", + "GITHUB_REPOSITORY": "owner/list", + "PR_NUMBER": "not-a-number", + } + with ( + patch.dict("os.environ", environment, clear=True), + patch("sys.argv", ["review_pr.py"]), + redirect_stdout(stdout), + redirect_stderr(stderr), + ): + result = main() + + self.assertEqual(result, 2) + self.assertIn("ERROR PR_NUMBER must be an integer", stderr.getvalue()) + + +if __name__ == "__main__": + unittest.main()