10980404551

Committed 22 Sep 2024 10:29AM UTC coverage: 94.595% (-0.1%) from 94.723%

Build # 10980404551

Build Type

Pull #380

github

Committed by

funilrys

Commit Message

Introduction of a way to provide a configuration file from the CLI.

This patch fixes #377.

Contributors:
  * @spirillen

Pull Request Pull Request #380: Introduction of a way to provide a configuration file from the CLI.

Run Details

17 of 34 new or added lines in 2 files covered. (50.0%)

25 existing lines in 3 files now uncovered.

11341 of 11989 relevant lines covered (94.6%)

14.15 hits per line

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

89.63

/PyFunceble/config/loader.py

"""
The tool to check the availability or syntax of domain, IP or URL.

::


    ██████╗ ██╗   ██╗███████╗██╗   ██╗███╗   ██╗ ██████╗███████╗██████╗ ██╗     ███████╗
    ██╔══██╗╚██╗ ██╔╝██╔════╝██║   ██║████╗  ██║██╔════╝██╔════╝██╔══██╗██║     ██╔════╝
    ██████╔╝ ╚████╔╝ █████╗  ██║   ██║██╔██╗ ██║██║     █████╗  ██████╔╝██║     █████╗
    ██╔═══╝   ╚██╔╝  ██╔══╝  ██║   ██║██║╚██╗██║██║     ██╔══╝  ██╔══██╗██║     ██╔══╝
    ██║        ██║   ██║     ╚██████╔╝██║ ╚████║╚██████╗███████╗██████╔╝███████╗███████╗
    ╚═╝        ╚═╝   ╚═╝      ╚═════╝ ╚═╝  ╚═══╝ ╚═════╝╚══════╝╚═════╝ ╚══════╝╚══════╝

Provides the configuration loader.

Author:
    Nissar Chababy, @funilrys, contactTATAfunilrysTODTODcom

Special thanks:
    https://pyfunceble.github.io/#/special-thanks

Contributors:
    https://pyfunceble.github.io/#/contributors

Project link:
    https://github.com/funilrys/PyFunceble

Project documentation:
    https://docs.pyfunceble.com

Project homepage:
    https://pyfunceble.github.io/

License:
::


    Copyright 2017, 2018, 2019, 2020, 2022, 2023, 2024 Nissar Chababy

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        https://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
"""

import functools
import os
from typing import Any, Optional

try:
    import importlib.resources as package_resources
except ImportError:  # pragma: no cover ## Retro compatibility
    import importlib_resources as package_resources

from box import Box
from yaml.error import MarkedYAMLError

import PyFunceble.cli.storage
import PyFunceble.storage
from PyFunceble.config.compare import ConfigComparison
from PyFunceble.downloader.iana import IANADownloader
from PyFunceble.downloader.public_suffix import PublicSuffixDownloader
from PyFunceble.downloader.user_agents import UserAgentsDownloader
from PyFunceble.helpers.dict import DictHelper
from PyFunceble.helpers.download import DownloadHelper
from PyFunceble.helpers.environment_variable import EnvironmentVariableHelper
from PyFunceble.helpers.file import FileHelper
from PyFunceble.helpers.merge import Merge


