feat(tests): add command to automatically update UI fixtures from CI results

Usage:
$ python tests/update_fixtures.py ci

[no-changelog]

tests/gitlab.py

@@ -0,0 +1,136 @@
+"""
+Helper functions for communication with Gitlab.
+
+Allowing for interaction with the test results, e.g. with UI tests.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any, Iterable, Iterator
+
+import requests
+
+AnyDict = dict[Any, Any]
+
+HERE = Path(__file__).parent
+
+BRANCHES_API_TEMPLATE = "https://gitlab.com/satoshilabs/trezor/trezor-firmware/-/pipelines.json?scope=branches&page={}"
+GRAPHQL_API = "https://gitlab.com/api/graphql"
+
+UI_JOB_NAMES = (
+    "core click R test",
+    "core device R test",
+    "core click test",
+    "core device test",
+    "core persistence test",
+    "legacy device test",
+)
+
+SAVE_GRAPHQL_RESULTS = False
+
+
+def _get_gitlab_branches(page: int) -> list[AnyDict]:
+    return requests.get(BRANCHES_API_TEMPLATE.format(page)).json()["pipelines"]
+
+
+def _get_branch_obj(branch_name: str) -> AnyDict:
+    # Trying first 10 pages of branches
+    for page in range(1, 11):
+        branches = _get_gitlab_branches(page)
+        for branch_obj in branches:
+            if branch_obj["ref"]["name"] == branch_name:
+                return branch_obj
+    raise ValueError(f"Branch {branch_name} not found")
+
+
+def _get_pipeline_jobs_info(pipeline_iid: int) -> AnyDict:
+    # Getting just the stuff we need - the job names and IDs
+    graphql_query = """
+query getJobsFromPipeline($projectPath: ID!, $iid: ID!) {
+  project(fullPath: $projectPath) {
+    pipeline(iid: $iid) {
+      stages {
+        nodes {
+          groups {
+            nodes {
+              jobs {
+                nodes {
+                  id
+                  name
+                }
+              }
+            }
+          }
+        }
+      }
+    }
+  }
+}
+    """
+    query = {
+        "query": graphql_query,
+        "variables": {
+            "projectPath": "satoshilabs/trezor/trezor-firmware",
+            "iid": pipeline_iid,
+        },
+    }
+    return requests.post(GRAPHQL_API, json=query).json()
+
+
+def _yield_pipeline_jobs(pipeline_iid: int) -> Iterator[AnyDict]:
+    jobs_info = _get_pipeline_jobs_info(pipeline_iid)
+    if SAVE_GRAPHQL_RESULTS:  # for development purposes
+        with open("jobs_info.json", "w") as f:
+            json.dump(jobs_info, f, indent=2)
+    stages = jobs_info["data"]["project"]["pipeline"]["stages"]["nodes"]
+    for stage in stages:
+        nodes = stage["groups"]["nodes"]
+        for node in nodes:
+            jobs = node["jobs"]["nodes"]
+            for job in jobs:
+                yield job
+
+
+def _get_job_ui_fixtures_results(job: AnyDict) -> AnyDict:
+    print(f"Checking job {job['name']}")
+    job_id = job["id"].split("/")[-1]
+    url = f"https://satoshilabs.gitlab.io/-/trezor/trezor-firmware/-/jobs/{job_id}/artifacts/tests/ui_tests/fixtures.results.json"
+    response = requests.get(url)
+    if response.status_code != 200:
+        print("No UI results found")
+        return {}
+    return response.json()
+
+
+def get_jobs_of_interest(
+    only_jobs: Iterable[str] | None, exclude_jobs: Iterable[str] | None
+) -> Iterable[str]:
+    if only_jobs and exclude_jobs:
+        raise ValueError("Cannot specify both only_jobs and exclude_jobs")
+    if only_jobs:
+        return [job for job in UI_JOB_NAMES if job in only_jobs]
+    if exclude_jobs:
+        return [job for job in UI_JOB_NAMES if job not in exclude_jobs]
+    return UI_JOB_NAMES
+
+
+def get_branch_ui_fixtures_results(
+    branch_name: str, jobs_of_interest: Iterable[str] | None = None
+) -> dict[str, AnyDict]:
+    print(f"Checking branch {branch_name}")
+
+    if jobs_of_interest is None:
+        jobs_of_interest = UI_JOB_NAMES
+
+    branch_obj = _get_branch_obj(branch_name)
+    pipeline_iid = branch_obj["iid"]
+
+    def yield_key_value() -> Iterator[tuple[str, AnyDict]]:
+        for job in _yield_pipeline_jobs(pipeline_iid):
+            for ui_job_name in jobs_of_interest:
+                if job["name"] == ui_job_name:
+                    yield job["name"], _get_job_ui_fixtures_results(job)
+
+    return dict(yield_key_value())

tests/update_fixtures.py

@@ -1,16 +1,109 @@
 #!/usr/bin/env python3
+
+from __future__ import annotations
+
+import json
+import subprocess
+from typing import Iterable
+
 import click
 
+from gitlab import UI_JOB_NAMES, get_branch_ui_fixtures_results, get_jobs_of_interest
 from ui_tests import update_fixtures
+from ui_tests.common import FIXTURES_FILE, get_current_fixtures
+
+
+@click.group()
+def cli():
+    pass
 
 
-@click.command()
+@cli.command()
 @click.option("-r", "--remove-missing", is_flag=True, help="Remove missing tests")
-def main(remove_missing: bool) -> None:
-    """Update fixtures file with results from latest test run."""
+def local(remove_missing: bool) -> None:
+    """Update fixtures file with results from latest local test run."""
+    print("Updating from local test run...")
     changes_amount = update_fixtures(remove_missing)
     print(f"Updated fixtures.json with data from {changes_amount} tests.")
 
 
+def _get_current_git_branch() -> str:
+    return (
+        subprocess.check_output(["git", "branch", "--show-current"])
+        .decode("ascii")
+        .strip()
+    )
+
+
+@cli.command()
+@click.option("-b", "--branch", help="Branch name")
+@click.option(
+    "-o",
+    "--only-jobs",
+    type=click.Choice(UI_JOB_NAMES),
+    help="Job names which to process",
+    multiple=True,
+)
+@click.option(
+    "-e",
+    "--exclude-jobs",
+    type=click.Choice(UI_JOB_NAMES),
+    help="Not take these jobs",
+    multiple=True,
+)
+def ci(
+    branch: str | None,
+    only_jobs: Iterable[str] | None,
+    exclude_jobs: Iterable[str] | None,
+) -> None:
+    """Update fixtures file with results from CI."""
+    print("Updating from CI...")
+
+    if only_jobs and exclude_jobs:
+        raise click.UsageError("Cannot use both --only-jobs and --exclude-jobs")
+
+    if branch is None:
+        branch = _get_current_git_branch()
+
+    print(f"Branch: {branch}")
+    if only_jobs:
+        print(f"Only jobs: {only_jobs}")
+    if exclude_jobs:
+        print(f"Exclude jobs: {exclude_jobs}")
+
+    jobs_of_interest = get_jobs_of_interest(only_jobs, exclude_jobs)
+    ui_results = get_branch_ui_fixtures_results(branch, jobs_of_interest)
+
+    current_fixtures = get_current_fixtures()
+
+    differing_total = 0
+    for job_name, ui_res_dict in ui_results.items():
+        print(f"Updating results from {job_name}...")
+        if not ui_res_dict:
+            print("No results found.")
+            continue
+        model = next(iter(ui_res_dict.keys()))
+        group = next(iter(ui_res_dict[model].keys()))
+        current_model = current_fixtures.setdefault(model, {})
+        current_group = current_model.setdefault(group, {})  # type: ignore
+
+        differing = 0
+        for test_name, res in ui_res_dict[model][group].items():
+            if current_group.get(test_name) != res:
+                differing += 1
+            current_group[test_name] = res
+
+        print(f"Updated {differing} tests.")
+        differing_total += differing
+
+    print(80 * "-")
+    print(f"Updated {differing_total} tests in total.")
+
+    FIXTURES_FILE.write_text(
+        json.dumps(current_fixtures, indent=0, sort_keys=True) + "\n"
+    )
+    print("Updated fixtures.json with data from CI.")
+
+
 if __name__ == "__main__":
-    main()
+    cli()