10524660746

Committed 23 Aug 2024 11:07AM UTC coverage: 47.177% (+0.4%) from 46.79%

Build # 10524660746

Build Type

Pull #710

github

Specific Base 74f4e8

Committed by

web-flow

Commit Message

Merge 222cebb88 into db04546ad

Pull Request Pull Request #710: 8.23.1

Run Details

40 of 86 new or added lines in 4 files covered. (46.51%)

989 existing lines in 22 files now uncovered.

4052 of 8589 relevant lines covered (47.18%)

0.94 hits per line

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

60.0

/iblrig/version_management.py

import logging
import re
from collections.abc import Callable
from functools import cache
from pathlib import Path
from subprocess import STDOUT, CalledProcessError, SubprocessError, check_call, check_output
from typing import Any, Literal

import requests
from packaging import version

from iblrig import __version__
from iblrig.constants import BASE_DIR, IS_GIT, IS_VENV
from iblrig.tools import cached_check_output, internet_available

log = logging.getLogger(__name__)


def check_for_updates() -> tuple[bool, str]:
    """
    Check for updates to the iblrig software.

    This function compares the locally installed version of iblrig with the
    latest available version to determine if an update is available.

    Returns
    -------
        tuple[bool, Union[str, None]]: A tuple containing two elements.
            - A boolean indicating whether an update is available.
            - A string representing the latest available version, or None if
              no remote version information is available.
    """
    log.debug('Checking for updates ...')

    update_available = False
    v_local = get_local_version()
    v_remote = get_remote_version()

    if v_local and v_remote:
        update_available = v_remote.base_version > v_local.base_version
        if update_available:
            log.info(f'Update to iblrig {v_remote.base_version} available')
        else:
            log.debug('No update available')

    return update_available, v_remote.base_version if v_remote else ''


def get_local_version() -> version.Version | None:
    """
    Parse the local version string to obtain a Version object.

    This function attempts to parse the local version string (__version__)
    and returns a Version object representing the parsed version. If the
    parsing fails, it logs an error and returns None.

    Returns
    -------
    Union[version.Version, None]
        A Version object representing the parsed local version, or None if
        parsing fails.
    """
    try:
        log.debug('Parsing local version string')
        return version.parse(__version__)
    except (version.InvalidVersion, TypeError):
        log.error(f'Could not parse local version string: {__version__}')
        return None


def get_detailed_version_string(v_basic: str) -> str:
    """
    Generate a detailed version string based on a basic version string.

    This function takes a basic version string (major.minor.patch) and generates
    a detailed version string by querying Git for additional version information.
    The detailed version includes commit number of commits since the last tag,
    and Git status (dirty or broken). It is designed to fail safely.

    Parameters
    ----------
    v_basic : str
        A basic version string in the format 'major.minor.patch'.

    Returns
    -------
    str
        A detailed version string containing version information retrieved
        from Git, or the original basic version string if Git information
        cannot be obtained.

    Notes
    -----
    This method will only work with installations managed through Git.
    """
    if not internet_available():
        return v_basic

    if not IS_GIT:
        log.error('This installation of IBLRIG is not managed through git.')
        return v_basic

    # sanitize & check if input only consists of three fields - major, minor and patch - separated by dots
    v_sanitized = re.sub(r'^(\d+\.\d+\.\d+).*$$', r'\1', v_basic)
    if not re.match(r'^\d+\.\d+\.\d+$', v_sanitized):
        log.error(f"Couldn't parse version string: {v_basic}")
        return v_basic

    # get details through `git describe`
    try:
        get_remote_tags()
        v_detailed = check_output(
            ['git', 'describe', '--dirty', '--broken', '--match', v_sanitized, '--tags', '--long'],
            cwd=BASE_DIR,
            text=True,
            timeout=1,
            stderr=STDOUT,
        )
    except (SubprocessError, CalledProcessError) as e:
        log.debug(e, exc_info=True)
        return v_basic

    # apply a bit of regex magic for formatting & return the detailed version string
    v_detailed = re.sub(r'^((?:[\d+\.])+)(-[1-9]\d*)?(?:-0\d*)?(?:-\w+)(-dirty|-broken)?\n?$', r'\1\2\3', v_detailed)
    v_detailed = re.sub(r'-(\d+)', r'.post\1', v_detailed)
    v_detailed = re.sub(r'\-(dirty|broken)', r'+\1', v_detailed)
    return v_detailed


