12839168347

Committed 18 Jan 2025 12:43AM UTC coverage: 99.136% (-0.03%) from 99.167%

Build # 12839168347

Build Type

Pull #311

github

Committed by

web-flow

Commit Message

Merge 343948eff into e9fd756f9

Pull Request Pull Request #311: chore(test): Test on ARM Ubuntu

Run Details

8 of 8 new or added lines in 2 files covered. (100.0%)

1 existing line in 1 file now uncovered.

2983 of 3009 relevant lines covered (99.14%)

0.99 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.95

/mybox/package/github.py

import json
import os
from collections.abc import Iterator
from dataclasses import dataclass
from subprocess import CalledProcessError
from typing import Any, Optional

from pydantic import Field

from ..driver import OS, Architecture
from ..filters import Filter, Filters, choose
from ..utils import (
    allow_singular_none,
    async_cached,
    async_cached_lock,
    http_get,
    run_output,
)
from .archive import ArchivePackage


@async_cached_lock
async def github_auth_token() -> Optional[str]:
    try:
        return os.environ["GITHUB_TOKEN"]
    except KeyError:
        pass

    try:
        return await run_output("gh", "auth", "token", silent=True)
    except (CalledProcessError, FileNotFoundError):
        pass

    return None  # pragma: no cover


async def github_api(url: str) -> Any:
    token = await github_auth_token()

    headers = {}
    if token:
        headers["Authorization"] = f"token {token}"

    result = await http_get(f"https://api.github.com/{url}", headers=headers)

    return json.loads(result)


@dataclass
class GitHubReleaseArtifact:
    name: str
    url: str


@dataclass
class GitHubRelease:
    id: int
    tag_name: str
    draft: bool
    prerelease: bool
    assets: list[GitHubReleaseArtifact]


ARCHITECTURE_FILTERS: dict[str, list[str]] = {
    "aarch64": ["arm64", "arm"],
    "i386": ["i686", "x86"],
    "mips": [],
    "powerpc": ["ppc"],
    "s390x": [],
    "x86_64": ["amd64", "x64"],
}

OS_FILTERS: dict[str, list[str]] = {
    "darwin": ["macOS", "macos", "osx"],
    "linux": [],
    "windows": [],
}


class GitHubPackage(ArchivePackage, Filters):
    repo: str
    skip_releases: list[str] = Field(default_factory=list, alias="skip_release")
    skip_releases_val = allow_singular_none("skip_releases")

    @async_cached
    async def releases(self) -> list[GitHubRelease]:
        result = await github_api(f"repos/{self.repo}/releases")
        return [
            GitHubRelease(
                id=release["id"],
                tag_name=release["tag_name"],
                draft=release["draft"],
                prerelease=release["prerelease"],
                assets=[
                    GitHubReleaseArtifact(
                        name=result["name"], url=result["browser_download_url"]
                    )
                    for result in release["assets"]
                ],
            )
            for release in result
        ]

    async def release(self) -> GitHubRelease:
        candidates = await self.releases()
        for candidate in candidates:
            if candidate.draft:
                # Can't find any stably draft releases to test with
                continue  # pragma: no cover
            if candidate.prerelease:
                continue
            if candidate.tag_name in self.skip_releases:
                continue
            return candidate
        raise ValueError(f"No releases found for {self.repo}.")

    @classmethod
    def environment_filters(
        cls, *, target_os: OS, target_arch: Architecture
    ) -> Iterator[Filter[str]]:
        for signature_hint in [".asc", ".sig", "sha256", "sha512", ".yml"]:
            yield cls.excludes_(signature_hint)

        for system_package_hint in [".deb", ".rpm", ".dmg", ".exe"]:
            yield cls.excludes_(system_package_hint)

        yield from cls.from_synonyms(
            OS_FILTERS, target_os.switch(linux="linux", macos="darwin")
        )

        yield from cls.from_synonyms(ARCHITECTURE_FILTERS, target_arch)

        if target_os.switch(linux=True, macos=False):
            yield cls.includes_("gnu")
            yield cls.excludes_("musl")

    def all_filters(
        self, *, target_os: OS, target_arch: Architecture
    ) -> Iterator[Filter[str]]:
        yield from self.filters()
        yield from self.environment_filters(
            target_os=target_os, target_arch=target_arch
        )

    async def artifact(self) -> GitHubReleaseArtifact:
        candidates = (await self.release()).assets

        def candidate_filter(
            name_filter: Filter,
        ) -> Filter[GitHubReleaseArtifact]:
            return lambda candidate: name_filter(candidate.name)

        target_os = await self.driver.os()
        target_arch = await self.driver.architecture()
        return choose(
            candidates,
            map(
                candidate_filter,
                self.all_filters(target_os=target_os, target_arch=target_arch),
            ),
        )

    async def archive_url(self) -> str:
        return (await self.artifact()).url

    def derive_name(self) -> str:
        return self.repo

    async def get_remote_version(self) -> str:
        release = await self.release()
        return str(release.id)

