feat: add cli to create github pull requests

pyproject.toml

@@ -34,6 +34,8 @@ semver = "3.0.0-dev.3"
 pyinstaller = { version = "*", optional = true }
 ops2deb = "*"
 pygit2 = "^1.9.0"
+gql = "^3.1.0"
+aiohttp = "^3.8.1"
 
 [tool.poetry.scripts]
 wakemebot = "wakemebot.cli:main"

setup.cfg

@@ -4,6 +4,12 @@ disallow_untyped_defs = True
 warn_redundant_casts = True
 strict_equality = True
 
+[mypy-pygit2]
+ignore_missing_imports = True
+
+[mypy-ruamel]
+ignore_missing_imports = True
+
 [mypy-semver]
 ignore_missing_imports = True
 

src/wakemebot/bot.py

@@ -0,0 +1,163 @@
+import contextlib
+import re
+import sys
+from pathlib import Path
+from typing import Any, Dict, Generator, List, Union
+
+import pygit2
+import ruamel.yaml
+from gql import Client, gql
+from gql.transport.aiohttp import AIOHTTPTransport
+from ops2deb.exceptions import Ops2debError
+from ops2deb.parser import parse
+from ops2deb.updater import FixIndentEmitter, LatestRelease, find_latest_releases
+from ruamel.yaml import YAML, YAMLError
+
+GIT_AUTHOR = pygit2.Signature("wakemebot", "[email protected]")
+GIT_COMMITTER = pygit2.Signature("wakemebot", "[email protected]")
+
+BRANCH_NAME_TEMPLATE = (
+    "chore/wakemebot-update-{release.blueprint.name}-"
+    "from-{release.blueprint.version}-to-{release.version}"
+)
+
+COMMIT_MESSAGE_TEMPLATE = (
+    "chore(bot): update {release.blueprint.name} "
+    "from {release.blueprint.version} to {release.version}"
+)
+
+
+def load(
+    configuration_content: str, yaml: YAML = YAML()
+) -> Union[List[Dict[str, Any]], Dict[str, Any]]:
+    try:
+        return yaml.load(configuration_content)
+    except YAMLError as e:
+        print(f"Invalid YAML file.\n{e}")
+        sys.exit(1)
+
+
+def yaml_factory() -> ruamel.yaml.YAML:
+    yaml = ruamel.yaml.YAML(typ="rt")
+    yaml.Emitter = FixIndentEmitter
+    return yaml
+
+
+def find_configuration_files(repository_path: Path) -> List[Path]:
+    paths = list(repository_path.glob("**/*ops2deb*.yml"))
+    print(f"Found {len(paths)} ops2deb configuration files")
+    return paths
+
+
+def find_blueprint_updates(
+    configuration_path: Path,
+    skip_names: List[str] = None,
+) -> Generator[LatestRelease, None, None]:
+    try:
+        blueprints = parse(configuration_path)
+    except Ops2debError as e:
+        print(e)
+        return
+
+    print(f"Looking for new releases in {configuration_path}")
+    releases, _ = find_latest_releases(blueprints, skip_names)
+    releases or print("Did not find any updates")
+
+    for release in releases:
+        yield release
+
+
+@contextlib.contextmanager
+def update_configuration(
+    configuration_path: Path, release: LatestRelease
+) -> Generator[None, None, None]:
+    original_content = configuration_path.read_text()
+    yaml = yaml_factory()
+    configuration_dict = yaml.load(configuration_path.read_text())
+    release.update_configuration(configuration_dict)
+    with configuration_path.open("w") as output:
+        yaml.dump(configuration_dict, output)
+    yield
+    configuration_path.write_text(original_content)
+
+
+def parse_git_remote(repo: pygit2.Repository) -> str:
+    remote = repo.remotes["origin"]
+    result = re.search(r"/([^/]+/[^/]+).git", remote.url)
+    if result is None:
+        raise ValueError("Not a Github repository")
+    return result.group(1)
+
+
+def create_branch_and_commit(
+    repo: pygit2.Repository,
+    configuration_path: Path,
+    branch_name: str,
+    commit_message: str,
+) -> None:
+    try:
+        commit = repo[repo.head.target]
+        branch = repo.branches.create(branch_name, commit, force=True)
+        repository_path = Path(
+            pygit2.discover_repository(configuration_path.parent)
+        ).parent
+        repo.index.add(configuration_path.relative_to(repository_path))
+        repo.index.write()
+        tree = repo.index.write_tree()
+        parent, ref = repo.resolve_refish(refish=branch.name)
+        repo.create_commit(
+            branch.name, GIT_AUTHOR, GIT_COMMITTER, commit_message, tree, [parent.oid]
+        )
+    except pygit2.GitError as e:
+        print(f"Failed to create branch {branch_name}: {e}")
+
+
+def push_branch(repo: pygit2.Repository, access_token: str, branch_name: str) -> None:
+    try:
+        remote = repo.remotes["origin"]
+        credentials = pygit2.UserPass("wakemebot", access_token)
+        callbacks = pygit2.RemoteCallbacks(credentials)
+        branch = repo.branches.get(branch_name)
+        remote.push([branch.name], callbacks)
+    except pygit2.GitError as e:
+        print(f"Failed to push branch {branch_name}: {e}")
+
+
+def create_pull_request(
+    repo: pygit2.Repository,
+    access_token: str,
+    branch_name: str,
+    commit_message: str,
+) -> None:
+    transport = AIOHTTPTransport(
+        url="https://api.github.com/graphql",
+        headers={"Authorization": f"Bearer {access_token}"},
+    )
+    client = Client(transport=transport, fetch_schema_from_transport=False)
+    query = gql(
+        """
+        query createPullRequest ($code: ID!) {
+          continent (code: $code) {
+            name
+          }
+        }
+    """
+    )
+    github_repo = parse_git_remote(repo)
+    result = client.execute(query, variable_values={"repo": github_repo})
+    print(result)
+
+
+def create_pull_requests(repository_path: Path, access_token: str) -> None:
+    configuration_paths = find_configuration_files(repository_path)
+    repo = pygit2.Repository(repository_path)
+    for configuration_path in configuration_paths:
+        for release in find_blueprint_updates(configuration_path):
+            with update_configuration(configuration_path, release):
+                branch_name = BRANCH_NAME_TEMPLATE.format(release=release)
+                commit_message = COMMIT_MESSAGE_TEMPLATE.format(release=release)
+                create_branch_and_commit(
+                    repo, configuration_path, branch_name, commit_message
+                )
+                push_branch(repo, access_token, branch_name)
+                create_pull_request(repo, access_token, branch_name, commit_message)