class ConfigLoader:
    """
    Provides the interface which loads and updates the configuration (if needed).

    :param merge_upstream:
        Authorizes the merging of the upstream configuration.

        .. note::
            If value is set to :py:class:`None` (default), we fallback to the
            :code:`PYFUNCEBLE_AUTO_CONFIGURATION` environment variable.
    """

    _path_to_config: Optional[str] = None
    _remote_config_location: Optional[str] = None
    path_to_default_config: Optional[str] = None
    path_to_overwrite_config: Optional[str] = None

    _custom_config: dict = {}
    _merge_upstream: bool = False

    file_helper: FileHelper = FileHelper()
    dict_helper: DictHelper = DictHelper()

    def __init__(self, merge_upstream: Optional[bool] = None) -> None:
        with package_resources.path(
            "PyFunceble.data.infrastructure",
            PyFunceble.storage.DISTRIBUTED_CONFIGURATION_FILENAME,
        ) as file_path:
            self.path_to_default_config = str(file_path)

        self.path_to_config = os.path.join(
            PyFunceble.storage.CONFIG_DIRECTORY,
            PyFunceble.storage.CONFIGURATION_FILENAME,
        )

        self.path_to_remote_config = None

        self.path_to_overwrite_config = os.path.join(
            PyFunceble.storage.CONFIG_DIRECTORY,
            PyFunceble.storage.CONFIGURATION_OVERWRITE_FILENAME,
        )

        if merge_upstream is not None:
            self.merge_upstream = merge_upstream
        elif EnvironmentVariableHelper("PYFUNCEBLE_AUTO_CONFIGURATION").exists():
            self.merge_upstream = True

    def __del__(self) -> None:
        self.destroy()

    def reload_config(func):  # pylint: disable=no-self-argument
        """
        Reload the configuration (if it was already loaded) after launching the
        decorated method.
        """

        @functools.wraps(func)
        def wrapper(self, *args, **kwargs):
            result = func(self, *args, **kwargs)  # pylint: disable=not-callable

            if self.is_already_loaded():
                self.start()

            return result

        return wrapper

    @staticmethod
    def conditional_switch(config: dict) -> dict:
        """
        Given the configuration that we are going to load, switches some of
        setting.

        :param config:
            The configuration we are going to load.
        """

        # pylint: disable=too-many-boolean-expressions
        if (
            "cli_testing" in config
            and "ci" in config["cli_testing"]
            and "active" in config["cli_testing"]["ci"]
            and "autocontinue" in config["cli_testing"]
            and bool(config["cli_testing"]["ci"]["active"])
            and not bool(config["cli_testing"]["autocontinue"])
        ):
            # Conditional autocontinue.
            # If we are under continuous integration, the autocontinue should be
            # activated.

            config["cli_testing"]["autocontinue"] = True

        if (
            "lookup" in config
            and "timeout" in config["lookup"]
            and config["lookup"]["timeout"]
            and config["lookup"]["timeout"] < 0
        ):
            # If timeout is set to a negative digit, switch to the default one.
            config["lookup"]["timeout"] = 5

        if (
            "cli_testing" in config
            and "testing_mode" in config["cli_testing"]
            and "platform_contribution" in config["cli_testing"]["testing_mode"]
            and config["cli_testing"]["testing_mode"]["platform_contribution"]
        ):
            # If we are under a special testing mode. We shouldn't generate
            # any files
            config["cli_testing"]["file_generation"]["no_file"] = True
            config["cli_testing"]["display_mode"]["dots"] = True
            config["cli_testing"]["autocontinue"] = False
            config["cli_testing"]["inactive_db"] = False
            config["cli_testing"]["mining"] = False
            config["cli_testing"]["local_network"] = False
            config["cli_testing"]["preload_file"] = False
            config["cli_testing"]["display_mode"]["percentage"] = False
            config["lookup"]["platform"] = False

        return config

    @staticmethod
    def is_already_loaded() -> bool:
        """
        Checks if the configuration was already loaded.
        """

        return bool(PyFunceble.storage.CONFIGURATION)

    @property
    def custom_config(self) -> dict:
        """
        Provides the current state of the :code:`_custom_config` attribute.
        """

        return self._custom_config

    @custom_config.setter
    @reload_config
    def custom_config(self, value: dict) -> None:
        """
        Sets the custom configuration to set after loading.

        Side Effect:
            Directly inject into the configuration variables if it was already
            loaded.

        :raise TypeError:
            When :code:`value` is not a :py:class:`dict`.
        """

        if not isinstance(value, dict):
            raise TypeError(f"<value> should be {dict}, {type(value)} given.")

        if not self._custom_config:
            self._custom_config = value
        else:
            self._custom_config.update(value)

    def set_custom_config(self, value: dict) -> "ConfigLoader":
        """
        Sets the custom configuration to set after loading.

        Side Effect:
            Directly inject into the configuration variables if it was already
            loaded.
        """

        self.custom_config = value

        return self

    @property
    def merge_upstream(self) -> bool:
        """
        Provides the current state of the :code:`_merge_upstream` attribute.
        """

        return self._merge_upstream

    @merge_upstream.setter
    def merge_upstream(self, value: bool) -> None:
        """
        Updates the value of :code:`_merge_upstream` attribute.

        :raise TypeError:
            When :code:`value` is not a :py:class:`bool`.
        """

        if not isinstance(value, bool):
            raise TypeError(f"<value> should be {bool}, {type(value)} given.")

        self._merge_upstream = value

    def set_merge_upstream(self, value: bool) -> "ConfigLoader":
        """
        Updates the value of :code:`_merge_upstream` attribute.
        """

        self.merge_upstream = value

        return self

    @property
    def remote_config_location(self) -> Optional[str]:
        """
        Provides the current state of the :code:`_remote_config_location` attribute.
        """

        return self._remote_config_location

    @remote_config_location.setter
    def remote_config_location(self, value: Optional[str]) -> None:
        """
        Updates the value of :code:`_remote_config_location` attribute.

        :raise TypeError:
            When :code:`value` is not a :py:class:`str`.
        """

        if value is not None and not isinstance(value, str):
            raise TypeError(f"<value> should be {str}, {type(value)} given.")

        if not value.startswith("http") and not value.startswith("https"):
            self.path_to_remote_config = os.path.realpath(value)
        else:
            self.path_to_remote_config = os.path.join(
                PyFunceble.storage.CONFIG_DIRECTORY,
                PyFunceble.storage.CONFIGURATION_REMOTE_FILENAME,
            )

        self._remote_config_location = value

    def set_remote_config_location(self, value: Optional[str]) -> "ConfigLoader":
        """
        Updates the value of :code:`_remote_config_location` attribute.
        """

        self.remote_config_location = value

        return self

    def install_missing_infrastructure_files(
        self,
    ) -> "ConfigLoader":  # pragma: no cover ## Copy method already tested
        """
        Installs the missing files (when needed).

        .. note::
            Installed if missing:
                - The configuration file.
                - The directory structure file.
        """

        if not self.is_already_loaded():
            if not self.file_helper.set_path(self.path_to_config).exists():
                self.file_helper.set_path(self.path_to_default_config).copy(
                    self.path_to_config
                )

        return self

    @classmethod
    def download_dynamic_infrastructure_files(
        cls,
    ) -> "ConfigLoader":
        """
        Downloads all the dynamicly (generated) infrastructure files.

        .. note::
            Downloaded if missing:
                - The IANA dump file.
                - The Public Suffix dump file.
        """

        ## pragma: no cover ## Underlying download methods already tested.

        if not cls.is_already_loaded():
            IANADownloader().start()
            PublicSuffixDownloader().start()
            UserAgentsDownloader().start()

    def get_config_file_content(self) -> dict:
        """
        Provides the content of the configuration file or the one already loaded.
        """

        def is_3_x_version(config: dict) -> bool:
            """
            Checks if the given configuration is an old one.

            :param config:
                The config to work with.
            """

            return config and "days_between_inactive_db_clean" in config

        def download_remote_config(src: str, dest: str = None) -> None:
            """
            Downloads the remote configuration.

            :param src:
                The source to download from.
            :param dest:
                The destination to download
            """

            if src and (src.startswith("http") or src.startswith("https")):
                if dest is None:
                    destination = os.path.join(
                        PyFunceble.storage.CONFIG_DIRECTORY,
                        os.path.basename(dest),
                    )
                else:
                    destination = dest

                DownloadHelper(src).download_text(destination=destination)

        if not self.is_already_loaded():
            self.install_missing_infrastructure_files()
            self.download_dynamic_infrastructure_files()
            download_remote_config(
                self.remote_config_location, self.path_to_remote_config
            )
            download_remote_config(self.path_to_config)

        try:
            config = self.dict_helper.from_yaml_file(self.path_to_config)
        except MarkedYAMLError:
            self.file_helper.set_path(self.path_to_default_config).copy(
                self.path_to_config
            )
            config = self.dict_helper.from_yaml_file(self.path_to_config)

        config_comparer = ConfigComparison(
            local_config=config,
            upstream_config=self.dict_helper.from_yaml_file(
                self.path_to_default_config
            ),
        )

        if (
            not config
            or not isinstance(config, dict)
            or self.merge_upstream
            or is_3_x_version(config)
            or not config_comparer.is_local_identical()
        ):  # pragma: no cover ## Testing the underlying comparison method is sufficent
            config = config_comparer.get_merged()

            self.dict_helper.set_subject(config).to_yaml_file(self.path_to_config)

        if (
            self.path_to_remote_config
            and self.file_helper.set_path(self.path_to_remote_config).exists()
        ):
            remote_data = self.dict_helper.from_yaml_file(self.path_to_remote_config)

            if isinstance(remote_data, dict):
                config = Merge(remote_data).into(config)

        if self.file_helper.set_path(self.path_to_overwrite_config).exists():
            overwrite_data = self.dict_helper.from_yaml_file(
                self.path_to_overwrite_config
            )

            if isinstance(overwrite_data, dict):
                config = Merge(overwrite_data).into(config)
        else:  # pragma: no cover  ## Just make it visible to end-user.
            self.file_helper.write("")

        return config

    def get_configured_value(self, entry: str) -> Any:
        """
        Provides the currently configured value.

        :param entry:
            An entry to check.

            multilevel should be separated with a point.

        :raise RuntimeError:
            When the configuration is not loaded yet.

        :raise ValueError:
            When the given :code:`entry` is not found.
        """

        if not self.is_already_loaded():
            raise RuntimeError("Configuration not loaded, yet.")

        if entry not in PyFunceble.storage.FLATTEN_CONFIGURATION:
            raise ValueError(f"<entry> ({entry!r}) not in loaded configuration.")

        return PyFunceble.storage.FLATTEN_CONFIGURATION[entry]

    def reload(self) -> "ConfigLoader":
        """
        Reloads the configuration.
        """

        self.destroy()
        self.start()

    def start(self) -> "ConfigLoader":
        """
        Starts the loading processIs.
        """

        config = self.get_config_file_content()

        if self.custom_config:
            config = Merge(self.custom_config).into(config)

        config = self.conditional_switch(config)

        PyFunceble.storage.CONFIGURATION = Box(
            config,
        )
        PyFunceble.storage.FLATTEN_CONFIGURATION = DictHelper(
            PyFunceble.storage.CONFIGURATION
        ).flatten()
        PyFunceble.storage.HTTP_CODES = Box(
            config["http_codes"],
        )
        if "platform" in config and config["platform"]:
            PyFunceble.storage.PLATFORM = Box(config["platform"])
        PyFunceble.storage.LINKS = Box(config["links"])

        if "proxy" in config and config["proxy"]:
            PyFunceble.storage.PROXY = Box(config["proxy"])

        return self

    def destroy(self) -> "ConfigLoader":
        """
        Destroys everything loaded.
        """

        try:
            PyFunceble.storage.CONFIGURATION = Box(
                {},
            )
            PyFunceble.storage.FLATTEN_CONFIGURATION = {}
            PyFunceble.storage.HTTP_CODES = Box({})
            PyFunceble.storage.PLATFORM = Box({})
            PyFunceble.storage.LINKS = Box({})
            PyFunceble.storage.PROXY = Box({})
        except (AttributeError, TypeError):  # pragma: no cover ## Safety.
            pass

        # This is not a mistake.
        self._custom_config = {}

        return self

