improved logger.py, events.py

conftest.py

@@ -8,7 +8,6 @@
 log = Logger(log_lvl=LogLevel.INFO)
 
 
-# Registering pytest options for command-line argument parsing
 def pytest_addoption(parser):
     parser.addoption(
         "--env", action="store", default="dev", help="Default environment"
@@ -62,21 +61,17 @@ def playwright():
 def browser(request, playwright):
     browser_type = request.config.getoption("--browser-type")
 
-    # Fetching options from pytest command-line arguments
-    headless = request.config.getoption("--headless")
-    slow_mo = request.config.getoption("--slow-mo")
-    launch_options = {"headless": headless, "slow_mo": slow_mo}
+    launch_options = get_browser_options(request)
 
     if browser_type == "chromium":
         args = ["--start-maximized"]
         browser_instance = playwright.chromium.launch(
             headless=launch_options["headless"],
             args=args,
             slow_mo=launch_options["slow_mo"],
-            devtools=request.config.getoption("--devtools"),
+            devtools=launch_options["devtools"],
         )
     else:
-        # For other browsers, launch normally
         browser_launch_func = {
             "firefox": playwright.firefox.launch,
             "webkit": playwright.webkit.launch,
@@ -88,6 +83,31 @@ def browser(request, playwright):
     browser_instance.close()
 
 
+@pytest.fixture
+def page(browser, request):
+    # Retrieve timeout options
+    navigation_timeout = request.config.getoption("--navigation-timeout", default=12000)
+    default_timeout = request.config.getoption("--default-timeout", default=6000)
+
+    log.annotate(f"Creating a new browser context with timeouts: "
+                 f"{navigation_timeout}, {default_timeout}")
+
+    # Create a new browser context
+    browser_context = browser.new_context(no_viewport=True)
+    browser_context.set_default_navigation_timeout(navigation_timeout)
+    browser_context.set_default_timeout(default_timeout)
+
+    # Create a new page
+    page = browser_context.new_page()
+    selected_listeners = request.config.getoption("--listeners").strip().split(",")
+    EventListenerManager(page, selected_listeners, log)
+
+    yield page
+
+    page.close()
+    browser_context.close()  # Optional: Close the context if you want to free resources
+
+
 @pytest.hookimpl(hookwrapper=True)
 def pytest_runtest_makereport(item):
     """
@@ -122,30 +142,16 @@ def pytest_runtest_makereport(item):
                 print(f"Error capturing screenshot: {e}")
 
 
-# Page fixture to open a new page and attach listeners dynamically
-@pytest.fixture
-def page(browser, request):
-    browser_context = browser.new_context(no_viewport=True)
-    browser_context.set_default_navigation_timeout(12000)
-    browser_context.set_default_timeout(6000)
-
-    page = browser.new_page(no_viewport=True)
-
-    # Fetching and attaching event listeners
-    selected_listeners = request.config.getoption("--listeners").strip().split(",")
-    EventListenerManager(page, selected_listeners, log)
-
-    yield page
-    page.close()
-
-
 def get_browser_options(request):
     """
     Returns browser launch options based on pytest command-line options.
     """
     return {
         "headless": request.config.getoption("--headless"),
         "devtools": request.config.getoption("--devtools"),
+        "slow_mo": float(
+            request.config.getoption("--slow-mo")
+        ),  # Include slow_mo here
         "proxy": {"server": request.config.getoption("--proxy")}
         if request.config.getoption("--proxy")
         else None,

src/drivers/events.py

@@ -9,28 +9,37 @@ def attach_listeners(self):
         """Attach event listeners based on selected options."""
         if "console" in self.selected_listeners:
             self.page.on("console", self.handle_console_log)
+            self.logger.annotate("Attached console log listener", "info")
 
         if "request" in self.selected_listeners:
             self.page.on("request", self.handle_request)
+            self.logger.annotate("Attached request listener", "info")
 
         if "response" in self.selected_listeners:
             self.page.on("response", self.handle_response)
+            self.logger.annotate("Attached response listener", "info")
 
         if "events" in self.selected_listeners:
             self.page.on("load", self.handle_page_load)
             self.page.on("close", self.handle_page_close)
+            self.logger.annotate("Attached page load and close listeners", "info")
 
     def handle_console_log(self, msg):
+        """Handle console log events."""
         self.logger.annotate(f"Console log: {msg.text()}", "info")
 
     def handle_request(self, req):
+        """Handle request events."""
         self.logger.annotate(f"Request: {req.url}", "info")
 
     def handle_response(self, res):
+        """Handle response events."""
         self.logger.annotate(f"Response: {res.url} - {res.status()}", "info")
 
     def handle_page_load(self):
+        """Handle page load events."""
         self.logger.annotate(f"Page loaded: {self.page.url}", "info")
 
     def handle_page_close(self):
+        """Handle page close events."""
         self.logger.annotate(f"Page closed: {self.page.url}", "info")

src/utils/logger.py

@@ -23,97 +23,96 @@ def __call__(cls, *args, **kwargs) -> Any:
 
 class Logger(metaclass=Singleton):
     def __init__(
-        self,
-        log_lvl: LogLevel = LogLevel.INFO,
-        log_base_directory: Optional[str] = None,
+            self,
+            log_lvl: LogLevel = LogLevel.INFO,
+            log_base_directory: Optional[str] = None,
+            log_mode: str = 'w',  # 'w' for overwrite, 'a' for append
+            console_logging: bool = True,
     ) -> None:
-        self._log = logging.getLogger("selenium")
+        self._log = logging.getLogger("playwrite")
         self._log.setLevel(LogLevel.DEBUG.value)
+
         self.log_base_directory = log_base_directory or os.path.abspath(
-            os.path.join(os.path.dirname(__file__), "../..")
-        )  # Default to the project's root directory
+            os.path.join(os.path.dirname(__file__), "../.."))
         self.log_file = self._create_log_file()
-        self._initialize_logging(log_lvl)
+        self._initialize_logging(log_lvl, log_mode, console_logging)
 
     def _create_log_file(self) -> str:
         current_time = time.strftime("%Y-%m-%d")
-        log_directory = os.path.join(
-            self.log_base_directory, "reports/logs"
-        )  # Build the path to reports/logs
-
-        try:
-            os.makedirs(
-                log_directory, exist_ok=True
-            )  # Create directory if it doesn't exist
-        except Exception as e:
-            raise RuntimeError(
-                f"Failed to create log directory '{log_directory}': {e}"
-            )
+        log_directory = os.path.join(self.log_base_directory, "reports/logs")
+
+        os.makedirs(log_directory,
+                    exist_ok=True)  # Create directory if it doesn't exist
 
         return os.path.join(log_directory, f"log_{current_time}.log")
 
-    def _initialize_logging(self, log_lvl: LogLevel) -> None:
+    def _initialize_logging(self, log_lvl: LogLevel, log_mode: str,
+                            console_logging: bool) -> None:
         formatter = logging.Formatter(
-            "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
-        )
-        fh = logging.FileHandler(self.log_file, mode="w")
+            "%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+
+        # File handler
+        fh = logging.FileHandler(self.log_file, mode=log_mode)
         fh.setFormatter(formatter)
         fh.setLevel(log_lvl.value)
         self._log.addHandler(fh)
 
+        # Console handler
+        if console_logging:
+            ch = logging.StreamHandler()
+            ch.setFormatter(formatter)
+            ch.setLevel(log_lvl.value)
+            self._log.addHandler(ch)
+
     def get_instance(self) -> logging.Logger:
         return self._log
 
     def annotate(
-        self, message: str, level: Literal["info", "warn", "debug", "error"]
+            self, message: str,
+            level: Literal["info", "warn", "debug", "error"] = "info"
     ) -> None:
         """Log a message at the specified level."""
-        if level == "info":
-            self._log.info(message)
-        elif level == "warn":
-            self._log.warning(message)
-        elif level == "debug":
-            self._log.debug(message)
-        elif level == "error":
-            self._log.error(message)
-        else:
+        log_methods = {
+            "info": self._log.info,
+            "warn": self._log.warning,
+            "debug": self._log.debug,
+            "error": self._log.error,
+        }
+
+        if level not in log_methods:
             raise ValueError(f"Invalid log level: {level}")
 
+        log_methods[level](message)
+
 
 def log(
-    data: Optional[str] = None,
-    level: Literal["info", "warn", "debug", "error"] = "info",
+        data: Optional[str] = None,
+        level: Literal["info", "warn", "debug", "error"] = "info",
 ) -> Callable:
     """Decorator to log the current method's execution.
 
     :param data: Custom log message to use if no docstring is provided.
     :param level: Level of the logs, e.g., info, warn, debug, error.
     """
-    logger_instance = Logger()  # Get the singleton instance of Logger
+    logger_instance = Logger()
 
     def decorator(func: Callable) -> Callable:
         def wrapper(self, *args, **kwargs) -> Any:
             # Get the method's docstring
             method_docs = format_method_doc_str(func.__doc__)
 
             if method_docs is None and data is None:
-                raise ValueError(
-                    f"No documentation available for :: {func.__name__}"
-                )
+                raise ValueError(f"No documentation available for :: {func.__name__}")
 
             # Construct the parameter string for logging
             params_str = ", ".join(repr(arg) for arg in args)
             kwargs_str = ", ".join(f"{k}={v!r}" for k, v in kwargs.items())
             all_params_str = ", ".join(filter(None, [params_str, kwargs_str]))
 
             logs = (
-                method_docs
-                + f" Method :: {func.__name__}()"
-                + f" with parameters: {all_params_str}"
-                if method_docs
-                else data
-                + f" Method :: {func.__name__}()"
-                + f" with parameters: {all_params_str}"
+                f"{method_docs + '.' if method_docs else data}"
+                f" Method :: {func.__name__}() "
+                f"with parameters: {all_params_str}"
             )
 
             logger_instance.annotate(logs, level)