5006990816

Build Type

push

github

Committed by GitHub

Commit Message

Merge pull request #1032 from nextcloud/pre_commit-config

Run Details

1487 of 1610 new or added lines in 99 files covered. (92.36%)

14 existing lines in 9 files now uncovered.

3996 of 4316 relevant lines covered (92.59%)

5.55 hits per line

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

90.32

/nextcloudappstore/core/github.py

from itertools import chain, takewhile
from typing import Iterable, List

import requests
from semantic_version import Version

from nextcloudappstore.core.models import NextcloudRelease


class GitHubClient:
    def __init__(self, base_url: str, api_token: str = None) -> None:
        self.base_url = base_url.rstrip("/")
        self.api_token = api_token
        self.headers = None if self.api_token else {"Authorization": "token %s" % self.api_token}

    def get_tags(self, page: int, size: int = 100):
        url = "%s/repos/nextcloud/server/tags" % self.base_url
        params = (("per_page", size), ("page", page))
        response = requests.get(url, params=params, headers=self.headers, timeout=21)
        response.raise_for_status()
        return response.json()


def sync_releases(versions: Iterable[str]) -> None:
    """
    All given versions have a release. If a release is absent, persisted
    releases are out of date and need to have their release flag removed.
    Finally the latest version must be marked as current.
    :param versions: an iterable yielding all retrieved GitHub tags
    :return:
    """
    current_releases = NextcloudRelease.objects.all()
    imported_releases = [NextcloudRelease.objects.get_or_create(version=version)[0] for version in versions]
    if imported_releases:
        # all imported releases have a release, existing ones don't
        for release in imported_releases:
            release.is_supported = True
            release.has_release = True
            release.save()
        for release in get_old_releases(current_releases, imported_releases):
            release.is_supported = False
            release.save()
        # set latest release
        NextcloudRelease.objects.update(is_current=False)
        latest = max(imported_releases, key=lambda v: Version(v.version))
        latest.is_current = True
        latest.save()


NextcloudReleases = List[NextcloudRelease]


def get_old_releases(current: NextcloudReleases, imported: NextcloudReleases) -> NextcloudReleases:
    imported_versions = {release.version for release in imported}
    return [release for release in current if release.version not in imported_versions]


def get_supported_releases(client: GitHubClient, oldest_supported: str) -> Iterable[str]:
    releases = get_stable_releases(client)
    return takewhile(lambda v: is_supported(v, oldest_supported), releases)


def get_stable_releases(client: GitHubClient) -> Iterable[str]:
    json = chain.from_iterable(TagPages(client))
    return (tag for tag in (release["name"].lstrip("v") for release in json if "name" in release) if is_stable(tag))


def is_supported(oldest_supported: str, version: str) -> bool:
    return Version(oldest_supported) >= Version(version)


def is_stable(release: str) -> bool:
    try:
        version = Version(release)
        return not version.prerelease
    except ValueError:
        return False


class TagPages:
    """
    The GitHub API is paginated which makes it a pain to fetch all results.
    This iterable returns a stream of json arrays until no further results
    are found. To iterate over all releases you need to flatten the results
    returned from this iterator first
    """

    def __init__(self, client: GitHubClient, size: int = 100) -> None:
        self.client = client
        self.size = size
        self.page = 1  # pages are 1 indexed

    def __iter__(self) -> "TagPages":
        return self

    def __next__(self):
        json = self.client.get_tags(self.page, self.size)
        if len(json) > 0:
            self.page += 1
            return json
        else:
            raise StopIteration

1	from itertools import chain, takewhile	6✔
2	from typing import Iterable, List	6✔
3
4	import requests	6✔
5	from semantic_version import Version	6✔
6
7	from nextcloudappstore.core.models import NextcloudRelease	6✔
8
9
10	class GitHubClient:	6✔
11	def __init__(self, base_url: str, api_token: str = None) -> None:	6✔
12	self.base_url = base_url.rstrip("/")	6✔
13	self.api_token = api_token	6✔
14	self.headers = None if self.api_token else {"Authorization": "token %s" % self.api_token}	6✔
15
16	def get_tags(self, page: int, size: int = 100):	6✔
NEW 17	url = "%s/repos/nextcloud/server/tags" % self.base_url	×
NEW 18	params = (("per_page", size), ("page", page))	×
NEW 19	response = requests.get(url, params=params, headers=self.headers, timeout=21)	×
20	response.raise_for_status()	×
21	return response.json()	×
22
23
24	def sync_releases(versions: Iterable[str]) -> None:	6✔
25	"""
26	All given versions have a release. If a release is absent, persisted
27	releases are out of date and need to have their release flag removed.
28	Finally the latest version must be marked as current.
29	:param versions: an iterable yielding all retrieved GitHub tags
30	:return:
31	"""
32	current_releases = NextcloudRelease.objects.all()	6✔
33	imported_releases = [NextcloudRelease.objects.get_or_create(version=version)[0] for version in versions]	6✔
34	if imported_releases:	6✔
35	# all imported releases have a release, existing ones don't
36	for release in imported_releases:	6✔
37	release.is_supported = True	6✔
38	release.has_release = True	6✔
39	release.save()	6✔
40	for release in get_old_releases(current_releases, imported_releases):	6✔
41	release.is_supported = False	6✔
42	release.save()	6✔
43	# set latest release
44	NextcloudRelease.objects.update(is_current=False)	6✔
45	latest = max(imported_releases, key=lambda v: Version(v.version))	6✔
46	latest.is_current = True	6✔
47	latest.save()	6✔
48
49
50	NextcloudReleases = List[NextcloudRelease]	6✔
51
52
53	def get_old_releases(current: NextcloudReleases, imported: NextcloudReleases) -> NextcloudReleases:	6✔
54	imported_versions = {release.version for release in imported}	6✔
55	return [release for release in current if release.version not in imported_versions]	6✔
56
57
58	def get_supported_releases(client: GitHubClient, oldest_supported: str) -> Iterable[str]:	6✔
59	releases = get_stable_releases(client)	6✔
60	return takewhile(lambda v: is_supported(v, oldest_supported), releases)	6✔
61
62
63	def get_stable_releases(client: GitHubClient) -> Iterable[str]:	6✔
64	json = chain.from_iterable(TagPages(client))	6✔
65	return (tag for tag in (release["name"].lstrip("v") for release in json if "name" in release) if is_stable(tag))	6✔
66
67
68	def is_supported(oldest_supported: str, version: str) -> bool:	6✔
69	return Version(oldest_supported) >= Version(version)	6✔
70
71
72	def is_stable(release: str) -> bool:	6✔
73	try:	6✔
74	version = Version(release)	6✔
75	return not version.prerelease	6✔
76	except ValueError:	6✔
77	return False	6✔
78
79
80	class TagPages:	6✔
81	"""
82	The GitHub API is paginated which makes it a pain to fetch all results.
83	This iterable returns a stream of json arrays until no further results
84	are found. To iterate over all releases you need to flatten the results
85	returned from this iterator first
86	"""
87
88	def __init__(self, client: GitHubClient, size: int = 100) -> None:	6✔
89	self.client = client	6✔
90	self.size = size	6✔
91	self.page = 1 # pages are 1 indexed	6✔
92
93	def __iter__(self) -> "TagPages":	6✔
94	return self	6✔
95
96	def __next__(self):	6✔
97	json = self.client.get_tags(self.page, self.size)	6✔
98	if len(json) > 0:	6✔
99	self.page += 1	6✔
100	return json	6✔
101	else:
102	raise StopIteration	×

nextcloud / appstore / 5006990816

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