Split updater.py in half for a 'library' file and a 'custom' file

.github/workflows/updater.py

@@ -2,38 +2,26 @@
 """
 This script is meant to be run by GitHub Actions.
 It comes with a Github Action updater.yml to run this script periodically.
-
-Since each app is different, maintainers can adapt its contents to perform
-automatic actions when a new upstream release is detected.
-
 You need to enable the action by removing `if ${{ false }}` in updater.yml!
 """
 
-import hashlib
-import json
-import logging
-import os
-from subprocess import run, PIPE
-import textwrap
 from typing import Tuple, Any
 import requests
 from packaging import version
+import updater_lib as lib
 
-logging.getLogger().setLevel(logging.INFO)
-
-
-# ========================================================================== #
-# Functions customizable by app maintainer
 
 def get_latest_version(repo: str) -> Tuple[version.Version, Any]:
     """
     May be customized by maintainers for other forges than Github.
-    Returns a tuple: a comparable version, and some data that will
-    be passed to get_asset_urls_of_release().
+    Returns a tuple:
+    - a comparable version
+    - some data that will be passed to get_asset_urls_of_release().
+
+    This example handles tags on github.
     """
     api_url = repo.replace("github.com", "api.github.com/repos")
 
-    # Maintainer: use either releases or tags
     tags = requests.get(f"{api_url}/tags", timeout=3).json()
     tag_info = next(
         tag for tag in tags
@@ -41,7 +29,13 @@ def get_latest_version(repo: str) -> Tuple[version.Version, Any]:
     )
     return version.Version(tag_info["name"]), tag_info
 