src/wakemebot/cli.py

@@ -1,9 +1,11 @@
 from pathlib import Path
+from typing import Optional
 
+import pygit2
 import typer
 from pydantic import BaseModel, Field, HttpUrl
 
-from wakemebot import aptly
+from wakemebot import aptly, bot
 
 from . import __version__
 
@@ -16,6 +18,36 @@ class DebianRepository(BaseModel):
     distribution: str = Field(..., regex=r"[a-zA-Z0-9]+")
 
 
+def check_access_token(value: Optional[str]) -> str:
+    if value is None:
+        raise typer.BadParameter("Missing Github Access Token")
+    return value
+
+
+def get_git_repo_path(path: Path) -> Path:
+    repo_path = pygit2.discover_repository(path)
+    if not repo_path:
+        raise typer.BadParameter(f"No repository found at '{path}'")
+    return Path(repo_path).parent
+
+
+@app.command(help="Look for updates and create pull requests")
+def create_pull_requests(
+    repository_path: Path = typer.Option(
+        Path("."),
+        help="Path to WakeMeOps repository",
+        callback=get_git_repo_path,
+    ),
+    access_token: str = typer.Option(
+        None,
+        help="Github Personal Access Token",
+        envvar="WAKEMEBOT_GITHUB_ACCESS_TOKEN",
+        callback=check_access_token,
+    ),
+) -> None:
+    bot.create_pull_requests(repository_path, access_token)
+
+
 @app.command(help="Output wakemebot version")
 def version() -> None:
     typer.secho(__version__)

tests/test_bot.py

