add support for expression in sensor_attributes

unit_tests/test_virtual_sensor_attribute_expressions.py

@@ -0,0 +1,103 @@
+"""Check reactive expressions alongside legacy entity-backed attributes."""
+import ast
+import re
+import sys
+import unittest
+from pathlib import Path
+
+
+MODULE_DIR = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(MODULE_DIR))
+import smartvalue as SmartValue
+from expressionparser import ParsingException
+
+
+source = MODULE_DIR / "virtualsensors.py"
+node = next(node for node in ast.parse(source.read_text()).body
+            if isinstance(node, ast.ClassDef) and node.name == "VirtualSensorBase")
+namespace = {"SmartValue": SmartValue, "ParsingException": ParsingException, "re": re}
+exec(compile(ast.Module(body=[node], type_ignores=[]), str(source), "exec"), namespace)
+VirtualSensorBase = namespace["VirtualSensorBase"]
+
+
+class FakeAPI:
+    def __init__(self):
+        self.states = {
+            "sensor.icon_source": {"state": "mdi:bike", "attributes": {}},
+            "sensor.next_rain": {"state": "unknown", "attributes": {"minutes_until_rain": "12"}},
+        }
+        self.listeners = []
+
+    def entity_exists(self, entity):
+        return entity in self.states
+
+    def get_state(self, entity, attribute=None):
+        data = self.states.get(entity)
+        if data is None:
+            return None
+        return data["state"] if attribute is None else data["attributes"].get(attribute)
+
+    def set_state(self, entity, state=None, attributes=None):
+        data = self.states.setdefault(entity, {"state": None, "attributes": {}})
+        if state is not None:
+            data["state"] = state
+        if attributes is not None:
+            data["attributes"].update(attributes)
+
+    def listen_state(self, callback, entity, attribute=None):
+        self.listeners.append((callback, entity, attribute))
+
+    def change(self, entity, value, attribute=None):
+        previous = self.get_state(entity, attribute)
+        if attribute is None:
+            self.states[entity]["state"] = value
+        else:
+            self.states[entity]["attributes"][attribute] = value
+        for callback, watched_entity, watched_attribute in list(self.listeners):
+            if (entity, attribute) == (watched_entity, watched_attribute):
+                callback(entity, attribute, previous, value, {})
+
+
+class FakeLogger:
+    def log_info(self, message):
+        pass
+
+    def log_warning(self, message):
+        pass
+
+    def log_error(self, message):
+        raise AssertionError(message)
+
+
+class ExampleSensor(VirtualSensorBase):
+    def initialize(self):
+        self.ad_api.set_state(self.sensor_name, state="off")
+
+
+class SensorAttributeExpressionsTests(unittest.TestCase):
+    def test_legacy_state_and_expressions_track_their_sources(self):
+        api = FakeAPI()
+        sensor = ExampleSensor(
+            api, FakeLogger(), virtual_sensor_name="example",
+            yaml_block={"sensor_attributes": {
+                "icon": "sensor.icon_source",
+                "minutes": "sensor.next_rain.minutes_until_rain",
+                "minutes_plus_five": "sensor.next_rain.minutes_until_rain + 5",
+            }},
+        )
+
+        self.assertEqual(set(sensor.generate_attributes_dependencies_list()),
+                         {"sensor.icon_source", "sensor.next_rain"})
+        sensor._initialize()
+        attrs = api.states["sensor.example"]["attributes"]
+        self.assertEqual((attrs["icon"], attrs["minutes"], attrs["minutes_plus_five"]),
+                         ("mdi:bike", 12, 17))
+
+        api.change("sensor.next_rain", "7", "minutes_until_rain")
+        api.change("sensor.icon_source", "mdi:weather-rainy")
+        self.assertEqual((attrs["icon"], attrs["minutes"], attrs["minutes_plus_five"]),
+                         ("mdi:weather-rainy", 7, 12))
+
+
+if __name__ == "__main__":
+    unittest.main()