flake.nix

@@ -39,6 +39,7 @@
         ./targets/flake-module.nix
         ./modules/flake-module.nix
         ./pkgs/flake-module.nix
+        ./pkgs/clan-merge/flake-module.nix
       ];
       perSystem = { pkgs, inputs', ... }: {
         treefmt = {

pkgs/clan-merge/.envrc

@@ -0,0 +1 @@
+use flake .#clan-merge

pkgs/clan-merge/.gitignore

@@ -0,0 +1,12 @@
+.direnv
+result*
+
+# python
+__pycache__
+.coverage
+.mypy_cache
+.pytest_cache
+.pythonenv
+.reports
+.ruff_cache
+htmlcov

pkgs/clan-merge/clan_merge/__init__.py

@@ -0,0 +1,126 @@
+import argparse
+import json
+import urllib.request
+from os import environ
+from typing import Optional
+
+
+def load_token() -> str:
+    GITEA_TOKEN_FILE = environ.get("GITEA_TOKEN_FILE", default=None)
+    assert GITEA_TOKEN_FILE is not None
+    with open(GITEA_TOKEN_FILE, "r") as f:
+        return f.read().strip()
+
+
+def is_ci_green(pr: dict) -> bool:
+    print("Checking CI status for PR " + str(pr["id"]))
+    url = (
+        "https://git.clan.lol/api/v1/repos/clan/"
+        + pr["base"]["repo"]["name"]
+        + "/commits/"
+        + pr["head"]["sha"]
+        + "/status"
+    )
+    response = urllib.request.urlopen(url)
+    data = json.loads(response.read())
+    # check for all commit statuses to have status "success"
+    for status in data["statuses"]:
+        if status["status"] != "success":
+            return False
+    return True
+
+
+def decide_merge(pr: dict, allowed_users: list[str]) -> bool:
+    if (
+        pr["user"]["login"] in allowed_users
+        and pr["mergeable"] is True
+        and not pr["title"].startswith("WIP:")
+        and pr["state"] == "open"
+        and is_ci_green(pr)
+    ):
+        return True
+    return False
+
+
+# python equivalent for: curl -X 'GET' https://git.clan.lol/api/v1/repos/clan/{repo}/pulls
+def list_prs(repo: str) -> list:
+    url = "https://git.clan.lol/api/v1/repos/clan/" + repo + "/pulls"
+    response = urllib.request.urlopen(url)
+    data = json.loads(response.read())
+    return data
+
+
+def list_prs_to_merge(prs: list, allowed_users: list[str]) -> list:
+    prs_to_merge = []
+    for pr in prs:
+        if decide_merge(pr, allowed_users) is True:
+            prs_to_merge.append(pr)
+    return prs_to_merge
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Merge PRs on clan.lol")
+    # parse a list of allowed users
+    parser.add_argument(
+        "--allowed-users",
+        nargs="+",
+        help="list of users allowed to merge",
+        required=True,
+    )
+    # parse list of repository names for which to merge PRs
+    parser.add_argument(
+        "--repos",
+        nargs="+",
+        help="list of repositories for which to merge PRs",
+        required=True,
+    )
+    # option for dry run
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="dry run",
+    )
+    return parser.parse_args()
+
+
+def clan_merge(
+    args: Optional[argparse.Namespace] = None, gitea_token: Optional[str] = None
+) -> None:
+    if gitea_token is None:
+        gitea_token = load_token()
+    if args is None:
+        args = parse_args()
+    allowed_users = args.allowed_users
+    repos = args.repos
+    dry_run = args.dry_run
+    for repo in repos:
+        prs = list_prs(repo)
+        prs_to_merge = list_prs_to_merge(prs, allowed_users)
+        for pr in prs_to_merge:
+            url = (
+                "https://git.clan.lol/api/v1/repos/clan/"
+                + repo
+                + "/pulls/"
+                + str(pr["number"])
+                + "/merge"
+                + f"?token={gitea_token}"
+            )
+            if dry_run is True:
+                print(
+                    f"Would merge PR {pr['number']} in repo {repo} from user {pr['user']['login']}"
+                )
+            else:
+                print("Merging PR " + str(pr["id"]))
+                data = dict(
+                    Do="merge",
+                )
+                data_encoded = json.dumps(data).encode("utf8")
+                print(data)
+                req = urllib.request.Request(
+                    url, data=data_encoded, headers={"Content-Type": "application/json"}
+                )
+                urllib.request.urlopen(req)
+
+
+if __name__ == "__main__":
+    clan_merge()

pkgs/clan-merge/default.nix

@@ -0,0 +1,60 @@
+{ pkgs ? import <nixpkgs> { }
+, lib ? pkgs.lib
+, python3 ? pkgs.python3
+, ruff ? pkgs.ruff
+, runCommand ? pkgs.runCommand
+,
+}:
+let
+  pyproject = builtins.fromTOML (builtins.readFile ./pyproject.toml);
+  name = pyproject.project.name;
+
+  src = lib.cleanSource ./.;
+
+  dependencies = lib.attrValues {
+    # inherit (python3.pkgs)
+    #   some-package
+    #   ;
+  };
+
+  devDependencies = lib.attrValues {
+    inherit (pkgs) ruff;
+    inherit (python3.pkgs)
+      black
+      mypy
+      pytest
+      pytest-cov
+      setuptools
+      wheel
+      ;
+  };
+
+  package = python3.pkgs.buildPythonPackage {
+    inherit name src;
+    format = "pyproject";
+    nativeBuildInputs = [
+      python3.pkgs.setuptools
+    ];
+    propagatedBuildInputs =
+      dependencies
+      ++ [ ];
+    passthru.tests = { inherit check; };
+    passthru.devDependencies = devDependencies;
+  };
+
+  checkPython = python3.withPackages (_ps: devDependencies ++ dependencies);
+
+  check = runCommand "${name}-check" { } ''
+    cp -r ${src} ./src
+    chmod +w -R ./src
+    cd src
+    export PYTHONPATH=.
+    echo -e "\x1b[32m## run mypy\x1b[0m"
+    ${checkPython}/bin/mypy .
+    echo -e "\x1b[32m## run pytest\x1b[0m"
+    ${checkPython}/bin/pytest
+    touch $out
+  '';
+
+in
+package

pkgs/clan-merge/flake-module.nix

@@ -0,0 +1,11 @@
+{
+  perSystem = { pkgs, ... }:
+    let
+      package = pkgs.callPackage ./default.nix { inherit pkgs; };
+    in
+    {
+      packages.clan-merge = package;
+      checks.clan-merge = package.tests.check;
+      devShells.clan-merge = import ./shell.nix { inherit pkgs; };
+    };
+}

pkgs/clan-merge/pyproject.toml

@@ -0,0 +1,58 @@
+[build-system]
+requires = ["setuptools"]
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools.packages.find]
+include = ["clan_merge*"]
+
+[project]
+name = "clan_merge"
+description = "cLAN internal merge bot for gitea"
+dynamic = ["version"]
+scripts = {clan-merge = "clan_merge:clan_merge"}
+
+[tool.pytest.ini_options]
+addopts = "--cov . --cov-report term --cov-report=html:.reports/html --no-cov-on-fail"
+
+[tool.mypy]
+python_version = "3.10"
+warn_redundant_casts = true
+disallow_untyped_calls = true
+disallow_untyped_defs = true
+no_implicit_optional = true
+
+[[tool.mypy.overrides]]
+module = "setuptools.*"
+ignore_missing_imports = true
+
+[[tool.mypy.overrides]]
+module = "pytest.*"
+ignore_missing_imports = true
+
+[tool.ruff]
+line-length = 88
+
+select = ["E", "F", "I"]
+ignore = [ "E501" ]
+
+[tool.black]
+line-length = 88
+target-version = ['py310']
+include = '\.pyi?$'
+exclude = '''
+/(
+    \.git
+  | \.hg
+  | \.mypy_cache
+  | \.tox
+  | \.venv
+  | _build
+  | buck-out
+  | build
+  | dist
+  # The following are specific to Black, you probably don't want those.
+  | blib2to3
+  | tests/data
+  | profiling
+)/
+'''

