treefmt: format Python with ruff

Add ruff to the tree formatter so Python sources are kept consistently formatted by 'nix fmt' and the CI format check. Test fixtures under tests/modules are excluded since some are intentionally invalid Python. Reformats the existing lib/python and tests scripts accordingly.
2026-06-19 10:00:25 -05:00
parent 6aaf73d18d
commit fea021792d
9 changed files with 302 additions and 119 deletions
@@ -17,52 +17,67 @@ def main():
    print("🔍 Checking for duplicate maintainers between HM and nixpkgs...")

    # Get home-manager maintainers
-    hm_result = subprocess.run(['nix', 'eval', '--file', 'modules/lib/maintainers.nix', '--json'],
-                              capture_output=True, text=True, check=True)
+    hm_result = subprocess.run(
+        ["nix", "eval", "--file", "modules/lib/maintainers.nix", "--json"],
+        capture_output=True,
+        text=True,
+        check=True,
+    )
    hm_maintainers = json.loads(hm_result.stdout)
    hm_github_users = set()
    for name, data in hm_maintainers.items():
-        if 'github' in data:
-            hm_github_users.add(data['github'])
+        if "github" in data:
+            hm_github_users.add(data["github"])

    # Read nixpkgs revision from flake.lock to ensure consistency between local and CI
-    flake_lock_path = Path('flake.lock')
-    with open(flake_lock_path, 'r') as f:
+    flake_lock_path = Path("flake.lock")
+    with open(flake_lock_path, "r") as f:
        flake_lock = json.load(f)
-        nixpkgs_rev = flake_lock['nodes']['nixpkgs']['locked']['rev']
+        nixpkgs_rev = flake_lock["nodes"]["nixpkgs"]["locked"]["rev"]

    print(f"📌 Using nixpkgs from flake.lock: {nixpkgs_rev[:7]}")

    # Get nixpkgs maintainers from the locked revision
    nixpkgs_result = subprocess.run(
-        ['nix', 'eval', f'github:NixOS/nixpkgs/{nixpkgs_rev}#lib.maintainers', '--json'],
-        capture_output=True, text=True, check=True
+        [
+            "nix",
+            "eval",
+            f"github:NixOS/nixpkgs/{nixpkgs_rev}#lib.maintainers",
+            "--json",
+        ],
+        capture_output=True,
+        text=True,
+        check=True,
    )
    nixpkgs_maintainers = json.loads(nixpkgs_result.stdout)
    nixpkgs_github_users = set()
    for name, data in nixpkgs_maintainers.items():
-        if isinstance(data, dict) and 'github' in data:
-            nixpkgs_github_users.add(data['github'])
+        if isinstance(data, dict) and "github" in data:
+            nixpkgs_github_users.add(data["github"])

    # Find duplicates
    duplicates = hm_github_users.intersection(nixpkgs_github_users)

    if duplicates:
-        print(f'❌ Found {len(duplicates)} duplicate maintainers between HM and nixpkgs:')
+        print(
+            f"❌ Found {len(duplicates)} duplicate maintainers between HM and nixpkgs:"
+        )
        for github_user in sorted(duplicates):
            # Find the HM attribute name for this github user
            hm_attr = None
            for attr_name, data in hm_maintainers.items():
-                if data.get('github') == github_user:
+                if data.get("github") == github_user:
                    hm_attr = attr_name
                    break
-            print(f'  - {github_user} (HM attribute: {hm_attr})')
+            print(f"  - {github_user} (HM attribute: {hm_attr})")
        print()
-        print('These maintainers should be removed from HM maintainers file to avoid duplication.')
-        print('They can be referenced directly from nixpkgs instead.')
+        print(
+            "These maintainers should be removed from HM maintainers file to avoid duplication."
+        )
+        print("They can be referenced directly from nixpkgs instead.")
        sys.exit(1)
    else:
-        print('✅ No duplicate maintainers found')
+        print("✅ No duplicate maintainers found")


 if __name__ == "__main__":
