Skip to content

Commit

Permalink
Refactoring: move keyword matches to helpers.py
Browse files Browse the repository at this point in the history
  • Loading branch information
HardNorth committed Jan 10, 2025
1 parent 5c4d5b8 commit dfbe52e
Show file tree
Hide file tree
Showing 2 changed files with 102 additions and 89 deletions.
111 changes: 87 additions & 24 deletions robotframework_reportportal/helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,31 +17,10 @@
import binascii
import fnmatch
import re
from typing import Iterable, Optional, Tuple
from abc import ABC, abstractmethod
from typing import Callable, Iterable, Optional, Tuple


def replace_patterns(text: str, patterns: Iterable[Tuple[re.Pattern, str]]) -> str:
"""Replace given patterns in the text."""
result = text
for p, repl in patterns:
result = p.sub(repl, result)
return result


BARE_LINK_PATTERN = re.compile(r"\[\s*([^]|]+)]")
NAMED_LINK_PATTERN = re.compile(r"\[\s*([^]|]+)\|\s*([^]]+)]")

ROBOT_MARKUP_REPLACEMENT_PATTERS = [
(BARE_LINK_PATTERN, r"<\1>"),
(NAMED_LINK_PATTERN, r"[\2](\1)"),
]

PATTERN_MATCHES_EMPTY_STRING: re.Pattern = re.compile("^$")


def robot_markup_to_markdown(text: str) -> str:
"""Convert Robot Framework's text markup to Markdown format."""
return replace_patterns(text, ROBOT_MARKUP_REPLACEMENT_PATTERS)
from robotframework_reportportal.model import Keyword


def translate_glob_to_regex(pattern: Optional[str]) -> Optional[re.Pattern]:
Expand Down Expand Up @@ -72,6 +51,90 @@ def match_pattern(pattern: Optional[re.Pattern], line: Optional[str]) -> bool:
return pattern.fullmatch(line) is not None


class KeywordMatch(ABC):
"""Base class for keyword matchers."""

@abstractmethod
def match(self, kw: Keyword) -> bool:
"""Check if the keyword matches the criteria."""


class KeywordEqual(KeywordMatch):
"""Match keyword based on a predicate."""

predicate: Callable[[Keyword], bool]

def __init__(self, predicate: Callable[[Keyword], bool] = None) -> None:
"""Initialize the matcher with the predicate."""
self.predicate = predicate

def match(self, kw: Keyword) -> bool:
"""Check if the keyword matches the criteria."""
return self.predicate(kw)


class KeywordNameMatch(KeywordEqual):
"""Match keyword based on the name pattern."""

def __init__(self, pattern: Optional[str]) -> None:
"""Initialize the matcher with the pattern."""
super().__init__(lambda kw: match_pattern(pattern, kw.name))


class KeywordTypeEqual(KeywordEqual):
"""Match keyword based on the type."""

def __init__(self, expected_value: Optional[str]) -> None:
"""Initialize the matcher with the expected value."""
super().__init__(lambda kw: kw.keyword_type == expected_value)


class KeywordTagMatch(KeywordMatch):
"""Match keyword based on the tag pattern."""

pattern: Optional[re.Pattern]

def __init__(self, pattern: Optional[str]) -> None:
"""Initialize the matcher with the pattern."""
self.pattern = translate_glob_to_regex(pattern)

def match(self, kw: Keyword) -> bool:
"""Check if the keyword matches the criteria."""
return next((True for t in kw.tags if match_pattern(self.pattern, t)), False)


class KeywordStatusEqual(KeywordEqual):
"""Match keyword based on the status."""

def __init__(self, status: str) -> None:
"""Initialize the matcher with the status."""
super().__init__(lambda kw: kw.status == status)


def replace_patterns(text: str, patterns: Iterable[Tuple[re.Pattern, str]]) -> str:
"""Replace given patterns in the text."""
result = text
for p, repl in patterns:
result = p.sub(repl, result)
return result


BARE_LINK_PATTERN = re.compile(r"\[\s*([^]|]+)]")
NAMED_LINK_PATTERN = re.compile(r"\[\s*([^]|]+)\|\s*([^]]+)]")

ROBOT_MARKUP_REPLACEMENT_PATTERS = [
(BARE_LINK_PATTERN, r"<\1>"),
(NAMED_LINK_PATTERN, r"[\2](\1)"),
]

PATTERN_MATCHES_EMPTY_STRING: re.Pattern = re.compile("^$")


def robot_markup_to_markdown(text: str) -> str:
"""Convert Robot Framework's text markup to Markdown format."""
return replace_patterns(text, ROBOT_MARKUP_REPLACEMENT_PATTERS)


def _unescape(binary_string: str, stop_at: int = -1):
result = bytearray()
join_list = list()
Expand Down
80 changes: 15 additions & 65 deletions robotframework_reportportal/listener.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,15 +18,20 @@
import os
import re
import uuid
from abc import ABC, abstractmethod
from functools import wraps
from mimetypes import guess_type
from typing import Any, Callable, Dict, List, Optional, Union
from typing import Any, Dict, List, Optional, Union
from warnings import warn

