Introduction of PyFunceble v4.0.2 (Blue Duckling: Grandiflora)!

Fixed:
  * Configuration Upgrade issue.
  * DNS Resolver: issue when an incorrect URL (not sub-domain) is provided.
  * URL Testing: History may be empty - due to weird server configuration.

New:
  * Introduction of the `fc2.com` SPECIAL rule.

Improvement:
  * Better handling of `ascii` default encoding - around the CLI.

Maintenance:
  * Update outdated URLs.
  * Fix typo.
  * Introduction of an internal way to profile memory usage.
  * DRY around requester and extra rules.
  * Fix failing tests.

Contributors:
  * @DandelionSprout
  * @mitchellkrogza
  * @spirillen
  * @T145
  * @ZeroDot1

Author: funilrys

PyFunceble/checker/availability/base.py

@@ -50,6 +50,8 @@
     limitations under the License.
 """
 
+# pylint: disable=too-many-lines
+
 import multiprocessing
 from datetime import datetime
 from typing import Dict, List, Optional
@@ -60,6 +62,7 @@
 import PyFunceble.facility
 import PyFunceble.factory
 import PyFunceble.storage
+from PyFunceble.checker.availability.extra_rules import ExtraRulesHandler
 from PyFunceble.checker.availability.params import AvailabilityCheckerParams
 from PyFunceble.checker.availability.status import AvailabilityCheckerStatus
 from PyFunceble.checker.base import CheckerBase
@@ -123,6 +126,7 @@ class AvailabilityCheckerBase(CheckerBase):
     domain_syntax_checker: Optional[DomainSyntaxChecker] = None
     ip_syntax_checker: Optional[IPSyntaxChecker] = None
     url_syntax_checker: Optional[URLSyntaxChecker] = None
+    extra_rules_handler: Optional[ExtraRulesHandler] = None
 
     _use_extra_rules: bool = False
     _use_whois_lookup: bool = False
@@ -148,15 +152,17 @@ def __init__(
         do_syntax_check_first: Optional[bool] = None,
         db_session: Optional[Session] = None,
         use_whois_db: Optional[bool] = None,
+        use_collection: Optional[bool] = None,
     ) -> None:
-        self.dns_query_tool = DNSQueryTool().guess_all_settings()
+        self.dns_query_tool = DNSQueryTool()
         self.whois_query_tool = WhoisQueryTool()
         self.addressinfo_query_tool = AddressInfo()
         self.hostbyaddr_query_tool = HostByAddrInfo()
         self.http_status_code_query_tool = HTTPStatusCode()
         self.domain_syntax_checker = DomainSyntaxChecker()
         self.ip_syntax_checker = IPSyntaxChecker()
         self.url_syntax_checker = URLSyntaxChecker()
+        self.extra_rules_handler = ExtraRulesHandler()
         self.db_session = db_session
 
         self.params = AvailabilityCheckerParams()
@@ -202,7 +208,10 @@ def __init__(
             self.guess_and_set_use_whois_db()
 
         super().__init__(
-            subject, do_syntax_check_first=do_syntax_check_first, db_session=db_session
+            subject,
+            do_syntax_check_first=do_syntax_check_first,
+            db_session=db_session,
+            use_collection=use_collection,
         )
 
     @property
@@ -940,6 +949,48 @@ def try_to_query_status_from_reputation(self) -> "AvailabilityCheckerBase":
 
         raise NotImplementedError()
 
+    def try_to_query_status_from_collection(self) -> "AvailabilityCheckerBase":
+        """
+        Tries to get and set the status from the Collection API.
+        """
+
+        PyFunceble.facility.Logger.info(
+            "Started to try to query the status of %r from: Collection Lookup",
+            self.status.idna_subject,
+        )
+
+        data = self.collection_query_tool.pull(self.idna_subject)
+
+        if data and "status" in data:
+            if (
+                self.collection_query_tool.preferred_status_origin == "frequent"
+                and data["status"]["availability"]["frequent"]
+            ):
+                self.status.status = data["status"]["availability"]["frequent"]
+                self.status.status_source = "COLLECTION"
+            elif (
+                self.collection_query_tool.preferred_status_origin == "latest"
+                and data["status"]["availability"]["latest"]
+            ):
+                self.status.status = data["status"]["availability"]["latest"]["status"]
+                self.status.status_source = "COLLECTION"
+            elif (
+                self.collection_query_tool.preferred_status_origin == "recommended"
+                and data["status"]["availability"]["recommended"]
+            ):
+                self.status.status = data["status"]["availability"]["recommended"]
+                self.status.status_source = "COLLECTION"
+
+            PyFunceble.facility.Logger.info(
+                "Could define the status of %r from: Collection Lookup",
+                self.status.idna_subject,
+            )
+
+        PyFunceble.facility.Logger.info(
+            "Finished to try to query the status of %r from: Collection Lookup",
+            self.status.idna_subject,
+        )
+
     @CheckerBase.ensure_subject_is_given
     @CheckerBase.update_status_date_after_query
     def query_status(self) -> "AvailabilityCheckerBase":

PyFunceble/checker/availability/domain.py

@@ -55,7 +55,6 @@
 import PyFunceble.factory
 import PyFunceble.storage
 from PyFunceble.checker.availability.base import AvailabilityCheckerBase
-from PyFunceble.checker.availability.extra_rules import ExtraRulesHandler
 from PyFunceble.checker.reputation.domain import DomainReputationChecker
 
 
@@ -131,6 +130,9 @@ def query_status(
 
         status_post_syntax_checker = None
 
+        if not self.status.status and self.use_collection:
+            self.try_to_query_status_from_collection()
+
         if not self.status.status and self.do_syntax_check_first:
             self.try_to_query_status_from_syntax_lookup()
 
@@ -176,7 +178,7 @@ def query_status(
             )
 
         if self.use_extra_rules:
-            ExtraRulesHandler(self.status).start()
+            self.extra_rules_handler.set_status(self.status).start()
 
         return self
 

PyFunceble/checker/availability/extra_rules.py

@@ -79,7 +79,7 @@ class ExtraRulesHandler:
 
     http_codes_dataset: Optional[Box] = None
 
-    def __init__(self, status: Optional[AvailabilityCheckerStatus]) -> None:
+    def __init__(self, status: Optional[AvailabilityCheckerStatus] = None) -> None:
         self.regex_active2inactive = {
             r"\.000webhostapp\.com": [
                 (self.__switch_to_down_if, 410),
@@ -88,6 +88,7 @@ def __init__(self, status: Optional[AvailabilityCheckerStatus]) -> None:
             r"\.angelfire\.com$": [(self.__switch_to_down_if, 404)],
             r"\.blogspot\.": [self.__handle_blogspot],
             r"\.canalblog\.com$": [(self.__switch_to_down_if, 404)],
+            r"\.fc2\.com$": [self.__handle_fc2_dot_com],
             r"\.github\.io$": [(self.__switch_to_down_if, 404)],
             r"\.godaddysites\.com$": [(self.__switch_to_down_if, 404)],
             r"\.hpg.com.br$": [(self.__switch_to_down_if, 404)],
@@ -294,6 +295,26 @@ def __handle_wordpress_dot_com(self) -> "ExtraRulesHandler":
 
         return self
 
+    def __handle_fc2_dot_com(self) -> "ExtraRulesHandler":
+        """
+        Handles the :code:`fc2.com` case.
+
+        .. warning::
+            This method assume that we know that we are handling a fc2 domain.
+        """
+
+        if self.status.subject.startswith("http:"):
+            url = self.status.subject
+        else:
+            url = f"http://{self.status.subject}:80"
+
+        req = PyFunceble.factory.Requester.get(url, allow_redirects=False)
+
+        if "Location" in req.headers and "error.fc2.com" in req.headers["Location"]:
+            self.__switch_to_down()
+
+        return self
+
     def __handle_active2inactive(self) -> "ExtraRulesHandler":
         """
         Handles the status deescalation.