OnErrorLiteral = Literal['raise', 'log', 'silence']


def call_git(*args: str, cache_output: bool = True, on_error: OnErrorLiteral = 'raise') -> str | None:
    """
    Call a git command with the specified arguments.

    This function executes a git command with the provided arguments. It can cache the output of the command
    and handle errors based on the specified behavior.

    Parameters
    ----------
    *args : str
        The arguments to pass to the git command.
    cache_output : bool, optional
        Whether to cache the output of the command. Default is True.
    on_error : str, optional
        The behavior when an error occurs. Either
        - 'raise': raise the exception (default),
        - 'log': log the exception, or
        - 'silence': suppress the exception.

    Returns
    -------
    str or None
        The output of the git command as a string, or None if an error occurred.

    Raises
    ------
    RuntimeError
        If the installation is not managed through git and on_error is set to 'raise'.
    SubprocessError
        If the command fails and on_error is set to 'raise'.
    """
    kwargs: dict[str, Any] = {'args': ('git', *args), 'cwd': BASE_DIR, 'timeout': 5, 'text': True}
    if not IS_GIT:
        message = 'This installation of iblrig is not managed through git'
        if on_error == 'raise':
            raise RuntimeError(message)
        elif on_error == 'log':
            log.error(message)
        return None
    try:
        output = cached_check_output(**kwargs) if cache_output else check_output(**kwargs)
        return str(output).strip()
    except SubprocessError as e:
        if on_error == 'raise':
            raise e
        elif on_error == 'log':
            log.exception(e)
        return None


def get_branch():
    """
    Get the Git branch of the iblrig installation.

    Returns
    -------
    str or None
        The Git branch of the iblrig installation, or None if it cannot be determined.
    """
    return call_git('rev-parse', '--abbrev-ref', 'HEAD', on_error='log')


def get_commit_hash(short: bool = True):
    """
    Get the hash of the currently checked out commit of the iblrig installation.

    Parameters
    ----------
    short : bool, optional
        Whether to return the short hash of the commit hash. Default is True.

    Returns
    -------
    str or None
        Hash of the currently checked out commit, or None if it cannot be determined.
    """
    args = ['rev-parse', '--short', 'HEAD'] if short else ['rev-parse', 'HEAD']
    return call_git(*args, on_error='log')


def get_remote_tags() -> None:
    """
    Fetch remote Git tags if not already fetched.

    This function fetches remote Git tags if they have not been fetched already.
    If tags are already fetched, it does nothing. If the installation is not
    managed through Git, it logs an error.

    Returns
    -------
    None

    Notes
    -----
    This method will only work with installations managed through Git.
    """
    if not internet_available():
        return
    if (branch := get_branch()) is None:
        return
    call_git('fetch', 'origin', branch, '-t', '-q', '-f', on_error='log')


@cache
def get_changelog() -> str:
    """
    Retrieve the changelog for the iblrig installation.

    This function retrieves and caches the changelog for the iblrig installation
    based on the current Git branch. If the changelog is already cached, it
    returns the cached value. If not, it attempts to fetch the changelog from
    the GitHub repository or read it locally if the remote fetch fails.

    Returns
    -------
    str
        The changelog for the iblrig installation.

    Notes
    -----
    This method relies on the presence of a CHANGELOG.md file either in the
    repository or locally.
    """
    try:
        if (branch := get_branch()) is None:
            raise RuntimeError()
        changelog = requests.get(
            f'https://raw.githubusercontent.com/int-brain-lab/iblrig/{branch}/CHANGELOG.md', allow_redirects=True
        ).text
    except (requests.RequestException, RuntimeError):
        with open(Path(BASE_DIR).joinpath('CHANGELOG.md')) as f:
            changelog = f.read()
    return changelog