pkgs/clan-merge/shell.nix

@@ -0,0 +1,44 @@
+{ pkgs ? import <nixpkgs> { } }:
+let
+  lib = pkgs.lib;
+  python3 = pkgs.python3;
+  package = import ./default.nix {
+    inherit lib pkgs python3;
+  };
+  pythonWithDeps = python3.withPackages (
+    ps:
+    package.propagatedBuildInputs
+    ++ package.devDependencies
+    ++ [
+      ps.pip
+    ]
+  );
+  checkScript = pkgs.writeScriptBin "check" ''
+    nix build -f . tests -L "$@"
+  '';
+  devShell = pkgs.mkShell {
+    packages = [
+      pkgs.ruff
+      pythonWithDeps
+    ];
+    # sets up an editable install and add enty points to $PATH
+    shellHook = ''
+      tmp_path=$(realpath ./.pythonenv)
+      repo_root=$(realpath .)
+      rm -rf $tmp_path
+      mkdir -p "$tmp_path/${pythonWithDeps.sitePackages}"
+
+      ${pythonWithDeps.interpreter} -m pip install \
+        --quiet \
+        --disable-pip-version-check \
+        --no-index \
+        --no-build-isolation \
+        --prefix "$tmp_path" \
+        --editable $repo_root
+
+      export PATH="$tmp_path/bin:${checkScript}/bin:$PATH"
+      export PYTHONPATH="$repo_root:$tmp_path/${pythonWithDeps.sitePackages}"
+    '';
+  };
+in
+devShell