@@ -16,6 +16,7 @@ from pathlib import Path

 class NixEvalError(Exception):
    """Custom exception for errors during Nix evaluation."""
+
    pass


@@ -39,7 +40,9 @@ def run_nix_eval(nix_file: Path, *args: str) -> str:
        )
        return result.stdout.strip()
    except FileNotFoundError:
-        logging.error("'nix-instantiate' command not found. Is Nix installed and in your PATH?")
+        logging.error(
+            "'nix-instantiate' command not found. Is Nix installed and in your PATH?"
+        )
        raise NixEvalError("'nix-instantiate' not found")
    except subprocess.CalledProcessError as e:
        logging.error(f"Nix evaluation failed with exit code {e.returncode}")
@@ -40,6 +40,7 @@ def get_project_root() -> Path:
        # Assumes this script is at: <root>/flake/dev/generate-all-maintainers/
        return Path(__file__).parent.parent.parent.parent.resolve()

+
 class MetaMaintainerGenerator:
    """Generates maintainers list using meta.maintainers from Home Manager evaluation."""

@@ -54,9 +55,12 @@ class MetaMaintainerGenerator:
        print("🔍 Extracting maintainers using meta.maintainers...")

        try:
-            result = subprocess.run([
-                "nix", "eval", "--file", str(self.extractor_script), "--json"
-            ], capture_output=True, text=True, timeout=60)
+            result = subprocess.run(
+                ["nix", "eval", "--file", str(self.extractor_script), "--json"],
+                capture_output=True,
+                text=True,
+                timeout=60,
+            )

            if result.returncode == 0:
                data = json.loads(result.stdout)
@@ -87,7 +91,7 @@ class MetaMaintainerGenerator:
        print(f"🏠 Home Manager maintainers: {len(hm_maintainers)}")
        print(f"📦 Nixpkgs maintainers: {len(nixpkgs_maintainers)}")