def get_remote_version() -> version.Version | None:
    """
    Retrieve the remote version of iblrig from the Git repository.

    This function fetches and parses the remote version of the iblrig software
    from the Git repository. It uses Git tags to identify available versions.

    Returns
    -------
    Union[version.Version, None]
        A Version object representing the remote version if successfully obtained,
        or None if the remote version cannot be retrieved.

    Notes
    -----
    This method will only work with installations managed through Git.
    """
    if not internet_available():
        log.error('Cannot obtain remote version: Not connected to internet')
        return None

    references = call_git('ls-remote', '-t', '-q', '--exit-code', '--refs', 'origin', 'tags', '*', on_error='log')

    try:
        log.debug('Parsing local version string')
        get_remote_version.remote_version = max([version.parse(v) for v in re.findall(r'/(\d+\.\d+\.\d+)', references)])
        return get_remote_version.remote_version
    except (version.InvalidVersion, TypeError):
        log.error('Could not parse remote version string')
        return None


def is_dirty() -> bool:
    """
    Check if the Git working directory is dirty (has uncommitted changes).

    Uses 'git diff --quiet' to determine if there are uncommitted changes in the Git repository.

    Returns
    -------
        bool: True if the directory is dirty (has uncommitted changes) or an error occurs during execution,
              False if the directory is clean (no uncommitted changes).
    """
    try:
        return check_call(['git', 'diff', '--quiet'], cwd=BASE_DIR) != 0
    except CalledProcessError:
        return True


def check_upgrade_prerequisites(exception_handler: Callable | None = None, *args, **kwargs) -> None:
    """Check prerequisites for upgrading IBLRIG.

    This function verifies the prerequisites necessary for upgrading IBLRIG. It checks for
    internet connectivity, whether the IBLRIG installation is managed through Git, and
    whether the script is running within the IBLRIG virtual environment.

    Parameters
    ----------
    exception_handler : Callable, optional
        An optional callable that handles exceptions if raised during the check.
        If provided, it will be called with the exception as the first argument,
        followed by any additional positional arguments (*args), and any
        additional keyword arguments (**kwargs).

    *args : Additional positional arguments
        Any additional positional arguments needed by the `exception_handler` callable.

    **kwargs : Additional keyword arguments
        Any additional keyword arguments needed by the `exception_handler` callable.


    Raises
    ------
    ConnectionError
        If there is no connection to the internet.
    RuntimeError
        If the IBLRIG installation is not managed through Git, or
        if the script is not running within the IBLRIG virtual environment.
    """
    try:
        if not internet_available():
            raise ConnectionError('No connection to internet.')
        if not IS_GIT:
            raise RuntimeError('This installation of IBLRIG is not managed through Git.')
        if not IS_VENV:
            raise RuntimeError('You need to be in the IBLRIG virtual environment in order to upgrade.')
    except (ConnectionError, RuntimeError) as e:
        if callable(exception_handler):
            exception_handler(e, *args, **kwargs)
        else:
            raise e

