810389142

Build Type

push

github

Committed by mborsetti

Commit Message

Release 3.5.0

Run Details

815 of 1333 branches covered (61.14%)

Branch coverage included in aggregate %.

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

473 existing lines in 12 files now uncovered.

2546 of 3314 relevant lines covered (76.83%)

12.06 hits per line

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

68.97

/webchanges/cli.py

#!/usr/bin/env python3

"""Module containing the entry point."""

# See config module for the command line arguments

import logging
import os.path
import shutil
import signal
import sys
import warnings
from typing import Optional, Union

from appdirs import AppDirs

from . import __copyright__, __docs_url__, __min_python_version__, __project_name__, __version__
from .command import UrlwatchCommand
from .config import CommandConfig
from .main import Urlwatch
from .storage import CacheDirStorage, CacheRedisStorage, CacheSQLite3Storage, YamlConfigStorage, YamlJobsStorage

# directory where the config, jobs and hooks files are located
if os.name != 'nt':
    # typically ~/.config/{__project_name__}
    config_dir = AppDirs(__project_name__).user_config_dir
else:
    config_dir = os.path.expanduser(os.path.join('~', 'Documents', __project_name__))

# directory where the database is located
# typically ~/.cache/{__project_name__} or %LOCALAPPDATA%\{__project_name__}\{__project_name__}\Cache
cache_dir = AppDirs(__project_name__).user_cache_dir

# Ignore SIGPIPE for stdout (see https://github.com/thp/urlwatch/issues/77)
try:
    signal.signal(signal.SIGPIPE, signal.SIG_DFL)
except AttributeError:
    # Windows does not have signal.SIGPIPE
    pass

logger = logging.getLogger(__name__)


def setup_logger_verbose(log_level: Union[str, int] = logging.DEBUG) -> None:
    """Set up the loggers for verbosity."""
    import platform

    # logging.basicConfig(format='%(asctime)s %(module)s %(levelname)s: %(message)s', level=log_level)
    console = logging.StreamHandler()
    console.setFormatter(logging.Formatter('%(asctime)s %(module)s %(levelname)s: %(message)s'))
    logger.addHandler(console)
    logger.setLevel(log_level)
    logger.debug(f'{__project_name__}: {__version__} {__copyright__}')
    logger.debug(f'Python: {platform.python_build()}')
    logger.debug(f'System: {platform.platform()}')


def locate_storage_file(filename: str, default_dir: str, ext: Optional[str] = None) -> str:
    """Searches for file as specified and in default directory; retry with 'ext' extension.

    :param filename: The filename
    :param default_dir: Default storage directory
    :param ext: The extension, e.g. '.yaml', to add for searching if first scan fails

    :returns: The filename, either original or with path where found
    """
    search_filenames = [filename]

    # if exp is given, iterate both on raw filename and the filename with ext if different
    if ext and os.path.splitext(filename)[1] != ext:
        search_filenames.append(filename + ext)

    for file in search_filenames:
        # return if found
        if os.path.isfile(file):
            return file

        # no directory specified: add default one
        if file == os.path.basename(file):
            new_file = os.path.join(default_dir, file)
            if os.path.isfile(new_file):
                return new_file

    return filename


def migrate_from_urlwatch(config_file: str, jobs_file: str, hooks_file: str, cache_file: str) -> None:
    """Check for existence of legacy (urlwatch 2.23) config, jobs and hooks files and migrate them (i.e. make a copy to
    new folder and or naming).  Original files are not deleted."""
    uw_urlwatch_dir = os.path.expanduser(os.path.join('~', '.' + 'urlwatch'))
    uw_config_file = os.path.join(uw_urlwatch_dir, 'urlwatch.yaml')
    uw_urls_file = os.path.join(uw_urlwatch_dir, 'urls.yaml')
    uw_hooks_file = os.path.join(uw_urlwatch_dir, 'hooks.py')
    uw_cache_dir = AppDirs('urlwatch').user_cache_dir
    uw_cache_file = os.path.join(uw_cache_dir, 'cache.db')
    for old_file, new_file in zip((uw_config_file, uw_urls_file, uw_hooks_file, uw_cache_file),
                                  (config_file, jobs_file, hooks_file, cache_file)):
        if os.path.isfile(old_file) and not os.path.isfile(new_file):
            os.makedirs(os.path.dirname(new_file), exist_ok=True)
            shutil.copyfile(old_file, new_file)
            logger.warning(f"Copied urlwatch '{old_file}' file to {__project_name__} '{new_file}'")


