Introduction of PyFunceble v3.3.2 (Teal Blauwbok: Tick)

New:

- Introduction of an option for sanity check into `PyFunceble.load_config`.
  This is mainly for those who may use the API.

Fixed:

- Warnings from dnspython. We now use the `resolve()` method instead of the
  `query()` method.
- Warnings from domain2idna. We now use `domain2idna.domain2idna()` instead of
  `domain2idna.get()`.

Improved:

- The way we detect if a package is the `dev` one.
- The tests which relate to the syntax check.
- The tests which relate to the DNS resolver.
- The way we get the latest user agent. We now ensure that our file was
  downloaded before trying to get it.
  This is mainly thought for those who may use the API.
- The presence of SQLAlchemy in the migration from the old to the new
  layout.

Removed:

- All occurrences of the old database connection method.

Author: funilrys

PyFunceble/__init__.py

@@ -672,7 +672,9 @@ def is_url(subject):  # pragma: no cover
     return None
 
 
-def load_config(generate_directory_structure=False, custom=None):  # pragma: no cover
+def load_config(
+    generate_directory_structure=False, custom=None, sanity_check=False
+):  # pragma: no cover
     """
     Load the configuration.
 
@@ -683,6 +685,9 @@ def load_config(generate_directory_structure=False, custom=None):  # pragma: no
     :param dict custom:
         A dict with the configuration index (from .PyFunceble.yaml) to update.
 
+    :param bool sanity_check:
+        Tell us to run the safety check of the configuration.
+
     .. note::
         If :code:`config` is given, the given :code:`dict` overwrite
         the last value of the given indexes in the configuration.
@@ -706,6 +711,9 @@ def load_config(generate_directory_structure=False, custom=None):  # pragma: no
     else:
         LOADER.set_custom_config(custom)
 
+    if sanity_check:
+        cconfig.Preset().init_all()
+
     if generate_directory_structure:
         output.Constructor()
 

PyFunceble/abstracts/package.py

@@ -64,7 +64,7 @@ class Package:
     :type: str
     """
 
-    VERSION = "3.3.1. (Teal Blauwbok: Grub)"
+    VERSION = "3.3.2. (Teal Blauwbok: Tick)"
     """
     Sets the package version.
 
@@ -213,7 +213,9 @@ def is_local_dev(cls):
         Checks if the local version is the development version.
         """
 
-        return "dev" in Package.VERSION
+        return cls.split_versions(Package.VERSION, return_non_digits=True)[
+            -1
+        ].startswith("dev")
 
     @classmethod
     def is_local_cloned(cls):  # pragma: no cover

PyFunceble/check.py

@@ -58,7 +58,7 @@
     ip_network,
 )
 
-from domain2idna import get as domain2idna
+import domain2idna
 
 import PyFunceble
 
