fixup! Implement ReceiveComponentDataStream method

Add tests for Sample

Author: llucax

tests/metrics/test_sample.py

@@ -0,0 +1,390 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Tests for the Sample class and related classes."""
+
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any
+from unittest.mock import Mock, patch
+
+import pytest
+from frequenz.api.common.v1.metrics import bounds_pb2, metric_sample_pb2
+from google.protobuf.timestamp_pb2 import Timestamp
+
+from frequenz.client.microgrid.metrics import (
+    AggregatedMetricValue,
+    AggregationMethod,
+    Bounds,
+    Metric,
+    MetricSample,
+)
+from frequenz.client.microgrid.metrics._sample_proto import (
+    aggregated_metric_sample_from_proto,
+    metric_sample_from_proto,
+)
+
+
+@dataclass(frozen=True, kw_only=True)
+class AggregatedValueTestCase:
+    """Test case for AggregatedMetricValue protobuf conversion."""
+
+    name: str
+    """The description of the test case."""
+
+    avg_value: float
+    """The average value to set."""
+
+    has_min: bool = True
+    """Whether to include min value."""
+
+    has_max: bool = True
+    """Whether to include max value."""
+
+    min_value: float | None = None
+    """The minimum value to set."""
+
+    max_value: float | None = None
+    """The maximum value to set."""
+
+    raw_values: list[float] = field(default_factory=list)
+    """The raw values to include."""
+
+
+@dataclass(frozen=True, kw_only=True)
+class MetricSampleTestCase:
+    """Test case for MetricSample protobuf conversion."""
+
+    name: str
+    """The description of the test case."""
+
+    has_simple_value: bool
+    """Whether to include a simple value."""
+
+    has_aggregated_value: bool
+    """Whether to include an aggregated value."""
+
+    has_source: bool
+    """Whether to include a source connection."""
+
+    metric: Metric | int
+    """The metric to set."""
+
+    expect_warning: bool = False
+    """Whether to expect a warning during conversion."""
+
+
+@pytest.fixture
+def now() -> datetime:
+    """Get the current time."""
+    return datetime.now(timezone.utc)
+
+
+def test_aggregation_method_values() -> None:
+    """Test that AggregationMethod enum has the expected values."""
+    assert AggregationMethod.AVG.value == "avg"
+    assert AggregationMethod.MIN.value == "min"
+    assert AggregationMethod.MAX.value == "max"
+
+
+def test_aggregated_metric_value() -> None:
+    """Test AggregatedMetricValue creation and string representation."""
+    # Test with full data
+    value = AggregatedMetricValue(
+        avg=5.0,
+        min=1.0,
+        max=10.0,
+        raw_values=[1.0, 5.0, 10.0],
+    )
+    assert value.avg == 5.0
+    assert value.min == 1.0
+    assert value.max == 10.0
+    assert list(value.raw_values) == [1.0, 5.0, 10.0]
+    assert str(value) == "avg:5.0<min:1.0 max:10.0 num_raw:3>"
+
+    # Test with minimal data (only avg required)
+    value = AggregatedMetricValue(
+        avg=5.0,
+        min=None,
+        max=None,
+        raw_values=[],
+    )
+    assert value.avg == 5.0
+    assert value.min is None
+    assert value.max is None
+    assert not value.raw_values
+    assert str(value) == "avg:5.0"
+
+
+def test_metric_sample_creation(now: datetime) -> None:
+    """Test MetricSample creation with different value types."""
+    bounds = [Bounds(lower=-10.0, upper=10.0)]
+
+    # Test with float value
+    sample = MetricSample(
+        sampled_at=now,
+        metric=Metric.AC_ACTIVE_POWER,
+        value=5.0,
+        bounds=bounds,
+    )
+    assert sample.sampled_at == now
+    assert sample.metric == Metric.AC_ACTIVE_POWER
+    assert sample.value == 5.0
+    assert sample.bounds == bounds
+    assert sample.connection is None
+
+    # Test with AggregatedMetricValue
+    agg_value = AggregatedMetricValue(
+        avg=5.0,
+        min=1.0,
+        max=10.0,
+        raw_values=[1.0, 5.0, 10.0],
+    )
+    sample = MetricSample(
+        sampled_at=now,
+        metric=Metric.AC_ACTIVE_POWER,
+        value=agg_value,
+        bounds=bounds,
+        connection="dc_battery_0",
+    )
+    assert sample.sampled_at == now
+    assert sample.metric == Metric.AC_ACTIVE_POWER
+    assert sample.value == agg_value
+    assert sample.bounds == bounds
+    assert sample.connection == "dc_battery_0"
+
+    # Test with None value
+    sample = MetricSample(
+        sampled_at=now,
+        metric=Metric.AC_ACTIVE_POWER,
+        value=None,
+        bounds=bounds,
+    )
+    assert sample.value is None
+
+
+def test_metric_sample_as_single_value(now: datetime) -> None:
+    """Test MetricSample.as_single_value with different value types and methods."""
+    bounds = [Bounds(lower=-10.0, upper=10.0)]
+    test_cases: list[tuple[Any, dict[AggregationMethod, float | None]]] = [
+        # (value, {method: expected_result})
+        (
+            5.0,
+            {
+                AggregationMethod.AVG: 5.0,
+                AggregationMethod.MIN: 5.0,
+                AggregationMethod.MAX: 5.0,
+            },
+        ),
+        (
+            AggregatedMetricValue(
+                avg=5.0,
+                min=1.0,
+                max=10.0,
+                raw_values=[1.0, 5.0, 10.0],
+            ),
+            {
+                AggregationMethod.AVG: 5.0,
+                AggregationMethod.MIN: 1.0,
+                AggregationMethod.MAX: 10.0,
+            },
+        ),
+        (
+            None,
+            {
+                AggregationMethod.AVG: None,
+                AggregationMethod.MIN: None,
+                AggregationMethod.MAX: None,
+            },
+        ),
+    ]
+
+    for value, method_results in test_cases:
+        sample = MetricSample(
+            sampled_at=now,
+            metric=Metric.AC_ACTIVE_POWER,
+            value=value,
+            bounds=bounds,
+        )
+
+        for method, expected in method_results.items():
+            assert sample.as_single_value(aggregation_method=method) == expected
+
+
+def test_metric_sample_multiple_bounds(now: datetime) -> None:
+    """Test MetricSample creation with multiple bounds."""
+    bounds = [
+        Bounds(lower=-10.0, upper=-5.0),
+        Bounds(lower=5.0, upper=10.0),
+    ]
+    sample = MetricSample(
+        sampled_at=now,
+        metric=Metric.AC_ACTIVE_POWER,
+        value=7.0,
+        bounds=bounds,
+    )
+    assert sample.bounds == bounds
+
+
+@pytest.mark.parametrize(
+    "case",
+    [
+        AggregatedValueTestCase(
+            name="complete data",
+            avg_value=5.0,
+            min_value=1.0,
+            max_value=10.0,
+            raw_values=[1.0, 5.0, 10.0],
+        ),
+        AggregatedValueTestCase(
+            name="minimal data",
+            avg_value=5.0,
+            has_min=False,
+            has_max=False,
+        ),
+        AggregatedValueTestCase(
+            name="only min additional",
+            avg_value=5.0,
+            has_max=False,
+            min_value=1.0,
+        ),
+        AggregatedValueTestCase(
+            name="only max additional",
+            avg_value=5.0,
+            has_min=False,
+            max_value=10.0,
+        ),
+    ],
+    ids=lambda case: case.name,
+)
+def test_aggregated_metric_value_from_proto(case: AggregatedValueTestCase) -> None:
+    """Test conversion from protobuf message to AggregatedMetricValue.
+
+    Args:
+        case: Test case parameters
+
+    Raises:
+        AssertionError: If the conversion doesn't match expected values
+    """
+    proto = metric_sample_pb2.AggregatedMetricValue(
+        avg_value=case.avg_value,
+    )
+    if case.has_min and case.min_value is not None:
+        proto.min_value = case.min_value
+    if case.has_max and case.max_value is not None:
+        proto.max_value = case.max_value
+    if case.raw_values:
+        proto.raw_values.extend(case.raw_values)
+
+    value = aggregated_metric_sample_from_proto(proto)
+
+    assert value.avg == case.avg_value
+    assert value.min == (case.min_value if case.has_min else None)
+    assert value.max == (case.max_value if case.has_max else None)
+    assert list(value.raw_values) == case.raw_values
+
+
+@pytest.mark.parametrize(
+    "case",
+    [
+        MetricSampleTestCase(
+            name="simple value",
+            has_simple_value=True,
+            has_aggregated_value=False,
+            has_source=False,
+            metric=Metric.AC_ACTIVE_POWER,
+        ),
+        MetricSampleTestCase(
+            name="aggregated value",
+            has_simple_value=False,
+            has_aggregated_value=True,
+            has_source=True,
+            metric=Metric.AC_ACTIVE_POWER,
+        ),
+        MetricSampleTestCase(
+            name="no value",
+            has_simple_value=False,
+            has_aggregated_value=False,
+            has_source=False,
+            metric=Metric.AC_ACTIVE_POWER,
+        ),
+        MetricSampleTestCase(
+            name="unrecognized metric",
+            has_simple_value=True,
+            has_aggregated_value=False,
+            has_source=False,
+            metric=999,
+            expect_warning=True,
+        ),
+    ],
+    ids=lambda case: case.name,
+)
+@patch("frequenz.client.microgrid.metrics._sample_proto.bounds_from_proto")
+@patch("frequenz.client.microgrid.metrics._sample_proto.enum_from_proto")
+@patch("frequenz.client.base.conversion.to_datetime")
+def test_metric_sample_from_proto(
+    mock_to_datetime: Mock,
+    mock_enum_from_proto: Mock,
+    mock_bounds_from_proto: Mock,
+    case: MetricSampleTestCase,
+) -> None:
+    """Test conversion from protobuf message to MetricSample.
+
+    Args:
+        mock_to_datetime: Mock for timestamp conversion
+        mock_enum_from_proto: Mock for enum conversion
+        mock_bounds_from_proto: Mock for bounds conversion
+        case: Test case parameters
+
+    Raises:
+        AssertionError: If the conversion doesn't match expected values
+    """
+    now = datetime.now(timezone.utc)
+    mock_to_datetime.return_value = now
+
+    if isinstance(case.metric, Metric):
+        mock_enum_from_proto.return_value = case.metric
+    else:
+        mock_enum_from_proto.return_value = case.metric
+
+    mock_bounds_from_proto.return_value = bounds_pb2.Bounds(lower=-10.0, upper=10.0)
+
+    timestamp = Timestamp()
+    timestamp.FromDatetime(now)
+
+    metric_value = case.metric.value if isinstance(case.metric, Metric) else case.metric
+    proto = metric_sample_pb2.MetricSample(
+        sampled_at=timestamp,
+        metric=metric_value,  # type: ignore[arg-type]
+    )
+
+    if case.has_simple_value:
+        proto.value.simple_metric.value = 5.0
+    elif case.has_aggregated_value:
+        proto.value.aggregated_metric.avg_value = 5.0
+        proto.value.aggregated_metric.min_value = 1.0
+        proto.value.aggregated_metric.max_value = 10.0
+
+    if case.has_source:
+        proto.source = "dc_battery_0"
+
+    proto.bounds.append(bounds_pb2.Bounds(lower=-10.0, upper=10.0))
+
+    sample = metric_sample_from_proto(proto)
+
+    assert sample.sampled_at == now
+    assert sample.metric == case.metric
+    if case.has_simple_value:
+        assert isinstance(sample.value, float)
+        assert sample.value == 5.0
+    elif case.has_aggregated_value:
+        assert isinstance(sample.value, AggregatedMetricValue)
+        assert sample.value.avg == 5.0
+        assert sample.value.min == 1.0
+        assert sample.value.max == 10.0
+    else:
+        assert sample.value is None
+
+    mock_to_datetime.assert_called_once()
+    mock_enum_from_proto.assert_called_once()
+    mock_bounds_from_proto.assert_called_once()