BUG: Handle invalid filters consistently

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

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=">",
+            )
+        )