10385944731

Committed 14 Aug 2024 10:31AM UTC coverage: 77.792% (-0.04%) from 77.832%

Build # 10385944731

Build Type

push

github

Committed by

mborsetti

Commit Message

Version 3.25.0rc1

Run Details

1747 of 2513 branches covered (69.52%)

Branch coverage included in aggregate %.

9 of 9 new or added lines in 5 files covered. (100.0%)

1 existing line in 1 file now uncovered.

4446 of 5448 relevant lines covered (81.61%)

6.33 hits per line

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

79.38

/webchanges/util.py

"""A few utilities used elsewhere."""

# The code below is subject to the license contained in the LICENSE file, which is part of the source code.
from __future__ import annotations

import getpass
import importlib.machinery
import importlib.util
import logging
import os
import re
import shlex
import stat
import subprocess  # noqa: S404 Consider possible security implications associated with the subprocess module.
import sys
import textwrap
from math import floor, log10
from os import PathLike
from pathlib import Path
from types import ModuleType
from typing import Callable, Iterable, Match, Optional, Union

from markdown2 import Markdown

from webchanges import __project_name__, __version__

try:
    import httpx
except ImportError:  # pragma: no cover
    httpx = None  # type: ignore[assignment]
if httpx is not None:
    try:
        import h2
    except ImportError:  # pragma: no cover
        h2 = None  # type: ignore[assignment]

try:
    from packaging.version import parse as parse_version
except ImportError:  # pragma: no cover
    from webchanges._vendored.packaging_version import parse as parse_version  # type: ignore[assignment]

logger = logging.getLogger(__name__)


class TrackSubClasses(type):
    """A metaclass that stores subclass name-to-class mappings in the base class."""

    # __subclasses__ gets redefined from default "Callable[[_TT], list[_TT]]
    __subclasses__: dict[str, TrackSubClasses]  # type: ignore[assignment]
    __anonymous_subclasses__: list[TrackSubClasses]
    __required__: tuple[str, ...] = ()
    __optional__: tuple[str, ...] = ()

    __kind__: str

    @staticmethod
    def sorted_by_kind(cls: TrackSubClasses) -> list[TrackSubClasses]:
        """Generates a list of all members of a class sorted by the value of their __kind__ attribute. Useful for
        documentation.

        :param cls: The class.
        :return: The sorted list of class members.
        """
        return [item for _, item in sorted((it.__kind__, it) for it in cls.__subclasses__.values() if it.__kind__)]

    def __init__(cls, name: str, bases: tuple[type, ...], namespace: dict) -> None:
        for base in bases:
            if base == object:
                continue

            for attr in {'__required__', '__optional__'}:
                if not hasattr(base, attr):
                    continue

                inherited = getattr(base, attr, ())
                new_value = tuple(namespace.get(attr, ())) + tuple(inherited)
                namespace[attr] = new_value
                setattr(cls, attr, new_value)

        for base in bases:
            if base == object:
                continue

            if hasattr(cls, '__kind__'):
                subclasses = getattr(base, '__subclasses__', None)
                if subclasses is not None:
                    logger.debug(f'Registering {cls} as {cls.__kind__}')
                    subclasses[cls.__kind__] = cls
                    break
            else:
                anonymous_subclasses = getattr(base, '__anonymous_subclasses__', None)
                if anonymous_subclasses is not None:
                    logger.debug(f'Registering {cls}')
                    anonymous_subclasses.append(cls)
                    break

        super().__init__(name, bases, namespace)


def edit_file(filename: Union[str, bytes, PathLike]) -> None:
    """Opens the editor to edit a file.

    :param filename: The filename.
    """
    editor = os.environ.get('EDITOR')
    if not editor:
        editor = os.environ.get('VISUAL')
    if not editor:
        if os.name == 'nt':
            editor = 'notepad.exe'
        else:
            print('Please set the path to the editor in the environment variable $EDITOR, e.g. "export EDITOR=nano"')
            raise SystemExit(1)

    subprocess.run(  # noqa: S603 subprocess call - check for execution of untrusted input.
        shlex.split(editor) + [str(filename)],
        check=True,
    )