def first_run(command_config: CommandConfig) -> None:
    """Create jobs and configuration files."""
    os.makedirs(config_dir, exist_ok=True)
    if not os.path.isfile(command_config.config):
        YamlConfigStorage.write_default_config(command_config.config)
        print(f'Created default config file at {command_config.config}\n> Edit it with {__project_name__} '
              f'--edit-config')
    if not os.path.isfile(command_config.jobs):
        with open(command_config.jobs, 'w') as fp:
            fp.write(f'# {__project_name__} jobs file. See {__docs_url__}jobs.html\n')
        command_config.edit = True
        print(f'Created default jobs file at {command_config.jobs}\n> Edit it with {__project_name__} --edit-config '
              f'(trying to launch it automatically)')


def main() -> None:  # pragma: no cover
    """The entry point."""
    # make sure that PendingDeprecationWarning are displayed from all modules (otherwise only those in __main__ are)
    warnings.filterwarnings('default', category=PendingDeprecationWarning)

    # Issue deprecation warning if running on minimum version supported
    if sys.version_info[0:2] == __min_python_version__:
        current_minor_version = '.'.join(str(n) for n in sys.version_info[0:2])
        next_minor_version = f'{__min_python_version__[0]}.{__min_python_version__[1] + 1}'
        warning = (f'Support for Python {current_minor_version} will be ending three years from the date Python '
                   f'{next_minor_version} was released')
        print(f'WARNING: {warning}\n')
        PendingDeprecationWarning(warning)

    # The config, jobs, hooks and cache files
    default_config_file = os.path.join(config_dir, 'config.yaml')
    default_jobs_file = os.path.join(config_dir, 'jobs.yaml')
    default_hooks_file = os.path.join(config_dir, 'hooks.py')
    default_cache_file = os.path.join(cache_dir, 'cache.db')

    # migrate legacy urlwatch 2.23 files
    migrate_from_urlwatch(default_config_file, default_jobs_file, default_hooks_file, default_cache_file)

    # load config files
    prefix, bindir = os.path.split(os.path.dirname(os.path.abspath(sys.argv[0])))
    command_config = CommandConfig(__project_name__, config_dir, bindir, prefix, default_config_file, default_jobs_file,
                                   default_hooks_file, default_cache_file, verbose=False)

    # set up the logger to verbose if needed
    if command_config.verbose:
        setup_logger_verbose(command_config.log_level)

    # check for location of config files entered in cli
    command_config.config = locate_storage_file(command_config.config, command_config.config_dir, '.yaml')
    command_config.jobs = locate_storage_file(command_config.jobs, command_config.config_dir, '.yaml')
    command_config.hooks = locate_storage_file(command_config.hooks, command_config.config_dir, '.py')

    # check for first run
    if command_config.config == default_config_file and not os.path.isfile(command_config.config):
        first_run(command_config)

    # setup config file API
    config_storage = YamlConfigStorage(command_config.config)  # storage.py

    # setup database API
    if command_config.database_engine == 'sqlite3':
        cache_storage = CacheSQLite3Storage(command_config.cache, command_config.max_snapshots)  # storage.py
    elif any(command_config.cache.startswith(prefix) for prefix in ('redis://', 'rediss://')):
        cache_storage = CacheRedisStorage(command_config.cache)  # storage.py
    elif command_config.database_engine == 'redis':
        raise RuntimeError("A redis URI (starting with 'redis://' or 'rediss://' needs to be specified with --cache")
    elif command_config.database_engine == 'textfiles':
        cache_storage = CacheDirStorage(command_config.cache)  # storage.py
    elif command_config.database_engine == 'minidb':
        from .storage_minidb import CacheMiniDBStorage  # legacy code imported only if needed (requires minidb)

        cache_storage = CacheMiniDBStorage(command_config.cache)  # storage.py
    else:
        raise NotImplementedError(f'Database engine {command_config.database_engine} not implemented')

    # setup jobs file API
    jobs_storage = YamlJobsStorage(command_config.jobs)  # storage.py

    # setup urlwatch
    urlwatcher = Urlwatch(command_config, config_storage, cache_storage, jobs_storage)  # main.py
    urlwatch_command = UrlwatchCommand(urlwatcher)  # command.py

    # run urlwatch
    urlwatch_command.run()


if __name__ == '__main__':

    main()

