2454619689

Build # 2454619689

Build Type

push

github

Committed by mborsetti

Commit Message

Version 3.10.2rc4

Run Details

1367 of 1917 branches covered (71.31%)

Branch coverage included in aggregate %.

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

156 existing lines in 5 files now uncovered.

3242 of 3974 relevant lines covered (81.58%)

6.41 hits per line

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

78.95

/webchanges/main.py

"""The main class.

For the entry point, see main() in the cli module."""

# 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 logging
import warnings
from typing import List, Optional, Tuple, Union

from . import __docs_url__
from .config import CommandConfig
from .handler import Report
from .jobs import JobBase
from .storage import CacheStorage, YamlConfigStorage, YamlJobsStorage
from .util import file_ownership_checks, get_new_version_number, import_module_from_source
from .worker import run_jobs

logger = logging.getLogger(__name__)


class Urlwatch(object):
    """The main class."""

    def __init__(
        self,
        urlwatch_config: CommandConfig,
        config_storage: YamlConfigStorage,
        cache_storage: CacheStorage,
        jobs_storage: YamlJobsStorage,
    ) -> None:
        """

        :param urlwatch_config: The CommandConfig object containing the program run information.
        :param config_storage: The YamlConfigStorage object containing the configuration information.
        :param cache_storage: The CacheStorage object containing snapshot database information
        :param jobs_storage: The YamlJobsStorage object containing the jobs information.
        """

        self.urlwatch_config = urlwatch_config

        self.config_storage = config_storage
        self.cache_storage = cache_storage
        self.jobs_storage = jobs_storage

        self.report = Report(self)
        self.jobs: List[JobBase] = []

        self._latest_release: Optional[Union[str, bool]] = None

        self.check_directories()

        if not self.urlwatch_config.edit_hooks:
            self.load_hooks()

        if not self.urlwatch_config.edit:
            self.load_jobs()

    def check_directories(self) -> None:
        """Check whether the configuration and jobs files directories exist. Create them and write default configuration
        files if one not found.
        """
        if (
            not (self.urlwatch_config.config and self.urlwatch_config.jobs)
            and not self.urlwatch_config.config_path.is_dir()
        ):
            self.urlwatch_config.config_path.mkdir(parents=True)
            if not self.urlwatch_config.config.is_file():
                self.config_storage.write_default_config(self.urlwatch_config.config)
                print(
                    f'A default config has been written to {self.urlwatch_config.config}.'
                    f'Use "{self.urlwatch_config.project_name} --edit-config" to customize it.'
                )

    def load_hooks(self) -> None:
        """Load hooks file."""
        if not self.urlwatch_config.hooks.is_file():
            warnings.warn(
                f'Hooks file not imported because {self.urlwatch_config.hooks} is not a file',
                ImportWarning,
            )
            return

        hooks_file_errors = file_ownership_checks(self.urlwatch_config.hooks)
        if hooks_file_errors:
            warnings.warn(
                f'Hooks file {self.urlwatch_config.hooks} not imported because '
                f" {' and '.join(hooks_file_errors)}.\n"
                f'(see {__docs_url__}en/stable/hooks.html#important-note-for-hooks-file)',
                ImportWarning,
            )
        else:
            import_module_from_source('hooks', self.urlwatch_config.hooks)
            logger.info(f'Imported hooks module from {self.urlwatch_config.hooks}')

    def load_jobs(self) -> None:
        """Load jobs from the file into self.jobs.

        :raises SystemExit: If job is not found, setting argument to 1.
        """
        if self.urlwatch_config.jobs.is_file():
            jobs = self.jobs_storage.load_secure()
        else:
            print(f'Jobs file not found: {self.urlwatch_config.jobs}')
            raise SystemExit(1)

        self.jobs = jobs

    def get_new_release_version(self, timeout: Optional[Union[float, Tuple[float, float]]] = None) -> Union[str, bool]:
        """Check PyPi to see if we're running the latest version. Memoized.

        :returns: Empty string if no higher version is available, otherwise the new version number.
        """
        if self._latest_release is not None:
            return self._latest_release

        self._latest_release = get_new_version_number(timeout)
        return self._latest_release

    def run_jobs(self) -> None:
        """Run all jobs."""
        run_jobs(self)

    def close(self) -> None:
        """Finalizer. Create reports ands close snapshots database."""
        self.report.finish(jobs_file=self.jobs_storage.filename)
        self.cache_storage.close()