def import_module_from_source(module_name: str, source_path: Union[str, bytes, PathLike]) -> ModuleType:
    """Loads a module and executes it in its own namespace.

    :param module_name: The name of the module to import.
    :param source_path: The path where the module is located.
    :return: A ModuleType object.
    """
    source_path = str(source_path)
    loader = importlib.machinery.SourceFileLoader(module_name, source_path)
    spec = importlib.util.spec_from_file_location(module_name, source_path, loader=loader)
    module = importlib.util.module_from_spec(spec)  # type: ignore[arg-type]
    sys.modules[module_name] = module
    try:
        loader.exec_module(module)
    except Exception:
        sys.tracebacklimit = 1000
        raise
    return module


def chunk_string(text: str, length: int, numbering: bool = False) -> list[str]:
    """Chunks a string.

    :param text: The text to be chunked.
    :param length: The length of the chunked text.
    :param numbering: Whether to number each chunk on the left if more than one chunk is generated.

    :returns: a list of chunked strings
    """
    if numbering and len(text) > length:
        try:
            text_length = length - 4 - 2
            digits_try = 1 if text_length <= 0 else floor(log10(len(text) / text_length))  # initialization floor
            digits_guess = digits_try + 1
            while digits_guess > digits_try:
                digits_try += 1
                text_length = length - 4 - 2 * digits_try
                if text_length <= 0:
                    raise ValueError('Not enough space to chunkify string with line numbering (1)')
                lines_guess = len(text) / text_length
                digits_guess = floor(log10(lines_guess)) + 1

            chunks = textwrap.wrap(text, text_length, replace_whitespace=False)
            actual_digits = floor(log10(len(chunks))) + 1
            while actual_digits > digits_try:
                digits_try += 1
                text_length = length - 4 - 2 * digits_try
                if text_length <= 0:
                    raise ValueError('Not enough space to chunkify string with line numbering (2)')
                chunks = textwrap.wrap(text, text_length, replace_whitespace=False)
                actual_digits = floor(log10(len(chunks))) + 1

            length = len(chunks)
            return [line + ' (' + f'{{:{digits_try}d}}'.format(i + 1) + f'/{length})' for i, line in enumerate(chunks)]

        except ValueError as e:
            logger.error(f'{e}')

    return textwrap.wrap(text, length, replace_whitespace=False)


def linkify(
    text: str,
    shorten: bool = False,
    extra_params: Union[str, Callable[[str], str]] = '',
    require_protocol: bool = False,
    permitted_protocols: tuple[str, ...] = (
        'http',
        'https',
        'mailto',
    ),
) -> str:
    """Converts plain text into HTML with links.

    For example linkify("Hello http://tornadoweb.org!") would return 'Hello
    <a href="http://tornadoweb.org">http://tornadoweb.org</a>!'.

    We are using a regex from tornado library https://github.com/tornadoweb/tornado/blob/master/tornado/escape.py.
    This regex should avoid character entities other than &amp; so that we won't pick up &quot;, etc., but it is
    vulnerable to Regular expression Denial of Service (ReDoS), which would divert computational resources to an
    expensive regex match. The risk in this application is limited.

    In the future, consider using linkify from the bleach project instead (requires importing another package).

    :parameter text: The text to linkify.
    :parameter shorten: Long urls will be shortened for display.
    :parameter extra_params: Extra text to include in the link tag, or a callable taking the link as an argument and
        returning the extra text, e.g. linkify(text, extra_params='rel="nofollow" class="external"').
    :parameter require_protocol: Only linkify urls which include a protocol; if this is False, urls such as
        www.facebook.com will also be linkified.
    :parameter permitted_protocols: Protocols which should be linkified, e.g. linkify(text,
        permitted_protocols=('http', 'ftp', 'mailto')); it is very unsafe to include protocols such as javascript.
    """
    # _URL_RE = re.compile(  # original re
    #     r'\b('
    #     r'(?:([\w-]+):(/{1,3})|www[.])'
    #     r'(?:('
    #     r'?:(?:[^\s&()]|&amp;|&quot;)*(?:[^!"#$%&'
    #     r"'()*+,.:;<=>?@\[\]^`{|}~\s])"
    #     r")"
    #     r'|(?:\((?:[^\s&()]|&amp;|&quot;)*\))'
    #     r')+'
    #     r')'
    # )  # noqa: DUO138 catastrophic "re" usage - denial-of-service possible.

    _URL_RE = re.compile(  # modified to catch all URL parameters
        r'\b('
        r'(?:([\w-]+):(/{1,3})|www[.])'
        r'(?:('
        r'?:(?:[^\s()])*(?:[^!"#$%&'
        r"'()*+,.:;<=>?@\[\]^`{|}~\s])"
        r')'
        r'|(?:\((?:[^\s()])*\))'
        r')+'
        r')'
    )  # noqa: DUO138 catastrophic "re" usage - denial-of-service possible.

    if extra_params and not callable(extra_params):
        extra_params = f' {extra_params.strip()}'

    def make_link(m: Match) -> str:
        """Replacement function for re.sub using re.match as input to convert plain text into HTML with links."""
        url: str = m.group(1)
        proto: str = m.group(2)
        if require_protocol and not proto:
            return url  # not protocol, no linkify

        if proto and proto not in permitted_protocols:
            return url  # bad protocol, no linkify

        href: str = m.group(1)
        if not proto:
            proto = 'https'
            href = f'https://{href}'  # no proto specified, use https

        if callable(extra_params):
            params = f' {extra_params(href).strip()}'
        else:
            params = extra_params

        # clip long urls. max_len is just an approximation
        max_len = 30
        if shorten and len(url) > max_len:
            before_clip = url
            if proto:
                proto_len = len(proto) + 1 + len(m.group(3) or '')  # +1 for :
            else:
                proto_len = 0

            parts = url[proto_len:].split('/')
            if len(parts) > 1:
                # Grab the whole host part plus the first bit of the path
                # The path is usually not that interesting once shortened
                # (no more slug, etc), so it really just provides a little
                # extra indication of shortening.
                url = url[:proto_len] + parts[0] + '/' + parts[1][:8].split('?')[0].split('.')[0]

            if len(url) > max_len * 1.5:  # still too long
                url = url[:max_len]

            if url != before_clip:
                amp = url.rfind('&')
                # avoid splitting html char entities
                if amp > max_len - 5:
                    url = url[:amp]
                url += '...'

                if len(url) >= len(before_clip):
                    url = before_clip
                else:
                    # full url is visible on mouse-over (for those who don't
                    # have a status bar, such as Safari by default)
                    params += f' title={href}'

        return f'<a href="{href}"{params}>{url}</a>'

    # text = html.escape(text)
    return _URL_RE.sub(make_link, text)