1	import logging	2✔
2	import re	2✔
3	from collections.abc import Callable	2✔
4	from functools import cache	2✔
5	from pathlib import Path	2✔
6	from subprocess import STDOUT, CalledProcessError, SubprocessError, check_call, check_output	2✔
7	from typing import Any, Literal	2✔
8
9	import requests	2✔
10	from packaging import version	2✔
11
12	from iblrig import __version__	2✔
13	from iblrig.constants import BASE_DIR, IS_GIT, IS_VENV	2✔
14	from iblrig.tools import cached_check_output, internet_available	2✔
15
16	log = logging.getLogger(__name__)	2✔
17
18
19	def check_for_updates() -> tuple[bool, str]:	2✔
20	"""
21	Check for updates to the iblrig software.
22
23	This function compares the locally installed version of iblrig with the
24	latest available version to determine if an update is available.
25
26	Returns
27	-------
28	tuple[bool, Union[str, None]]: A tuple containing two elements.
29	- A boolean indicating whether an update is available.
30	- A string representing the latest available version, or None if
31	no remote version information is available.
32	"""
33	log.debug('Checking for updates ...')	2✔
34
35	update_available = False	2✔
36	v_local = get_local_version()	2✔
37	v_remote = get_remote_version()	2✔
38
39	if v_local and v_remote:	2✔
40	update_available = v_remote.base_version > v_local.base_version	2✔
41	if update_available:	2✔
42	log.info(f'Update to iblrig {v_remote.base_version} available')	2✔
43	else:
44	log.debug('No update available')	2✔
45
46	return update_available, v_remote.base_version if v_remote else ''	2✔
47
48
49	def get_local_version() -> version.Version \| None:	2✔
50	"""
51	Parse the local version string to obtain a Version object.
52
53	This function attempts to parse the local version string (__version__)
54	and returns a Version object representing the parsed version. If the
55	parsing fails, it logs an error and returns None.
56
57	Returns
58	-------
59	Union[version.Version, None]
60	A Version object representing the parsed local version, or None if
61	parsing fails.
62	"""
63	try:	2✔
64	log.debug('Parsing local version string')	2✔
65	return version.parse(__version__)	2✔
66	except (version.InvalidVersion, TypeError):	2✔
67	log.error(f'Could not parse local version string: {__version__}')	2✔
68	return None	2✔
69
70
71	def get_detailed_version_string(v_basic: str) -> str:	2✔
72	"""
73	Generate a detailed version string based on a basic version string.
74
75	This function takes a basic version string (major.minor.patch) and generates
76	a detailed version string by querying Git for additional version information.
77	The detailed version includes commit number of commits since the last tag,
78	and Git status (dirty or broken). It is designed to fail safely.
79
80	Parameters
81	----------
82	v_basic : str
83	A basic version string in the format 'major.minor.patch'.
84
85	Returns
86	-------
87	str
88	A detailed version string containing version information retrieved
89	from Git, or the original basic version string if Git information
90	cannot be obtained.
91
92	Notes
93	-----
94	This method will only work with installations managed through Git.
95	"""
96	if not internet_available():	2✔
97	return v_basic	2✔
98
99	if not IS_GIT:	2✔
100	log.error('This installation of IBLRIG is not managed through git.')	2✔
101	return v_basic	2✔
102
103	# sanitize & check if input only consists of three fields - major, minor and patch - separated by dots
104	v_sanitized = re.sub(r'^(\d+\.\d+\.\d+).*$$', r'\1', v_basic)	2✔
105	if not re.match(r'^\d+\.\d+\.\d+$', v_sanitized):	2✔
106	log.error(f"Couldn't parse version string: {v_basic}")	×
107	return v_basic	×
108
109	# get details through `git describe`
110	try:	2✔
111	get_remote_tags()	2✔
112	v_detailed = check_output(	2✔
113	['git', 'describe', '--dirty', '--broken', '--match', v_sanitized, '--tags', '--long'],
114	cwd=BASE_DIR,
115	text=True,
116	timeout=1,
117	stderr=STDOUT,
118	)
119	except (SubprocessError, CalledProcessError) as e:	2✔
120	log.debug(e, exc_info=True)	2✔
121	return v_basic	2✔
122
123	# apply a bit of regex magic for formatting & return the detailed version string
124	v_detailed = re.sub(r'^((?:[\d+\.])+)(-[1-9]\d)?(?:-0\d)?(?:-\w+)(-dirty\|-broken)?\n?$', r'\1\2\3', v_detailed)	2✔
125	v_detailed = re.sub(r'-(\d+)', r'.post\1', v_detailed)	2✔
126	v_detailed = re.sub(r'\-(dirty\|broken)', r'+\1', v_detailed)	2✔
127	return v_detailed	2✔
128
129
130	OnErrorLiteral = Literal['raise', 'log', 'silence']	2✔
131
132
133	def call_git(*args: str, cache_output: bool = True, on_error: OnErrorLiteral = 'raise') -> str \| None:	2✔
134	"""
135	Call a git command with the specified arguments.
136
137	This function executes a git command with the provided arguments. It can cache the output of the command
138	and handle errors based on the specified behavior.
139
140	Parameters
141	----------
142	*args : str
143	The arguments to pass to the git command.
144	cache_output : bool, optional
145	Whether to cache the output of the command. Default is True.
146	on_error : str, optional
147	The behavior when an error occurs. Either
148	- 'raise': raise the exception (default),
149	- 'log': log the exception, or
150	- 'silence': suppress the exception.
151
152	Returns
153	-------
154	str or None
155	The output of the git command as a string, or None if an error occurred.
156
157	Raises
158	------
159	RuntimeError
160	If the installation is not managed through git and on_error is set to 'raise'.
161	SubprocessError
162	If the command fails and on_error is set to 'raise'.
163	"""
164	kwargs: dict[str, Any] = {'args': ('git', *args), 'cwd': BASE_DIR, 'timeout': 5, 'text': True}	2✔
165	if not IS_GIT:	2✔
NEW 166	message = 'This installation of iblrig is not managed through git'	×
NEW 167	if on_error == 'raise':	×
NEW 168	raise RuntimeError(message)	×
NEW 169	elif on_error == 'log':	×
NEW 170	log.error(message)	×
NEW 171	return None	×
172	try:	2✔
173	output = cached_check_output(kwargs) if cache_output else check_output(kwargs)	2✔
174	return str(output).strip()	2✔
NEW 175	except SubprocessError as e:	×
NEW 176	if on_error == 'raise':	×
NEW 177	raise e	×
NEW 178	elif on_error == 'log':	×
NEW 179	log.exception(e)	×
UNCOV 180	return None	×
181
182
183	def get_branch():	2✔
184	"""
185	Get the Git branch of the iblrig installation.
186
187	Returns
188	-------
189	str or None
190	The Git branch of the iblrig installation, or None if it cannot be determined.
191	"""
192	return call_git('rev-parse', '--abbrev-ref', 'HEAD', on_error='log')	2✔
193
194
195	def get_commit_hash(short: bool = True):	2✔
196	"""
197	Get the hash of the currently checked out commit of the iblrig installation.
198
199	Parameters
200	----------
201	short : bool, optional
202	Whether to return the short hash of the commit hash. Default is True.
203
204	Returns
205	-------
206	str or None
207	Hash of the currently checked out commit, or None if it cannot be determined.
208	"""
NEW 209	args = ['rev-parse', '--short', 'HEAD'] if short else ['rev-parse', 'HEAD']	×
NEW 210	return call_git(*args, on_error='log')	×
211
212
213	def get_remote_tags() -> None:	2✔
214	"""
215	Fetch remote Git tags if not already fetched.
216
217	This function fetches remote Git tags if they have not been fetched already.
218	If tags are already fetched, it does nothing. If the installation is not
219	managed through Git, it logs an error.
220
221	Returns
222	-------
223	None
224
225	Notes
226	-----
227	This method will only work with installations managed through Git.
228	"""
229	if not internet_available():	2✔
230	return	×
231	if (branch := get_branch()) is None:	2✔
232	return	×
233	call_git('fetch', 'origin', branch, '-t', '-q', '-f', on_error='log')	2✔
234
235
236	@cache	2✔
237	def get_changelog() -> str:	2✔
238	"""
239	Retrieve the changelog for the iblrig installation.
240
241	This function retrieves and caches the changelog for the iblrig installation
242	based on the current Git branch. If the changelog is already cached, it
243	returns the cached value. If not, it attempts to fetch the changelog from
244	the GitHub repository or read it locally if the remote fetch fails.
245
246	Returns
247	-------
248	str
249	The changelog for the iblrig installation.
250
251	Notes
252	-----
253	This method relies on the presence of a CHANGELOG.md file either in the
254	repository or locally.
255	"""
UNCOV 256	try:	×
NEW 257	if (branch := get_branch()) is None:	×
NEW 258	raise RuntimeError()	×
259	changelog = requests.get(	×
260	f'https://raw.githubusercontent.com/int-brain-lab/iblrig/{branch}/CHANGELOG.md', allow_redirects=True
261	).text
NEW 262	except (requests.RequestException, RuntimeError):	×
263	with open(Path(BASE_DIR).joinpath('CHANGELOG.md')) as f:	×
264	changelog = f.read()	×
NEW 265	return changelog	×
266
267
268	def get_remote_version() -> version.Version \| None:	2✔
269	"""
270	Retrieve the remote version of iblrig from the Git repository.
271
272	This function fetches and parses the remote version of the iblrig software
273	from the Git repository. It uses Git tags to identify available versions.
274
275	Returns
276	-------
277	Union[version.Version, None]
278	A Version object representing the remote version if successfully obtained,
279	or None if the remote version cannot be retrieved.
280
281	Notes
282	-----
283	This method will only work with installations managed through Git.
284	"""
285	if not internet_available():	×
286	log.error('Cannot obtain remote version: Not connected to internet')	×
287	return None	×
288
NEW 289	references = call_git('ls-remote', '-t', '-q', '--exit-code', '--refs', 'origin', 'tags', '*', on_error='log')	×
290
UNCOV 291	try:	×
UNCOV 292	log.debug('Parsing local version string')	×
293	get_remote_version.remote_version = max([version.parse(v) for v in re.findall(r'/(\d+\.\d+\.\d+)', references)])	×
294	return get_remote_version.remote_version	×
295	except (version.InvalidVersion, TypeError):	×
296	log.error('Could not parse remote version string')	×
UNCOV 297	return None	×
298
299
300	def is_dirty() -> bool:	2✔
301	"""
302	Check if the Git working directory is dirty (has uncommitted changes).
303
304	Uses 'git diff --quiet' to determine if there are uncommitted changes in the Git repository.
305
306	Returns
307	-------
308	bool: True if the directory is dirty (has uncommitted changes) or an error occurs during execution,
309	False if the directory is clean (no uncommitted changes).
310	"""
311	try:	2✔
312	return check_call(['git', 'diff', '--quiet'], cwd=BASE_DIR) != 0	2✔
313	except CalledProcessError:	2✔
314	return True	2✔
315
316
317	def check_upgrade_prerequisites(exception_handler: Callable \| None = None, args, *kwargs) -> None:	2✔
318	"""Check prerequisites for upgrading IBLRIG.
319
320	This function verifies the prerequisites necessary for upgrading IBLRIG. It checks for
321	internet connectivity, whether the IBLRIG installation is managed through Git, and
322	whether the script is running within the IBLRIG virtual environment.
323
324	Parameters
325	----------
326	exception_handler : Callable, optional
327	An optional callable that handles exceptions if raised during the check.
328	If provided, it will be called with the exception as the first argument,
329	followed by any additional positional arguments (*args), and any
330	additional keyword arguments (**kwargs).
331
332	*args : Additional positional arguments
333	Any additional positional arguments needed by the `exception_handler` callable.
334
335	**kwargs : Additional keyword arguments
336	Any additional keyword arguments needed by the `exception_handler` callable.
337
338
339	Raises
340	------
341	ConnectionError
342	If there is no connection to the internet.
343	RuntimeError
344	If the IBLRIG installation is not managed through Git, or
345	if the script is not running within the IBLRIG virtual environment.
346	"""
347	try:	×
348	if not internet_available():	×
349	raise ConnectionError('No connection to internet.')	×
UNCOV 350	if not IS_GIT:	×
351	raise RuntimeError('This installation of IBLRIG is not managed through Git.')	×
352	if not IS_VENV:	×
353	raise RuntimeError('You need to be in the IBLRIG virtual environment in order to upgrade.')	×
354	except (ConnectionError, RuntimeError) as e:	×
355	if callable(exception_handler):	×
UNCOV 356	exception_handler(e, args, *kwargs)	×
357	else:
358	raise e	×

int-brain-lab / iblrig / 10524660746

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