PyFunceble/checker/availability/ip.py

@@ -55,7 +55,6 @@
 import PyFunceble.factory
 import PyFunceble.storage
 from PyFunceble.checker.availability.base import AvailabilityCheckerBase
-from PyFunceble.checker.availability.extra_rules import ExtraRulesHandler
 from PyFunceble.checker.reputation.ip import IPReputationChecker
 
 
@@ -171,7 +170,7 @@ def query_status(
             )
 
         if self.use_extra_rules:
-            ExtraRulesHandler(self.status).start()
+            self.extra_rules_handler.set_status(self.status).start()
 
         return self
 

PyFunceble/checker/base.py

@@ -57,8 +57,11 @@
 import domain2idna
 from sqlalchemy.orm import Session
 
+import PyFunceble.facility
+import PyFunceble.storage
 from PyFunceble.checker.params_base import CheckerParamsBase
 from PyFunceble.checker.status_base import CheckerStatusBase
+from PyFunceble.query.collection import CollectionQueryTool
 
 
 class CheckerBase:
@@ -75,13 +78,16 @@ class CheckerBase:
     """
 
     STD_DO_SYNTAX_CHECK_FIRST: bool = False
+    STD_USE_COLLECTION: bool = False
 
     _do_syntax_check_first: bool = False
+    _use_collection: bool = False
 
     _subject: Optional[str] = None
     _idna_subject: Optional[str] = None
 
     db_session: Optional[Session] = None
+    collection_query_tool: Optional[CollectionQueryTool] = None
 
     status: Optional[CheckerStatusBase] = None
     params: Optional[CheckerParamsBase] = None
@@ -92,7 +98,10 @@ def __init__(
         *,
         do_syntax_check_first: Optional[bool] = None,
         db_session: Optional[Session] = None,
+        use_collection: Optional[bool] = None,
     ) -> None:
+        self.collection_query_tool = CollectionQueryTool()
+
         if self.params is None:
             self.params = CheckerParamsBase()
 
@@ -107,6 +116,11 @@ def __init__(
         else:
             self.do_syntax_check_first = self.STD_DO_SYNTAX_CHECK_FIRST
 
+        if use_collection is not None:
+            self.use_collection = use_collection
+        else:
+            self.guess_and_set_use_collection()
+
         self.db_session = db_session
 
     def propagate_subject(func):  # pylint: disable=no-self-argument
@@ -301,6 +315,56 @@ def set_do_syntax_check_first(self, value: bool) -> "CheckerBase":
 
         return self
 
+    @property
+    def use_collection(self) -> bool:
+        """
+        Provides the current value of the :code:`_use_collection` attribute.
+        """
+
+        return self._use_collection
+
+    @use_collection.setter
+    def use_collection(self, value: bool) -> None:
+        """
+        Sets the value which authorizes the usage of the Collection.
+
+        :param value:
+            The value to set.
+
+        :param TypeError:
+            When the given :code:`value` is not a :py:class:`bool`.
+        """
+
+        if not isinstance(value, bool):
+            raise TypeError(f"<value> should be {bool}, {type(value)} given.")
+
+        self._use_collection = self.params.use_collection = value
+
+    def set_use_collection(self, value: bool) -> "CheckerBase":
+        """
+        Sets the value which authorizes the usage of the Collection.
+
+        :param value:
+            The value to set.
+        """
+
+        self.use_collection = value
+
+        return self
+
+    def guess_and_set_use_collection(self) -> "CheckerBase":
+        """
+        Try to guess and set the value of the :code:`use_collection` attribute.
+        """
+
+        if PyFunceble.facility.ConfigLoader.is_already_loaded():
+            if isinstance(PyFunceble.storage.CONFIGURATION.lookup.collection, bool):
+                self.use_collection = PyFunceble.storage.CONFIGURATION.lookup.collection
+            else:
+                self.use_collection = self.STD_USE_COLLECTION
+        else:
+            self.use_collection = self.STD_USE_COLLECTION
+
     def subject_propagator(self) -> "CheckerBase":
         """
         Propagate the currently set subject.