def get_new_version_number(timeout: Optional[float] = None) -> Union[str, bool]:
    """Check PyPi for newer version of project.

    :parameter timeout: Timeout in seconds after which empty string is returned.
    :returns: The new version number if a newer version of project is found on PyPi, empty string otherwise, False if
      error retrieving the new version number is encountered.
    """
    if httpx is None:
        logger.info('Cannot query PyPi for latest release: HTTPX not installed')
        return False

    try:
        r = httpx.Client(http2=h2 is not None).get(f'https://pypi.org/pypi/{__project_name__}/json', timeout=timeout)
    except httpx.RequestError as e:
        logger.info(f'Exception when querying PyPi for latest release: {e}')
        return False

    if r.is_success:
        latest_release: str = r.json()['info']['version']
        if parse_version(latest_release) > parse_version(__version__):
            return latest_release
    else:
        logger.info(f'HTTP error when querying PyPi for latest release: {r}')

    return ''


def dur_text(duration: float) -> str:
    """Returns a formatted string optimized to the number of seconds for use in footers.

    :parameter duration: The duration in seconds.
    :returns: The formatted string.
    """
    if duration < 60:
        return f'{float(f"{duration:.2g}"):g} seconds'
    else:
        m, s = divmod(duration, 60)
        return f'{m:.0f}:{s:02.0f}'


def file_ownership_checks(filename: Path) -> list[str]:
    """Check security of file and its directory, i.e. that they belong to the current UID and only the owner
    can write to them. Return list of errors if any. Linux only.

    :returns: List of errors encountered (if any).
    """

    if os.name == 'nt':
        return []

    file_ownership_errors = []
    current_uid = os.getuid()  # type: ignore[attr-defined]  # not defined in Windows

    dirname = filename.parent
    dir_st = dirname.stat()
    if (dir_st.st_mode & (stat.S_IWGRP | stat.S_IWOTH)) != 0:
        file_ownership_errors.append(f'{dirname} is group/world-writable')
    if dir_st.st_uid != current_uid:
        file_ownership_errors.append(f'{dirname} not owned by {getpass.getuser()}')

    file_st = filename.stat()
    if (file_st.st_mode & (stat.S_IWGRP | stat.S_IWOTH)) != 0:
        file_ownership_errors.append(f'{filename} is group/world-writable')
    if file_st.st_uid != current_uid:
        file_ownership_errors.append(f'{filename} not owned by {getpass.getuser()}')

    return file_ownership_errors