@@ -134,7 +134,7 @@ def is_url(
                     # We have to convert the domain to IDNA.
 
                     # We convert the initial base to IDNA.
-                    base = domain2idna(base)
+                    base = domain2idna.domain2idna(base)
 
                 if ":" in base:
                     # The port is explicitly given.
@@ -424,7 +424,7 @@ def is_ipv6(self):
         except ValueError:
             return False
 
-    def is_ip_range(self):  # pragma: no cover
+    def is_ip_range(self):
         """
         Checks if the given subject is a valid IPv4 or IPv6 range.
 
@@ -464,7 +464,7 @@ def is_ipv6_range(self):
                 return 0 <= block <= 128
         return False
 
-    def is_reserved_ip(self):  # pragma: no cover
+    def is_reserved_ip(self):
         """
         Checks if the given subject is a reserved IPv4 or IPv6.
 

PyFunceble/core/file.py

@@ -51,7 +51,7 @@
 
 from tempfile import NamedTemporaryFile
 
-from domain2idna import get as domain2idna
+import domain2idna
 from sqlalchemy.orm.exc import NoResultFound
 
 import PyFunceble
@@ -264,7 +264,7 @@ def test(self, subject):
         """
 
         if PyFunceble.CONFIGURATION.idna_conversion:
-            subject = domain2idna(subject)
+            subject = domain2idna.domain2idna(subject)
 
         if isinstance(PyFunceble.CONFIGURATION.cooldown_time, (float, int)):
             PyFunceble.sleep(PyFunceble.CONFIGURATION.cooldown_time)

PyFunceble/core/multiprocess.py

@@ -56,9 +56,9 @@
 from tempfile import NamedTemporaryFile
 from traceback import format_exc
 
+import domain2idna
 from colorama import Fore, Style
 from colorama import init as initiate_colorama
-from domain2idna import get as domain2idna
 
 import PyFunceble
 
@@ -169,7 +169,7 @@ def test(
         PyFunceble.INTERN.update(intern)
 
         if PyFunceble.CONFIGURATION.idna_conversion:
-            subject = domain2idna(subject)
+            subject = domain2idna.domain2idna(subject)
 
         if not self.should_be_ignored(
             subject,

PyFunceble/core/simple.py

@@ -50,7 +50,7 @@
     limitations under the License.
 """
 
-from domain2idna import get as domain2idna
+import domain2idna
 
 import PyFunceble
 
@@ -70,7 +70,7 @@ def __init__(self, subject):
         self.mining = PyFunceble.engine.Mining("simple")
 
         if PyFunceble.CONFIGURATION.idna_conversion:
-            self.subject = domain2idna(subject)
+            self.subject = domain2idna.domain2idna(subject)
         else:
             self.subject = subject
 

PyFunceble/engine/database/migrations/old2new.py

@@ -52,10 +52,7 @@
 
 from datetime import datetime
 from multiprocessing import active_children
-from os import sep as directory_separator
 
-import pymysql
-import pymysql.cursors
 from colorama import Fore, Style
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm.exc import MultipleResultsFound, NoResultFound
@@ -98,58 +95,25 @@ def authorized(self):
             [self.does_table_exists(x) for x in self.old_tables]
         )
 
-    def __get_rows(self, statement, limit=20):
+    @classmethod
+    def __get_rows(cls, statement, limit=20):
         """
         Get the row of the database.
         """
 
         statement += f" LIMIT {limit}"
 
         while True:
-            fetcher_connection = self.get_old_connection()
-            with fetcher_connection.cursor() as cursor:
-                cursor.execute(statement)
-
-                db_result = cursor.fetchall()
-
-            fetcher_connection.close()
+            with PyFunceble.engine.database.loader.session.Session() as db_session:
+                db_result = db_session.execute(statement)
+                db_result = [dict(x) for x in db_result.fetchall()]
 
             if not db_result:
                 break
 
             for result in db_result:
                 yield result
 
-    def get_old_connection(self):
-        """
-        Provides a connection, the old way.
-        """
-
-        if (
-            directory_separator not in self.credentials["host"]
-            or "/" not in self.credentials["host"]
-        ):
-            return pymysql.connect(
-                host=self.credentials["host"],
-                port=self.credentials["port"],
-                user=self.credentials["username"],
-                password=self.credentials["password"],
-                db=self.credentials["name"],
-                charset=self.credentials["charset"],
-                cursorclass=pymysql.cursors.DictCursor,
-                autocommit=True,
-            )
-
-        return pymysql.connect(
-            unix_socket=self.credentials["host"],
-            user=self.credentials["username"],
-            password=self.credentials["password"],
-            db=self.credentials["name"],
-            charset=self.credentials["charset"],
-            cursorclass=pymysql.cursors.DictCursor,
-            autocommit=True,
-        )
-
     def does_table_exists(self, table_name):
         """
         Checks if the table exists.
@@ -158,25 +122,23 @@ def does_table_exists(self, table_name):
             The name of the table to check.
         """
 
-        old_connection = self.get_old_connection()
-        with old_connection.cursor() as cursor:
+        with PyFunceble.engine.database.loader.session.Session() as db_session:
             statement = (
                 "SELECT COUNT(*) "
                 "FROM information_schema.tables "
-                "WHERE table_schema = %(database_name)s "
-                "AND table_name = %(table_name)s "
+                "WHERE table_schema = :database_name "
+                "AND table_name = :table_name "
             )
 
