diff --git a/elementary/monitor/data_monitoring/schema.py b/elementary/monitor/data_monitoring/schema.py index abe556635..b4d4f2cf8 100644 --- a/elementary/monitor/data_monitoring/schema.py +++ b/elementary/monitor/data_monitoring/schema.py @@ -31,6 +31,7 @@ class ResourceType(str, Enum): class FilterType(str, Enum): IS = "is" IS_NOT = "is_not" + CONTAINS = "contains" def apply_filter(filter_type: FilterType, value: Any, filter_value: Any) -> bool: @@ -38,13 +39,15 @@ def apply_filter(filter_type: FilterType, value: Any, filter_value: Any) -> bool return value == filter_value elif filter_type == FilterType.IS_NOT: return value != filter_value + elif filter_type == FilterType.CONTAINS: + return str(filter_value).lower() in str(value).lower() raise ValueError(f"Unsupported filter type: {filter_type}") ValueT = TypeVar("ValueT") -ANY_OPERATORS = [FilterType.IS] +ANY_OPERATORS = [FilterType.IS, FilterType.CONTAINS] ALL_OPERATORS = [FilterType.IS_NOT] diff --git a/tests/unit/monitor/data_monitoring/test_filter_schema.py b/tests/unit/monitor/data_monitoring/test_filter_schema.py index 41df630cd..34b7de270 100644 --- a/tests/unit/monitor/data_monitoring/test_filter_schema.py +++ b/tests/unit/monitor/data_monitoring/test_filter_schema.py @@ -44,3 +44,30 @@ def test_filter_schema_apply_filter_on_values_is_not_operator(): def test_filter_schema_invalid_filter_type(): with pytest.raises(ValueError): FilterSchema(values=["test1"], type="invalid") # type: ignore[arg-type] + + +def test_filter_schema_contains_operator(): + filter_schema = FilterSchema(values=["test"], type=FilterType.CONTAINS) + + # Should match when value contains the filter value + assert filter_schema.apply_filter_on_value("test123") is True + assert filter_schema.apply_filter_on_value("123test") is True + assert filter_schema.apply_filter_on_value("123test456") is True + + # Should match case-insensitive + assert filter_schema.apply_filter_on_value("TEST123") is True + assert filter_schema.apply_filter_on_value("123TEST") is True + + # Should not match when value doesn't contain filter value + assert filter_schema.apply_filter_on_value("123") is False + + +def test_filter_schema_apply_filter_on_values_contains_operator(): + filter_schema = FilterSchema(values=["test1", "test2"], type=FilterType.CONTAINS) + + # Should match when any value contains any filter value + assert filter_schema.apply_filter_on_values(["abc_test1_def", "xyz"]) is True + assert filter_schema.apply_filter_on_values(["abc", "xyz_test2"]) is True + + # Should not match when no values contain any filter values + assert filter_schema.apply_filter_on_values(["abc", "xyz"]) is False