1	"""The main class.
2
3	For the entry point, see main() in the cli module."""
4
5	# The code below is subject to the license contained in the LICENSE file, which is part of the source code.
6
7	from __future__ import annotations	8✔
8
9	import logging	8✔
10	import warnings	8✔
11	from typing import List, Optional, Tuple, Union	8✔
12
13	from . import __docs_url__	8✔
14	from .config import CommandConfig	8✔
15	from .handler import Report	8✔
16	from .jobs import JobBase	8✔
17	from .storage import CacheStorage, YamlConfigStorage, YamlJobsStorage	8✔
18	from .util import file_ownership_checks, get_new_version_number, import_module_from_source	8✔
19	from .worker import run_jobs	8✔
20
21	logger = logging.getLogger(__name__)	8✔
22
23
24	class Urlwatch(object):	8✔
25	"""The main class."""
26
27	def __init__(	8✔
28	self,
29	urlwatch_config: CommandConfig,
30	config_storage: YamlConfigStorage,
31	cache_storage: CacheStorage,
32	jobs_storage: YamlJobsStorage,
33	) -> None:
34	"""
35
36	:param urlwatch_config: The CommandConfig object containing the program run information.
37	:param config_storage: The YamlConfigStorage object containing the configuration information.
38	:param cache_storage: The CacheStorage object containing snapshot database information
39	:param jobs_storage: The YamlJobsStorage object containing the jobs information.
40	"""
41
42	self.urlwatch_config = urlwatch_config	8✔
43
44	self.config_storage = config_storage	8✔
45	self.cache_storage = cache_storage	8✔
46	self.jobs_storage = jobs_storage	8✔
47
48	self.report = Report(self)	8✔
49	self.jobs: List[JobBase] = []	8✔
50
51	self._latest_release: Optional[Union[str, bool]] = None	8✔
52
53	self.check_directories()	8✔
54
55	if not self.urlwatch_config.edit_hooks:	8!
56	self.load_hooks()	8✔
57
58	if not self.urlwatch_config.edit:	8!
59	self.load_jobs()	8✔
60
61	def check_directories(self) -> None:	8✔
62	"""Check whether the configuration and jobs files directories exist. Create them and write default configuration
63	files if one not found.
64	"""
65	if (	8!
66	not (self.urlwatch_config.config and self.urlwatch_config.jobs)
67	and not self.urlwatch_config.config_path.is_dir()
68	):
69	self.urlwatch_config.config_path.mkdir(parents=True)	×
70	if not self.urlwatch_config.config.is_file():	×
71	self.config_storage.write_default_config(self.urlwatch_config.config)	×
72	print(	×
73	f'A default config has been written to {self.urlwatch_config.config}.'
74	f'Use "{self.urlwatch_config.project_name} --edit-config" to customize it.'
75	)
76
77	def load_hooks(self) -> None:	8✔
78	"""Load hooks file."""
79	if not self.urlwatch_config.hooks.is_file():	8✔
80	warnings.warn(	8✔
81	f'Hooks file not imported because {self.urlwatch_config.hooks} is not a file',
82	ImportWarning,
83	)
84	return	8✔
85
86	hooks_file_errors = file_ownership_checks(self.urlwatch_config.hooks)	8✔
87	if hooks_file_errors:	8!
88	warnings.warn(	×
89	f'Hooks file {self.urlwatch_config.hooks} not imported because '
90	f" {' and '.join(hooks_file_errors)}.\n"
91	f'(see {__docs_url__}en/stable/hooks.html#important-note-for-hooks-file)',
92	ImportWarning,
93	)
94	else:
95	import_module_from_source('hooks', self.urlwatch_config.hooks)	8✔
96	logger.info(f'Imported hooks module from {self.urlwatch_config.hooks}')	8✔
97
98	def load_jobs(self) -> None:	8✔
99	"""Load jobs from the file into self.jobs.
100
101	:raises SystemExit: If job is not found, setting argument to 1.
102	"""
103	if self.urlwatch_config.jobs.is_file():	8!
104	jobs = self.jobs_storage.load_secure()	8✔
105	else:
UNCOV 106	print(f'Jobs file not found: {self.urlwatch_config.jobs}')	×
107	raise SystemExit(1)	×
108
109	self.jobs = jobs	8✔
110
111	def get_new_release_version(self, timeout: Optional[Union[float, Tuple[float, float]]] = None) -> Union[str, bool]:	8✔
112	"""Check PyPi to see if we're running the latest version. Memoized.
113
114	:returns: Empty string if no higher version is available, otherwise the new version number.
115	"""
116	if self._latest_release is not None:	8!
UNCOV 117	return self._latest_release	×
118
119	self._latest_release = get_new_version_number(timeout)	8✔
120	return self._latest_release	8✔
121
122	def run_jobs(self) -> None:	8✔
123	"""Run all jobs."""
124	run_jobs(self)	8✔
125
126	def close(self) -> None:	8✔
127	"""Finalizer. Create reports ands close snapshots database."""
128	self.report.finish(jobs_file=self.jobs_storage.filename)	8✔
129	self.cache_storage.close()	8✔

mborsetti / webchanges / 2454619689

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