From eb04e801232edc23d576604a73666798140cd326 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rg=20Zimmermann?= Date: Sun, 2 Feb 2025 18:17:15 +0100 Subject: [PATCH] fix typing --- logprep/processor/base/rule.py | 23 ------- logprep/util/rule_loader.py | 6 +- .../processor/labeler/test_labeler_rule.py | 14 ----- tests/unit/processor/test_processor_rule.py | 63 ------------------- 4 files changed, 3 insertions(+), 103 deletions(-) delete mode 100644 tests/unit/processor/test_processor_rule.py diff --git a/logprep/processor/base/rule.py b/logprep/processor/base/rule.py index acbdb9137..799bcf2ab 100644 --- a/logprep/processor/base/rule.py +++ b/logprep/processor/base/rule.py @@ -127,9 +127,7 @@ """ import hashlib -import json from functools import cached_property -from os.path import basename, splitext from typing import Dict, List, Optional, Set from attrs import define, field, validators @@ -140,7 +138,6 @@ from logprep.filter.lucene_filter import LuceneFilter from logprep.metrics.metrics import CounterMetric, HistogramMetric from logprep.processor.base.exceptions import InvalidRuleDefinitionError -from logprep.util.getter import GetterFactory from logprep.util.helper import camel_to_snake yaml = YAML(typ="safe", pure=True) @@ -318,26 +315,6 @@ def lucene_filter(self): # pylint: enable=C0111 - @classmethod - def create_rules_from_target(cls, rule_target: str, processor_name: str = None) -> list: - """Create a rule from a file.""" - if isinstance(rule_target, dict): - return [cls._create_from_dict(rule_target, processor_name)] - content = GetterFactory.from_string(rule_target).get() - try: - rule_data = json.loads(content) - except ValueError: - rule_data = yaml.load_all(content) - try: - rules = [cls._create_from_dict(rule, processor_name) for rule in rule_data] - except InvalidRuleDefinitionError as error: - raise InvalidRuleDefinitionError(f"{rule_target}: {error}") from error - if len(rules) == 0: - raise InvalidRuleDefinitionError(f"no rules in file {rule_target}") - for rule in rules: - rule.file_name = splitext(basename(rule_target))[0] - return rules - @classmethod def normalize_rule_dict(cls, rule: dict) -> None: """normalizes rule dict before create rule config object diff --git a/logprep/util/rule_loader.py b/logprep/util/rule_loader.py index 75d394d02..6b7de4669 100644 --- a/logprep/util/rule_loader.py +++ b/logprep/util/rule_loader.py @@ -31,7 +31,7 @@ class DirectoryRuleLoader(RuleLoader): Returns the list of rules loaded from the directory. """ - def __init__(self, directory: str, rule_class: type = Rule): + def __init__(self, directory: str, rule_class: Rule = Rule): self.source = directory self.rule_class = rule_class @@ -65,7 +65,7 @@ class FileRuleLoader(RuleLoader): Returns the list of rules loaded from the file. """ - def __init__(self, file: str, rule_class: type = Rule): + def __init__(self, file: str, rule_class: Rule = Rule): self.rule_class = rule_class self.source = file @@ -98,7 +98,7 @@ class ListRuleLoader(RuleLoader): Returns a list of Rule objects created from the source list of dictionaries. """ - def __init__(self, rules: List[dict], rule_class: type = Rule): + def __init__(self, rules: List[dict], rule_class: Rule = Rule): self.rule_class = rule_class self.source = rules diff --git a/tests/unit/processor/labeler/test_labeler_rule.py b/tests/unit/processor/labeler/test_labeler_rule.py index aade96836..fb1b2d608 100644 --- a/tests/unit/processor/labeler/test_labeler_rule.py +++ b/tests/unit/processor/labeler/test_labeler_rule.py @@ -11,7 +11,6 @@ from logprep.processor.base.exceptions import InvalidRuleDefinitionError from logprep.processor.labeler.labeling_schema import LabelingSchema from logprep.processor.labeler.rule import LabelerRule -from tests.testdata.FilledTempFile import JsonTempFile from tests.testdata.ruledata import simple_rule_dict @@ -30,19 +29,6 @@ def get_parent_labels(self, category, label): class TestRule: - def test_create_from_file_fails_if_document_does_not_contain_filter_and_label(self): - with pytest.raises(InvalidRuleDefinitionError): - with JsonTempFile({}) as rule_path: - LabelerRule.create_rules_from_target(rule_path) - - for missing_field in ["filter", "labeler"]: - invalid_rule_dict = deepcopy(simple_rule_dict) - del invalid_rule_dict[missing_field] - - with pytest.raises(InvalidRuleDefinitionError): - with JsonTempFile([invalid_rule_dict]) as rule_path: - LabelerRule.create_rules_from_target(rule_path) - def test_create_from_dict_creates_expected_rule(self): rule_definition = { "filter": 'applyrule: "yes"', diff --git a/tests/unit/processor/test_processor_rule.py b/tests/unit/processor/test_processor_rule.py deleted file mode 100644 index a6f7ca73d..000000000 --- a/tests/unit/processor/test_processor_rule.py +++ /dev/null @@ -1,63 +0,0 @@ -# pylint: disable=missing-docstring -# pylint: disable=protected-access -from unittest import mock - -import pytest - -from logprep.processor.base.exceptions import InvalidRuleDefinitionError -from logprep.processor.base.rule import Rule - - -class TestRule: - @pytest.mark.parametrize( - "file_data, raises", - [ - ( - b""" - filter: test_filter - rule: - regex_fields: [] - description: this is a test rule - """, - None, - ), - ( - b""" - filter: - rule: - regex_fields: [] - description: this is a test rule - """, - (InvalidRuleDefinitionError, "no filter defined"), - ), - ( - b""" - [ - { - "filter": "test_filter", - "rule": {} - } - ] - """, - None, - ), - ( - b""" - """, - (InvalidRuleDefinitionError, "no rules in file"), - ), - ], - ) - def test_create_rules_from_file(self, file_data, raises): - mock_open = mock.mock_open(read_data=file_data) - with mock.patch("pathlib.Path.open", mock_open): - if raises: - error, message = raises - with pytest.raises(error, match=message): - rules = Rule.create_rules_from_target("mock_path.json") - else: - rules = Rule.create_rules_from_target("mock_path.json") - assert isinstance(rules, list) - assert isinstance(rules[0], Rule) - rule = rules[0] - assert "test_filter" in rule.filter_str