feat: add cli to create github pull requests

setup.cfg

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

src/wakemebot/bot.py

@@ -0,0 +1,104 @@
+import sys
+from pathlib import Path
+from typing import Any, Dict, Generator, List, Union
+
+import pygit2
+import ruamel.yaml
+from ops2deb.parser import validate
+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 dump_updated_configuration_files(
+    configuration_path: Path,
+    skip_names: List[str] = None,
+) -> Generator[LatestRelease, None, None]:
+    original_configuration = configuration_path.read_text()
+    blueprints = validate(load(original_configuration))
+
+    print("Looking for new releases...")
+    releases, _ = find_latest_releases(blueprints, skip_names)
+    releases or print("Did not find any updates")
+
+    for release in releases:
+        yaml = yaml_factory()
+        configuration_dict = yaml.load(original_configuration)
+        release.update_configuration(configuration_dict)
+        with configuration_path.open("w") as output:
+            yaml.dump(configuration_dict, output)
+        yield release
+
+
+def create_branch(
+    repo: pygit2.Repository,
+    configuration_path: Path,
+    branch_name: str,
+    commit_message: str,
+) -> str:
+
+    # create branch
+    commit = repo[repo.head.target]
+    branch = repo.branches.create(branch_name, commit)
+
+    # add configuration
+    repo.index.add(configuration_path.name)
+    repo.index.write()
+    tree = repo.index.write_tree()
+
+    # commit configuration
+    parent, ref = repo.resolve_refish(refish=branch.name)
+    repo.create_commit(
+        branch.name, GIT_AUTHOR, GIT_COMMITTER, commit_message, tree, [parent.oid]
+    )
+
+    return branch.name
+
+
+def push_branch(repo: pygit2.Repository, branch_name: str, access_token: str) -> None:
+    remote = repo.remotes["origin"]
+    credentials = pygit2.UserPass("wakemebot", access_token)
+    callbacks = pygit2.RemoteCallbacks(credentials)
+    remote.push([branch_name], callbacks)
+
+
+def update(configuration_path: Path, access_token: str) -> None:
+    repo = pygit2.Repository(configuration_path.parent)
+    for release in dump_updated_configuration_files(configuration_path):
+        branch_name = BRANCH_NAME_TEMPLATE.format(release=release)
+        commit_message = COMMIT_MESSAGE_TEMPLATE.format(release=release)
+        try:
+            branch_name = create_branch(
+                repo, configuration_path, branch_name, commit_message
+            )
+            push_branch(repo, branch_name, access_token)
+        except pygit2.GitError as e:
+            print(f"Failed to create PR {branch_name}: {e}")

src/wakemebot/cli.py

@@ -1,9 +1,10 @@
 from pathlib import Path
+from typing import Optional
 
 import typer
 from pydantic import BaseModel, Field, HttpUrl
 
-from wakemebot import aptly
+from wakemebot import aptly, bot
 
 from . import __version__
 
@@ -16,6 +17,28 @@ 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
+
+
+@app.command(help="Look for updates and create pull requests.")
+def update(
+    configuration_path: Path = typer.Argument(
+        ..., help="Path to ops2deb configuration path"
+    ),
+    access_token: str = typer.Option(
+        None,
+        "--access-token",
+        help="Github Personal Access Token",
+        envvar="WAKEMEBOT_GITHUB_ACCESS_TOKEN",
+        callback=check_access_token,
+    ),
+) -> None:
+    bot.update(configuration_path, access_token)
+
+
 @app.command(help="Output wakemebot version")
 def version() -> None:
     typer.secho(__version__)

tests/test_bot.py

@@ -0,0 +1,108 @@
+import base64
+from textwrap import dedent
+from unittest.mock import Mock, patch
+
+import httpx
+import pygit2
+import pytest
+from ops2deb.parser import parse
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.responses import Response
+
+from wakemebot.bot import (
+    GIT_AUTHOR,
+    GIT_COMMITTER,
+    dump_updated_configuration_files,
+    update,
+)
+
+
+@pytest.fixture
+def ops2deb_configuration() -> str:
+    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
+    """
+    return dedent(configuration)
+
+
+@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):
+    repo = pygit2.init_repository(tmp_path, False)
+    configuration_path = tmp_path / "ops2deb.yml"
+    configuration_path.write_text(ops2deb_configuration)
+    repo.index.add(configuration_path.name)
+    repo.index.write()
+    tree = repo.index.write_tree()
+    repo.create_commit("HEAD", GIT_AUTHOR, GIT_COMMITTER, "initial commit", tree, [])
+    return repo
+
+
+def test_dump_updated_configuration_files_should_dump_updated_configurations(
+    client, ops2deb_configuration, tmp_path
+):
+    configuration_path = tmp_path / "ops2deb.yml"
+    configuration_path.write_text(ops2deb_configuration)
+    for release, i in zip(dump_updated_configuration_files(configuration_path), [0, 1]):
+        blueprints = parse(configuration_path)
+        assert blueprints[i].version == "1.1.1"
+
+
+@patch("wakemebot.bot.push_branch", Mock())
+def test_create_branches_should_use_proper_commit_titles_and_branch_names(
+    client, ops2deb_configuration, tmp_path, repo
+):
+    configuration_path = tmp_path / "ops2deb.yml"
+    update(configuration_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")