def mark_to_html(
    text: str, markdown_padded_tables: Optional[bool] = False, extras: Optional[Iterable[str]] = None
) -> str:
    """Converts a line of Markdown (e.g. as generated by html2text filter) to html.

    :param text: The text in Markdown format.
    :param markdown_padded_tables: If true, monospace the tables for alignment.
    :param extras: Additional extras for Markdown.
    :return: The text in html format.
    """
    markdowner_extras = set(extras) if extras else set()
    markdowner_extras.add('strike')  # text marked by double tildes is ~~strikethrough~~
    markdowner_extras.add('target-blank-links')  # <a> tags have rel="noopener" for added security
    markdowner = Markdown(extras=markdowner_extras)
    if text == '* * *':  # manually expand horizontal ruler since <hr> is used to separate jobs
        return '-' * 80
    pre = ''
    post = ''
    if text.lstrip()[:2] == '* ':  # item of unordered list
        lstripped = text.lstrip(' ')
        indent = len(text) - len(lstripped)
        pre += '&nbsp;' * indent
        pre += '● ' if indent == 2 else '⯀ ' if indent == 4 else '○ '
        text = text.split('* ', 1)[1]
    if text[:1] == ' ':
        # replace leading spaces with NBSP or converter will strip them all
        stripped = text.lstrip()
        text = '&nbsp;' * (len(text) - len(stripped)) + stripped
    text = text.replace('` ', '`&nbsp;')  # replace leading spaces within code blocks
    if markdown_padded_tables and '|' in text:
        # a padded row in a table; keep it monospaced for alignment
        pre += '<span style="font-family:monospace;white-space:pre-wrap">'
        post += '</span>'
    url_re = r'(?<!\]\()(http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+(?<![.,;!?)]))'
    text = re.sub(url_re, r'[[_Link with no text_]](\1)', text)  # linkify markdown lacking links
    text = text.replace('[](', '[[_Link with no text_]](')  # Add link text where missing
    html_out = str(markdowner.convert(text)).rstrip('\n')  # convert markdown to html
    # fixes for Gmail
    html_out = html_out.replace('<a', '<a style="font-family:inherit"')  # fix <a> tag styling
    html_out = html_out.replace('<img', '<img style="max-width:100%;height:auto;max-height:100%"')
    html_out = html_out.replace('<code>', '<span style="font-family:monospace;white-space:pre-wrap">')
    html_out = html_out.replace('</code>', '</span>')
    if 'tables' in markdowner_extras:
        html_out = html_out.replace('<table>', '<table border="1" cellspacing="0">')
    # remove <p> tags wrapping
    html_out, sub = re.subn(r'^<p>|</p>$', '', html_out)  # remove paragraph tags
    if sub:
        return pre + html_out + post
    html_out = re.sub(r'<(/?)h\d>', r'<\g<1>strong>', html_out)  # replace heading tags with <strong>
    return pre + html_out + post