PyFunceble/checker/params_base.py

@@ -64,6 +64,7 @@ class CheckerParamsBase:
     """
 
     do_syntax_check_first: Optional[bool] = None
+    use_collection: Optional[bool] = None
 
     def to_dict(self) -> dict:
         """

PyFunceble/checker/reputation/base.py

@@ -93,8 +93,9 @@ def __init__(
         subject: Optional[str] = None,
         do_syntax_check_first: Optional[bool] = None,
         db_session: Optional[Session] = None,
+        use_collection: Optional[bool] = None,
     ) -> None:
-        self.dns_query_tool = DNSQueryTool().guess_all_settings()
+        self.dns_query_tool = DNSQueryTool()
         self.ipv4_reputation_query_tool = IPV4ReputationDataset()
         self.domain_syntax_checker = DomainSyntaxChecker()
         self.ip_syntax_checker = IPSyntaxChecker()
@@ -107,7 +108,10 @@ def __init__(
         self.status.dns_lookup_record = self.dns_query_tool.lookup_record
 
         super().__init__(
-            subject, do_syntax_check_first=do_syntax_check_first, db_session=db_session
+            subject,
+            do_syntax_check_first=do_syntax_check_first,
+            db_session=db_session,
+            use_collection=use_collection,
         )
 
     @staticmethod
@@ -244,6 +248,48 @@ def try_to_query_status_from_syntax_lookup(self) -> "ReputationCheckerBase":
 
         return self
 
+    def try_to_query_status_from_collection(self) -> "ReputationCheckerBase":
+        """
+        Tries to get and set the status from the Collection API.
+        """
+
+        PyFunceble.facility.Logger.info(
+            "Started to try to query the status of %r from: Collection Lookup",
+            self.status.idna_subject,
+        )
+
+        data = self.collection_query_tool[self.idna_subject]
+
+        if data and "status" in data:
+            if (
+                self.collection_query_tool.preferred_status_origin == "frequent"
+                and data["status"]["reputation"]["frequent"]
+            ):
+                self.status.status = data["status"]["reputation"]["frequent"]
+                self.status.status_source = "COLLECTION"
+            elif (
+                self.collection_query_tool.preferred_status_origin == "latest"
+                and data["status"]["reputation"]["latest"]
+            ):
+                self.status.status = data["status"]["reputation"]["latest"]["status"]
+                self.status.status_source = "COLLECTION"
+            elif (
+                self.collection_query_tool.preferred_status_origin == "recommended"
+                and data["status"]["reputation"]["recommended"]
+            ):
+                self.status.status = data["status"]["reputation"]["recommended"]
+                self.status.status_source = "COLLECTION"
+
+            PyFunceble.facility.Logger.info(
+                "Could define the status of %r from: Collection Lookup",
+                self.status.idna_subject,
+            )
+
+        PyFunceble.facility.Logger.info(
+            "Finished to try to query the status of %r from: Collection Lookup",
+            self.status.idna_subject,
+        )
+
     @CheckerBase.ensure_subject_is_given
     @CheckerBase.update_status_date_after_query
     def query_status(self) -> "ReputationCheckerBase":
@@ -253,6 +299,11 @@ def query_status(self) -> "ReputationCheckerBase":
 
         status_post_syntax_checker = None
 
+        if (
+            not self.status.status and self.use_collection
+        ):  # pragma: no cover ## Underlying tested
+            self.try_to_query_status_from_collection()
+
         if not self.status.status and self.do_syntax_check_first:
             self.try_to_query_status_from_syntax_lookup()