104 lines
3.6 KiB
Python
104 lines
3.6 KiB
Python
"""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()
|