1	"""A few utilities used elsewhere."""
2
3	# The code below is subject to the license contained in the LICENSE file, which is part of the source code.
4	from __future__ import annotations	8✔
5
6	import getpass	8✔
7	import importlib.machinery	8✔
8	import importlib.util	8✔
9	import logging	8✔
10	import os	8✔
11	import re	8✔
12	import shlex	8✔
13	import stat	8✔
14	import subprocess # noqa: S404 Consider possible security implications associated with the subprocess module.	8✔
15	import sys	8✔
16	import textwrap	8✔
17	from math import floor, log10	8✔
18	from os import PathLike	8✔
19	from pathlib import Path	8✔
20	from types import ModuleType	8✔
21	from typing import Callable, Iterable, Match, Optional, Union	8✔
22
23	from markdown2 import Markdown	8✔
24
25	from webchanges import __project_name__, __version__	8✔
26
27	try:	8✔
28	import httpx	8✔
29	except ImportError: # pragma: no cover
30	httpx = None # type: ignore[assignment]
31	if httpx is not None:	8!
32	try:	8✔
33	import h2	8✔
34	except ImportError: # pragma: no cover
35	h2 = None # type: ignore[assignment]
36
37	try:	8✔
38	from packaging.version import parse as parse_version	8✔
39	except ImportError: # pragma: no cover
40	from webchanges._vendored.packaging_version import parse as parse_version # type: ignore[assignment]
41
42	logger = logging.getLogger(__name__)	8✔
43
44
45	class TrackSubClasses(type):	8✔
46	"""A metaclass that stores subclass name-to-class mappings in the base class."""
47
48	# __subclasses__ gets redefined from default "Callable[[_TT], list[_TT]]
49	__subclasses__: dict[str, TrackSubClasses] # type: ignore[assignment]	8✔
50	__anonymous_subclasses__: list[TrackSubClasses]	8✔
51	__required__: tuple[str, ...] = ()	8✔
52	__optional__: tuple[str, ...] = ()	8✔
53
54	__kind__: str	8✔
55
56	@staticmethod	8✔
57	def sorted_by_kind(cls: TrackSubClasses) -> list[TrackSubClasses]:	8✔
58	"""Generates a list of all members of a class sorted by the value of their __kind__ attribute. Useful for
59	documentation.
60
61	:param cls: The class.
62	:return: The sorted list of class members.
63	"""
64	return [item for _, item in sorted((it.__kind__, it) for it in cls.__subclasses__.values() if it.__kind__)]	8✔
65
66	def __init__(cls, name: str, bases: tuple[type, ...], namespace: dict) -> None:	8✔
67	for base in bases:	8✔
68	if base == object:	8!
69	continue	×
70
71	for attr in {'__required__', '__optional__'}:	8✔
72	if not hasattr(base, attr):	8!
73	continue	×
74
75	inherited = getattr(base, attr, ())	8✔
76	new_value = tuple(namespace.get(attr, ())) + tuple(inherited)	8✔
77	namespace[attr] = new_value	8✔
78	setattr(cls, attr, new_value)	8✔
79
80	for base in bases:	8✔
81	if base == object:	8!
82	continue	×
83
84	if hasattr(cls, '__kind__'):	8!
85	subclasses = getattr(base, '__subclasses__', None)	8✔
86	if subclasses is not None:	8!
87	logger.debug(f'Registering {cls} as {cls.__kind__}')	8✔
88	subclasses[cls.__kind__] = cls	8✔
89	break	8✔
90	else:
91	anonymous_subclasses = getattr(base, '__anonymous_subclasses__', None)	×
92	if anonymous_subclasses is not None:	×
93	logger.debug(f'Registering {cls}')	×
94	anonymous_subclasses.append(cls)	×
95	break	×
96
97	super().__init__(name, bases, namespace)	8✔
98
99
100	def edit_file(filename: Union[str, bytes, PathLike]) -> None:	8✔
101	"""Opens the editor to edit a file.
102
103	:param filename: The filename.
104	"""
105	editor = os.environ.get('EDITOR')	8✔
106	if not editor:	8✔
107	editor = os.environ.get('VISUAL')	8✔
108	if not editor:	8!
109	if os.name == 'nt':	×
110	editor = 'notepad.exe'	×
111	else:
112	print('Please set the path to the editor in the environment variable $EDITOR, e.g. "export EDITOR=nano"')	×
113	raise SystemExit(1)	×
114
115	subprocess.run( # noqa: S603 subprocess call - check for execution of untrusted input.	8✔
116	shlex.split(editor) + [str(filename)],
117	check=True,
118	)
119
120
121	def import_module_from_source(module_name: str, source_path: Union[str, bytes, PathLike]) -> ModuleType:	8✔
122	"""Loads a module and executes it in its own namespace.
123
124	:param module_name: The name of the module to import.
125	:param source_path: The path where the module is located.
126	:return: A ModuleType object.
127	"""
128	source_path = str(source_path)	8✔
129	loader = importlib.machinery.SourceFileLoader(module_name, source_path)	8✔
130	spec = importlib.util.spec_from_file_location(module_name, source_path, loader=loader)	8✔
131	module = importlib.util.module_from_spec(spec) # type: ignore[arg-type]	8✔
132	sys.modules[module_name] = module	8✔
133	try:	8✔
134	loader.exec_module(module)	8✔
135	except Exception:	×
136	sys.tracebacklimit = 1000	×
137	raise	×
138	return module	8✔
139
140
141	def chunk_string(text: str, length: int, numbering: bool = False) -> list[str]:	8✔
142	"""Chunks a string.
143
144	:param text: The text to be chunked.
145	:param length: The length of the chunked text.
146	:param numbering: Whether to number each chunk on the left if more than one chunk is generated.
147
148	:returns: a list of chunked strings
149	"""
150	if numbering and len(text) > length:	8✔
151	try:	8✔
152	text_length = length - 4 - 2	8✔
153	digits_try = 1 if text_length <= 0 else floor(log10(len(text) / text_length)) # initialization floor	8✔
154	digits_guess = digits_try + 1	8✔
155	while digits_guess > digits_try:	8✔
156	digits_try += 1	8✔
157	text_length = length - 4 - 2 * digits_try	8✔
158	if text_length <= 0:	8✔
159	raise ValueError('Not enough space to chunkify string with line numbering (1)')	8✔
160	lines_guess = len(text) / text_length	8✔
161	digits_guess = floor(log10(lines_guess)) + 1	8✔
162
163	chunks = textwrap.wrap(text, text_length, replace_whitespace=False)	8✔
164	actual_digits = floor(log10(len(chunks))) + 1	8✔
165	while actual_digits > digits_try:	8!
166	digits_try += 1	×
167	text_length = length - 4 - 2 * digits_try	×
168	if text_length <= 0:	×
169	raise ValueError('Not enough space to chunkify string with line numbering (2)')	×
170	chunks = textwrap.wrap(text, text_length, replace_whitespace=False)	×
171	actual_digits = floor(log10(len(chunks))) + 1	×
172
173	length = len(chunks)	8✔
174	return [line + ' (' + f'{{:{digits_try}d}}'.format(i + 1) + f'/{length})' for i, line in enumerate(chunks)]	8✔
175
176	except ValueError as e:	8✔
177	logger.error(f'{e}')	8✔
178
179	return textwrap.wrap(text, length, replace_whitespace=False)	8✔
180
181
182	def linkify(	8✔
183	text: str,
184	shorten: bool = False,
185	extra_params: Union[str, Callable[[str], str]] = '',
186	require_protocol: bool = False,
187	permitted_protocols: tuple[str, ...] = (
188	'http',
189	'https',
190	'mailto',
191	),
192	) -> str:
193	"""Converts plain text into HTML with links.
194
195	For example linkify("Hello http://tornadoweb.org!") would return 'Hello
196	<a href="http://tornadoweb.org">http://tornadoweb.org</a>!'.
197
198	We are using a regex from tornado library https://github.com/tornadoweb/tornado/blob/master/tornado/escape.py.
199	This regex should avoid character entities other than & so that we won't pick up ", etc., but it is
200	vulnerable to Regular expression Denial of Service (ReDoS), which would divert computational resources to an
201	expensive regex match. The risk in this application is limited.
202
203	In the future, consider using linkify from the bleach project instead (requires importing another package).
204
205	:parameter text: The text to linkify.
206	:parameter shorten: Long urls will be shortened for display.
207	:parameter extra_params: Extra text to include in the link tag, or a callable taking the link as an argument and
208	returning the extra text, e.g. linkify(text, extra_params='rel="nofollow" class="external"').
209	:parameter require_protocol: Only linkify urls which include a protocol; if this is False, urls such as
210	www.facebook.com will also be linkified.
211	:parameter permitted_protocols: Protocols which should be linkified, e.g. linkify(text,
212	permitted_protocols=('http', 'ftp', 'mailto')); it is very unsafe to include protocols such as javascript.
213	"""
214	# _URL_RE = re.compile( # original re
215	# r'\b('
216	# r'(?:([\w-]+):(/{1,3})\|www[.])'
217	# r'(?:('
218	# r'?:(?:[^\s&()]\|&\|")*(?:[^!"#$%&'
219	# r"'()*+,.:;<=>?@\[\]^`{\|}~\s])"
220	# r")"
221	# r'\|(?:\((?:[^\s&()]\|&\|")*\))'
222	# r')+'
223	# r')'
224	# ) # noqa: DUO138 catastrophic "re" usage - denial-of-service possible.
225
226	_URL_RE = re.compile( # modified to catch all URL parameters	8✔
227	r'\b('
228	r'(?:([\w-]+):(/{1,3})\|www[.])'
229	r'(?:('
230	r'?:(?:[^\s()])*(?:[^!"#$%&'
231	r"'()*+,.:;<=>?@\[\]^`{\|}~\s])"
232	r')'
233	r'\|(?:\((?:[^\s()])*\))'
234	r')+'
235	r')'
236	) # noqa: DUO138 catastrophic "re" usage - denial-of-service possible.
237
238	if extra_params and not callable(extra_params):	8✔
239	extra_params = f' {extra_params.strip()}'	8✔
240
241	def make_link(m: Match) -> str:	8✔
242	"""Replacement function for re.sub using re.match as input to convert plain text into HTML with links."""
243	url: str = m.group(1)	8✔
244	proto: str = m.group(2)	8✔
245	if require_protocol and not proto:	8!
246	return url # not protocol, no linkify	×
247
248	if proto and proto not in permitted_protocols:	8✔
249	return url # bad protocol, no linkify	8✔
250
251	href: str = m.group(1)	8✔
252	if not proto:	8✔
253	proto = 'https'	8✔
254	href = f'https://{href}' # no proto specified, use https	8✔
255
256	if callable(extra_params):	8!
257	params = f' {extra_params(href).strip()}'	×
258	else:
259	params = extra_params	8✔
260
261	# clip long urls. max_len is just an approximation
262	max_len = 30	8✔
263	if shorten and len(url) > max_len:	8✔
264	before_clip = url	8✔
265	if proto:	8!
266	proto_len = len(proto) + 1 + len(m.group(3) or '') # +1 for :	8✔
267	else:
UNCOV 268	proto_len = 0	×
269
270	parts = url[proto_len:].split('/')	8✔
271	if len(parts) > 1:	8!
272	# Grab the whole host part plus the first bit of the path
273	# The path is usually not that interesting once shortened
274	# (no more slug, etc), so it really just provides a little
275	# extra indication of shortening.
276	url = url[:proto_len] + parts[0] + '/' + parts[1][:8].split('?')[0].split('.')[0]	8✔
277
278	if len(url) > max_len * 1.5: # still too long	8!
279	url = url[:max_len]	×
280
281	if url != before_clip:	8!
282	amp = url.rfind('&')	8✔
283	# avoid splitting html char entities
284	if amp > max_len - 5:	8!
285	url = url[:amp]	×
286	url += '...'	8✔
287
288	if len(url) >= len(before_clip):	8!
289	url = before_clip	×
290	else:
291	# full url is visible on mouse-over (for those who don't
292	# have a status bar, such as Safari by default)
293	params += f' title={href}'	8✔
294
295	return f'<a href="{href}"{params}>{url}</a>'	8✔
296
297	# text = html.escape(text)
298	return _URL_RE.sub(make_link, text)	8✔
299
300
301	def get_new_version_number(timeout: Optional[float] = None) -> Union[str, bool]:	8✔
302	"""Check PyPi for newer version of project.
303
304	:parameter timeout: Timeout in seconds after which empty string is returned.
305	:returns: The new version number if a newer version of project is found on PyPi, empty string otherwise, False if
306	error retrieving the new version number is encountered.
307	"""
308	if httpx is None:	8!
309	logger.info('Cannot query PyPi for latest release: HTTPX not installed')	×
310	return False	×
311
312	try:	8✔
313	r = httpx.Client(http2=h2 is not None).get(f'https://pypi.org/pypi/{__project_name__}/json', timeout=timeout)	8✔
314	except httpx.RequestError as e:	×
315	logger.info(f'Exception when querying PyPi for latest release: {e}')	×
316	return False	×
317
318	if r.is_success:	8!
319	latest_release: str = r.json()['info']['version']	8✔
320	if parse_version(latest_release) > parse_version(__version__):	8!
321	return latest_release	×
322	else:
323	logger.info(f'HTTP error when querying PyPi for latest release: {r}')	×
324
325	return ''	8✔
326
327
328	def dur_text(duration: float) -> str:	8✔
329	"""Returns a formatted string optimized to the number of seconds for use in footers.
330
331	:parameter duration: The duration in seconds.
332	:returns: The formatted string.
333	"""
334	if duration < 60:	8!
335	return f'{float(f"{duration:.2g}"):g} seconds'	8✔
336	else:
337	m, s = divmod(duration, 60)	×
338	return f'{m:.0f}:{s:02.0f}'	×
339
340
341	def file_ownership_checks(filename: Path) -> list[str]:	8✔
342	"""Check security of file and its directory, i.e. that they belong to the current UID and only the owner
343	can write to them. Return list of errors if any. Linux only.
344
345	:returns: List of errors encountered (if any).
346	"""
347
348	if os.name == 'nt':	8!
349	return []	×
350
351	file_ownership_errors = []	8✔
352	current_uid = os.getuid() # type: ignore[attr-defined] # not defined in Windows	8✔
353
354	dirname = filename.parent	8✔
355	dir_st = dirname.stat()	8✔
356	if (dir_st.st_mode & (stat.S_IWGRP \| stat.S_IWOTH)) != 0:	8✔
357	file_ownership_errors.append(f'{dirname} is group/world-writable')	4✔
358	if dir_st.st_uid != current_uid:	8✔
359	file_ownership_errors.append(f'{dirname} not owned by {getpass.getuser()}')	4✔
360
361	file_st = filename.stat()	8✔
362	if (file_st.st_mode & (stat.S_IWGRP \| stat.S_IWOTH)) != 0:	8✔
363	file_ownership_errors.append(f'{filename} is group/world-writable')	8✔
364	if file_st.st_uid != current_uid:	8!
365	file_ownership_errors.append(f'{filename} not owned by {getpass.getuser()}')	×
366
367	return file_ownership_errors	8✔
368
369
370	def mark_to_html(	8✔
371	text: str, markdown_padded_tables: Optional[bool] = False, extras: Optional[Iterable[str]] = None
372	) -> str:
373	"""Converts a line of Markdown (e.g. as generated by html2text filter) to html.
374
375	:param text: The text in Markdown format.
376	:param markdown_padded_tables: If true, monospace the tables for alignment.
377	:param extras: Additional extras for Markdown.
378	:return: The text in html format.
379	"""
380	markdowner_extras = set(extras) if extras else set()	8✔
381	markdowner_extras.add('strike') # text marked by double tildes is ~~strikethrough~~	8✔
382	markdowner_extras.add('target-blank-links') # <a> tags have rel="noopener" for added security	8✔
383	markdowner = Markdown(extras=markdowner_extras)	8✔
384	if text == '* * *': # manually expand horizontal ruler since <hr> is used to separate jobs	8✔
385	return '-' * 80	8✔
386	pre = ''	8✔
387	post = ''	8✔
388	if text.lstrip()[:2] == '* ': # item of unordered list	8✔
389	lstripped = text.lstrip(' ')	8✔
390	indent = len(text) - len(lstripped)	8✔
391	pre += ' ' * indent	8✔
392	pre += '● ' if indent == 2 else '⯀ ' if indent == 4 else '○ '	8✔
393	text = text.split('* ', 1)[1]	8✔
394	if text[:1] == ' ':	8✔
395	# replace leading spaces with NBSP or converter will strip them all
396	stripped = text.lstrip()	8✔
397	text = ' ' * (len(text) - len(stripped)) + stripped	8✔
398	text = text.replace('` ', '` ') # replace leading spaces within code blocks	8✔
399	if markdown_padded_tables and '\|' in text:	8✔
400	# a padded row in a table; keep it monospaced for alignment
401	pre += '<span style="font-family:monospace;white-space:pre-wrap">'	8✔
402	post += '</span>'	8✔
403	url_re = r'(?<!\]\()(http[s]?://(?:[a-zA-Z]\|[0-9]\|[$-_@.&+]\|[!*\\(\\),]\|(?:%[0-9a-fA-F][0-9a-fA-F]))+(?<![.,;!?)]))'	8✔
404	text = re.sub(url_re, r'[[_Link with no text_]](\1)', text) # linkify markdown lacking links	8✔
405	text = text.replace('[](', '[[_Link with no text_]](') # Add link text where missing	8✔
406	html_out = str(markdowner.convert(text)).rstrip('\n') # convert markdown to html	8✔
407	# fixes for Gmail
408	html_out = html_out.replace('<a', '<a style="font-family:inherit"') # fix <a> tag styling	8✔
409	html_out = html_out.replace('<img', '<img style="max-width:100%;height:auto;max-height:100%"')	8✔
410	html_out = html_out.replace('<code>', '<span style="font-family:monospace;white-space:pre-wrap">')	8✔
411	html_out = html_out.replace('</code>', '</span>')	8✔
412	if 'tables' in markdowner_extras:	8✔
413	html_out = html_out.replace('<table>', '<table border="1" cellspacing="0">')	8✔
414	# remove <p> tags wrapping
415	html_out, sub = re.subn(r'^<p>\|</p>$', '', html_out) # remove paragraph tags	8✔
416	if sub:	8✔
417	return pre + html_out + post	8✔
418	html_out = re.sub(r'<(/?)h\d>', r'<\g<1>strong>', html_out) # replace heading tags with <strong>	8✔
419	return pre + html_out + post	8✔

mborsetti / webchanges / 10385944731

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