Skip to content

Commit 4bbbe3c

Browse files
ela-kotulska-frequenzela-kotulska-frequenz
committed
Add wrappers for the ComponentData in Unit tests
In the unit tests we have to mock component data with only few parameters. Current ComponentData definition doesn't allow for it because we have to give all class parameters. We used to avoid that by creating protobuf message with only necessary data and use it to create ComponentData. But this approach is complex and not resilient for the changes in the protobuf api. This commit wraps ComponentData. Each parameter has default protobuf value that will be used if no argument was given. Signed-off-by: ela-kotulska-frequenz <[email protected]>
1 parent a7aedc8 commit 4bbbe3c

File tree

3 files changed

+186
-112
lines changed

3 files changed

+186
-112
lines changed

‎tests/actor/test_battery_status.py‎

Lines changed: 14 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -7,24 +7,17 @@
77
from datetime import datetime, timedelta, timezone
88
from typing import Generic, Iterable, List, Optional, Set, Tuple, TypeVar
99

10-
import frequenz.api.microgrid.microgrid_pb2 as microgrid_pb
1110
import pytest
12-
import pytz
1311
import time_machine
14-
from frequenz.api.microgrid.battery_pb2 import Battery as PbBattery
1512
from frequenz.api.microgrid.battery_pb2 import ComponentState as BatteryState
1613
from frequenz.api.microgrid.battery_pb2 import Error as BatteryError
1714
from frequenz.api.microgrid.battery_pb2 import ErrorCode as BatteryErrorCode
1815
from frequenz.api.microgrid.battery_pb2 import RelayState as BatteryRelayState
19-
from frequenz.api.microgrid.battery_pb2 import State as PbBatteryState
2016
from frequenz.api.microgrid.common_pb2 import ErrorLevel
2117
from frequenz.api.microgrid.inverter_pb2 import ComponentState as InverterState
2218
from frequenz.api.microgrid.inverter_pb2 import Error as InverterError
2319
from frequenz.api.microgrid.inverter_pb2 import ErrorCode as InverterErrorCode
24-
from frequenz.api.microgrid.inverter_pb2 import Inverter as PbInverter
25-
from frequenz.api.microgrid.inverter_pb2 import State as PbInverterState
2620
from frequenz.channels import Broadcast
27-
from google.protobuf.timestamp_pb2 import Timestamp # pylint: disable=no-name-in-module
2821
from pytest_mock import MockerFixture
2922

3023
from frequenz.sdk.actor.power_distributing._battery_status import (
@@ -40,6 +33,7 @@
4033
InverterData,
4134
)
4235

36+
from ..utils.component_data_wrapper import BatteryDataWrapper, InverterDataWrapper
4337
from ..utils.mock_microgrid import MockMicrogridClient
4438

4539

@@ -69,29 +63,14 @@ def battery_data(
6963
Returns:
7064
BatteryData with given arguments.
7165
"""
72-
if timestamp is None:
73-
timestamp = datetime.now(tz=pytz.utc)
74-
75-
# Create protobuf message first. Thanks to that I don't have to specify
76-
# all class variables.
77-
pb_timestamp = Timestamp()
78-
pb_timestamp.FromDatetime(timestamp)
79-
80-
pb_data = microgrid_pb.ComponentData(
81-
id=component_id,
82-
ts=pb_timestamp,
83-
battery=PbBattery(
84-
state=PbBatteryState(
85-
component_state=component_state,
86-
relay_state=relay_state,
87-
),
88-
# We don't need this method in public api. But we need it in unit tests.
89-
# pylint: disable=protected-access
90-
errors=[] if errors is None else errors,
91-
),
92-
)
9366

94-
return BatteryData.from_proto(pb_data)
67+
return BatteryDataWrapper(
68+
component_id=component_id,
69+
timestamp=datetime.now(tz=timezone.utc) if timestamp is None else timestamp,
70+
_relay_state=relay_state,
71+
_component_state=component_state,
72+
_errors=list(errors) if errors is not None else [],
73+
)
9574

9675

9776
def inverter_data(
@@ -117,26 +96,13 @@ def inverter_data(
11796
Returns:
11897
InverterData with given arguments.
11998
"""
120-
if timestamp is None:
121-
timestamp = datetime.now(tz=pytz.utc)
122-
123-
# Create protobuf message first. Thanks to that I don't have to specify
124-
# all InverterData class variable.
125-
pb_timestamp = Timestamp()
126-
pb_timestamp.FromDatetime(timestamp)
127-
128-
pb_data = microgrid_pb.ComponentData(
129-
id=component_id,
130-
ts=pb_timestamp,
131-
inverter=PbInverter(
132-
state=PbInverterState(
133-
component_state=component_state,
134-
),
135-
errors=[] if errors is None else errors,
136-
),
137-
)
13899

139-
return InverterData.from_proto(pb_data)
100+
return InverterDataWrapper(
101+
component_id=component_id,
102+
timestamp=datetime.now(tz=timezone.utc) if timestamp is None else timestamp,
103+
_component_state=component_state,
104+
_errors=list(errors) if errors is not None else [],
105+
)
140106

141107

142108
def component_graph() -> Tuple[Set[Component], Set[Connection]]:

‎tests/power/test_distribution_algorithm.py‎

Lines changed: 25 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -7,21 +7,14 @@
77
from datetime import datetime, timezone
88
from typing import Dict, List, Optional
99

10-
import frequenz.api.microgrid.microgrid_pb2 as microgrid_pb
1110
import pytest
12-
from frequenz.api.microgrid.battery_pb2 import Battery
13-
from frequenz.api.microgrid.battery_pb2 import Data as PbBatteryData
14-
from frequenz.api.microgrid.battery_pb2 import Properties as BatteryProperties
15-
from frequenz.api.microgrid.common_pb2 import AC, DC, Bounds
16-
from frequenz.api.microgrid.common_pb2 import Metric as PbMetric
17-
from frequenz.api.microgrid.common_pb2 import MetricAggregation
18-
from frequenz.api.microgrid.inverter_pb2 import Data as PbInverterData
19-
from frequenz.api.microgrid.inverter_pb2 import Inverter
20-
from google.protobuf.timestamp_pb2 import Timestamp # pylint: disable=no-name-in-module
11+
from frequenz.api.microgrid.common_pb2 import Bounds
2112

2213
from frequenz.sdk.microgrid.component import BatteryData, InverterData
2314
from frequenz.sdk.power import DistributionAlgorithm, InvBatPair
2415

16+
from ..utils.component_data_wrapper import BatteryDataWrapper, InverterDataWrapper
17+
2518

2619
@dataclass
2720
class Bound:
@@ -46,21 +39,6 @@ class Metric:
4639
now: Optional[float]
4740
bound: Optional[Bound] = None
4841

49-
def to_protobuf(self) -> PbMetric:
50-
"""Create protobuf Metric message from that instance.
51-
52-
Returns:
53-
Protobuf Metric
54-
"""
55-
if self.now is None:
56-
if self.bound is None:
57-
return PbMetric()
58-
return PbMetric(system_bounds=self.bound.to_protobuf())
59-
if self.bound is None:
60-
return PbMetric(value=self.now)
61-
62-
return PbMetric(value=self.now, system_bounds=self.bound.to_protobuf())
63-
6442

6543
def battery_msg( # pylint: disable=too-many-arguments
6644
component_id: int,
@@ -82,26 +60,16 @@ def battery_msg( # pylint: disable=too-many-arguments
8260
Returns:
8361
Protobuf battery component with data above
8462
"""
85-
pb_timestamp = Timestamp()
86-
pb_timestamp.FromDatetime(timestamp)
87-
capacitypb = capacity.to_protobuf()
88-
socpb = soc.to_protobuf()
89-
pb_data = microgrid_pb.ComponentData(
90-
id=component_id,
91-
ts=pb_timestamp,
92-
battery=Battery(
93-
properties=BatteryProperties(capacity=capacitypb.value),
94-
data=PbBatteryData(
95-
soc=MetricAggregation(
96-
avg=socpb.value, system_bounds=socpb.system_bounds
97-
),
98-
dc=DC(
99-
power=PbMetric(system_bounds=power.to_protobuf()),
100-
),
101-
),
102-
),
63+
return BatteryDataWrapper(
64+
component_id=component_id,
65+
capacity=capacity.now if capacity.now is not None else float("NaN"),
66+
soc=soc.now if soc.now is not None else float("NaN"),
67+
soc_lower_bound=soc.bound.lower if soc.bound is not None else float("NaN"),
68+
soc_upper_bound=soc.bound.upper if soc.bound is not None else float("NaN"),
69+
power_lower_bound=power.lower,
70+
power_upper_bound=power.upper,
71+
timestamp=timestamp,
10372
)
104-
return BatteryData.from_proto(pb_data)
10573

10674

10775
def inverter_msg(
@@ -120,20 +88,13 @@ def inverter_msg(
12088
Returns:
12189
Protobuf inverter component with data above.
12290
"""
123-
pb_timestamp = Timestamp()
124-
pb_timestamp.FromDatetime(timestamp)
125-
pb_data = microgrid_pb.ComponentData(
126-
id=component_id,
127-
ts=pb_timestamp,
128-
inverter=Inverter(
129-
data=PbInverterData(
130-
ac=AC(power_active=PbMetric(system_bounds=power.to_protobuf())),
131-
)
132-
),
91+
return InverterDataWrapper(
92+
component_id=component_id,
93+
timestamp=timestamp,
94+
active_power_lower_bound=power.lower,
95+
active_power_upper_bound=power.upper,
13396
)
13497

135-
return InverterData.from_proto(pb_data)
136-
13798

13899
class TestDistributionAlgorithm: # pylint: disable=too-many-public-methods
139100
"""Test whether the algorithm works as expected."""
@@ -147,16 +108,16 @@ def create_components_with_capacity(
147108

148109
components: List[InvBatPair] = []
149110
for i in range(0, num):
150-
bat_msg = microgrid_pb.ComponentData(
151-
id=2 * i,
152-
battery=Battery(properties=BatteryProperties(capacity=capacity[i])),
111+
battery_data = BatteryDataWrapper(
112+
component_id=2 * i,
113+
timestamp=datetime.now(tz=timezone.utc),
114+
capacity=capacity[i],
115+
)
116+
inverter_data = InverterDataWrapper(
117+
component_id=2 * i + 1, timestamp=datetime.now(tz=timezone.utc)
153118
)
154-
battery = BatteryData.from_proto(bat_msg)
155-
156-
inv_msg = microgrid_pb.ComponentData(id=2 * i + 1, inverter=Inverter())
157-
inverter = InverterData.from_proto(inv_msg)
158119

159-
components.append(InvBatPair(battery, inverter))
120+
components.append(InvBatPair(battery_data, inverter_data))
160121
return components
161122

162123
def test_total_capacity_all_0(self) -> None:

‎tests/utils/component_data_wrapper.py‎

Lines changed: 147 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,147 @@
1+
# License: MIT
2+
# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
3+
4+
"""Wrappers for the ComponentData.
5+
6+
This module allows the easy construction of mock ComponentData instances to be used in
7+
unit tests. Usually, all parameters for an instance need to be defined, this module
8+
helps by allowing users to only specify the parameters they are interested in for their
9+
tests. The rest will be filled with default protobuf values.
10+
This also abstracts away changes in the protobuf definition and minimizes the places
11+
that will need to be updated in such cases.
12+
"""
13+
from __future__ import annotations
14+
15+
from dataclasses import dataclass, field, replace
16+
from datetime import datetime
17+
from typing import Tuple
18+
19+
import frequenz.api.microgrid.battery_pb2 as battery_pb
20+
import frequenz.api.microgrid.inverter_pb2 as inverter_pb
21+
22+
from frequenz.sdk.microgrid.component import (
23+
BatteryData,
24+
EVChargerCableState,
25+
EVChargerComponentState,
26+
EVChargerData,
27+
InverterData,
28+
MeterData,
29+
)
30+
31+
32+
@dataclass(frozen=True)
33+
class BatteryDataWrapper(BatteryData):
34+
"""Wrapper for the BatteryData with default arguments."""
35+
36+
soc: float = float("NaN")
37+
soc_lower_bound: float = float("NaN")
38+
soc_upper_bound: float = float("NaN")
39+
capacity: float = float("NaN")
40+
power_lower_bound: float = float("NaN")
41+
power_upper_bound: float = float("NaN")
42+
temperature_max: float = float("NaN")
43+
_relay_state: battery_pb.RelayState.ValueType = (
44+
battery_pb.RelayState.RELAY_STATE_UNSPECIFIED
45+
)
46+
_component_state: battery_pb.ComponentState.ValueType = (
47+
battery_pb.ComponentState.COMPONENT_STATE_UNSPECIFIED
48+
)
49+
_errors: list[battery_pb.Error] = field(default_factory=list)
50+
51+
def copy_with_new_timestamp(self, new_timestamp: datetime) -> BatteryDataWrapper:
52+
"""Copy the component data but insert new timestamp.
53+
54+
Because the dataclass is frozen, we can't just replace the timestamp.
55+
We have to copy it.
56+
57+
Args:
58+
new_timestamp: New timestamp.
59+
60+
Returns:
61+
Copied component data.
62+
"""
63+
return replace(self, timestamp=new_timestamp)
64+
65+
66+
@dataclass(frozen=True)
67+
class InverterDataWrapper(InverterData):
68+
"""Wrapper for the InverterData with default arguments."""
69+
70+
active_power: float = float("NaN")
71+
active_power_lower_bound: float = float("NaN")
72+
active_power_upper_bound: float = float("NaN")
73+
_component_state: inverter_pb.ComponentState.ValueType = (
74+
inverter_pb.ComponentState.COMPONENT_STATE_UNSPECIFIED
75+
)
76+
_errors: list[inverter_pb.Error] = field(default_factory=list)
77+
78+
def copy_with_new_timestamp(self, new_timestamp: datetime) -> InverterDataWrapper:
79+
"""Copy the component data but insert new timestamp.
80+
81+
Because the dataclass is frozen, we can't just replace the timestamp.
82+
We have to copy it.
83+
84+
Args:
85+
new_timestamp: New timestamp.
86+
87+
Returns:
88+
Copied component data.
89+
"""
90+
return replace(self, timestamp=new_timestamp)
91+
92+
93+
@dataclass(frozen=True)
94+
class EvChargerDataWrapper(EVChargerData):
95+
"""Wrapper for the EvChargerData with default arguments."""
96+
97+
active_power: float = float("NaN")
98+
current_per_phase: Tuple[float, float, float] = field(
99+
default_factory=lambda: (float("NaN"), float("NaN"), float("NaN"))
100+
)
101+
voltage_per_phase: Tuple[float, float, float] = field(
102+
default_factory=lambda: (float("NaN"), float("NaN"), float("NaN"))
103+
)
104+
cable_state: EVChargerCableState = EVChargerCableState.UNSPECIFIED
105+
component_state: EVChargerComponentState = EVChargerComponentState.UNSPECIFIED
106+
107+
def copy_with_new_timestamp(self, new_timestamp: datetime) -> EvChargerDataWrapper:
108+
"""Copy the component data but insert new timestamp.
109+
110+
Because the dataclass is frozen, we can't just replace the timestamp.
111+
We have to copy it.
112+
113+
Args:
114+
new_timestamp: New timestamp.
115+
116+
Returns:
117+
Copied component data.
118+
"""
119+
return replace(self, timestamp=new_timestamp)
120+
121+
122+
@dataclass(frozen=True)
123+
class MeterDataWrapper(MeterData):
124+
"""Wrapper for the MeterData with default arguments."""
125+
126+
active_power: float = float("NaN")
127+
current_per_phase: Tuple[float, float, float] = field(
128+
default_factory=lambda: (float("NaN"), float("NaN"), float("NaN"))
129+
)
130+
voltage_per_phase: Tuple[float, float, float] = field(
131+
default_factory=lambda: (float("NaN"), float("NaN"), float("NaN"))
132+
)
133+
frequency: float = float("NaN")
134+
135+
def copy_with_new_timestamp(self, new_timestamp: datetime) -> MeterDataWrapper:
136+
"""Copy the component data but insert new timestamp.
137+
138+
Because the dataclass is frozen, we can't just replace the timestamp.
139+
We have to copy it.
140+
141+
Args:
142+
new_timestamp: New timestamp.
143+
144+
Returns:
145+
Copied component data.
146+
"""
147+
return replace(self, timestamp=new_timestamp)

0 commit comments

Comments
 (0)