75 lines
3.5 KiB
Python
75 lines
3.5 KiB
Python
"""Offline weather inputs selection, converter, fallback, migration and source order."""
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import sys
|
|
import types
|
|
import unittest
|
|
|
|
COMPONENTS = Path(__file__).resolve().parents[1] / 'custom_components' / 'heating_mpc'
|
|
# Fake only the package container, to avoid importing Home Assistant from __init__.
|
|
sys.modules.setdefault('heating_mpc', types.ModuleType('heating_mpc'))
|
|
sys.modules['heating_mpc'].__path__ = [str(COMPONENTS)]
|
|
for name in ('const', 'weather_sources'):
|
|
spec = importlib.util.spec_from_file_location(f'heating_mpc.{name}', COMPONENTS / f'{name}.py')
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
|
|
from heating_mpc import weather_sources as w
|
|
|
|
|
|
class WeatherInputs(unittest.TestCase):
|
|
def test_priority_fallback_and_compare(self):
|
|
data = {"outside_temperature_sources": ["sensor.owm", "sensor.alt"]}
|
|
states = {
|
|
"sensor.owm": ("unavailable", "°C"),
|
|
"sensor.alt": (53.6, "°F"),
|
|
}
|
|
values, active, comparisons = w.choose_weather_sources(data, states)
|
|
self.assertEqual(values["temperature"], 12.0)
|
|
self.assertEqual(active["temperature"], "sensor.alt")
|
|
self.assertEqual(comparisons["temperature"], {"sensor.alt": 12.0})
|
|
states["sensor.owm"] = (10.3, "°C")
|
|
values, active, comparisons = w.choose_weather_sources(data, states)
|
|
self.assertEqual(values["temperature"], 10.3)
|
|
self.assertEqual(active["temperature"], "sensor.owm")
|
|
self.assertEqual(comparisons["temperature"]["sensor.alt"], 12.0)
|
|
|
|
def test_source_reorder_changes_selected_source(self):
|
|
data = {"outside_temperature_sources": ["sensor.alt", "sensor.owm"]}
|
|
states = {"sensor.owm": (11.0, "°C"), "sensor.alt": (12.0, "°C")}
|
|
values, active, _ = w.choose_weather_sources(data, states)
|
|
self.assertEqual(active["temperature"], "sensor.alt")
|
|
self.assertEqual(values["temperature"], 12.0)
|
|
|
|
def test_legacy_v01_temperature_migration(self):
|
|
self.assertEqual(w.sources_for({"outside_entity": "sensor.temperature"}, "temperature"),
|
|
["sensor.temperature"])
|
|
self.assertEqual(w.sources_for({"outside_temperature_sources": ["sensor.b", "sensor.b"]}, "temperature"),
|
|
["sensor.b"])
|
|
|
|
def test_optional_weather_and_unit_conversions(self):
|
|
data = {"outside_temperature_sources": ["sensor.t"],
|
|
"outside_wind_sources": ["sensor.wind"],
|
|
"outside_pressure_sources": ["sensor.pressure"],
|
|
"outside_clouds_sources": ["sensor.clouds"]}
|
|
states = {"sensor.t": (12, "°C"), "sensor.wind": (36, "km/h"),
|
|
"sensor.pressure": (101325, "Pa"), "sensor.clouds": (40, "%")}
|
|
values, _, _ = w.choose_weather_sources(data, states)
|
|
self.assertEqual(values["wind"], 10.0)
|
|
self.assertEqual(values["pressure"], 1013.25)
|
|
self.assertEqual(values["clouds"], 40.0)
|
|
self.assertIsNone(values["snow"])
|
|
|
|
def test_bad_values_rejected(self):
|
|
for value in (None, 'unavailable', 'nan', 'inf', 1000):
|
|
self.assertIsNone(w.normalize("temperature", value, "°C"))
|
|
self.assertIsNone(w.normalize("temperature", 22.0, "km/h"))
|
|
self.assertIsNone(w.normalize("wind", -3, "m/s"))
|
|
self.assertIsNone(w.normalize("temperature", 20, 42))
|
|
self.assertIsNone(w.normalize("clouds", 150, "%"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|