1
0
Fork 0
ynh-apps_tools/update_app_levels/update_app_levels.py

232 lines
7.4 KiB
Python
Raw Normal View History

#!/usr/bin/env python3
"""
Update app catalog: commit, and create a merge request
"""
import argparse
import json
import logging
import tempfile
import textwrap
import time
2023-01-20 17:14:30 +01:00
from collections import OrderedDict
from typing import Any
from pathlib import Path
import jinja2
import requests
import toml
from git import Repo
# APPS_REPO = "YunoHost/apps"
APPS_REPO = "Salamandar/apps"
CI_RESULTS_URL = "https://ci-apps.yunohost.org/ci/api/results"
REPO_APPS_ROOT = Path(Repo(__file__, search_parent_directories=True).working_dir)
VERBOSE = False
def github_token() -> str | None:
github_token_path = REPO_APPS_ROOT.parent / ".github_token"
if github_token_path.exists():
return github_token_path.open("r", encoding="utf-8").read().strip()
return None
def get_ci_results() -> dict[str, dict[str, Any]]:
return requests.get(CI_RESULTS_URL, timeout=10).json()
def ci_result_is_outdated(result) -> bool:
# 3600 * 24 * 60 = ~2 months
return (int(time.time()) - result.get("timestamp", 0)) > 3600 * 24 * 60
def update_catalog(catalog, ci_results) -> dict:
"""
Actually change the catalog data
"""
# Re-sort the catalog keys / subkeys
for app, infos in catalog.items():
catalog[app] = OrderedDict(sorted(infos.items()))
catalog = OrderedDict(sorted(catalog.items()))
def app_level(app):
if app not in ci_results:
return 0
if ci_result_is_outdated(ci_results[app]):
return 0
return ci_results[app]["level"]
for app, info in catalog.items():
info["level"] = app_level(app)
return catalog
def list_changes(catalog, ci_results) -> dict[str, list[tuple[str, int, int]]]:
"""
Lists changes for a pull request
"""
changes = {
"major_regressions": [],
"minor_regressions": [],
"improvements": [],
"outdated": [],
"missing": [],
}
for app, infos in catalog.items():
if infos.get("state") != "working":
continue
if app not in ci_results:
changes["missing"].append(app)
continue
if ci_result_is_outdated(ci_results[app]):
changes["outdated"].append(app)
continue
ci_level = ci_results[app]["level"]
current_level = infos.get("level")
if ci_level == current_level:
continue
if current_level is None or ci_level > current_level:
changes["improvements"].append((app, current_level, ci_level))
continue
if ci_level < current_level:
if ci_level <= 4 < current_level:
changes["major_regressions"].append((app, current_level, ci_level))
else:
changes["minor_regressions"].append((app, current_level, ci_level))
return changes
def pretty_changes(changes: dict[str, list[tuple[str, int, int]]]) -> str:
pr_body_template = textwrap.dedent("""
{%- if changes["major_regressions"] %}
### Major regressions
{% for app in changes["major_regressions"] %}
- [ ] [{{app.0}}: {{app.1}} -> {{app.2}}](https://ci-apps.yunohost.org/ci/apps/{{app.0}}/latestjob)
{%- endfor %}
{% endif %}
{%- if changes["minor_regressions"] %}
### Minor regressions
{% for app in changes["minor_regressions"] %}
- [ ] [{{app.0}}: {{app.1}} -> {{app.2}}](https://ci-apps.yunohost.org/ci/apps/{{app.0}}/latestjob)
{%- endfor %}
{% endif %}
{%- if changes["improvements"] %}
### Improvements
{% for app in changes["improvements"] %}
- [{{app.0}}: {{app.1}} -> {{app.2}}](https://ci-apps.yunohost.org/ci/apps/{{app.0}}/latestjob)
{%- endfor %}
{% endif %}
{%- if changes["missing"] %}
### Missing
{% for app in changes["missing"] %}
- [{{app}} (See latest job if it exists)](https://ci-apps.yunohost.org/ci/apps/{{app.0}}/latestjob)
{%- endfor %}
{% endif %}
{%- if changes["outdated"] %}
### Outdated
{% for app in changes["outdated"] %}
- [ ] [{{app}} (See latest job if it exists)](https://ci-apps.yunohost.org/ci/apps/{{app.0}}/latestjob)
{%- endfor %}
{% endif %}
""")
return jinja2.Environment().from_string(pr_body_template).render(changes=changes)
def make_pull_request(pr_body: str) -> None:
pr_data = {
"title": "Update app levels according to CI results",
"body": pr_body,
"head": "update_app_levels",
"base": "master"
}
with requests.Session() as s:
s.headers.update({"Authorization": f"token {github_token()}"})
2024-02-07 17:25:16 +01:00
response = s.post(f"https://api.github.com/repos/{APPS_REPO}/pulls", json=pr_data)
if response.status_code == 422:
response = s.get(f"https://api.github.com/repos/{APPS_REPO}/pulls", data={"head": "update_app_levels"})
2024-02-07 17:25:16 +01:00
response.raise_for_status()
pr_number = response.json()[0]["number"]
# head can't be updated
del pr_data["head"]
response = s.patch(f"https://api.github.com/repos/{APPS_REPO}/pulls/{pr_number}", json=pr_data)
response.raise_for_status()
existing_url = response.json()["html_url"]
logging.warning(f"An existing Pull Request has been updated at {existing_url} !")
else:
2024-02-07 17:25:16 +01:00
response.raise_for_status()
new_url = response.json()["html_url"]
logging.info(f"Opened a Pull Request at {new_url} !")
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--commit", action=argparse.BooleanOptionalAction, default=True)
parser.add_argument("--pr", action=argparse.BooleanOptionalAction, default=True)
parser.add_argument("-v", "--verbose", action=argparse.BooleanOptionalAction)
args = parser.parse_args()
logging.getLogger().setLevel(logging.INFO)
global VERBOSE
if args.verbose:
VERBOSE = True
logging.getLogger().setLevel(logging.DEBUG)
with tempfile.TemporaryDirectory(prefix="update_app_levels_") as tmpdir:
logging.info("Cloning the repository...")
apps_repo = Repo.clone_from(f"git@github.com:{APPS_REPO}", to_path=tmpdir)
# Load the app catalog and filter out the non-working ones
catalog = toml.load((Path(apps_repo.working_tree_dir) / "apps.toml").open("r", encoding="utf-8"))
new_branch = apps_repo.create_head("update_app_levels", apps_repo.refs.master)
apps_repo.head.reference = new_branch
logging.info("Retrieving the CI results...")
ci_results = get_ci_results()
# Now compute changes, then update the catalog
changes = list_changes(catalog, ci_results)
pr_body = pretty_changes(changes)
catalog = update_catalog(catalog, ci_results)
# Save the new catalog
updated_catalog = toml.dumps(catalog)
updated_catalog = updated_catalog.replace(",]", " ]")
(Path(apps_repo.working_tree_dir) / "apps.toml").open("w", encoding="utf-8").write(updated_catalog)
if args.commit:
logging.info("Committing and pushing the new catalog...")
apps_repo.index.add("apps.toml")
apps_repo.index.commit("Update app levels according to CI results")
apps_repo.remote().push(force=True)
if VERBOSE:
print(pr_body)
if args.pr:
logging.info("Opening a pull request...")
make_pull_request(pr_body)
if __name__ == "__main__":
main()