1	"""
2	The tool to check the availability or syntax of domain, IP or URL.
3
4	::
5
6
7	██████╗ ██╗ ██╗███████╗██╗ ██╗███╗ ██╗ ██████╗███████╗██████╗ ██╗ ███████╗
8	██╔══██╗╚██╗ ██╔╝██╔════╝██║ ██║████╗ ██║██╔════╝██╔════╝██╔══██╗██║ ██╔════╝
9	██████╔╝ ╚████╔╝ █████╗ ██║ ██║██╔██╗ ██║██║ █████╗ ██████╔╝██║ █████╗
10	██╔═══╝ ╚██╔╝ ██╔══╝ ██║ ██║██║╚██╗██║██║ ██╔══╝ ██╔══██╗██║ ██╔══╝
11	██║ ██║ ██║ ╚██████╔╝██║ ╚████║╚██████╗███████╗██████╔╝███████╗███████╗
12	╚═╝ ╚═╝ ╚═╝ ╚═════╝ ╚═╝ ╚═══╝ ╚═════╝╚══════╝╚═════╝ ╚══════╝╚══════╝
13
14	Provides the configuration loader.
15
16	Author:
17	Nissar Chababy, @funilrys, contactTATAfunilrysTODTODcom
18
19	Special thanks:
20	https://pyfunceble.github.io/#/special-thanks
21
22	Contributors:
23	https://pyfunceble.github.io/#/contributors
24
25	Project link:
26	https://github.com/funilrys/PyFunceble
27
28	Project documentation:
29	https://docs.pyfunceble.com
30
31	Project homepage:
32	https://pyfunceble.github.io/
33
34	License:
35	::
36
37
38	Copyright 2017, 2018, 2019, 2020, 2022, 2023, 2024 Nissar Chababy
39
40	Licensed under the Apache License, Version 2.0 (the "License");
41	you may not use this file except in compliance with the License.
42	You may obtain a copy of the License at
43
44	https://www.apache.org/licenses/LICENSE-2.0
45
46	Unless required by applicable law or agreed to in writing, software
47	distributed under the License is distributed on an "AS IS" BASIS,
48	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
49	See the License for the specific language governing permissions and
50	limitations under the License.
51	"""
52
53	import functools	15✔
54	import os	15✔
55	from typing import Any, Optional	15✔
56
57	try:	15✔
58	import importlib.resources as package_resources	15✔
59	except ImportError: # pragma: no cover ## Retro compatibility
60	import importlib_resources as package_resources
61
62	from box import Box	15✔
63	from yaml.error import MarkedYAMLError	15✔
64
65	import PyFunceble.cli.storage	15✔
66	import PyFunceble.storage	15✔
67	from PyFunceble.config.compare import ConfigComparison	15✔
68	from PyFunceble.downloader.iana import IANADownloader	15✔
69	from PyFunceble.downloader.public_suffix import PublicSuffixDownloader	15✔
70	from PyFunceble.downloader.user_agents import UserAgentsDownloader	15✔
71	from PyFunceble.helpers.dict import DictHelper	15✔
72	from PyFunceble.helpers.download import DownloadHelper	15✔
73	from PyFunceble.helpers.environment_variable import EnvironmentVariableHelper	15✔
74	from PyFunceble.helpers.file import FileHelper	15✔
75	from PyFunceble.helpers.merge import Merge	15✔
76
77
78	class ConfigLoader:	15✔
79	"""
80	Provides the interface which loads and updates the configuration (if needed).
81
82	:param merge_upstream:
83	Authorizes the merging of the upstream configuration.
84
85	.. note::
86	If value is set to :py:class:`None` (default), we fallback to the
87	:code:`PYFUNCEBLE_AUTO_CONFIGURATION` environment variable.
88	"""
89
90	_path_to_config: Optional[str] = None	15✔
91	_remote_config_location: Optional[str] = None	15✔
92	path_to_default_config: Optional[str] = None	15✔
93	path_to_overwrite_config: Optional[str] = None	15✔
94
95	_custom_config: dict = {}	15✔
96	_merge_upstream: bool = False	15✔
97
98	file_helper: FileHelper = FileHelper()	15✔
99	dict_helper: DictHelper = DictHelper()	15✔
100
101	def __init__(self, merge_upstream: Optional[bool] = None) -> None:
102	with package_resources.path(
103	"PyFunceble.data.infrastructure",
104	PyFunceble.storage.DISTRIBUTED_CONFIGURATION_FILENAME,
105	) as file_path:
106	self.path_to_default_config = str(file_path)
107
108	self.path_to_config = os.path.join(
109	PyFunceble.storage.CONFIG_DIRECTORY,
110	PyFunceble.storage.CONFIGURATION_FILENAME,
111	)
112
113	self.path_to_remote_config = None
114
115	self.path_to_overwrite_config = os.path.join(
116	PyFunceble.storage.CONFIG_DIRECTORY,
117	PyFunceble.storage.CONFIGURATION_OVERWRITE_FILENAME,
118	)
119
120	if merge_upstream is not None:
121	self.merge_upstream = merge_upstream
122	elif EnvironmentVariableHelper("PYFUNCEBLE_AUTO_CONFIGURATION").exists():
123	self.merge_upstream = True
124
125	def __del__(self) -> None:	15✔
126	self.destroy()	15✔
127
128	def reload_config(func): # pylint: disable=no-self-argument	15✔
129	"""
130	Reload the configuration (if it was already loaded) after launching the
131	decorated method.
132	"""
133
134	@functools.wraps(func)	15✔
135	def wrapper(self, args, *kwargs):	15✔
136	result = func(self, args, *kwargs) # pylint: disable=not-callable	15✔
137
138	if self.is_already_loaded():	15✔
139	self.start()	15✔
140
141	return result	15✔
142
143	return wrapper	15✔
144
145	@staticmethod	15✔
146	def conditional_switch(config: dict) -> dict:	15✔
147	"""
148	Given the configuration that we are going to load, switches some of
149	setting.
150
151	:param config:
152	The configuration we are going to load.
153	"""
154
155	# pylint: disable=too-many-boolean-expressions
156	if (	15✔
157	"cli_testing" in config
158	and "ci" in config["cli_testing"]
159	and "active" in config["cli_testing"]["ci"]
160	and "autocontinue" in config["cli_testing"]
161	and bool(config["cli_testing"]["ci"]["active"])
162	and not bool(config["cli_testing"]["autocontinue"])
163	):
164	# Conditional autocontinue.
165	# If we are under continuous integration, the autocontinue should be
166	# activated.
167
168	config["cli_testing"]["autocontinue"] = True	15✔
169
170	if (	15✔
171	"lookup" in config
172	and "timeout" in config["lookup"]
173	and config["lookup"]["timeout"]
174	and config["lookup"]["timeout"] < 0
175	):
176	# If timeout is set to a negative digit, switch to the default one.
177	config["lookup"]["timeout"] = 5	15✔
178
179	if (	15✔
180	"cli_testing" in config
181	and "testing_mode" in config["cli_testing"]
182	and "platform_contribution" in config["cli_testing"]["testing_mode"]
183	and config["cli_testing"]["testing_mode"]["platform_contribution"]
184	):
185	# If we are under a special testing mode. We shouldn't generate
186	# any files
187	config["cli_testing"]["file_generation"]["no_file"] = True	15✔
188	config["cli_testing"]["display_mode"]["dots"] = True	15✔
189	config["cli_testing"]["autocontinue"] = False	15✔
190	config["cli_testing"]["inactive_db"] = False	15✔
191	config["cli_testing"]["mining"] = False	15✔
192	config["cli_testing"]["local_network"] = False	15✔
193	config["cli_testing"]["preload_file"] = False	15✔
194	config["cli_testing"]["display_mode"]["percentage"] = False	15✔
195	config["lookup"]["platform"] = False	15✔
196
197	return config	15✔
198
199	@staticmethod	15✔
200	def is_already_loaded() -> bool:	15✔
201	"""
202	Checks if the configuration was already loaded.
203	"""
204
205	return bool(PyFunceble.storage.CONFIGURATION)	15✔
206
207	@property	15✔
208	def custom_config(self) -> dict:	15✔
209	"""
210	Provides the current state of the :code:`_custom_config` attribute.
211	"""
212
213	return self._custom_config	15✔
214
215	@custom_config.setter	15✔
216	@reload_config	15✔
217	def custom_config(self, value: dict) -> None:	15✔
218	"""
219	Sets the custom configuration to set after loading.
220
221	Side Effect:
222	Directly inject into the configuration variables if it was already
223	loaded.
224
225	:raise TypeError:
226	When :code:`value` is not a :py:class:`dict`.
227	"""
228
229	if not isinstance(value, dict):	15✔
230	raise TypeError(f"<value> should be {dict}, {type(value)} given.")	15✔
231
232	if not self._custom_config:	15✔
233	self._custom_config = value	15✔
234	else:
235	self._custom_config.update(value)	15✔
236
237	def set_custom_config(self, value: dict) -> "ConfigLoader":	15✔
238	"""
239	Sets the custom configuration to set after loading.
240
241	Side Effect:
242	Directly inject into the configuration variables if it was already
243	loaded.
244	"""
245
246	self.custom_config = value	15✔
247
248	return self	15✔
249
250	@property	15✔
251	def merge_upstream(self) -> bool:	15✔
252	"""
253	Provides the current state of the :code:`_merge_upstream` attribute.
254	"""
255
256	return self._merge_upstream	15✔
257
258	@merge_upstream.setter	15✔
259	def merge_upstream(self, value: bool) -> None:	15✔
260	"""
261	Updates the value of :code:`_merge_upstream` attribute.
262
263	:raise TypeError:
264	When :code:`value` is not a :py:class:`bool`.
265	"""
266
267	if not isinstance(value, bool):	15✔
268	raise TypeError(f"<value> should be {bool}, {type(value)} given.")	15✔
269
270	self._merge_upstream = value	15✔
271
272	def set_merge_upstream(self, value: bool) -> "ConfigLoader":	15✔
273	"""
274	Updates the value of :code:`_merge_upstream` attribute.
275	"""
276
277	self.merge_upstream = value	15✔
278
279	return self	15✔
280
281	@property	15✔
282	def remote_config_location(self) -> Optional[str]:	15✔
283	"""
284	Provides the current state of the :code:`_remote_config_location` attribute.
285	"""
286
287	return self._remote_config_location	15✔
288
289	@remote_config_location.setter	15✔
290	def remote_config_location(self, value: Optional[str]) -> None:	15✔
291	"""
292	Updates the value of :code:`_remote_config_location` attribute.
293
294	:raise TypeError:
295	When :code:`value` is not a :py:class:`str`.
296	"""
297
NEW UNCOV 298	if value is not None and not isinstance(value, str):	×
NEW UNCOV 299	raise TypeError(f"<value> should be {str}, {type(value)} given.")	×
300
NEW UNCOV 301	if not value.startswith("http") and not value.startswith("https"):	×
NEW UNCOV 302	self.path_to_remote_config = os.path.realpath(value)	×
303	else:
NEW UNCOV 304	self.path_to_remote_config = os.path.join(	×
305	PyFunceble.storage.CONFIG_DIRECTORY,
306	PyFunceble.storage.CONFIGURATION_REMOTE_FILENAME,
307	)
308
NEW UNCOV 309	self._remote_config_location = value	×
310
311	def set_remote_config_location(self, value: Optional[str]) -> "ConfigLoader":	15✔
312	"""
313	Updates the value of :code:`_remote_config_location` attribute.
314	"""
315
NEW UNCOV 316	self.remote_config_location = value	×
317
NEW UNCOV 318	return self	×
319
320	def install_missing_infrastructure_files(
321	self,
322	) -> "ConfigLoader": # pragma: no cover ## Copy method already tested
323	"""
324	Installs the missing files (when needed).
325
326	.. note::
327	Installed if missing:
328	- The configuration file.
329	- The directory structure file.
330	"""
331
332	if not self.is_already_loaded():
333	if not self.file_helper.set_path(self.path_to_config).exists():
334	self.file_helper.set_path(self.path_to_default_config).copy(
335	self.path_to_config
336	)
337
338	return self
339
340	@classmethod	15✔
341	def download_dynamic_infrastructure_files(	15✔
342	cls,
343	) -> "ConfigLoader":
344	"""
345	Downloads all the dynamicly (generated) infrastructure files.
346
347	.. note::
348	Downloaded if missing:
349	- The IANA dump file.
350	- The Public Suffix dump file.
351	"""
352
353	## pragma: no cover ## Underlying download methods already tested.
354
355	if not cls.is_already_loaded():	15✔
356	IANADownloader().start()	15✔
357	PublicSuffixDownloader().start()	15✔
358	UserAgentsDownloader().start()	15✔
359
360	def get_config_file_content(self) -> dict:	15✔
361	"""
362	Provides the content of the configuration file or the one already loaded.
363	"""
364
365	def is_3_x_version(config: dict) -> bool:	15✔
366	"""
367	Checks if the given configuration is an old one.
368
369	:param config:
370	The config to work with.
371	"""
372
373	return config and "days_between_inactive_db_clean" in config	15✔
374
375	def download_remote_config(src: str, dest: str = None) -> None:	15✔
376	"""
377	Downloads the remote configuration.
378
379	:param src:
380	The source to download from.
381	:param dest:
382	The destination to download
383	"""
384
385	if src and (src.startswith("http") or src.startswith("https")):	15✔
NEW UNCOV 386	if dest is None:	×
NEW UNCOV 387	destination = os.path.join(	×
388	PyFunceble.storage.CONFIG_DIRECTORY,
389	os.path.basename(dest),
390	)
391	else:
NEW UNCOV 392	destination = dest	×
393
NEW UNCOV 394	DownloadHelper(src).download_text(destination=destination)	×
395
396	if not self.is_already_loaded():	15✔
397	self.install_missing_infrastructure_files()	15✔
398	self.download_dynamic_infrastructure_files()	15✔
399	download_remote_config(	15✔
400	self.remote_config_location, self.path_to_remote_config
401	)
402	download_remote_config(self.path_to_config)	15✔
403
404	try:	15✔
405	config = self.dict_helper.from_yaml_file(self.path_to_config)	15✔
406	except MarkedYAMLError:	15✔
407	self.file_helper.set_path(self.path_to_default_config).copy(	15✔
408	self.path_to_config
409	)
410	config = self.dict_helper.from_yaml_file(self.path_to_config)	15✔
411
412	config_comparer = ConfigComparison(	15✔
413	local_config=config,
414	upstream_config=self.dict_helper.from_yaml_file(
415	self.path_to_default_config
416	),
417	)
418
419	if (
420	not config
421	or not isinstance(config, dict)
422	or self.merge_upstream
423	or is_3_x_version(config)
424	or not config_comparer.is_local_identical()
425	): # pragma: no cover ## Testing the underlying comparison method is sufficent
426	config = config_comparer.get_merged()
427
428	self.dict_helper.set_subject(config).to_yaml_file(self.path_to_config)
429
430	if (	15✔
431	self.path_to_remote_config
432	and self.file_helper.set_path(self.path_to_remote_config).exists()
433	):
NEW 434	remote_data = self.dict_helper.from_yaml_file(self.path_to_remote_config)	×
435
NEW 436	if isinstance(remote_data, dict):	×
NEW 437	config = Merge(remote_data).into(config)	×
438
439	if self.file_helper.set_path(self.path_to_overwrite_config).exists():	15✔
440	overwrite_data = self.dict_helper.from_yaml_file(	15✔
441	self.path_to_overwrite_config
442	)
443
444	if isinstance(overwrite_data, dict):	15✔
445	config = Merge(overwrite_data).into(config)	15✔
446	else: # pragma: no cover ## Just make it visible to end-user.
447	self.file_helper.write("")
448
449	return config	15✔
450
451	def get_configured_value(self, entry: str) -> Any:	15✔
452	"""
453	Provides the currently configured value.
454
455	:param entry:
456	An entry to check.
457
458	multilevel should be separated with a point.
459
460	:raise RuntimeError:
461	When the configuration is not loaded yet.
462
463	:raise ValueError:
464	When the given :code:`entry` is not found.
465	"""
466
467	if not self.is_already_loaded():	15✔
468	raise RuntimeError("Configuration not loaded, yet.")	15✔
469
470	if entry not in PyFunceble.storage.FLATTEN_CONFIGURATION:	15✔
471	raise ValueError(f"<entry> ({entry!r}) not in loaded configuration.")	15✔
472
473	return PyFunceble.storage.FLATTEN_CONFIGURATION[entry]	15✔
474
475	def reload(self) -> "ConfigLoader":	15✔
476	"""
477	Reloads the configuration.
478	"""
479
NEW 480	self.destroy()	×
NEW 481	self.start()	×
482
483	def start(self) -> "ConfigLoader":	15✔
484	"""
485	Starts the loading processIs.
486	"""
487
488	config = self.get_config_file_content()	15✔
489
490	if self.custom_config:	15✔
491	config = Merge(self.custom_config).into(config)	15✔
492
493	config = self.conditional_switch(config)	15✔
494
495	PyFunceble.storage.CONFIGURATION = Box(	15✔
496	config,
497	)
498	PyFunceble.storage.FLATTEN_CONFIGURATION = DictHelper(	15✔
499	PyFunceble.storage.CONFIGURATION
500	).flatten()
501	PyFunceble.storage.HTTP_CODES = Box(	15✔
502	config["http_codes"],
503	)
504	if "platform" in config and config["platform"]:	15✔
505	PyFunceble.storage.PLATFORM = Box(config["platform"])	15✔
506	PyFunceble.storage.LINKS = Box(config["links"])	15✔
507
508	if "proxy" in config and config["proxy"]:	15✔
509	PyFunceble.storage.PROXY = Box(config["proxy"])	15✔
510
511	return self	15✔
512
513	def destroy(self) -> "ConfigLoader":	15✔
514	"""
515	Destroys everything loaded.
516	"""
517
518	try:	15✔
519	PyFunceble.storage.CONFIGURATION = Box(	15✔
520	{},
521	)
522	PyFunceble.storage.FLATTEN_CONFIGURATION = {}	15✔
523	PyFunceble.storage.HTTP_CODES = Box({})	15✔
524	PyFunceble.storage.PLATFORM = Box({})	15✔
525	PyFunceble.storage.LINKS = Box({})	15✔
526	PyFunceble.storage.PROXY = Box({})	15✔
527	except (AttributeError, TypeError): # pragma: no cover ## Safety.
528	pass
529
530	# This is not a mistake.
531	self._custom_config = {}	15✔
532
533	return self	15✔

funilrys / PyFunceble / 10980404551

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