-            cursor.execute(
+            result = db_session.execute(
                 statement,
                 {
                     "database_name": self.credentials["name"],
                     "table_name": table_name,
                 },
             )
 
-            result = cursor.fetchone()
-        old_connection.close()
+            result = dict(result.fetchone())
 
         if result["COUNT(*)"] != 1:
             return False
@@ -287,12 +249,11 @@ def __tested_migration(self, data):
             except IntegrityError:
                 pass
 
-        old_connection = self.get_old_connection()
-        with old_connection.cursor() as cursor:
-            statement = "DELETE FROM pyfunceble_tested WHERE id = %(status_id)s"
+        with PyFunceble.engine.database.loader.session.Session() as db_session:
+            statement = "DELETE FROM pyfunceble_tested WHERE id = :status_id"
+
             # pylint: disable=no-member
-            cursor.execute(statement, {"status_id": status.id})
-        old_connection.close()
+            db_session.execute(statement, {"status_id": status.id})
 
         if self.autosave.authorized or PyFunceble.CONFIGURATION.print_dots:
             PyFunceble.LOGGER.info(f'Switched {data["tested"]} to SQLAlchemy.')
@@ -364,11 +325,10 @@ def __autocontinue_migration(self, data):
             except IntegrityError:
                 pass
 
-        old_connection = self.get_old_connection()
-        with old_connection.cursor() as cursor:
-            statement = "DELETE FROM pyfunceble_auto_continue WHERE id = %(id)s"
-            cursor.execute(statement, {"id": data["id"]})
-        old_connection.close()
+        with PyFunceble.engine.database.loader.session.Session() as db_session:
+            statement = "DELETE FROM pyfunceble_auto_continue WHERE id = :id"
+
+            db_session.execute(statement, {"id": data["id"]})
 
         if self.autosave.authorized or PyFunceble.CONFIGURATION.print_dots:
             PyFunceble.LOGGER.info(
@@ -440,11 +400,10 @@ def __whois_migration(self, data):
             except IntegrityError:
                 pass
 
-        old_connection = self.get_old_connection()
-        with old_connection.cursor() as cursor:
-            statement = "DELETE FROM pyfunceble_whois WHERE id = %(id)s"
-            cursor.execute(statement, {"id": data["id"]})
-        old_connection.close()
+        with PyFunceble.engine.database.loader.session.Session() as db_session:
+            statement = "DELETE FROM pyfunceble_whois WHERE id = :id"
+
+            db_session.execute(statement, {"id": data["id"]})
 
         if self.autosave.authorized or PyFunceble.CONFIGURATION.print_dots:
             PyFunceble.LOGGER.info(f'Switched {data["subject"]} (WHOIS) to SQLAlchemy.')
@@ -482,12 +441,12 @@ def __delete_old_tables(self):
         for table in self.old_tables:
             if self.does_table_exists(table):
                 PyFunceble.LOGGER.info(f"Starting deletion of {table}.")
-                old_connection = self.get_old_connection()
-                with old_connection.cursor() as cursor:
+
+                with PyFunceble.engine.database.loader.session.Session() as db_session:
                     statement = f"DROP TABLE {table}"
-                    cursor.execute(statement)
-                old_connection.close()
-                PyFunceble.LOGGER.info(f"Finished deletion of {table}.")
+
+                    db_session.execute(statement)
+                    PyFunceble.LOGGER.info(f"Finished deletion of {table}.")
 
             if self.autosave.authorized or PyFunceble.CONFIGURATION.print_dots:
                 print(".", end="")

PyFunceble/engine/user_agent.py

@@ -60,11 +60,16 @@ class UserAgent:
     """
 
     def __init__(self):
-        self.dumped = PyFunceble.helpers.Dict().from_json_file(
+        file_instance = PyFunceble.helpers.File(
             PyFunceble.CONFIG_DIRECTORY
             + PyFunceble.abstracts.Infrastructure.USER_AGENT_FILENAME
         )
 
+        if not file_instance.exists():
+            PyFunceble.downloader.UserAgents()
+
+        self.dumped = PyFunceble.helpers.Dict().from_json_file(file_instance.path)
+
     def get(self):
         """
         Provides the user agent to use.