1	import json	1✔
2	import os	1✔
3	from collections.abc import Iterator	1✔
4	from dataclasses import dataclass	1✔
5	from subprocess import CalledProcessError	1✔
6	from typing import Any, Optional	1✔
7
8	from pydantic import Field	1✔
9
10	from ..driver import OS, Architecture	1✔
11	from ..filters import Filter, Filters, choose	1✔
12	from ..utils import (	1✔
13	allow_singular_none,
14	async_cached,
15	async_cached_lock,
16	http_get,
17	run_output,
18	)
19	from .archive import ArchivePackage	1✔
20
21
22	@async_cached_lock	1✔
23	async def github_auth_token() -> Optional[str]:	1✔
24	try:	1✔
25	return os.environ["GITHUB_TOKEN"]	1✔
26	except KeyError:	×
27	pass	×
28
29	try:	×
30	return await run_output("gh", "auth", "token", silent=True)	×
31	except (CalledProcessError, FileNotFoundError):	×
32	pass	×
33
34	return None # pragma: no cover
35
36
37	async def github_api(url: str) -> Any:	1✔
38	token = await github_auth_token()	1✔
39
40	headers = {}	1✔
41	if token:	1✔
42	headers["Authorization"] = f"token {token}"	1✔
43
44	result = await http_get(f"https://api.github.com/{url}", headers=headers)	1✔
45
46	return json.loads(result)	1✔
47
48
49	@dataclass	1✔
50	class GitHubReleaseArtifact:	1✔
51	name: str	1✔
52	url: str	1✔
53
54
55	@dataclass	1✔
56	class GitHubRelease:	1✔
57	id: int	1✔
58	tag_name: str	1✔
59	draft: bool	1✔
60	prerelease: bool	1✔
61	assets: list[GitHubReleaseArtifact]	1✔
62
63
64	ARCHITECTURE_FILTERS: dict[str, list[str]] = {	1✔
65	"aarch64": ["arm64", "arm"],
66	"i386": ["i686", "x86"],
67	"mips": [],
68	"powerpc": ["ppc"],
69	"s390x": [],
70	"x86_64": ["amd64", "x64"],
71	}
72
73	OS_FILTERS: dict[str, list[str]] = {	1✔
74	"darwin": ["macOS", "macos", "osx"],
75	"linux": [],
76	"windows": [],
77	}
78
79
80	class GitHubPackage(ArchivePackage, Filters):	1✔
81	repo: str	1✔
82	skip_releases: list[str] = Field(default_factory=list, alias="skip_release")	1✔
83	skip_releases_val = allow_singular_none("skip_releases")	1✔
84
85	@async_cached	1✔
86	async def releases(self) -> list[GitHubRelease]:	1✔
87	result = await github_api(f"repos/{self.repo}/releases")	1✔
88	return [	1✔
89	GitHubRelease(
90	id=release["id"],
91	tag_name=release["tag_name"],
92	draft=release["draft"],
93	prerelease=release["prerelease"],
94	assets=[
95	GitHubReleaseArtifact(
96	name=result["name"], url=result["browser_download_url"]
97	)
98	for result in release["assets"]
99	],
100	)
101	for release in result
102	]
103
104	async def release(self) -> GitHubRelease:	1✔
105	candidates = await self.releases()	1✔
106	for candidate in candidates:	1✔
107	if candidate.draft:	1✔
108	# Can't find any stably draft releases to test with
109	continue # pragma: no cover
110	if candidate.prerelease:	1✔
111	continue	1✔
112	if candidate.tag_name in self.skip_releases:	1✔
113	continue	1✔
114	return candidate	1✔
115	raise ValueError(f"No releases found for {self.repo}.")	1✔
116
117	@classmethod	1✔
118	def environment_filters(	1✔
119	cls, *, target_os: OS, target_arch: Architecture
120	) -> Iterator[Filter[str]]:
121	for signature_hint in [".asc", ".sig", "sha256", "sha512", ".yml"]:	1✔
122	yield cls.excludes_(signature_hint)	1✔
123
124	for system_package_hint in [".deb", ".rpm", ".dmg", ".exe"]:	1✔
125	yield cls.excludes_(system_package_hint)	1✔
126
127	yield from cls.from_synonyms(	1✔
128	OS_FILTERS, target_os.switch(linux="linux", macos="darwin")
129	)
130
131	yield from cls.from_synonyms(ARCHITECTURE_FILTERS, target_arch)	1✔
132
133	if target_os.switch(linux=True, macos=False):	1✔
134	yield cls.includes_("gnu")	1✔
UNCOV 135	yield cls.excludes_("musl")	×
136
137	def all_filters(	1✔
138	self, *, target_os: OS, target_arch: Architecture
139	) -> Iterator[Filter[str]]:
140	yield from self.filters()	1✔
141	yield from self.environment_filters(	1✔
142	target_os=target_os, target_arch=target_arch
143	)
144
145	async def artifact(self) -> GitHubReleaseArtifact:	1✔
146	candidates = (await self.release()).assets	1✔
147
148	def candidate_filter(	1✔
149	name_filter: Filter,
150	) -> Filter[GitHubReleaseArtifact]:
151	return lambda candidate: name_filter(candidate.name)	1✔
152
153	target_os = await self.driver.os()	1✔
154	target_arch = await self.driver.architecture()	1✔
155	return choose(	1✔
156	candidates,
157	map(
158	candidate_filter,
159	self.all_filters(target_os=target_os, target_arch=target_arch),
160	),
161	)
162
163	async def archive_url(self) -> str:	1✔
164	return (await self.artifact()).url	1✔
165
166	def derive_name(self) -> str:	1✔
167	return self.repo	1✔
168
169	async def get_remote_version(self) -> str:	1✔
170	release = await self.release()	1✔
171	return str(release.id)	1✔

koterpillar / mybox / 12839168347

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous