fixup! Implement ListConnections

Add tests for ComponentConnection.

Author: llucax

tests/component/test_connection.py

@@ -0,0 +1,246 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Tests for ComponentConnection class and related functionality."""
+
+from datetime import datetime, timedelta, timezone
+from typing import Any
+from unittest.mock import Mock, PropertyMock, patch
+
+import pytest
+from frequenz.api.common.v1.microgrid import lifetime_pb2
+from frequenz.api.common.v1.microgrid.components import components_pb2
+from google.protobuf import timestamp_pb2
+
+from frequenz.client.microgrid import ComponentId, Lifetime
+from frequenz.client.microgrid.component import ComponentConnection
+from frequenz.client.microgrid.component._connection_proto import (
+    component_connection_from_proto,
+)
+
+
+def test_component_connection_construction() -> None:
+    """Test basic ComponentConnection construction and validation.
+
+    Raises:
+        AssertionError: If any of the assertions fail.
+    """
+    now = datetime.now(timezone.utc)
+    lifetime = Lifetime(start=now)
+    connection = ComponentConnection(
+        source=ComponentId(1),
+        destination=ComponentId(2),
+        operational_lifetime=lifetime,
+    )
+
+    assert connection.source == ComponentId(1)
+    assert connection.destination == ComponentId(2)
+    assert connection.operational_lifetime == lifetime
+
+
+def test_component_connection_validation() -> None:
+    """Test validation of source and destination components."""
+    with pytest.raises(
+        ValueError, match="Source and destination components must be different"
+    ):
+        ComponentConnection(
+            source=ComponentId(1),
+            destination=ComponentId(1),
+        )
+
+
+@pytest.mark.parametrize(
+    "lifetime_active",
+    [True, False],
+)
+def test_component_connection_active_property(lifetime_active: bool) -> None:
+    """Test the active property of ComponentConnection.
+
+    Args:
+        lifetime_active: Whether the lifetime should be active or not.
+
+    Raises:
+        AssertionError: If any of the assertions fail.
+    """
+    mock_lifetime = Mock(spec=Lifetime)
+    mock_lifetime.active_at.return_value = lifetime_active
+
+    connection = ComponentConnection(
+        source=ComponentId(1),
+        destination=ComponentId(2),
+        operational_lifetime=mock_lifetime,
+    )
+
+    test_time = datetime.now(timezone.utc)
+    assert connection.active_at(test_time) is lifetime_active
+    mock_lifetime.active_at.assert_called_once_with(test_time)
+
+    # Test that the active property uses active_at with the current time
+    with patch("datetime.datetime") as mock_datetime:
+        mock_datetime.now.return_value = test_time
+        assert connection.active is lifetime_active
+
+
+def test_component_connection_string_representation() -> None:
+    """Test string representation of ComponentConnection.
+
+    Raises:
+        AssertionError: If the string representation is incorrect.
+    """
+    connection = ComponentConnection(
+        source=ComponentId(1),
+        destination=ComponentId(2),
+    )
+    assert str(connection) == "CID1->CID2"
+
+
+@pytest.mark.parametrize(
+    "proto_data",
+    [
+        pytest.param(
+            {
+                "source_component_id": 1,
+                "destination_component_id": 2,
+                "has_lifetime": True,
+            },
+            id="complete connection",
+        ),
+        pytest.param(
+            {
+                "source_component_id": 1,
+                "destination_component_id": 2,
+                "has_lifetime": False,
+            },
+            id="missing lifetime",
+        ),
+    ],
+)
+def test_component_connection_from_proto(
+    proto_data: dict[str, Any],
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test conversion from protobuf message to ComponentConnection.
+
+    Args:
+        proto_data: Test case parameters
+        caplog: Fixture to capture log messages
+
+    Raises:
+        AssertionError: If any of the assertions fail.
+    """
+    caplog.set_level("DEBUG")  # Ensure we capture DEBUG level messages
+
+    proto = components_pb2.ComponentConnection(
+        source_component_id=proto_data["source_component_id"],
+        destination_component_id=proto_data["destination_component_id"],
+    )
+
+    if proto_data["has_lifetime"]:
+        now = datetime.now(timezone.utc)
+        start_time = timestamp_pb2.Timestamp()
+        start_time.FromDatetime(now)
+        lifetime = lifetime_pb2.Lifetime()
+        lifetime.start_timestamp.CopyFrom(start_time)
+        proto.operational_lifetime.CopyFrom(lifetime)
+
+    connection = component_connection_from_proto(proto)
+
+    assert connection.source == ComponentId(proto_data["source_component_id"])
+    assert connection.destination == ComponentId(proto_data["destination_component_id"])
+
+    if not proto_data["has_lifetime"]:
+        assert "missing operational lifetime" in caplog.text.lower()
+
+
+def test_component_connection_from_proto_same_ids() -> None:
+    """Test proto conversion with same source and destination IDs."""
+    proto = components_pb2.ComponentConnection(
+        source_component_id=1,
+        destination_component_id=1,
+    )
+
+    with pytest.raises(
+        ValueError, match="Source and destination components must be different"
+    ):
+        component_connection_from_proto(proto)
+
+
+@patch("frequenz.client.microgrid.component._connection_proto.lifetime_from_proto")
+def test_component_connection_from_proto_invalid_lifetime(
+    mock_lifetime_from_proto: Mock,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test proto conversion with invalid lifetime data.
+
+    Args:
+        mock_lifetime_from_proto: Mock for lifetime conversion
+        caplog: Fixture to capture log messages
+
+    Raises:
+        AssertionError: If any of the assertions fail.
+    """
+    mock_lifetime_from_proto.side_effect = ValueError("Invalid lifetime")
+
+    proto = components_pb2.ComponentConnection(
+        source_component_id=1,
+        destination_component_id=2,
+    )
+    now = datetime.now(timezone.utc)
+    start_time = timestamp_pb2.Timestamp()
+    start_time.FromDatetime(now)
+    lifetime = lifetime_pb2.Lifetime()
+    lifetime.start_timestamp.CopyFrom(start_time)
+    proto.operational_lifetime.CopyFrom(lifetime)
+
+    connection = component_connection_from_proto(proto)
+
+    assert connection.source == ComponentId(1)
+    assert connection.destination == ComponentId(2)
+    assert "invalid operational lifetime" in caplog.text.lower()
+    mock_lifetime_from_proto.assert_called_once_with(proto.operational_lifetime)
+
+
+@pytest.mark.parametrize(
+    "case",
+    [
+        pytest.param(
+            {"test_time": "past", "expected": True},
+            id="past timestamp",
+        ),
+        pytest.param(
+            {"test_time": "now", "expected": True},
+            id="current timestamp",
+        ),
+        pytest.param(
+            {"test_time": "future", "expected": True},
+            id="future timestamp",
+        ),
+    ],
+)
+def test_component_connection_active_at(case: dict[str, Any]) -> None:
+    """Test checking activity at specific timestamps.
+
+    Args:
+        case: Test case parameters
+
+    Raises:
+        AssertionError: If any of the assertions fail.
+    """
+    mock_lifetime = Mock(spec=Lifetime)
+    mock_lifetime.active_at = Mock(return_value=case["expected"])
+
+    connection = ComponentConnection(
+        source=ComponentId(1),
+        destination=ComponentId(2),
+        operational_lifetime=mock_lifetime,
+    )
+
+    now = datetime.now(timezone.utc)
+    test_time = {
+        "past": now - timedelta(hours=1),
+        "now": now,
+        "future": now + timedelta(hours=1),
+    }[case["test_time"]]
+
+    assert connection.active_at(test_time) == case["expected"]
+    mock_lifetime.active_at.assert_called_once_with(test_time)