842240498

Build Type

push

github

Committed by mborsetti

Commit Message

Release 3.6.0

Run Details

860 of 1340 branches covered (64.18%)

Branch coverage included in aggregate %.

1 of 1 new or added line in 1 file covered. (100.0%)

316 existing lines in 11 files now uncovered.

2640 of 3348 relevant lines covered (78.85%)

6.19 hits per line

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

78.53

/webchanges/util.py

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

import html
import importlib.machinery
import importlib.util
import logging
import os
import re
import shlex
import subprocess
import sys
import textwrap
from math import floor, log10
from os import PathLike
from types import ModuleType
from typing import Callable, Iterable, List, Match, Tuple, Type, TypeVar, Union

logger = logging.getLogger(__name__)


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

    @staticmethod
    def sorted_by_kind(cls: T) -> List[Type[T]]:
        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:
                    # see https://github.com/thp/urlwatch/issues/634
                    logger.info(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:
                    # see https://github.com/thp/urlwatch/issues/634
                    logger.info(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 the file."""
    editor = os.environ.get('EDITOR', None)
    if not editor:
        editor = os.environ.get('VISUAL', None)
    if not editor:
        if os.name == 'nt':
            editor = 'notepad.exe'
        else:
            raise SystemExit('Please set the path to the editor in the environment variable $EDITOR'
                             ' e.g. "export EDITOR=nano"')

    subprocess.run(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."""
    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)
    sys.modules[module_name] = module
    loader.exec_module(module)
    return module


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

    :param string: The string
    :param length: The length of the chunked string
    :param numbering: Whether to number each line on the right

    :returns: a list of strings
    """
    if numbering and len(string) > length:
        try:
            text_length = length - 4 - 2
            digits_try = (1 if text_length <= 0
                          else floor(log10(len(string) / 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(string) / text_length
                digits_guess = floor(log10(lines_guess)) + 1

            chunks = textwrap.wrap(string, text_length)
            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(string, text_length)
                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(string, length)


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


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 and we can't find anything better.

    :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.
    """
    if extra_params and not callable(extra_params):
        extra_params = ' ' + extra_params.strip()

    def make_link(m: Match) -> str:
        """Replacement function for re.sub to convert plain text into HTML with links."""
        url = m.group(1)
        proto = 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 = m.group(1)
        if not proto:
            href = 'http://' + href  # no proto specified, use http

        if callable(extra_params):
            params = ' ' + 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 += ' title=' + href

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

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

1	"""A few utilities used elsewhere."""
2
3	import html	8✔
4	import importlib.machinery	8✔
5	import importlib.util	8✔
6	import logging	8✔
7	import os	8✔
8	import re	8✔
9	import shlex	8✔
10	import subprocess	8✔
11	import sys	8✔
12	import textwrap	8✔
13	from math import floor, log10	8✔
14	from os import PathLike	8✔
15	from types import ModuleType	8✔
16	from typing import Callable, Iterable, List, Match, Tuple, Type, TypeVar, Union	8✔
17
18	logger = logging.getLogger(__name__)	8✔
19
20
21	class TrackSubClasses(type):	8✔
22	"""A metaclass that stores subclass name-to-class mappings in the base class."""
23	T = TypeVar('T')	8✔
24
25	@staticmethod	8✔
26	def sorted_by_kind(cls: T) -> List[Type[T]]:	8✔
27	return [item for _, item in sorted((it.__kind__, it) for it in cls.__subclasses__.values() if it.__kind__)]	8✔
28
29	def __init__(cls, name: str, bases: Tuple[type], namespace: dict) -> None:	8✔
30	for base in bases:	8✔
31	if base == object:	8✔
32	continue	8✔
33
34	for attr in ('__required__', '__optional__'):	8✔
35	if not hasattr(base, attr):	8✔
36	continue	8✔
37
38	inherited = getattr(base, attr, ())	8✔
39	new_value = tuple(namespace.get(attr, ())) + tuple(inherited)	8✔
40	namespace[attr] = new_value	8✔
41	setattr(cls, attr, new_value)	8✔
42
43	for base in bases:	8✔
44	if base == object:	8✔
45	continue	8✔
46
47	if hasattr(cls, '__kind__'):	8✔
48	subclasses = getattr(base, '__subclasses__', None)	8✔
49	if subclasses is not None:	8!
50	# see https://github.com/thp/urlwatch/issues/634
51	logger.info(f'Registering {cls} as {cls.__kind__}')	8✔
52	subclasses[cls.__kind__] = cls	8✔
53	break	8✔
54	else:
55	anonymous_subclasses = getattr(base, '__anonymous_subclasses__', None)	8✔
56	if anonymous_subclasses is not None:	8✔
57	# see https://github.com/thp/urlwatch/issues/634
58	logger.info(f'Registering {cls}')	8✔
59	anonymous_subclasses.append(cls)	8✔
60	break	8✔
61
62	super().__init__(name, bases, namespace)	8✔
63
64
65	def edit_file(filename: Union[str, bytes, PathLike]) -> None:	8✔
66	"""Opens the editor to edit the file."""
67	editor = os.environ.get('EDITOR', None)	8✔
68	if not editor:	8!
UNCOV 69	editor = os.environ.get('VISUAL', None)	×
70	if not editor:	8!
71	if os.name == 'nt':	×
UNCOV 72	editor = 'notepad.exe'	×
73	else:
UNCOV 74	raise SystemExit('Please set the path to the editor in the environment variable $EDITOR'	×
75	' e.g. "export EDITOR=nano"')
76
77	subprocess.run(shlex.split(editor) + [str(filename)], check=True)	8✔
78
79
80	def import_module_from_source(module_name: str, source_path: Union[str, bytes, PathLike]) -> ModuleType:	8✔
81	"""Loads a module and executes it in its own namespace."""
82	source_path = str(source_path)	8✔
83	loader = importlib.machinery.SourceFileLoader(module_name, source_path)	8✔
84	spec = importlib.util.spec_from_file_location(module_name, source_path, loader=loader)	8✔
85	module = importlib.util.module_from_spec(spec)	8✔
86	sys.modules[module_name] = module	8✔
87	loader.exec_module(module)	8✔
88	return module	8✔
89
90
91	def chunk_string(string: str, length: int, numbering: bool = False) -> Iterable[str]:	8✔
92	"""Chunks a string.
93
94	:param string: The string
95	:param length: The length of the chunked string
96	:param numbering: Whether to number each line on the right
97
98	:returns: a list of strings
99	"""
100	if numbering and len(string) > length:	8✔
101	try:	8✔
102	text_length = length - 4 - 2	8✔
103	digits_try = (1 if text_length <= 0	8✔
104	else floor(log10(len(string) / text_length))) # initialization floor
105	digits_guess = digits_try + 1	8✔
106	while digits_guess > digits_try:	8✔
107	digits_try += 1	8✔
108	text_length = length - 4 - 2 * digits_try	8✔
109	if text_length <= 0:	8!
UNCOV 110	raise ValueError('Not enough space to chunkify string with line numbering (1)')	×
111	lines_guess = len(string) / text_length	8✔
112	digits_guess = floor(log10(lines_guess)) + 1	8✔
113
114	chunks = textwrap.wrap(string, text_length)	8✔
115	actual_digits = floor(log10(len(chunks))) + 1	8✔
116	while actual_digits > digits_try:	8!
117	digits_try += 1	×
118	text_length = length - 4 - 2 * digits_try	×
119	if text_length <= 0:	×
120	raise ValueError('Not enough space to chunkify string with line numbering (2)')	×
UNCOV 121	chunks = textwrap.wrap(string, text_length)	×
UNCOV 122	actual_digits = floor(log10(len(chunks))) + 1	×
123
124	length = len(chunks)	8✔
125	return [line + ' (' + f'{{:{digits_try}d}}'.format(i + 1) + f'/{length})' for i, line in enumerate(chunks)]	8✔
126
UNCOV 127	except ValueError as e:	×
UNCOV 128	logger.error(f'{e}')	×
129
130	return textwrap.wrap(string, length)	8✔
131
132
133	_URL_RE = re.compile(r"""\b((?:([\w-]+):(/{1,3})\|www[.])(?:(?:(?:[^\s&()]\|	8✔
134	&\|")(?:[^!"#$%&'()+,.:;<=>?@\[\]^`{\|}~\s]))\|(?:\((?:[^\s&()]\|&\|
135	")*\)))+)""") # noqa: DUO138 catastrophic "re" usage - denial-of-service possible
136
137
138	def linkify(	8✔
139	text: str,
140	shorten: bool = False,
141	extra_params: Union[str, Callable[[str], str]] = '',
142	require_protocol: bool = False,
143	permitted_protocols: Tuple[str] = ('http', 'https', 'mailto',)
144	) -> str:
145	"""Converts plain text into HTML with links.
146
147	For example linkify("Hello http://tornadoweb.org!") would return 'Hello
148	<a href="http://tornadoweb.org">http://tornadoweb.org</a>!'.
149
150	We are using a regex from tornado library https://github.com/tornadoweb/tornado/blob/master/tornado/escape.py.
151	This regex should avoid character entities other than & so that we won't pick up ", etc., but it is
152	vulnerable to Regular expression Denial of Service (ReDoS), which would divert computational resources to an
153	expensive regex match. The risk in this application is limited and we can't find anything better.
154
155	:parameter text: The text to linkify.
156	:parameter shorten: Long urls will be shortened for display.
157	:parameter extra_params: Extra text to include in the link tag, or a callable taking the link as an argument and
158	returning the extra text, e.g. linkify(text, extra_params='rel="nofollow" class="external"').
159	:parameter require_protocol: Only linkify urls which include a protocol; if this is False, urls such as
160	www.facebook.com will also be linkified.
161	:parameter permitted_protocols: Protocols which should be linkified, e.g. linkify(text,
162	permitted_protocols=('http', 'ftp', 'mailto')); it is very unsafe to include protocols such as javascript.
163	"""
164	if extra_params and not callable(extra_params):	8!
UNCOV 165	extra_params = ' ' + extra_params.strip()	×
166
167	def make_link(m: Match) -> str:	8✔
168	"""Replacement function for re.sub to convert plain text into HTML with links."""
169	url = m.group(1)	8✔
170	proto = m.group(2)	8✔
171	if require_protocol and not proto:	8!
UNCOV 172	return url # not protocol, no linkify	×
173
174	if proto and proto not in permitted_protocols:	8!
UNCOV 175	return url # bad protocol, no linkify	×
176
177	href = m.group(1)	8✔
178	if not proto:	8!
179	href = 'http://' + href # no proto specified, use http	8✔
180
181	if callable(extra_params):	8!
UNCOV 182	params = ' ' + extra_params(href).strip()	×
183	else:
184	params = extra_params	8✔
185
186	# clip long urls. max_len is just an approximation
187	max_len = 30	8✔
188	if shorten and len(url) > max_len:	8✔
189	before_clip = url	8✔
190	if proto:	8!
UNCOV 191	proto_len = len(proto) + 1 + len(m.group(3) or '') # +1 for :	×
192	else:
193	proto_len = 0	8✔
194
195	parts = url[proto_len:].split('/')	8✔
196	if len(parts) > 1:	8!
197	# Grab the whole host part plus the first bit of the path
198	# The path is usually not that interesting once shortened
199	# (no more slug, etc), so it really just provides a little
200	# extra indication of shortening.
201	url = (	8✔
202	url[:proto_len]
203	+ parts[0]
204	+ '/'
205	+ parts[1][:8].split('?')[0].split('.')[0]
206	)
207
208	if len(url) > max_len * 1.5: # still too long	8!
UNCOV 209	url = url[:max_len]	×
210
211	if url != before_clip:	8!
212	amp = url.rfind('&')	8✔
213	# avoid splitting html char entities
214	if amp > max_len - 5:	8!
215	url = url[:amp]	×
216	url += '...'	8✔
217
218	if len(url) >= len(before_clip):	8!
UNCOV 219	url = before_clip	×
220	else:
221	# full url is visible on mouse-over (for those who don't
222	# have a status bar, such as Safari by default)
223	params += ' title=' + href	8✔
224
225	return f'<a href="{href}"{params}>{url}</a>'	8✔
226
227	text = html.escape(text)	8✔
228	return _URL_RE.sub(make_link, text)	8✔

mborsetti / webchanges / 842240498

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