12279337432

Committed 11 Dec 2024 03:15PM UTC coverage: 47.031% (+0.2%) from 46.79%

Build # 12279337432

Build Type

Pull #751

github

Specific Base d4edef

Committed by

web-flow

Commit Message

Merge eea51f2f7 into 2f9d65d86

Pull Request Pull Request #751: Fiber trajectory GUI

Run Details

0 of 114 new or added lines in 1 file covered. (0.0%)

1076 existing lines in 22 files now uncovered.

4246 of 9028 relevant lines covered (47.03%)

0.94 hits per line

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

50.98

/iblrig/version_management.py

import logging
import re
from collections.abc import Callable
from functools import cache
from pathlib import Path
from subprocess import 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

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 call_git(*args: str, cache_output: bool = True, on_error: Literal['raise', 'log', 'silence'] = '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:
        from iblrig.tools import cached_check_output

        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.
    """
    from iblrig.tools import internet_available

    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.
    """
    from iblrig.tools import internet_available

    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:
        from iblrig.tools import internet_available

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

int-brain-lab / iblrig / 12279337432

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