diff --git a/setup.cfg b/setup.cfg index 952951b..0903918 100644 --- a/setup.cfg +++ b/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 diff --git a/src/wakemebot/bot.py b/src/wakemebot/bot.py new file mode 100644 index 0000000..49c5bd1 --- /dev/null +++ b/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", "wakemebot@users.noreply.github.com") +GIT_COMMITTER = pygit2.Signature("wakemebot", "wakemebot@protonmail.com") + +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}") diff --git a/src/wakemebot/cli.py b/src/wakemebot/cli.py index 990d34a..15e2025 100644 --- a/src/wakemebot/cli.py +++ b/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__) diff --git a/tests/test_bot.py b/tests/test_bot.py new file mode 100644 index 0000000..262cda2 --- /dev/null +++ b/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")