1	#!/usr/bin/env python3
2
3	"""Module containing the entry point."""	16✔
4
5	# See config module for the command line arguments
6
7	import logging	16✔
8	import os.path	16✔
9	import shutil	16✔
10	import signal	16✔
11	import sys	16✔
12	import warnings	16✔
13	from typing import Optional, Union	16✔
14
15	from appdirs import AppDirs	16✔
16
17	from . import __copyright__, __docs_url__, __min_python_version__, __project_name__, __version__	16✔
18	from .command import UrlwatchCommand	16✔
19	from .config import CommandConfig	16✔
20	from .main import Urlwatch	16✔
21	from .storage import CacheDirStorage, CacheRedisStorage, CacheSQLite3Storage, YamlConfigStorage, YamlJobsStorage	16✔
22
23	# directory where the config, jobs and hooks files are located
24	if os.name != 'nt':	16!
25	# typically ~/.config/{__project_name__}
26	config_dir = AppDirs(__project_name__).user_config_dir	16✔
27	else:
UNCOV 28	config_dir = os.path.expanduser(os.path.join('~', 'Documents', __project_name__))	×
29
30	# directory where the database is located
31	# typically ~/.cache/{__project_name__} or %LOCALAPPDATA%\{__project_name__}\{__project_name__}\Cache
32	cache_dir = AppDirs(__project_name__).user_cache_dir	16✔
33
34	# Ignore SIGPIPE for stdout (see https://github.com/thp/urlwatch/issues/77)
35	try:	16✔
36	signal.signal(signal.SIGPIPE, signal.SIG_DFL)	16✔
37	except AttributeError:	×
38	# Windows does not have signal.SIGPIPE
UNCOV 39	pass	×
40
41	logger = logging.getLogger(__name__)	16✔
42
43
44	def setup_logger_verbose(log_level: Union[str, int] = logging.DEBUG) -> None:	16✔
45	"""Set up the loggers for verbosity."""
46	import platform	16✔
47
48	# logging.basicConfig(format='%(asctime)s %(module)s %(levelname)s: %(message)s', level=log_level)
49	console = logging.StreamHandler()	16✔
50	console.setFormatter(logging.Formatter('%(asctime)s %(module)s %(levelname)s: %(message)s'))	16✔
51	logger.addHandler(console)	16✔
52	logger.setLevel(log_level)	16✔
53	logger.debug(f'{__project_name__}: {__version__} {__copyright__}')	16✔
54	logger.debug(f'Python: {platform.python_build()}')	16✔
55	logger.debug(f'System: {platform.platform()}')	16✔
56
57
58	def locate_storage_file(filename: str, default_dir: str, ext: Optional[str] = None) -> str:	16✔
59	"""Searches for file as specified and in default directory; retry with 'ext' extension.
60
61	:param filename: The filename
62	:param default_dir: Default storage directory
63	:param ext: The extension, e.g. '.yaml', to add for searching if first scan fails
64
65	:returns: The filename, either original or with path where found
66	"""
67	search_filenames = [filename]	16✔
68
69	# if exp is given, iterate both on raw filename and the filename with ext if different
70	if ext and os.path.splitext(filename)[1] != ext:	16!
71	search_filenames.append(filename + ext)	16✔
72
73	for file in search_filenames:	16✔
74	# return if found
75	if os.path.isfile(file):	16!
76	return file	×
77
78	# no directory specified: add default one
79	if file == os.path.basename(file):	16!
80	new_file = os.path.join(default_dir, file)	16✔
81	if os.path.isfile(new_file):	16!
UNCOV 82	return new_file	×
83
84	return filename	16✔
85
86
87	def migrate_from_urlwatch(config_file: str, jobs_file: str, hooks_file: str, cache_file: str) -> None:	16✔
88	"""Check for existence of legacy (urlwatch 2.23) config, jobs and hooks files and migrate them (i.e. make a copy to
89	new folder and or naming). Original files are not deleted."""
90	uw_urlwatch_dir = os.path.expanduser(os.path.join('~', '.' + 'urlwatch'))	16✔
91	uw_config_file = os.path.join(uw_urlwatch_dir, 'urlwatch.yaml')	16✔
92	uw_urls_file = os.path.join(uw_urlwatch_dir, 'urls.yaml')	16✔
93	uw_hooks_file = os.path.join(uw_urlwatch_dir, 'hooks.py')	16✔
94	uw_cache_dir = AppDirs('urlwatch').user_cache_dir	16✔
95	uw_cache_file = os.path.join(uw_cache_dir, 'cache.db')	16✔
96	for old_file, new_file in zip((uw_config_file, uw_urls_file, uw_hooks_file, uw_cache_file),	16✔
97	(config_file, jobs_file, hooks_file, cache_file)):
98	if os.path.isfile(old_file) and not os.path.isfile(new_file):	16!
UNCOV 99	os.makedirs(os.path.dirname(new_file), exist_ok=True)	×
UNCOV 100	shutil.copyfile(old_file, new_file)	×
UNCOV 101	logger.warning(f"Copied urlwatch '{old_file}' file to {__project_name__} '{new_file}'")	×
102
103
104	def first_run(command_config: CommandConfig) -> None:	16✔
105	"""Create jobs and configuration files."""
106	os.makedirs(config_dir, exist_ok=True)	×
UNCOV 107	if not os.path.isfile(command_config.config):	×
UNCOV 108	YamlConfigStorage.write_default_config(command_config.config)	×
UNCOV 109	print(f'Created default config file at {command_config.config}\n> Edit it with {__project_name__} '	×
110	f'--edit-config')
UNCOV 111	if not os.path.isfile(command_config.jobs):	×
UNCOV 112	with open(command_config.jobs, 'w') as fp:	×
UNCOV 113	fp.write(f'# {__project_name__} jobs file. See {__docs_url__}jobs.html\n')	×
UNCOV 114	command_config.edit = True	×
UNCOV 115	print(f'Created default jobs file at {command_config.jobs}\n> Edit it with {__project_name__} --edit-config '	×
116	f'(trying to launch it automatically)')
117
118
119	def main() -> None: # pragma: no cover
120	"""The entry point."""
121	# make sure that PendingDeprecationWarning are displayed from all modules (otherwise only those in __main__ are)
122	warnings.filterwarnings('default', category=PendingDeprecationWarning)
123
124	# Issue deprecation warning if running on minimum version supported
125	if sys.version_info[0:2] == __min_python_version__:
126	current_minor_version = '.'.join(str(n) for n in sys.version_info[0:2])
127	next_minor_version = f'{__min_python_version__[0]}.{__min_python_version__[1] + 1}'
128	warning = (f'Support for Python {current_minor_version} will be ending three years from the date Python '
129	f'{next_minor_version} was released')
130	print(f'WARNING: {warning}\n')
131	PendingDeprecationWarning(warning)
132
133	# The config, jobs, hooks and cache files
134	default_config_file = os.path.join(config_dir, 'config.yaml')
135	default_jobs_file = os.path.join(config_dir, 'jobs.yaml')
136	default_hooks_file = os.path.join(config_dir, 'hooks.py')
137	default_cache_file = os.path.join(cache_dir, 'cache.db')
138
139	# migrate legacy urlwatch 2.23 files
140	migrate_from_urlwatch(default_config_file, default_jobs_file, default_hooks_file, default_cache_file)
141
142	# load config files
143	prefix, bindir = os.path.split(os.path.dirname(os.path.abspath(sys.argv[0])))
144	command_config = CommandConfig(__project_name__, config_dir, bindir, prefix, default_config_file, default_jobs_file,
145	default_hooks_file, default_cache_file, verbose=False)
146
147	# set up the logger to verbose if needed
148	if command_config.verbose:
149	setup_logger_verbose(command_config.log_level)
150
151	# check for location of config files entered in cli
152	command_config.config = locate_storage_file(command_config.config, command_config.config_dir, '.yaml')
153	command_config.jobs = locate_storage_file(command_config.jobs, command_config.config_dir, '.yaml')
154	command_config.hooks = locate_storage_file(command_config.hooks, command_config.config_dir, '.py')
155
156	# check for first run
157	if command_config.config == default_config_file and not os.path.isfile(command_config.config):
158	first_run(command_config)
159
160	# setup config file API
161	config_storage = YamlConfigStorage(command_config.config) # storage.py
162
163	# setup database API
164	if command_config.database_engine == 'sqlite3':
165	cache_storage = CacheSQLite3Storage(command_config.cache, command_config.max_snapshots) # storage.py
166	elif any(command_config.cache.startswith(prefix) for prefix in ('redis://', 'rediss://')):
167	cache_storage = CacheRedisStorage(command_config.cache) # storage.py
168	elif command_config.database_engine == 'redis':
169	raise RuntimeError("A redis URI (starting with 'redis://' or 'rediss://' needs to be specified with --cache")
170	elif command_config.database_engine == 'textfiles':
171	cache_storage = CacheDirStorage(command_config.cache) # storage.py
172	elif command_config.database_engine == 'minidb':
173	from .storage_minidb import CacheMiniDBStorage # legacy code imported only if needed (requires minidb)
174
175	cache_storage = CacheMiniDBStorage(command_config.cache) # storage.py
176	else:
177	raise NotImplementedError(f'Database engine {command_config.database_engine} not implemented')
178
179	# setup jobs file API
180	jobs_storage = YamlJobsStorage(command_config.jobs) # storage.py
181
182	# setup urlwatch
183	urlwatcher = Urlwatch(command_config, config_storage, cache_storage, jobs_storage) # main.py
184	urlwatch_command = UrlwatchCommand(urlwatcher) # command.py
185
186	# run urlwatch
187	urlwatch_command.run()
188
189
190	if __name__ == '__main__':
191
192	main()

mborsetti / webchanges / 810389142

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