Road cleanup

pytest_response/app.py

@@ -112,9 +112,6 @@ def __init__(
         log_level: str = "debug",
         database: str = "database.db",
     ) -> None:
-        log.setLevel(log_level.upper())
-        log.info("<------------------------------------------------------------------->")
-
         self._basepath = pathlib.Path(__file__).parent
 
         self._path_to_mocks = self._basepath.joinpath(path)

pytest_response/database.py

@@ -5,6 +5,7 @@
 from typing import List, Tuple, Optional
 from datetime import date
 from urllib.parse import urljoin, urlparse
+from loguru import logger as log
 
 from pytest_response.exceptions import MalformedUrl
 
@@ -38,8 +39,7 @@ class ResponseDB:
 
     def __init__(self, path: str) -> None:
         self._path = path
-        self._database = sqlite3.connect(path)
-        self.setup()
+        self._database = None
         return
 
     def __repr__(self) -> str:
@@ -49,6 +49,8 @@ def setup(self) -> bool:
         """
         Function to setup the database table.
         """
+        log.debug("setting up database")
+        self._database = sqlite3.connect(self._path)
         self._database.cursor().executescript(
             """
             CREATE TABLE IF NOT EXISTS records (
@@ -78,6 +80,8 @@ def index(self, index: Optional[str] = "url") -> List[str]:
         _occurances : `list`
             All occurances of the selected column `index`.
         """
+        if not self._database:
+            self.setup()
         elements = self._database.execute("SELECT * FROM records;").fetchall()
         _occurances = []
         for element in elements:
@@ -125,6 +129,8 @@ def insert(self, url: str, response: bytes, headers: dict, status: Optional[int]
         kwargs : `dict`
             Any additional parameter to be dumped.
         """
+        if not self._database:
+            self.setup()
         _kwargs = [
             b64encode(self._sanatize_url(url).encode()).decode(),
             self.today,
@@ -157,6 +163,8 @@ def get(self, url: str, **kwargs) -> Tuple[int, bytes, dict]:
         headers : `dict`
             Response header.
         """
+        if not self._database:
+            self.setup()
         url = b64encode(self._sanatize_url(url).encode()).decode()
         element = self._database.execute(f"SELECT * FROM records WHERE url='{url}';").fetchall()
         if element:
@@ -178,21 +186,27 @@ def all(self) -> dict:
         -------
         Return list of all elements.
         """
+        if not self._database:
+            self.setup()
         return self._database.execute("SELECT * FROM records").fetchall()
 
     def truncate(self) -> bool:
         """
         Method to purge all records in the database.
         """
+        if not self._database:
+            self.setup()
         self._database.execute("DELETE FROM records;")
         return self._database.commit()
 
     def close(self) -> None:
-        self._database.close()
-        return
+        if not self._database:
+            return True
+        return self._database.close()
 
     def __del__(self) -> None:
-        self._database.close()
-        return
+        if not self._database:
+            return True
+        return self._database.close()
 
     pass

pytest_response/logger.py

@@ -1,45 +1 @@
-import logging
-
-
-class Formatter(logging.Formatter):
-    """Logging Formatter to add colors and count warning / errors"""
-
-    grey = "\033[38;1m"
-    cyan = "\033[36m"
-    green = "\033[32m"
-    yellow = "\033[33m"
-    red = "\033[31m"
-    bold_red = "\033[31;1m"
-    reset = "\033[0m"
-    format = "%(asctime)s - %(module)s.%(funcName)s() - %(levelname)s - %(message)s (%(filename)s:%(lineno)d)"
-
-    FORMATS = {
-        logging.DEBUG: "".join([cyan, format, reset]),
-        logging.INFO: "".join([green, format, reset]),
-        logging.WARNING: "".join([yellow, format, reset]),
-        logging.ERROR: "".join([red, format, reset]),
-        logging.CRITICAL: "".join([bold_red, format, reset]),
-    }
-
-    def format(self, record):
-        log_fmt = self.FORMATS.get(record.levelno)
-        formatter = logging.Formatter(log_fmt)
-        return formatter.format(record)
-
-
-def _init_log(level="info"):
-    log = logging.getLogger("pytest_response")
-    log.setLevel(getattr(logging, level.upper()))
-    fh = logging.FileHandler(".pytest_response.log")
-    fh.setLevel(logging.DEBUG)
-    ch = logging.StreamHandler()
-    ch.setLevel(logging.ERROR)
-    # formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
-    fh.setFormatter(Formatter())
-    ch.setFormatter(Formatter())
-    log.addHandler(fh)
-    log.addHandler(ch)
-    return log
-
-
-log = _init_log()
+from loguru import logger as log

pytest_response/plugin.py

@@ -30,6 +30,12 @@ def pytest_addoption(parser):
         default=True,
         help="Blocks remote connection requests for all interceptors.",
     )
+    parser.addini(
+        "enable_remote_response",
+        type="bool",
+        default=False,
+        help="Activate remote-response plugin",
+    )
     parser.addini(
         "remote_response_database",
         type="string",
@@ -42,6 +48,8 @@ def pytest_configure(config):
     """
     Pytest hook for setting up :class:`pytest_response.app.Response`
     """
+    if not config.getini("enable_remote_response"):
+        return
     # either remote_capture or remote_response
     if config.option.remote_capture and config.option.remote_response:
         assert not config.option.remote_capture and config.option.remote_response

setup.cfg

@@ -31,6 +31,7 @@ install_requires =
     pytest>=4.6
     requests>=2.25.1
     aiohttp
+    loguru
 
 [options.entry_points]
 pytest11 =
@@ -53,6 +54,7 @@ max-line-length = 110
 norecursedirs = ".tox" "*.egg-info"
 minversion = 4.6
 testpaths = tests
+enable_remote_response = true
 
 [coverage:run]
 disable_warnings = no-data-collected