From 2f402f48355fcc487757d00d0747ea53de780dcd Mon Sep 17 00:00:00 2001 From: Karim shoair Date: Sun, 17 Aug 2025 01:02:14 +0300 Subject: [PATCH] style: add flags for tests coverage - Some are already tested but the coverage report can't see it. - Some are not necessary to test or too hard to test on GitHub's CI --- scrapling/cli.py | 6 +- scrapling/core/_html_utils.py | 2 +- scrapling/core/_types.py | 2 +- scrapling/core/custom_types.py | 62 ++++++++++++-------- scrapling/core/shell.py | 42 +++++++------ scrapling/core/storage.py | 2 +- scrapling/core/translator.py | 18 +++--- scrapling/engines/_browsers/_camoufox.py | 32 +++++----- scrapling/engines/_browsers/_config_tools.py | 2 +- scrapling/engines/_browsers/_controllers.py | 24 ++++---- scrapling/engines/static.py | 12 ++-- scrapling/engines/toolbelt/convertor.py | 12 ++-- scrapling/parser.py | 6 +- 13 files changed, 121 insertions(+), 101 deletions(-) diff --git a/scrapling/cli.py b/scrapling/cli.py index 7b5d703..dd012cd 100644 --- a/scrapling/cli.py +++ b/scrapling/cli.py @@ -15,7 +15,7 @@ __OUTPUT_FILE_HELP__ = "Output file path can be HTML content, Markdown of the HT __PACKAGE_DIR__ = Path(__file__).parent -def __Execute(cmd: List[str], help_line: str) -> None: +def __Execute(cmd: List[str], help_line: str) -> None: # pragma: no cover print(f"Installing {help_line}...") _ = check_output(cmd, shell=False) # nosec B603 # I meant to not use try except here @@ -28,7 +28,7 @@ def __ParseJSONData(json_string: Optional[str] = None) -> Optional[Dict[str, Any try: return json_loads(json_string) - except JSONDecodeError as e: + except JSONDecodeError as e: # pragma: no cover raise ValueError(f"Invalid JSON data '{json_string}': {e}") @@ -105,7 +105,7 @@ def __BuildRequest( type=bool, help="Force Scrapling to reinstall all Fetchers dependencies", ) -def install(force): +def install(force): # pragma: no cover if ( force or not __PACKAGE_DIR__.joinpath(".scrapling_dependencies_installed").exists() diff --git a/scrapling/core/_html_utils.py b/scrapling/core/_html_utils.py index c0cd45c..99776af 100644 --- a/scrapling/core/_html_utils.py +++ b/scrapling/core/_html_utils.py @@ -340,7 +340,7 @@ def _replace_entities( if 0x80 <= number <= 0x9F: return bytes((number,)).decode("cp1252") return chr(number) - except (ValueError, OverflowError): + except (ValueError, OverflowError): # pragma: no cover pass return "" if remove_illegal and groups.get("semicolon") else m.group(0) diff --git a/scrapling/core/_types.py b/scrapling/core/_types.py index a114e37..cd8c9c0 100644 --- a/scrapling/core/_types.py +++ b/scrapling/core/_types.py @@ -35,7 +35,7 @@ StrOrBytes = Union[str, bytes] try: # Python 3.11+ from typing import Self # novermin -except ImportError: +except ImportError: # pragma: no cover try: from typing_extensions import Self # Backport except ImportError: diff --git a/scrapling/core/custom_types.py b/scrapling/core/custom_types.py index d46fd0c..ef76880 100644 --- a/scrapling/core/custom_types.py +++ b/scrapling/core/custom_types.py @@ -31,11 +31,15 @@ class TextHandler(str): __slots__ = () - def __getitem__(self, key: SupportsIndex | slice) -> "TextHandler": + def __getitem__( + self, key: SupportsIndex | slice + ) -> "TextHandler": # pragma: no cover lst = super().__getitem__(key) return cast(_TextHandlerType, TextHandler(lst)) - def split(self, sep: str = None, maxsplit: SupportsIndex = -1) -> "TextHandlers": + def split( + self, sep: str = None, maxsplit: SupportsIndex = -1 + ) -> "TextHandlers": # pragma: no cover return TextHandlers( cast( List[_TextHandlerType], @@ -43,58 +47,70 @@ class TextHandler(str): ) ) - def strip(self, chars: str = None) -> Union[str, "TextHandler"]: + def strip(self, chars: str = None) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().strip(chars)) - def lstrip(self, chars: str = None) -> Union[str, "TextHandler"]: + def lstrip( + self, chars: str = None + ) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().lstrip(chars)) - def rstrip(self, chars: str = None) -> Union[str, "TextHandler"]: + def rstrip( + self, chars: str = None + ) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().rstrip(chars)) - def capitalize(self) -> Union[str, "TextHandler"]: + def capitalize(self) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().capitalize()) - def casefold(self) -> Union[str, "TextHandler"]: + def casefold(self) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().casefold()) def center( self, width: SupportsIndex, fillchar: str = " " - ) -> Union[str, "TextHandler"]: + ) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().center(width, fillchar)) - def expandtabs(self, tabsize: SupportsIndex = 8) -> Union[str, "TextHandler"]: + def expandtabs( + self, tabsize: SupportsIndex = 8 + ) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().expandtabs(tabsize)) - def format(self, *args: str, **kwargs: str) -> Union[str, "TextHandler"]: + def format( + self, *args: str, **kwargs: str + ) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().format(*args, **kwargs)) - def format_map(self, mapping) -> Union[str, "TextHandler"]: + def format_map(self, mapping) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().format_map(mapping)) - def join(self, iterable: Iterable[str]) -> Union[str, "TextHandler"]: + def join( + self, iterable: Iterable[str] + ) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().join(iterable)) def ljust( self, width: SupportsIndex, fillchar: str = " " - ) -> Union[str, "TextHandler"]: + ) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().ljust(width, fillchar)) def rjust( self, width: SupportsIndex, fillchar: str = " " - ) -> Union[str, "TextHandler"]: + ) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().rjust(width, fillchar)) - def swapcase(self) -> Union[str, "TextHandler"]: + def swapcase(self) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().swapcase()) - def title(self) -> Union[str, "TextHandler"]: + def title(self) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().title()) - def translate(self, table) -> Union[str, "TextHandler"]: + def translate(self, table) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().translate(table)) - def zfill(self, width: SupportsIndex) -> Union[str, "TextHandler"]: + def zfill( + self, width: SupportsIndex + ) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().zfill(width)) def replace( @@ -120,10 +136,10 @@ class TextHandler(str): return self.__class__(__CONSECUTIVE_SPACES_REGEX__.sub(" ", data).strip()) # For easy copy-paste from Scrapy/parsel code when needed :) - def get(self, default=None): + def get(self, default=None): # pragma: no cover return self - def get_all(self): + def get_all(self): # pragma: no cover return self extract = get_all @@ -234,11 +250,11 @@ class TextHandlers(List[TextHandler]): __slots__ = () @overload - def __getitem__(self, pos: SupportsIndex) -> TextHandler: + def __getitem__(self, pos: SupportsIndex) -> TextHandler: # pragma: no cover pass @overload - def __getitem__(self, pos: slice) -> "TextHandlers": + def __getitem__(self, pos: slice) -> "TextHandlers": # pragma: no cover pass def __getitem__( @@ -276,7 +292,7 @@ class TextHandlers(List[TextHandler]): replace_entities: bool = True, clean_match: bool = False, case_sensitive: bool = True, - ) -> TextHandler: + ) -> TextHandler: # pragma: no cover """Call the ``.re_first()`` method for each element in this list and return the first result or the default value otherwise. diff --git a/scrapling/core/shell.py b/scrapling/core/shell.py index 5020194..0125ee0 100644 --- a/scrapling/core/shell.py +++ b/scrapling/core/shell.py @@ -108,7 +108,7 @@ def _ParseHeaders( cookie_dict = { key: value for key, value in _CookieParser(header_value) } - except Exception as e: + except Exception as e: # pragma: no cover raise ValueError( f"Could not parse cookie string from header '{header_value}': {e}" ) @@ -121,7 +121,7 @@ def _ParseHeaders( # Suppress exit on error to handle parsing errors gracefully -class NoExitArgumentParser(ArgumentParser): +class NoExitArgumentParser(ArgumentParser): # pragma: no cover def error(self, message): log.error(f"Curl arguments parsing error: {message}") raise ValueError(f"Curl arguments parsing error: {message}") @@ -188,8 +188,6 @@ class CurlParser: self.parser: NoExitArgumentParser = _parser self._supported_methods = ("get", "post", "put", "delete") - # --- Helper Functions --- - # --- Main Parsing Logic --- def parse(self, curl_command: str) -> Optional[Request]: """Parses the curl command string into a structured context for Fetcher.""" @@ -200,7 +198,7 @@ class CurlParser: tokens = shlex_split( clean_command ) # Split the string using shell-like syntax - except ValueError as e: + except ValueError as e: # pragma: no cover log.error(f"Could not split command line: {e}") return None @@ -209,13 +207,13 @@ class CurlParser: if unknown: raise AttributeError(f"Unknown/Unsupported curl arguments: {unknown}") - except ValueError: + except ValueError: # pragma: no cover return None except AttributeError: raise - except Exception as e: + except Exception as e: # pragma: no cover log.error( f"An unexpected error occurred during curl arguments parsing: {e}" ) @@ -249,7 +247,7 @@ class CurlParser: # Update the cookie dict, potentially overwriting cookies with the same name from -H 'cookie:' cookies[key] = value log.debug(f"Parsed cookies from -b argument: {list(cookies.keys())}") - except Exception as e: + except Exception as e: # pragma: no cover log.error( f"Could not parse cookie string from -b '{parsed_args.cookie}': {e}" ) @@ -261,7 +259,7 @@ class CurlParser: # DevTools often uses --data-raw for JSON bodies # Precedence: --data-binary > --data-raw / -d > --data-urlencode - if parsed_args.data_binary is not None: + if parsed_args.data_binary is not None: # pragma: no cover try: data_payload = parsed_args.data_binary.encode("utf-8") log.debug("Using data from --data-binary as bytes.") @@ -277,7 +275,7 @@ class CurlParser: elif parsed_args.data is not None: data_payload = parsed_args.data - elif parsed_args.data_urlencode: + elif parsed_args.data_urlencode: # pragma: no cover # Combine and parse urlencoded data combined_data = "&".join(parsed_args.data_urlencode) try: @@ -299,7 +297,7 @@ class CurlParser: pass # Not JSON, keep it in data_payload # Handle `-G`: Move data to params if the method is GET - if method == "get" and data_payload: + if method == "get" and data_payload: # pragma: no cover if isinstance(data_payload, dict): # From --data-urlencode likely params.update(data_payload) elif isinstance(data_payload, str): @@ -369,7 +367,7 @@ class CurlParser: ) # Ensure request parsing was successful before proceeding - if request is None: + if request is None: # pragma: no cover log.error("Failed to parse curl command, cannot convert to fetcher.") return None @@ -385,22 +383,22 @@ class CurlParser: try: return getattr(Fetcher, method)(**request_args) - except Exception as e: + except Exception as e: # pragma: no cover log.error(f"Error calling Fetcher.{method}: {e}") return None - else: + else: # pragma: no cover log.error( f'Request method "{method}" isn\'t supported by Scrapling yet' ) return None - else: + else: # pragma: no cover log.error("Input must be a valid curl command string or a Request object.") return None -def show_page_in_browser(page: Selector): +def show_page_in_browser(page: Selector): # pragma: no cover if not page or not isinstance(page, Selector): log.error("Input must be of type `Selector`") return @@ -429,7 +427,7 @@ class CustomShell: if _known_logging_levels.get(log_level): self.log_level = _known_logging_levels[log_level] - else: + else: # pragma: no cover log.warning(f'Unknown log level "{log_level}", defaulting to "DEBUG"') self.log_level = DEBUG @@ -480,7 +478,7 @@ class CustomShell: Type 'exit' or press Ctrl+D to exit. """ - def update_page(self, result): + def update_page(self, result): # pragma: no cover """Update the current page and add to pages history""" self.page = result if isinstance(result, (Response, Selector)): @@ -540,11 +538,11 @@ Type 'exit' or press Ctrl+D to exit. "help": self.show_help, } - def show_help(self): + def show_help(self): # pragma: no cover """Show help information""" print(self.banner()) - def start(self): + def start(self): # pragma: no cover """Start the interactive shell""" # Get our namespace with application objects namespace = self.get_namespace() @@ -594,7 +592,7 @@ class Convertor: main_content_only: bool = False, ) -> Generator[str, None, None]: """Extract the content of a Selector""" - if not page or not isinstance(page, Selector): + if not page or not isinstance(page, Selector): # pragma: no cover raise TypeError("Input must be of type `Selector`") elif not extraction_type or extraction_type not in cls._extension_map.values(): raise ValueError(f"Unknown extraction type: {extraction_type}") @@ -627,7 +625,7 @@ class Convertor: cls, page: Selector, filename: str, css_selector: Optional[str] = None ) -> None: """Write a Selector's content to a file""" - if not page or not isinstance(page, Selector): + if not page or not isinstance(page, Selector): # pragma: no cover raise TypeError("Input must be of type `Selector`") elif not filename or not isinstance(filename, str) or not filename.strip(): raise ValueError("Filename must be provided") diff --git a/scrapling/core/storage.py b/scrapling/core/storage.py index 096b688..089a9ec 100644 --- a/scrapling/core/storage.py +++ b/scrapling/core/storage.py @@ -12,7 +12,7 @@ from scrapling.core.utils import _StorageTools, log from scrapling.core._types import Dict, Optional, Any -class StorageSystemMixin(ABC): +class StorageSystemMixin(ABC): # pragma: no cover # If you want to make your own storage system, you have to inherit from this def __init__(self, url: Optional[str] = None): """ diff --git a/scrapling/core/translator.py b/scrapling/core/translator.py index bdab7a5..e0a91bc 100644 --- a/scrapling/core/translator.py +++ b/scrapling/core/translator.py @@ -37,15 +37,15 @@ class XPathExpr(OriginalXPathExpr): def __str__(self) -> str: path = super().__str__() if self.textnode: - if path == "*": + if path == "*": # pragma: no cover path = "text()" - elif path.endswith("::*/*"): + elif path.endswith("::*/*"): # pragma: no cover path = path[:-3] + "text()" else: path += "/text()" if self.attribute is not None: - if path.endswith("::*/*"): + if path.endswith("::*/*"): # pragma: no cover path = path[:-2] path += f"/@{self.attribute}" @@ -59,7 +59,7 @@ class XPathExpr(OriginalXPathExpr): **kwargs: Any, ) -> Self: if not isinstance(other, XPathExpr): - raise ValueError( + raise ValueError( # pragma: no cover f"Expressions of type {__name__}.XPathExpr can ony join expressions" f" of the same type (or its descendants), got {type(other)}" ) @@ -71,10 +71,10 @@ class XPathExpr(OriginalXPathExpr): # e.g. cssselect.GenericTranslator, cssselect.HTMLTranslator class TranslatorProtocol(Protocol): - def xpath_element(self, selector: Element) -> OriginalXPathExpr: + def xpath_element(self, selector: Element) -> OriginalXPathExpr: # pragma: no cover pass - def css_to_xpath(self, css: str, prefix: str = ...) -> str: + def css_to_xpath(self, css: str, prefix: str = ...) -> str: # pragma: no cover pass @@ -98,7 +98,7 @@ class TranslatorMixin: if isinstance(pseudo_element, FunctionalPseudoElement): method_name = f"xpath_{pseudo_element.name.replace('-', '_')}_functional_pseudo_element" method = getattr(self, method_name, None) - if not method: + if not method: # pragma: no cover raise ExpressionError( f"The functional pseudo-element ::{pseudo_element.name}() is unknown" ) @@ -108,7 +108,7 @@ class TranslatorMixin: f"xpath_{pseudo_element.replace('-', '_')}_simple_pseudo_element" ) method = getattr(self, method_name, None) - if not method: + if not method: # pragma: no cover raise ExpressionError( f"The pseudo-element ::{pseudo_element} is unknown" ) @@ -120,7 +120,7 @@ class TranslatorMixin: xpath: OriginalXPathExpr, function: FunctionalPseudoElement ) -> XPathExpr: """Support selecting attribute values using ::attr() pseudo-element""" - if function.argument_types() not in (["STRING"], ["IDENT"]): + if function.argument_types() not in (["STRING"], ["IDENT"]): # pragma: no cover raise ExpressionError( f"Expected a single string or ident for ::attr(), got {function.arguments!r}" ) diff --git a/scrapling/engines/_browsers/_camoufox.py b/scrapling/engines/_browsers/_camoufox.py index b8e33bd..55f185d 100644 --- a/scrapling/engines/_browsers/_camoufox.py +++ b/scrapling/engines/_browsers/_camoufox.py @@ -229,22 +229,24 @@ class StealthySession: def __create__(self): """Create a browser for this instance and context.""" self.playwright = sync_playwright().start() - self.context = self.playwright.firefox.launch_persistent_context( - **self.launch_options + self.context = ( + self.playwright.firefox.launch_persistent_context( # pragma: no cover + **self.launch_options + ) ) - if self.cookies: + if self.cookies: # pragma: no cover self.context.add_cookies(self.cookies) - def __enter__(self): + def __enter__(self): # pragma: no cover self.__create__() return self def __exit__(self, exc_type, exc_val, exc_tb): self.close() - def close(self): + def close(self): # pragma: no cover """Close all resources""" - if self._closed: + if self._closed: # pragma: no cover return if self.context: @@ -257,7 +259,7 @@ class StealthySession: self._closed = True - def _get_or_create_page(self) -> PageInfo: + def _get_or_create_page(self) -> PageInfo: # pragma: no cover """Get an available page or create a new one""" # Try to get a ready page first page_info = self.page_pool.get_ready_page() @@ -319,7 +321,7 @@ class StealthySession: return None - def _solve_cloudflare(self, page: Page) -> None: + def _solve_cloudflare(self, page: Page) -> None: # pragma: no cover """Solve the cloudflare challenge displayed on the playwright page passed :param page: The targeted page @@ -371,7 +373,7 @@ class StealthySession: :param url: The Target url. :return: A `Response` object. """ - if self._closed: + if self._closed: # pragma: no cover raise RuntimeError("Context manager has been closed") final_response = None @@ -392,7 +394,7 @@ class StealthySession: page_info = self._get_or_create_page() page_info.mark_busy(url=url) - try: + try: # pragma: no cover # Navigate to URL and wait for a specified state page_info.page.on("response", handle_response) first_response = page_info.page.goto(url, referer=referer) @@ -440,7 +442,7 @@ class StealthySession: return response - except Exception as e: + except Exception as e: # pragma: no cover page_info.mark_error() raise e @@ -567,7 +569,7 @@ class AsyncStealthySession(StealthySession): async def close(self): """Close all resources""" - if self._closed: + if self._closed: # pragma: no cover return if self.context: @@ -605,7 +607,7 @@ class AsyncStealthySession(StealthySession): max_wait = 30 start_time = time() - while time() - start_time < max_wait: + while time() - start_time < max_wait: # pragma: no cover page_info = self.page_pool.get_ready_page() if page_info: return page_info @@ -625,7 +627,7 @@ class AsyncStealthySession(StealthySession): return else: log.info(f'The turnstile version discovered is "{challenge_type}"') - if challenge_type == "non-interactive": + if challenge_type == "non-interactive": # pragma: no cover while "Just a moment..." in (await page.content()): log.info("Waiting for Cloudflare wait page to disappear.") await page.wait_for_timeout(1000) @@ -667,7 +669,7 @@ class AsyncStealthySession(StealthySession): :param url: The Target url. :return: A `Response` object. """ - if self._closed: + if self._closed: # pragma: no cover raise RuntimeError("Context manager has been closed") final_response = None diff --git a/scrapling/engines/_browsers/_config_tools.py b/scrapling/engines/_browsers/_config_tools.py index a2ee057..46d2ad6 100644 --- a/scrapling/engines/_browsers/_config_tools.py +++ b/scrapling/engines/_browsers/_config_tools.py @@ -40,7 +40,7 @@ def _compiled_stealth_scripts(): @lru_cache(2, typed=True) -def _set_flags(hide_canvas, disable_webgl): +def _set_flags(hide_canvas, disable_webgl): # pragma: no cover """Returns the flags that will be used while launching the browser if stealth mode is enabled""" flags = DEFAULT_STEALTH_FLAGS if hide_canvas: diff --git a/scrapling/engines/_browsers/_controllers.py b/scrapling/engines/_browsers/_controllers.py index 382fdcc..649c761 100644 --- a/scrapling/engines/_browsers/_controllers.py +++ b/scrapling/engines/_browsers/_controllers.py @@ -234,7 +234,7 @@ class DynamicSession: self.playwright = sync_context().start() - if self.cdp_url: + if self.cdp_url: # pragma: no cover self.context = self.playwright.chromium.connect_over_cdp( endpoint_url=self.cdp_url ).new_context(**self.context_options) @@ -243,7 +243,7 @@ class DynamicSession: user_data_dir="", **self.launch_options ) - if self.cookies: + if self.cookies: # pragma: no cover self.context.add_cookies(self.cookies) def __enter__(self): @@ -253,7 +253,7 @@ class DynamicSession: def __exit__(self, exc_type, exc_val, exc_tb): self.close() - def close(self): + def close(self): # pragma: no cover """Close all resources""" if self._closed: return @@ -268,7 +268,7 @@ class DynamicSession: self._closed = True - def _get_or_create_page(self) -> PageInfo: + def _get_or_create_page(self) -> PageInfo: # pragma: no cover """Get an available page or create a new one""" # Try to get a ready page first page_info = self.page_pool.get_ready_page() @@ -310,7 +310,7 @@ class DynamicSession: :param url: The Target url. :return: A `Response` object. """ - if self._closed: + if self._closed: # pragma: no cover raise RuntimeError("Context manager has been closed") final_response = None @@ -331,7 +331,7 @@ class DynamicSession: page_info = self._get_or_create_page() page_info.mark_busy(url=url) - try: + try: # pragma: no cover # Navigate to URL and wait for a specified state page_info.page.on("response", handle_response) first_response = page_info.page.goto(url, referer=referer) @@ -346,7 +346,7 @@ class DynamicSession: if self.page_action is not None: try: page_info.page = self.page_action(page_info.page) - except Exception as e: + except Exception as e: # pragma: no cover log.error(f"Error executing page_action: {e}") if self.wait_selector: @@ -358,7 +358,7 @@ class DynamicSession: page_info.page.wait_for_load_state(state="domcontentloaded") if self.network_idle: page_info.page.wait_for_load_state("networkidle") - except Exception as e: + except Exception as e: # pragma: no cover log.error(f"Error waiting for selector {self.wait_selector}: {e}") page_info.page.wait_for_timeout(self.wait) @@ -506,7 +506,7 @@ class AsyncDynamicSession(DynamicSession): async def close(self): """Close all resources""" - if self._closed: + if self._closed: # pragma: no cover return if self.context: @@ -548,7 +548,7 @@ class AsyncDynamicSession(DynamicSession): max_wait = 30 # seconds start_time = time() - while time() - start_time < max_wait: + while time() - start_time < max_wait: # pragma: no cover page_info = self.page_pool.get_ready_page() if page_info: return page_info @@ -562,7 +562,7 @@ class AsyncDynamicSession(DynamicSession): :param url: The Target url. :return: A `Response` object. """ - if self._closed: + if self._closed: # pragma: no cover raise RuntimeError("Context manager has been closed") final_response = None @@ -625,6 +625,6 @@ class AsyncDynamicSession(DynamicSession): return response - except Exception as e: + except Exception as e: # pragma: no cover page_info.mark_error() raise e diff --git a/scrapling/engines/static.py b/scrapling/engines/static.py index 9e82774..59f68fa 100644 --- a/scrapling/engines/static.py +++ b/scrapling/engines/static.py @@ -112,7 +112,9 @@ class FetcherSession: kwargs, "impersonate", self.default_impersonate ) - if self.get_with_precedence(kwargs, "http3", self.default_http3): + if self.get_with_precedence( + kwargs, "http3", self.default_http3 + ): # pragma: no cover request_args["http_version"] = CurlHttpVersion.V3ONLY if impersonate: log.warning( @@ -286,7 +288,7 @@ class FetcherSession: response = session.request(method, **request_args) # response.raise_for_status() # Retry responses with a status code between 200-400 return ResponseFactory.from_http_request(response, selector_config) - except CurlError as e: + except CurlError as e: # pragma: no cover if attempt < max_retries - 1: log.error( f"Attempt {attempt + 1} failed: {e}. Retrying in {retry_delay} seconds..." @@ -296,7 +298,7 @@ class FetcherSession: log.error(f"Failed after {max_retries} attempts: {e}") raise # Raise the exception if all retries fail - raise RuntimeError("No active session available.") + raise RuntimeError("No active session available.") # pragma: no cover async def __make_async_request( self, @@ -333,7 +335,7 @@ class FetcherSession: response = await session.request(method, **request_args) # response.raise_for_status() # Retry responses with a status code between 200-400 return ResponseFactory.from_http_request(response, selector_config) - except CurlError as e: + except CurlError as e: # pragma: no cover if attempt < max_retries - 1: log.error( f"Attempt {attempt + 1} failed: {e}. Retrying in {retry_delay} seconds..." @@ -343,7 +345,7 @@ class FetcherSession: log.error(f"Failed after {max_retries} attempts: {e}") raise # Raise the exception if all retries fail - raise RuntimeError("No active session available.") + raise RuntimeError("No active session available.") # pragma: no cover @staticmethod def get_with_precedence(kwargs, key, default_value): diff --git a/scrapling/engines/toolbelt/convertor.py b/scrapling/engines/toolbelt/convertor.py index b7da02a..12bfd55 100644 --- a/scrapling/engines/toolbelt/convertor.py +++ b/scrapling/engines/toolbelt/convertor.py @@ -52,12 +52,12 @@ class ResponseFactory: **parser_arguments, ), ) - except Exception as e: + except Exception as e: # pragma: no cover log.error(f"Error processing redirect: {e}") break current_request = current_request.redirected_from - except Exception as e: + except Exception as e: # pragma: no cover log.error(f"Error processing response history: {e}") return history @@ -105,7 +105,7 @@ class ResponseFactory: history = cls._process_response_history(first_response, parser_arguments) try: page_content = page.content() - except Exception as e: + except Exception as e: # pragma: no cover log.error(f"Error getting page content: {e}") page_content = "" @@ -157,12 +157,12 @@ class ResponseFactory: **parser_arguments, ), ) - except Exception as e: + except Exception as e: # pragma: no cover log.error(f"Error processing redirect: {e}") break current_request = current_request.redirected_from - except Exception as e: + except Exception as e: # pragma: no cover log.error(f"Error processing response history: {e}") return history @@ -212,7 +212,7 @@ class ResponseFactory: ) try: page_content = await page.content() - except Exception as e: + except Exception as e: # pragma: no cover log.error(f"Error getting page content in async: {e}") page_content = "" diff --git a/scrapling/parser.py b/scrapling/parser.py index aa43100..296cb8a 100644 --- a/scrapling/parser.py +++ b/scrapling/parser.py @@ -169,7 +169,9 @@ class Selector(SelectorsGeneration): "Storage class must be wrapped with lru_cache decorator, see docs for info" ) - if not issubclass(storage.__wrapped__, StorageSystemMixin): + if not issubclass( + storage.__wrapped__, StorageSystemMixin + ): # pragma: no cover raise ValueError( "Storage system must be inherited from class `StorageSystemMixin`" ) @@ -1400,7 +1402,7 @@ class Selectors(List[Selector]): """Returns the length of the current list""" return len(self) - def __getstate__(self) -> Any: + def __getstate__(self) -> Any: # pragma: no cover # lxml don't like it :) raise TypeError("Can't pickle Selectors object")