-        with open(self.output_file, 'w') as f:
+        with open(self.output_file, "w") as f:
            f.write(
                inspect.cleandoc("""
              # Home Manager all maintainers list.
@@ -114,9 +118,12 @@ class MetaMaintainerGenerator:
    def validate_generated_file(self) -> bool:
        """Validate the generated Nix file syntax."""
        try:
-            result = subprocess.run([
-                'nix-instantiate', '--eval', str(self.output_file), '--strict'
-            ], capture_output=True, text=True, timeout=10)
+            result = subprocess.run(
+                ["nix-instantiate", "--eval", str(self.output_file), "--strict"],
+                capture_output=True,
+                text=True,
+                timeout=10,
+            )

            if result.returncode == 0:
                print("✅ Generated file has valid Nix syntax")
@@ -148,16 +155,16 @@ def main():
        description="Generate Home Manager all-maintainers.nix using meta.maintainers"
    )
    parser.add_argument(
-        '--root',
+        "--root",
        type=Path,
        default=None,
-        help='Path to Home Manager root (default: auto-detect)'
+        help="Path to Home Manager root (default: auto-detect)",
    )
    parser.add_argument(
-        '--output',
+        "--output",
        type=Path,
        default=None,
-        help='Output file path (default: <root>/all-maintainers.nix)'
+        help="Output file path (default: <root>/all-maintainers.nix)",
    )

    args = parser.parse_args()
@@ -60,6 +60,7 @@ query($owner: String!, $repo: String!, $prNumber: Int!) {

 class GHError(Exception):
    """Custom exception for errors related to 'gh' CLI commands."""
+
    pass


@@ -94,15 +95,28 @@ def get_manual_reviewer_actions(
        tuple: (manually_requested, manually_removed) sets of usernames
    """
    try:
-        result = run_gh_command([
-            "api", "graphql",
-            "-f", f"query={MANUAL_REVIEW_REQUEST_QUERY}",
-            "-F", f"owner={owner}",
-            "-F", f"repo={repo}",
-            "-F", f"prNumber={pr_number}",
-        ])
+        result = run_gh_command(
+            [
+                "api",
+                "graphql",
+                "-f",
+                f"query={MANUAL_REVIEW_REQUEST_QUERY}",
+                "-F",
+                f"owner={owner}",
+                "-F",
+                f"repo={repo}",
+                "-F",
+                f"prNumber={pr_number}",
+            ]
+        )
        data = json.loads(result.stdout)
-        nodes = data.get("data", {}).get("repository", {}).get("pullRequest", {}).get("timelineItems", {}).get("nodes", [])
+        nodes = (
+            data.get("data", {})
+            .get("repository", {})
+            .get("pullRequest", {})
+            .get("timelineItems", {})
+            .get("nodes", [])
+        )

        manually_requested = set()
        manually_removed = set()
@@ -143,7 +157,15 @@ def get_users_from_gh(args: list[str], error_message: str) -> set[str]:
 def get_pending_reviewers(pr_number: int) -> set[str]:
    """Gets the set of currently pending reviewers for a PR."""
    return get_users_from_gh(
-        ["pr", "view", str(pr_number), "--json", "reviewRequests", "--jq", ".reviewRequests[].login"],
+        [
+            "pr",
+            "view",
+            str(pr_number),
+            "--json",
+            "reviewRequests",
+            "--jq",
+            ".reviewRequests[].login",
+        ],
        "Error getting pending reviewers",
    )

@@ -151,7 +173,12 @@ def get_pending_reviewers(pr_number: int) -> set[str]:
 def get_past_reviewers(owner: str, repo: str, pr_number: int) -> set[str]:
    """Gets the set of users who have already reviewed the PR."""
    return get_users_from_gh(
-        ["api", f"repos/{owner}/{repo}/pulls/{pr_number}/reviews", "--jq", ".[].user.login"],
+        [
+            "api",
+            f"repos/{owner}/{repo}/pulls/{pr_number}/reviews",
+            "--jq",
+            ".[].user.login",
+        ],
        "Error getting past reviewers",
    )

@@ -162,17 +189,14 @@ def is_collaborator(owner: str, repo: str, username: str) -> bool:
    Handles 404 as a non-collaborator, while other errors are raised.
    """
    result = run_gh_command(
-        ["api", f"repos/{owner}/{repo}/collaborators/{username}"],
-        check=False
+        ["api", f"repos/{owner}/{repo}/collaborators/{username}"], check=False
    )

    if result.returncode == 0:
        return True

    if "HTTP 404" in result.stderr:
-        logging.error(
-            "'%s' is not a collaborator in this repository.", username
-        )
+        logging.error("'%s' is not a collaborator in this repository.", username)
        return False
    else:
        logging.error(
@@ -195,22 +219,32 @@ def update_reviewers(
    if reviewers_to_add:
        logging.info("Requesting reviews from: %s", ", ".join(reviewers_to_add))
        try:
-            run_gh_command([
-                "pr", "edit", str(pr_number),
-                "--add-reviewer", ",".join(reviewers_to_add)
-            ])
+            run_gh_command(
+                [
+                    "pr",
+                    "edit",
+                    str(pr_number),
+                    "--add-reviewer",
+                    ",".join(reviewers_to_add),
+                ]
+            )
        except GHError as e:
            logging.error("Failed to add reviewers: %s", e)

    if reviewers_to_remove and owner and repo:
-        logging.info("Removing review requests from: %s", ", ".join(reviewers_to_remove))
+        logging.info(
+            "Removing review requests from: %s", ", ".join(reviewers_to_remove)
+        )
        payload = json.dumps({"reviewers": list(reviewers_to_remove)})
        try:
            run_gh_command(
                [
-                    "api", "--method", "DELETE",
+                    "api",
+                    "--method",
+                    "DELETE",
                    f"repos/{owner}/{repo}/pulls/{pr_number}/requested_reviewers",
-                    "--input", "-",
+                    "--input",
+                    "-",
                ],
                input_data=payload,
            )
@@ -220,15 +254,33 @@ def update_reviewers(

 def main() -> None:
    """Main function to handle command-line arguments and manage reviewers."""
-    parser = argparse.ArgumentParser(description="Manage pull request reviewers for Home Manager.")
+    parser = argparse.ArgumentParser(
+        description="Manage pull request reviewers for Home Manager."
+    )
    parser.add_argument("--owner", required=True, help="Repository owner.")
    parser.add_argument("--repo", required=True, help="Repository name.")
-    parser.add_argument("--pr-number", type=int, required=True, help="Pull request number.")
+    parser.add_argument(
+        "--pr-number", type=int, required=True, help="Pull request number."
+    )
    parser.add_argument("--pr-author", required=True, help="PR author's username.")
-    parser.add_argument("--current-maintainers", default="", help="Space-separated list of maintainers for the changed files.")
-    parser.add_argument("--changed-files", default="", help="Newline-separated list of changed files.")
-    parser.add_argument("--bot-user-name", default="", help="Bot user name to distinguish manual vs automated review requests.")
-    parser.add_argument("--dry-run", action="store_true", help="Show what would be done without making actual changes.")
+    parser.add_argument(
+        "--current-maintainers",
+        default="",
+        help="Space-separated list of maintainers for the changed files.",
+    )
+    parser.add_argument(
+        "--changed-files", default="", help="Newline-separated list of changed files."
+    )
+    parser.add_argument(
+        "--bot-user-name",
+        default="",
+        help="Bot user name to distinguish manual vs automated review requests.",
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Show what would be done without making actual changes.",
+    )
    args = parser.parse_args()

    no_changed_files = not args.changed_files.strip()
@@ -237,13 +289,15 @@ def main() -> None:
    maintainers: set[str] = set(args.current_maintainers.split())
    pending_reviewers = get_pending_reviewers(args.pr_number)
    past_reviewers = get_past_reviewers(args.owner, args.repo, args.pr_number)
-    manually_requested, manually_removed = get_manual_reviewer_actions(args.owner, args.repo, args.pr_number, args.bot_user_name)
+    manually_requested, manually_removed = get_manual_reviewer_actions(
+        args.owner, args.repo, args.pr_number, args.bot_user_name
+    )

-    logging.info("File Maintainers: %s", ' '.join(maintainers) or "None")
-    logging.info("Pending Reviewers: %s", ' '.join(pending_reviewers) or "None")
-    logging.info("Past Reviewers: %s", ' '.join(past_reviewers) or "None")
-    logging.info("Manually Requested: %s", ' '.join(manually_requested) or "None")
-    logging.info("Manually Removed: %s", ' '.join(manually_removed) or "None")
+    logging.info("File Maintainers: %s", " ".join(maintainers) or "None")
+    logging.info("Pending Reviewers: %s", " ".join(pending_reviewers) or "None")
+    logging.info("Past Reviewers: %s", " ".join(past_reviewers) or "None")
+    logging.info("Manually Requested: %s", " ".join(manually_requested) or "None")
+    logging.info("Manually Removed: %s", " ".join(manually_removed) or "None")

    # --- 2. Determine reviewers to remove ---
    reviewers_to_remove: set[str] = set()
@@ -257,13 +311,15 @@ def main() -> None:

    if reviewers_to_remove:
        if args.dry_run:
-            logging.info("DRY RUN: Would remove reviewers: %s", ", ".join(reviewers_to_remove))
+            logging.info(
+                "DRY RUN: Would remove reviewers: %s", ", ".join(reviewers_to_remove)
+            )
        else:
            update_reviewers(
                args.pr_number,
                owner=args.owner,
                repo=args.repo,
-                reviewers_to_remove=reviewers_to_remove
+                reviewers_to_remove=reviewers_to_remove,
            )
    else:
        logging.info("No reviewers to remove.")
@@ -278,20 +334,29 @@ def main() -> None:
                MAX_MAINTAINERS_THRESHOLD,
            )
        else:
-            users_to_exclude = {args.pr_author} | past_reviewers | pending_reviewers | manually_removed
+            users_to_exclude = (
+                {args.pr_author} | past_reviewers | pending_reviewers | manually_removed
+            )
            potential_reviewers = maintainers - users_to_exclude

            reviewers_to_add = {
-                user for user in potential_reviewers if is_collaborator(args.owner, args.repo, user)
+                user
+                for user in potential_reviewers
+                if is_collaborator(args.owner, args.repo, user)
            }

            non_collaborators = potential_reviewers - reviewers_to_add
            if non_collaborators:
-                logging.warning("Ignoring non-collaborators: %s", ", ".join(non_collaborators))
+                logging.warning(
+                    "Ignoring non-collaborators: %s", ", ".join(non_collaborators)
+                )

            manually_removed_maintainers = reviewers_to_add & manually_removed
            if manually_removed_maintainers:
-                logging.info("Not re-adding manually removed maintainers: %s", ", ".join(manually_removed_maintainers))
+                logging.info(
+                    "Not re-adding manually removed maintainers: %s",
+                    ", ".join(manually_removed_maintainers),
+                )
                reviewers_to_add -= manually_removed

            if len(reviewers_to_add) > MAX_REVIEWERS:
@@ -304,7 +369,9 @@ def main() -> None:

    if reviewers_to_add:
        if args.dry_run:
-            logging.info("DRY RUN: Would add reviewers: %s", ", ".join(reviewers_to_add))
+            logging.info(
+                "DRY RUN: Would add reviewers: %s", ", ".join(reviewers_to_add)
+            )
        else:
            update_reviewers(args.pr_number, reviewers_to_add=reviewers_to_add)
    else:
@@ -14,32 +14,38 @@ import sys
 def main():
    print("🔍 Validating maintainer entries...")

-    result = subprocess.run(['nix', 'eval', '--file', 'modules/lib/maintainers.nix', '--json'],
-                           capture_output=True, text=True, check=True)
+    result = subprocess.run(
+        ["nix", "eval", "--file", "modules/lib/maintainers.nix", "--json"],
+        capture_output=True,
+        text=True,
+        check=True,
+    )
    maintainers = json.loads(result.stdout)
    errors = []

    for name, data in maintainers.items():
-        if 'github' not in data:
+        if "github" not in data:
            errors.append(f'{name}: Missing required field "github"')
-        if 'githubId' not in data:
+        if "githubId" not in data:
            errors.append(f'{name}: Missing required field "githubId"')

-        if 'githubId' in data:
-            github_id = data['githubId']
+        if "githubId" in data:
+            github_id = data["githubId"]
            if not isinstance(github_id, int):
-                errors.append(f'{name}: githubId must be a number, not a string: {github_id} (type: {type(github_id).__name__})')
+                errors.append(
+                    f"{name}: githubId must be a number, not a string: {github_id} (type: {type(github_id).__name__})"
+                )
            elif github_id <= 0:
-                errors.append(f'{name}: githubId must be positive: {github_id}')
+                errors.append(f"{name}: githubId must be positive: {github_id}")

    if errors:
-        print('❌ Validation errors found:')
+        print("❌ Validation errors found:")
        for error in errors:
-            print(f'  - {error}')
+            print(f"  - {error}")
        sys.exit(1)
    else:
-        print('✅ All maintainer entries are valid')
-        print(f'✅ Validated {len(maintainers)} maintainer entries')
+        print("✅ All maintainer entries are valid")
+        print(f"✅ Validated {len(maintainers)} maintainer entries")


 if __name__ == "__main__":
@@ -13,8 +13,12 @@ def main():
    print("🔍 Validating maintainers.nix syntax...")

    try:
-        subprocess.run(['nix', 'eval', '--file', 'modules/lib/maintainers.nix', '--json'],
-                               capture_output=True, text=True, check=True)
+        subprocess.run(
+            ["nix", "eval", "--file", "modules/lib/maintainers.nix", "--json"],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
        print("✅ Valid Nix syntax")
    except subprocess.CalledProcessError:
        print("❌ Invalid Nix syntax")