from reportportal_client.helpers import LifoQueue, guess_content_type_from_bytes, is_binary

from robotframework_reportportal.helpers import _unescape, match_pattern, translate_glob_to_regex
from robotframework_reportportal.helpers import (
KeywordMatch,
KeywordNameMatch,
KeywordTagMatch,
KeywordTypeEqual,
_unescape,
)
from robotframework_reportportal.model import Entity, Keyword, Launch, LogMessage, Suite, Test
from robotframework_reportportal.service import RobotService
from robotframework_reportportal.static import MAIN_SUITE_ID, PABOT_WITHOUT_LAUNCH_ID_MSG
Expand All @@ -49,6 +54,10 @@
FOR_KEYWORD_TYPE = "FOR"
WHILE_KEYWORD_TYPE = "WHILE"

WUKS_KEYWORD_MATCH = KeywordNameMatch(WUKS_KEYWORD_NAME)
FOR_KEYWORD_MATCH = KeywordTypeEqual(FOR_KEYWORD_TYPE)
WHILE_KEYWORD_NAME = KeywordTypeEqual(WHILE_KEYWORD_TYPE)


def check_rp_enabled(func):
"""Verify is RP is enabled in config."""
Expand All @@ -63,73 +72,14 @@ def wrap(*args, **kwargs):
return wrap


class _KeywordMatch(ABC):
@abstractmethod
def match(self, kw: Keyword) -> bool: ...


class _KeywordFieldEqual(_KeywordMatch):
expected_value: Optional[str]
extract_func: Callable[[Keyword], str]

def __init__(self, expected_value: Optional[str], extract_func: Callable[[Keyword], str] = None) -> None:
self.expected_value = expected_value
self.extract_func = extract_func

def match(self, kw: Keyword) -> bool:
return self.extract_func(kw) == self.expected_value


class _KeywordPatternMatch(_KeywordMatch):
pattern: Optional[re.Pattern]
extract_func: Optional[Callable[[Keyword], str]]

def __init__(self, pattern: Optional[str], extract_func: Callable[[Keyword], str] = None):
self.pattern = translate_glob_to_regex(pattern)
self.extract_func = extract_func

def match(self, kw: Keyword) -> bool:
return match_pattern(self.pattern, self.extract_func(kw))


class _KeywordNameMatch(_KeywordPatternMatch):
def __init__(self, pattern: Optional[str]) -> None:
super().__init__(pattern, lambda kw: kw.name)


class _KeywordTypeEqual(_KeywordFieldEqual):
def __init__(self, expected_value: Optional[str]) -> None:
super().__init__(expected_value, lambda kw: kw.keyword_type)


class _KeywordTagMatch(_KeywordMatch):
pattern: Optional[re.Pattern]

def __init__(self, pattern: Optional[str]) -> None:
self.pattern = translate_glob_to_regex(pattern)

def match(self, kw: Keyword) -> bool:
return next((True for t in kw.tags if match_pattern(self.pattern, t)), False)


class _KeywordStatusEqual(_KeywordFieldEqual):
def __init__(self, status: str) -> None:
super().__init__(status, lambda kw: kw.status)


WUKS_KEYWORD_MATCH = _KeywordNameMatch(WUKS_KEYWORD_NAME)
FOR_KEYWORD_MATCH = _KeywordTypeEqual(FOR_KEYWORD_TYPE)
WHILE_KEYWORD_NAME = _KeywordTypeEqual(WHILE_KEYWORD_TYPE)


# noinspection PyPep8Naming
class listener:
"""Robot Framework listener interface for reporting to ReportPortal."""

_items: LifoQueue[Union[Keyword, Launch, Suite, Test]]
_service: Optional[RobotService]
_variables: Optional[Variables]
_remove_keyword_filters: List[_KeywordMatch] = []
_remove_keyword_filters: List[KeywordMatch] = []
_remove_all_keyword_content: bool = False
_remove_data_passed_tests: bool = False
ROBOT_LISTENER_API_VERSION = 2
Expand Down Expand Up @@ -360,9 +310,9 @@ def _process_keyword_skip(self):
pattern_type, pattern = pattern_str.split(":", 1)
pattern_type = pattern_type.strip().upper()
if "NAME" == pattern_type.upper():
self._remove_keyword_filters.append(_KeywordNameMatch(pattern.strip()))
self._remove_keyword_filters.append(KeywordNameMatch(pattern.strip()))
elif "TAG" == pattern_type.upper():
self._remove_keyword_filters.append(_KeywordTagMatch(pattern.strip()))
self._remove_keyword_filters.append(KeywordTagMatch(pattern.strip()))
except ImportError:
warn('Unable to locate Robot Framework context. "--remove-keywords" feature will not work.', stacklevel=2)

Expand Down

0 comments on commit dfbe52e

Please sign in to comment.