4145649460

Build Type

push

github-actions

Committed by GitHub

Commit Message

Merge branch 'develop' into allow-ref-target-for-release-action

Run Details

25096 of 28903 relevant lines covered (86.83%)

4.53 hits per line

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

43.04

/renku/command/version.py

# -*- coding: utf-8 -*-
#
# Copyright 2017-2022 - Swiss Data Science Center (SDSC)
# A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
# Eidgenössische Technische Hochschule Zürich (ETHZ).
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Versioning utilities."""

import datetime
import json
import sys

import attr
import click
import lockfile


def print_version(ctx, param, value):
    """Print version number.

    Args:
        ctx: Current click context.
        param: Not used.
        value: Not used.
    """
    if not value or ctx.resilient_parsing:
        return

    from renku.version import __version__

    click.echo(__version__)
    ctx.exit()


def find_latest_version(name: str, allow_prereleases: bool = False):
    """Find a latest version on PyPI.

    Args:
        name (str): Package name.
        allow_prereleases (bool): Whether to also check for prereleases (Default value = False).

    Returns:
        Version: Newest version of package.
    """
    from renku.core.util import requests

    response = requests.get("https://pypi.org/pypi/{name}/json".format(name=name))

    if response.status_code != 200:
        return

    description = response.json()

    from packaging.version import Version

    return max(
        version
        for version in (Version(version) for version in description["releases"].keys())
        if allow_prereleases or not version.is_prerelease
    )


@attr.s
class VersionCache:
    """Cache information about package version."""

    STATE_NAME = "selfcheck.json"

    DATE_FMT = "%Y-%m-%dT%H:%M:%SZ"

    MAX_AGE = 24 * 60 * 60  # 1 day

    last_check = attr.ib()

    pypi_version = attr.ib(default=None)

    @last_check.default
    def default_last_check(self):
        """Format default last check."""
        return datetime.datetime.utcnow().strftime(self.DATE_FMT)

    @property
    def is_fresh(self):
        """Check if we need to refresh the state."""
        if self.last_check and self.pypi_version:
            current_time = datetime.datetime.utcnow()
            last_check = datetime.datetime.strptime(self.last_check, self.DATE_FMT)
            return (current_time - last_check).total_seconds() < self.MAX_AGE

    @classmethod
    def _cache(cls, app_name):
        """Return cache file.

        Args:
            app_name: Current application name.

        Returns:
            Directory of the cache.
        """
        from pathlib import Path

        from appdirs import user_cache_dir

        cache_dir = Path(user_cache_dir(app_name, None))
        cache_dir.mkdir(parents=True, exist_ok=True)

        return cache_dir / cls.STATE_NAME

    @classmethod
    def load(cls, app_name):
        """Load information from a cache.

        Args:
            app_name:Current application name.

        Returns:
            Cache object.
        """
        cache = cls._cache(app_name)

        try:
            with cache.open() as fp:
                return cls(**json.load(fp)[sys.prefix])
        except (IOError, ValueError, KeyError):
            return cls()

    def dump(self, app_name):
        """Store information in a cache.

        Args:
            app_name: Current application name.
        """
        cache = self._cache(app_name)

        # Attempt to write out our version check file
        with lockfile.LockFile(str(cache)):
            if cache.exists():
                with cache.open() as fp:
                    state = json.load(fp)
            else:
                state = {}

            state[sys.prefix] = attr.asdict(self)

            with cache.open("w") as fp:
                json.dump(state, fp, sort_keys=True)


def _check_version():
    """Check renku version."""
    from renku.core.constant import APP_NAME

    if VersionCache.load(APP_NAME).is_fresh:
        return

    from packaging.version import Version

    from renku.version import __version__

    version = Version(__version__)
    allow_prereleases = version.is_prerelease

    latest_version = find_latest_version("renku", allow_prereleases=allow_prereleases)

    if version < latest_version:
        click.secho(
            f"You are using renku version {__version__}, however version {latest_version} is available.\n"
            "You should consider upgrading ...",
            fg="yellow",
            bold=True,
            err=True,
        )

    VersionCache(pypi_version=str(latest_version)).dump(APP_NAME)


def check_version(ctx, param, value):
    """Check for latest version of renku on PyPI."""
    if ctx.resilient_parsing:
        return

    if not value and ctx.invoked_subcommand != "run":
        ctx.call_on_close(_check_version)

1	# -- coding: utf-8 --
2	#
3	# Copyright 2017-2022 - Swiss Data Science Center (SDSC)
4	# A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
5	# Eidgenössische Technische Hochschule Zürich (ETHZ).
6	#
7	# Licensed under the Apache License, Version 2.0 (the "License");
8	# you may not use this file except in compliance with the License.
9	# You may obtain a copy of the License at
10	#
11	# http://www.apache.org/licenses/LICENSE-2.0
12	#
13	# Unless required by applicable law or agreed to in writing, software
14	# distributed under the License is distributed on an "AS IS" BASIS,
15	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16	# See the License for the specific language governing permissions and
17	# limitations under the License.
18	"""Versioning utilities."""	10✔
19
20	import datetime	10✔
21	import json	10✔
22	import sys	10✔
23
24	import attr	10✔
25	import click	10✔
26	import lockfile	10✔
27
28
29	def print_version(ctx, param, value):	10✔
30	"""Print version number.
31
32	Args:
33	ctx: Current click context.
34	param: Not used.
35	value: Not used.
36	"""
37	if not value or ctx.resilient_parsing:	9✔
38	return	9✔
39
40	from renku.version import __version__	2✔
41
42	click.echo(__version__)	2✔
43	ctx.exit()	2✔
44
45
46	def find_latest_version(name: str, allow_prereleases: bool = False):	10✔
47	"""Find a latest version on PyPI.
48
49	Args:
50	name (str): Package name.
51	allow_prereleases (bool): Whether to also check for prereleases (Default value = False).
52
53	Returns:
54	Version: Newest version of package.
55	"""
56	from renku.core.util import requests	×
57
58	response = requests.get("https://pypi.org/pypi/{name}/json".format(name=name))	×
59
60	if response.status_code != 200:	×
61	return	×
62
63	description = response.json()	×
64
65	from packaging.version import Version	×
66
67	return max(	×
68	version
69	for version in (Version(version) for version in description["releases"].keys())
70	if allow_prereleases or not version.is_prerelease
71	)
72
73
74	@attr.s	10✔
75	class VersionCache:	10✔
76	"""Cache information about package version."""
77
78	STATE_NAME = "selfcheck.json"	10✔
79
80	DATE_FMT = "%Y-%m-%dT%H:%M:%SZ"	10✔
81
82	MAX_AGE = 24 * 60 * 60 # 1 day	10✔
83
84	last_check = attr.ib()	10✔
85
86	pypi_version = attr.ib(default=None)	10✔
87
88	@last_check.default	10✔
89	def default_last_check(self):	10✔
90	"""Format default last check."""
91	return datetime.datetime.utcnow().strftime(self.DATE_FMT)	×
92
93	@property	10✔
94	def is_fresh(self):	10✔
95	"""Check if we need to refresh the state."""
96	if self.last_check and self.pypi_version:	×
97	current_time = datetime.datetime.utcnow()	×
98	last_check = datetime.datetime.strptime(self.last_check, self.DATE_FMT)	×
99	return (current_time - last_check).total_seconds() < self.MAX_AGE	×
100
101	@classmethod	10✔
102	def _cache(cls, app_name):	10✔
103	"""Return cache file.
104
105	Args:
106	app_name: Current application name.
107
108	Returns:
109	Directory of the cache.
110	"""
111	from pathlib import Path	×
112
113	from appdirs import user_cache_dir	×
114
115	cache_dir = Path(user_cache_dir(app_name, None))	×
116	cache_dir.mkdir(parents=True, exist_ok=True)	×
117
118	return cache_dir / cls.STATE_NAME	×
119
120	@classmethod	10✔
121	def load(cls, app_name):	10✔
122	"""Load information from a cache.
123
124	Args:
125	app_name:Current application name.
126
127	Returns:
128	Cache object.
129	"""
130	cache = cls._cache(app_name)	×
131
132	try:	×
133	with cache.open() as fp:	×
134	return cls(**json.load(fp)[sys.prefix])	×
135	except (IOError, ValueError, KeyError):	×
136	return cls()	×
137
138	def dump(self, app_name):	10✔
139	"""Store information in a cache.
140
141	Args:
142	app_name: Current application name.
143	"""
144	cache = self._cache(app_name)	×
145
146	# Attempt to write out our version check file
147	with lockfile.LockFile(str(cache)):	×
148	if cache.exists():	×
149	with cache.open() as fp:	×
150	state = json.load(fp)	×
151	else:
152	state = {}	×
153
154	state[sys.prefix] = attr.asdict(self)	×
155
156	with cache.open("w") as fp:	×
157	json.dump(state, fp, sort_keys=True)	×
158
159
160	def _check_version():	10✔
161	"""Check renku version."""
162	from renku.core.constant import APP_NAME	×
163
164	if VersionCache.load(APP_NAME).is_fresh:	×
165	return	×
166
167	from packaging.version import Version	×
168
169	from renku.version import __version__	×
170
171	version = Version(__version__)	×
172	allow_prereleases = version.is_prerelease	×
173
174	latest_version = find_latest_version("renku", allow_prereleases=allow_prereleases)	×
175
176	if version < latest_version:	×
177	click.secho(	×
178	f"You are using renku version {__version__}, however version {latest_version} is available.\n"
179	"You should consider upgrading ...",
180	fg="yellow",
181	bold=True,
182	err=True,
183	)
184
185	VersionCache(pypi_version=str(latest_version)).dump(APP_NAME)	×
186
187
188	def check_version(ctx, param, value):	10✔
189	"""Check for latest version of renku on PyPI."""
190	if ctx.resilient_parsing:	9✔
191	return	×
192
193	if not value and ctx.invoked_subcommand != "run":	9✔
194	ctx.call_on_close(_check_version)	×

SwissDataScienceCenter / renku-python / 4145649460

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