pkgs/clan-merge/tests/test_cli.py

@@ -0,0 +1,95 @@
+import pytest
+
+import clan_merge
+
+
+def test_no_args(capsys: pytest.CaptureFixture) -> None:
+    # handle EsystemExit via pytest.raises
+    with pytest.raises(SystemExit):
+        clan_merge.clan_merge(gitea_token="")
+    captured = capsys.readouterr()
+    assert captured.err.startswith("usage:")
+
+
+def test_decide_merge_allowed(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(clan_merge, "is_ci_green", lambda x: True)
+    allowed_users = ["foo"]
+    pr = dict(
+        id=1,
+        user=dict(login="foo"),
+        title="Some PR Title",
+        mergeable=True,
+        state="open",
+    )
+    assert clan_merge.decide_merge(pr, allowed_users) is True
+
+
+def test_decide_merge_not_allowed(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(clan_merge, "is_ci_green", lambda x: True)
+    allowed_users = ["foo"]
+    pr1 = dict(
+        id=1,
+        user=dict(login="bar"),
+        title="Some PR Title",
+        mergeable=True,
+        state="open",
+    )
+    pr2 = dict(
+        id=1,
+        user=dict(login="foo"),
+        title="WIP: xyz",
+        mergeable=True,
+        state="open",
+    )
+    pr3 = dict(
+        id=1,
+        user=dict(login="foo"),
+        title="Some PR Title",
+        mergeable=False,
+        state="open",
+    )
+    pr4 = dict(
+        id=1,
+        user=dict(login="foo"),
+        title="Some PR Title",
+        mergeable=True,
+        state="closed",
+    )
+    assert clan_merge.decide_merge(pr1, allowed_users) is False
+    assert clan_merge.decide_merge(pr2, allowed_users) is False
+    assert clan_merge.decide_merge(pr3, allowed_users) is False
+    assert clan_merge.decide_merge(pr4, allowed_users) is False
+
+
+def test_list_prs_to_merge(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(clan_merge, "is_ci_green", lambda x: True)
+    prs = [
+        dict(
+            id=1,
+            base=dict(repo=dict(name="repo1")),
+            head=dict(sha="1234567890"),
+            user=dict(login="foo"),
+            state="open",
+            title="PR 1",
+            mergeable=True,
+        ),
+        dict(
+            id=2,
+            base=dict(repo=dict(name="repo1")),
+            head=dict(sha="1234567890"),
+            user=dict(login="foo"),
+            state="open",
+            title="WIP: xyz",
+            mergeable=True,
+        ),
+        dict(
+            id=3,
+            base=dict(repo=dict(name="repo1")),
+            head=dict(sha="1234567890"),
+            user=dict(login="bar"),
+            state="open",
+            title="PR 2",
+            mergeable=True,
+        ),
+    ]
+    assert clan_merge.list_prs_to_merge(prs, ["foo"]) == [prs[0]]