9032957364

Committed 10 May 2024 01:25PM UTC coverage: 48.538% (+1.7%) from 46.79%

Build # 9032957364

Build Type

Pull #643

github

Specific Base 74d2ec

Committed by

web-flow

Commit Message

Merge aebf2c9af into ec2d8e4fe

Pull Request Pull Request #643: 8.19.0

Run Details

377 of 1074 new or added lines in 38 files covered. (35.1%)

977 existing lines in 19 files now uncovered.

3253 of 6702 relevant lines covered (48.54%)

0.97 hits per line

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

57.26

/iblrig/version_management.py

import logging
import re
from collections.abc import Callable
from pathlib import Path
from subprocess import STDOUT, CalledProcessError, SubprocessError, check_call, check_output

import requests
from packaging import version

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

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


@static_vars(branch=None)
def get_branch() -> str | None:
    """
    Get the Git branch of the iblrig installation.

    This function retrieves and caches the Git branch of the iblrig installation.
    If the branch is already cached, it returns the cached value. If not, it
    attempts to obtain the branch from the Git repository.

    Returns
    -------
    Union[str, None]
        The Git branch of the iblrig installation, or None if it cannot be determined.

    Notes
    -----
    This method will only work with installations managed through Git.
    """
    if get_branch.branch is not None:
        return get_branch.branch
    if not IS_GIT:
        log.error('This installation of iblrig is not managed through git')
    try:
        get_branch.branch = check_output(
            ['git', 'rev-parse', '--abbrev-ref', 'HEAD'], cwd=BASE_DIR, timeout=5, text=True
        ).removesuffix('\n')
        return get_branch.branch
    except (SubprocessError, CalledProcessError):
        return None


@static_vars(is_fetched_already=False)
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 get_remote_tags.is_fetched_already or not internet_available():
        return
    if not IS_GIT:
        log.error('This installation of iblrig is not managed through git')
    try:
        check_call(['git', 'fetch', 'origin', get_branch(), '-t', '-q', '-f'], cwd=BASE_DIR, timeout=5)
    except (SubprocessError, CalledProcessError):
        return
    get_remote_tags.is_fetched_already = True


@static_vars(changelog=None)
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.
    """
    if get_changelog.changelog is not None:
        return get_changelog.changelog
    try:
        changelog = requests.get(
            f'https://raw.githubusercontent.com/int-brain-lab/iblrig/{get_branch()}/CHANGELOG.md', allow_redirects=True
        ).text
    except requests.RequestException:
        with open(Path(BASE_DIR).joinpath('CHANGELOG.md')) as f:
            changelog = f.read()
    get_changelog.changelog = changelog
    return get_changelog.changelog


@static_vars(remote_version=None)
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 get_remote_version.remote_version is not None:
        log.debug(f'Using cached remote version: {get_remote_version.remote_version}')
        return get_remote_version.remote_version

    if not IS_GIT:
        log.error('Cannot obtain remote version: This installation of iblrig is not managed through git')
        return None

    if not internet_available():
        log.error('Cannot obtain remote version: Not connected to internet')
        return None

    try:
        log.debug('Obtaining remote version from github')
        get_remote_tags()
        references = check_output(
            ['git', 'ls-remote', '-t', '-q', '--exit-code', '--refs', 'origin', 'tags', '*'],
            cwd=BASE_DIR,
            timeout=5,
            encoding='UTF-8',
        )

    except (SubprocessError, CalledProcessError, FileNotFoundError):
        log.error('Could not obtain remote version string')
        return None

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

int-brain-lab / iblrig / 9032957364

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