@@ -0,0 +1,126 @@
+import base64
+from pathlib import Path
+from textwrap import dedent
+from unittest.mock import Mock, patch
+
+import httpx
+import pygit2
+import pytest
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.responses import Response
+
+from wakemebot.bot import (
+    GIT_AUTHOR,
+    GIT_COMMITTER,
+    create_pull_requests,
+    parse_git_remote,
+)
+
+
+@pytest.fixture
+def ops2deb_configuration(tmp_path) -> Path:
+    configuration = """\
+    - name: great-app
+      version: 1.0.0
+      summary: Great package
+      description: A detailed description of the great package.
+      fetch:
+        url: http://testserver/{{version}}/great-app.tar.gz
+        sha256: f1be6dd36b503641d633765655e81cdae1ff8f7f73a2582b7468adceb5e212a9
+      script:
+        - mv great-app {{src}}/usr/bin/great-app
+
+    - name: super-app
+      version: 1.0.0
+      summary: Super package
+      description: A detailed description of the great package.
+      fetch:
+        url: http://testserver/{{version}}/great-app.tar.gz
+        sha256: f1be6dd36b503641d633765655e81cdae1ff8f7f73a2582b7468adceb5e212a9
+      script:
+        - mv super-app {{src}}/usr/bin/super-app
+    """
+    configuration_path = tmp_path / "ops2deb.yml"
+    configuration_path.write_text(dedent(configuration))
+    return configuration_path
+
+
+@pytest.fixture
+def server() -> Starlette:
+    starlette_app = Starlette(debug=True)
+
+    @starlette_app.route("/1.1.0/great-app.tar.gz")
+    @starlette_app.route("/1.1.1/great-app.tar.gz")
+    async def server_great_app_tar_gz(request: Request):
+        response = b"""H4sIAAAAAAAAA+3OMQ7CMBAEQD/FH0CyjSy/xwVCFJAoCf/HFCAqqEI1U9yudF
+            fceTn17dDnOewnDa3VZ+ZW02e+hHxsrYxRagkp59FDTDv+9HZft77EGNbLdbp9uf
+            u1BwAAAAAAAAAAgD96AGPmdYsAKAAA"""
+        return Response(
+            base64.b64decode(response),
+            status_code=200,
+            media_type="application/x-gzip",
+        )
+
+    return starlette_app
+
+
+@pytest.fixture(scope="function")
+def client(server):
+    real_async_client = httpx.AsyncClient
+
+    def async_client_mock(**kwargs):
+        kwargs.pop("transport", None)
+        return real_async_client(app=server, **kwargs)
+
+    httpx.AsyncClient = async_client_mock
+    yield
+    httpx.AsyncClient = real_async_client
+
+
+@pytest.fixture
+def repo(tmp_path, ops2deb_configuration) -> pygit2.init_repository:
+    repo = pygit2.init_repository(tmp_path, False)
+    repo.index.add(ops2deb_configuration.name)
+    repo.index.write()
+    tree = repo.index.write_tree()
+    repo.create_commit("HEAD", GIT_AUTHOR, GIT_COMMITTER, "initial commit", tree, [])
+    repo.remotes.create("origin", "https://github.com/upciti/wakemeops.git")
+    return repo
+
+
+def test_parse_git_remote_should_return_organization_and_repo_names(repo):
+    assert parse_git_remote(repo) == "upciti/wakemeops"
+
+
+@patch("wakemebot.bot.push_branch", Mock())
+@patch("wakemebot.bot.create_pull_request", Mock())
+def test_create_pull_requests_should_create_as_many_branches_as_they_are_updates(
+    client, tmp_path, repo
+):
+    create_pull_requests(tmp_path, "")
+    assert repo.branches.get("chore/wakemebot-update-great-app-from-1.0.0-to-1.1.1")
+    assert repo.branches.get("chore/wakemebot-update-super-app-from-1.0.0-to-1.1.1")
+
+
+@patch("wakemebot.bot.push_branch", Mock())
+@patch("wakemebot.bot.create_pull_request", Mock())
+def test_create_pull_requests_should_reset_branch_if_it_already_exists(
+    client, tmp_path, repo
+):
+    branch_name = "chore/wakemebot-update-great-app-from-1.0.0-to-1.1.1"
+    commit = repo[repo.head.target]
+    repo.branches.create(branch_name, commit)
+    create_pull_requests(tmp_path, "")
+
+
+@patch("wakemebot.bot.push_branch", Mock())
+@patch("wakemebot.bot.create_pull_request", Mock())
+def test_create_pull_requests_commit_message_should_mention_blueprint_and_versions(
+    client, tmp_path, repo
+):
+    create_pull_requests(tmp_path, "")
+    branch = repo.branches.get("chore/wakemebot-update-super-app-from-1.0.0-to-1.1.1")
+    commit = repo[branch.target]
+    commit_message = "chore(bot): update super-app from 1.0.0 to 1.1.1"
+    assert commit.message == commit_message