-    # Maintainer: use either releases or tags
+
+def get_latest_version_releases(repo: str) -> Tuple[version.Version, Any]:
+    """
+    Another example of get_latest_version that handles github releases
+    """
+    api_url = repo.replace("github.com", "api.github.com/repos")
+
     releases = requests.get(f"{api_url}/releases", timeout=3).json()
     release_info = next(
         release for release in releases
@@ -52,87 +46,13 @@ def get_latest_version(repo: str) -> Tuple[version.Version, Any]:
 
 def generate_src_files(repo: str, release: Any):
     """
-    Should call write_src_file() for every asset/binary/... to download.
+    You should call write_src_file() for every asset/binary/... to download.
+    It will handle downloading and computing of the sha256.
     """
 
     built_release = f"{repo}/archive/refs/tags/{release['name']}.tar.gz"
-    logging.info("Handling main tarball at %s", built_release)
-    write_src_file("app.src", built_release, "tar.gz")
-
-
-# ========================================================================== #
-# Core generic code of the script, app maintainers should not edit this part
-
-def sha256sum_of_url(url: str) -> str:
-    """Compute checksum without saving the file"""
-    checksum = hashlib.sha256()
-    for chunk in requests.get(url, stream=True, timeout=1000).iter_content(10*1024):
-        checksum.update(chunk)
-    return checksum.hexdigest()
-
-
-def write_src_file(name: str, asset_url: str, extension: str,
-                   extract: bool = True, subdir: bool = True) -> None:
-    """Rewrite conf/app.src"""
-    logging.info("Writing %s...", name)
-
-    with open(f"conf/{name}", "w", encoding="utf-8") as conf_file:
-        conf_file.write(textwrap.dedent(f"""\
-            SOURCE_URL={asset_url}
-            SOURCE_SUM={sha256sum_of_url(asset_url)}
-            SOURCE_SUM_PRG=sha256sum
-            SOURCE_FORMAT={extension}
-            SOURCE_IN_SUBDIR={str(subdir).lower()}
-            SOURCE_EXTRACT={str(extract).lower()}
-        """))
-
-
-def write_github_env(proceed: bool, new_version: str, branch: str):
-    """Those values will be used later in the workflow"""
-    if "GITHUB_ENV" not in os.environ:
-        logging.warning("GITHUB_ENV is not in the envvars, assuming not in CI")
-        return
-    with open(os.environ["GITHUB_ENV"], "w", encoding="utf-8") as github_env:
-        github_env.write(textwrap.dedent(f"""\
-            VERSION={new_version}
-            BRANCH={branch}
-            PROCEED={str(proceed).lower()}
-        """))
-
-
-def main():
-    with open("manifest.json", "r", encoding="utf-8") as manifest_file:
-        manifest = json.load(manifest_file)
-    repo = manifest["upstream"]["code"]
-
-    current_version = version.Version(manifest["version"].split("~")[0])
-    latest_version, release_info = get_latest_version(repo)
-    logging.info("Current version: %s", current_version)
-    logging.info("Latest upstream version: %s", latest_version)
-
-    # Proceed only if the retrieved version is greater than the current one
-    if latest_version <= current_version:
-        logging.warning("No new version available")
-        write_github_env(False, "", "")
-        return
-
-    # Proceed only if a PR for this new version does not already exist
-    branch = f"ci-auto-update-v{latest_version}"
-    command = ["git", "ls-remote", "--exit-code", "-h", repo, branch]
-    if run(command, stderr=PIPE, stdout=PIPE, check=False).returncode == 0:
-        logging.warning("A branch already exists for this update")
-        write_github_env(False, "", "")
-        return
-
-    generate_src_files(repo, release_info)
-
-    manifest["version"] = f"{latest_version}~ynh1"
-    with open("manifest.json", "w", encoding="utf-8") as manifest_file:
-        json.dump(manifest, manifest_file, indent=4, ensure_ascii=False)
-        manifest_file.write("\n")
-
-    write_github_env(True, latest_version, branch)
+    lib.write_src_file("app.src", built_release, "tar.gz")
 
 
 if __name__ == "__main__":
-    main()
+    lib.run(get_latest_version, generate_src_files)

.github/workflows/updater_lib.py

@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+"""
+This script is meant to be imported by updater.py.
+
+MAINTAINERS: DO NOT EDIT THIS FILE! EDIT updater.py INSTEAD.
+"""
+
+from typing import Callable, Tuple, Any
+import hashlib
+import json
+import logging
+import os
+import subprocess
+import textwrap
+import requests
+from packaging import version
+
+logging.getLogger().setLevel(logging.INFO)
+
+
+def sha256sum_of_url(url: str) -> str:
+    """Compute checksum without saving the file"""
+    checksum = hashlib.sha256()
+    for chunk in requests.get(url, stream=True, timeout=1000).iter_content(10 * 1024):
+        checksum.update(chunk)
+    return checksum.hexdigest()
+
+
+def write_src_file(name: str, asset_url: str, extension: str,
+                   extract: bool = True, subdir: bool = True) -> None:
+    """Rewrite conf/app.src"""
+    logging.info("Writing %s...", name)
+
+    with open(f"conf/{name}", "w", encoding="utf-8") as conf_file:
+        conf_file.write(textwrap.dedent(f"""\
+            SOURCE_URL={asset_url}
+            SOURCE_SUM={sha256sum_of_url(asset_url)}
+            SOURCE_SUM_PRG=sha256sum
+            SOURCE_FORMAT={extension}
+            SOURCE_IN_SUBDIR={str(subdir).lower()}
+            SOURCE_EXTRACT={str(extract).lower()}
+        """))
+
+
+def write_github_env(proceed: bool, new_version: str, branch: str):
+    """Those values will be used later in the workflow"""
+    if "GITHUB_ENV" not in os.environ:
+        logging.warning("GITHUB_ENV is not in the envvars, assuming not in CI")
+        return
+    with open(os.environ["GITHUB_ENV"], "w", encoding="utf-8") as github_env:
+        github_env.write(textwrap.dedent(f"""\
+            VERSION={new_version}
+            BRANCH={branch}
+            PROCEED={str(proceed).lower()}
+        """))
+
+
+def run(
+    get_latest_version: Callable[[str], Tuple[version.Version, Any]],
+    generate_src_files: Callable[[str, Any], Any]
+):
+    with open("manifest.json", "r", encoding="utf-8") as manifest_file:
+        manifest = json.load(manifest_file)
+    repo = manifest["upstream"]["code"]
+
+    current_version = version.Version(manifest["version"].split("~")[0])
+    latest_version, release_info = get_latest_version(repo)
+    logging.info("Current version: %s", current_version)
+    logging.info("Latest upstream version: %s", latest_version)
+
+    # Proceed only if the retrieved version is greater than the current one
+    if latest_version <= current_version:
+        logging.warning("No new version available")
+        write_github_env(False, "", "")
+        return
+
+    # Proceed only if a PR for this new version does not already exist
+    branch = f"ci-auto-update-v{latest_version}"
+    command = ["git", "ls-remote", "--exit-code", "-h", repo, branch]
+    if subprocess.run(command, capture_output=True, check=False).returncode == 0:
+        logging.warning("A branch already exists for this update")
+        write_github_env(False, "", "")
+        return
+
+    generate_src_files(repo, release_info)
+
+    manifest["version"] = f"{latest_version}~ynh1"
+    with open("manifest.json", "w", encoding="utf-8") as manifest_file:
+        json.dump(manifest, manifest_file, indent=4, ensure_ascii=False)
+        manifest_file.write("\n")
+
+    write_github_env(True, str(latest_version), branch)