diff --git a/src/fmu/settings/_resources/log_manager.py b/src/fmu/settings/_resources/log_manager.py index 53f5201..bac99c7 100644 --- a/src/fmu/settings/_resources/log_manager.py +++ b/src/fmu/settings/_resources/log_manager.py @@ -1,9 +1,10 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Generic, Self +from datetime import datetime +from typing import TYPE_CHECKING, Generic, Self, get_args, get_origin import pandas as pd -from pydantic import ValidationError +from pydantic import AwareDatetime, ValidationError from fmu.settings._resources.pydantic_resource_manager import PydanticResourceManager from fmu.settings.models._enums import FilterType @@ -48,10 +49,18 @@ def filter_log(self: Self, filter: Filter) -> Log[LogEntryType]: """Filters the log resource with the provided filter.""" if self._cached_dataframe is None: log_model: Log[LogEntryType] = self.load() - df_log = pd.DataFrame([entry.model_dump() for entry in log_model]) - self._cached_dataframe = df_log + if len(log_model) == 0: + self._cached_dataframe = pd.DataFrame() + else: + df_log = pd.DataFrame([entry.model_dump() for entry in log_model]) + self._cached_dataframe = df_log df_log = self._cached_dataframe + self._validate_filter_field(filter) + + if df_log.empty: + return self.model_class([]) + if filter.filter_type == FilterType.text and filter.operator not in { "==", "!=", @@ -96,3 +105,52 @@ def filter_log(self: Self, filter: Filter) -> Log[LogEntryType]: filtered_dict = filtered_df.to_dict("records") return self.model_class.model_validate(filtered_dict) + + def _validate_filter_field(self: Self, filter: Filter) -> None: + """Validate that the filter matches a field on the log entry model.""" + entry_model = self._entry_model_class() + if filter.field_name not in entry_model.model_fields: + raise ValueError( + f"Invalid filter field '{filter.field_name}' when filtering " + f"log resource {self.model_class.__name__}." + ) + + field = entry_model.model_fields[filter.field_name] + expected_filter_type = self._filter_type_for_annotation(field.annotation) + if expected_filter_type is None or filter.filter_type != expected_filter_type: + raise ValueError( + f"Invalid filter type '{filter.filter_type}' applied to field " + f"'{filter.field_name}' when filtering log resource " + f"{self.model_class.__name__}." + ) + + def _entry_model_class(self: Self) -> type[LogEntryType]: + """Return the concrete Pydantic model used for log entries.""" + model_args = self.model_class.__pydantic_generic_metadata__["args"] + return model_args[0] + + @staticmethod + def _filter_type_for_annotation(annotation: object) -> FilterType | None: + """Return the supported filter type for a model field annotation.""" + origin = get_origin(annotation) + if origin is not None: + non_none_annotation_args = [ + arg for arg in get_args(annotation) if arg is not type(None) + ] + if len(non_none_annotation_args) == 1: + return LogManager._filter_type_for_annotation( + non_none_annotation_args[0] + ) + + if annotation in {datetime, AwareDatetime}: + return FilterType.date + if isinstance(annotation, type): + if issubclass(annotation, str): + return FilterType.text + if issubclass(annotation, (int, float)) and not issubclass( + annotation, bool + ): + return FilterType.number + if issubclass(annotation, datetime): + return FilterType.date + return None diff --git a/tests/test_resources/test_log_manager.py b/tests/test_resources/test_log_manager.py index b0ac01d..5aeee0e 100644 --- a/tests/test_resources/test_log_manager.py +++ b/tests/test_resources/test_log_manager.py @@ -1,5 +1,6 @@ """Tests for LogManager.""" +from datetime import UTC, datetime from pathlib import Path from typing import Self @@ -12,6 +13,27 @@ from fmu.settings.models.log import Filter, Log +class MixedLogEntry(BaseModel): + """Log entry with fields for testing all filter types.""" + + user: str = "test_user" + timestamp: datetime = datetime(2026, 1, 1, tzinfo=UTC) + count: int = 1 + + +class MixedLogManager(LogManager[MixedLogEntry]): + """Log manager with fields for testing all filter types.""" + + def __init__(self: Self, fmu_dir: ProjectFMUDirectory) -> None: + """Initializes the mixed log manager.""" + super().__init__(fmu_dir, Log[MixedLogEntry]) + + @property + def relative_path(self: Self) -> Path: + """Returns the relative path to the mixed log file.""" + return Path("logs") / "mixedlog.json" + + def test_log_manager_instantiation(fmu_dir: ProjectFMUDirectory) -> None: """Tests basic facts about the LogManager.""" @@ -94,3 +116,107 @@ def relative_path(self: Self) -> Path: filtered_log = log_manager.filter_log(filter) assert len(filtered_log) == 1 assert all(entry.count >= filter_value for entry in filtered_log) + + +def test_filtering_empty_log_returns_empty_log( + fmu_dir: ProjectFMUDirectory, +) -> None: + """Tests filtering an existing empty log with a valid filter.""" + log_manager = MixedLogManager(fmu_dir=fmu_dir) + log_manager.save(Log[MixedLogEntry]([])) + + filtered_log = log_manager.filter_log( + Filter( + field_name="user", + filter_value="test_user", + filter_type=FilterType.text, + operator="==", + ) + ) + + assert filtered_log.__class__ is log_manager.model_class + assert len(filtered_log) == 0 + + +@pytest.mark.parametrize( + ("filter", "match"), + [ + ( + Filter( + field_name="unknown", + filter_value="test_user", + filter_type=FilterType.text, + operator="==", + ), + "unknown", + ), + ( + Filter( + field_name="user", + filter_value="1", + filter_type=FilterType.number, + operator="==", + ), + "user", + ), + ], +) +def test_filtering_empty_log_with_invalid_filter_raises_value_error( + fmu_dir: ProjectFMUDirectory, filter: Filter, match: str +) -> None: + """Tests filtering an existing empty log with an invalid filter.""" + log_manager = MixedLogManager(fmu_dir=fmu_dir) + log_manager.save(Log[MixedLogEntry]([])) + + with pytest.raises(ValueError, match=match): + log_manager.filter_log(filter) + + +def test_filtering_unknown_field_raises_value_error( + fmu_dir: ProjectFMUDirectory, +) -> None: + """Tests filtering on fields that are not part of the log entry model.""" + log_manager = MixedLogManager(fmu_dir=fmu_dir) + log_manager.add_log_entry(MixedLogEntry()) + + with pytest.raises(ValueError, match="unknown"): + log_manager.filter_log( + Filter( + field_name="unknown", + filter_value="test_user", + filter_type=FilterType.text, + operator="==", + ) + ) + + +@pytest.mark.parametrize( + ("field_name", "filter_value", "filter_type"), + [ + ("user", "10", FilterType.number), + ("user", "2026-01-01T00:00:00+00:00", FilterType.date), + ("timestamp", "2026-01-01T00:00:00+00:00", FilterType.text), + ("timestamp", "10", FilterType.number), + ("count", "1", FilterType.text), + ("count", "2026-01-01T00:00:00+00:00", FilterType.date), + ], +) +def test_filtering_with_incompatible_filter_type_raises_value_error( + fmu_dir: ProjectFMUDirectory, + field_name: str, + filter_value: str, + filter_type: FilterType, +) -> None: + """Tests filter type validation against the selected model field.""" + log_manager = MixedLogManager(fmu_dir=fmu_dir) + log_manager.add_log_entry(MixedLogEntry()) + + with pytest.raises(ValueError, match=field_name): + log_manager.filter_log( + Filter( + field_name=field_name, + filter_value=filter_value, + filter_type=filter_type, + operator=">", + ) + )