Skip to content

Commit

Permalink
Use Github /latest endpoint (#132)
Browse files Browse the repository at this point in the history
* umu_proton: refactor to use the /latest endpoint

- Refactor to use the /latest endpoint from Github's REST API and to guarantee order of Github release assets. While the launcher specifies to use a versioned Github API, thus guaranteeing API stability of its JSON structure, it's not documented that the API will guarantee stability for the order of its Github assets.

* umu_test: update test
  • Loading branch information
R1kaB3rN authored Jun 30, 2024
1 parent fb28c60 commit 53d0073
Show file tree
Hide file tree
Showing 2 changed files with 42 additions and 54 deletions.
86 changes: 37 additions & 49 deletions umu/umu_proton.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from ssl import SSLContext, create_default_context
from tarfile import open as tar_open
from tempfile import mkdtemp
from typing import Any
from urllib.error import URLError
from urllib.request import Request, urlopen

Expand Down Expand Up @@ -43,7 +44,7 @@ def get_umu_proton(
# Subset of Github release assets from the Github API (ver. 2022-11-28)
# First element is the digest asset, second is the Proton asset. Each asset
# will contain the asset's name and the URL that hosts it.
assets: list[tuple[str, str]] = []
assets: tuple[tuple[str, str], tuple[str, str]] | tuple[()] = ()
tmp: Path = Path(mkdtemp())

STEAM_COMPAT.mkdir(exist_ok=True, parents=True)
Expand All @@ -65,76 +66,63 @@ def get_umu_proton(
return env


def _fetch_releases() -> list[tuple[str, str]]:
def _fetch_releases() -> tuple[tuple[str, str], tuple[str, str]] | tuple[()]:
"""Fetch the latest releases from the Github API."""
assets: list[tuple[str, str]] = []
digest_asset: tuple[str, str]
proton_asset: tuple[str, str]
asset_count: int = 0
url: str = "https://api.github.com"
repo: str = "/repos/Open-Wine-Components/umu-proton/releases"
repo: str = "/repos/Open-Wine-Components/umu-proton/releases/latest"
headers: dict[str, str] = {
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2022-11-28",
"User-Agent": "",
}

if os.environ.get("PROTONPATH") == "GE-Proton":
repo = "/repos/GloriousEggroll/proton-ge-custom/releases"
repo = "/repos/GloriousEggroll/proton-ge-custom/releases/latest"

with urlopen( # noqa: S310
Request(f"{url}{repo}", headers=headers), # noqa: S310
context=ssl_default_context,
) as resp:
releases: list[dict[str, Any]]

if resp.status != 200:
return assets

for release in loads(resp.read().decode("utf-8")):
if not release.get("assets"):
continue
for asset in release.get("assets"):
if (
asset.get("name")
and (
asset.get("name").endswith("sum")
or (
asset.get("name").endswith("tar.gz")
and asset.get("name").startswith(
("UMU-Proton", "GE-Proton")
)
)
)
and asset.get("browser_download_url")
):
if asset["name"].endswith("sum"):
assets.append(
(
asset["name"],
asset["browser_download_url"],
)
)
asset_count += 1
else:
assets.append(
(
asset["name"],
asset["browser_download_url"],
)
)
asset_count += 1
if asset_count == 2:
break
break
return ()

releases = loads(resp.read().decode("utf-8")).get("assets", [])

for release in releases:
if release["name"].endswith("sum"):
digest_asset = (
release["name"],
release["browser_download_url"],
)
asset_count += 1
elif release["name"].endswith("tar.gz") and release[
"name"
].startswith(("UMU-Proton", "GE-Proton")):
proton_asset = (
release["name"],
release["browser_download_url"],
)
asset_count += 1

if asset_count == 2:
break

if asset_count != 2:
err: str = (
"Failed to acquire all assets from api.github.com: " f"{assets}"
)
err: str = "Failed to acquire all assets from api.github.com"
raise RuntimeError(err)

return assets
return digest_asset, proton_asset


def _fetch_proton(
env: dict[str, str], tmp: Path, assets: list[tuple[str, str]]
env: dict[str, str],
tmp: Path,
assets: tuple[tuple[str, str], tuple[str, str]],
) -> dict[str, str]:
"""Download the latest UMU-Proton or GE-Proton."""
hash, hash_url = assets[0]
Expand Down Expand Up @@ -287,7 +275,7 @@ def _get_latest(
env: dict[str, str],
steam_compat: Path,
tmp: Path,
assets: list[tuple[str, str]],
assets: tuple[tuple[str, str], tuple[str, str]] | tuple[()],
thread_pool: ThreadPoolExecutor,
) -> dict[str, str] | None:
"""Download the latest Proton for new installs.
Expand Down
10 changes: 5 additions & 5 deletions umu/umu_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -626,7 +626,7 @@ def test_latest_interrupt(self):
# In the real usage, should be populated after successful callout
# for latest Proton releases
# In this case, assume the test variable will be downloaded
files = [("", ""), (self.test_archive.name, "")]
files = (("", ""), (self.test_archive.name, ""))
thread_pool = ThreadPoolExecutor()

with patch("umu_proton._fetch_proton") as mock_function:
Expand Down Expand Up @@ -669,7 +669,7 @@ def test_latest_val_err(self):
# latest Proton releases
# When empty, it means the callout failed for some reason (e.g. no
# internet)
files = [("", ""), (self.test_archive.name, "")]
files = (("", ""), (self.test_archive.name, ""))
thread_pool = ThreadPoolExecutor()

self.assertTrue(
Expand Down Expand Up @@ -702,7 +702,7 @@ def test_latest_offline(self):
# latest Proton releases
# When empty, it means the callout failed for some reason (e.g. no
# internet)
files = []
files = ()
thread_pool = ThreadPoolExecutor()

os.environ["PROTONPATH"] = ""
Expand Down Expand Up @@ -733,7 +733,7 @@ def test_link_umu(self):
latest = Path("UMU-Proton-9.0-beta15")
latest.mkdir()
Path(f"{latest}.sha512sum").touch()
files = [(f"{latest}.sha512sum", ""), (f"{latest}.tar.gz", "")]
files = ((f"{latest}.sha512sum", ""), (f"{latest}.tar.gz", ""))
thread_pool = ThreadPoolExecutor()

# Mock the latest Proton in /tmp
Expand Down Expand Up @@ -788,7 +788,7 @@ def test_latest_umu(self):
latest = Path("UMU-Proton-9.0-beta16")
latest.mkdir()
Path(f"{latest}.sha512sum").touch()
files = [(f"{latest}.sha512sum", ""), (f"{latest}.tar.gz", "")]
files = ((f"{latest}.sha512sum", ""), (f"{latest}.tar.gz", ""))
thread_pool = ThreadPoolExecutor()

# Mock the latest Proton in /tmp
Expand Down

0 comments on commit 53d0073

Please sign in to comment.