Merge pull request smarthomeNG#1045 from Morg42/utc

uzsu: use shng configured timezone in line with lib.orb

Author: Morg42

uzsu/__init__.py

@@ -180,7 +180,7 @@ def run(self):
             # set activeToday to false if entry is in the future (and wasn't reset correctly at midnight)
             for i, entry in enumerate(self._items[item].get('list')):
                 next, _, _ = self._get_time(entry, 'next', item, i, 'dry_run')
-                if next and next.time() > datetime.now().time() and entry.get('activeToday'):
+                if next and next.time() > datetime.now(self._timezone).time() and entry.get('activeToday'):
                     entry['activeToday'] = False
                     update = 'activeToday'
                     self.logger.debug(
@@ -1085,7 +1085,7 @@ def _get_time(self, entry, timescan, item=None, entryindex=None, caller=None):
                 self._ignore_once_entries is False and entry.get('activeToday') is True and timescan == 'previous'
             )
             active = True if caller == 'dry_run' or cond_activetoday else entry.get('active')
-            today = datetime.today()
+            today = datetime.now(self._timezone)
             tomorrow = today + timedelta(days=1)
             yesterday = today - timedelta(days=1)
             weekbefore = today - timedelta(days=7)
@@ -1123,7 +1123,7 @@ def _get_time(self, entry, timescan, item=None, entryindex=None, caller=None):
                             rrule = rrulestr(entry['rrule'], dtstart=datetime.combine(weekbefore, datetime.min.time()))
                             rstr = str(rrule).replace('\n', ';')
                             self.logger.debug(f'{item}: Looking for {timescan} time. Found rrule: {rstr}')
-                dt = datetime.now()
+                dt = datetime.now(self._timezone).replace(tzinfo=None)
                 while self.alive:
                     dt = rrule.before(dt) if timescan == 'previous' else rrule.after(dt)
                     if dt is None:
@@ -1145,12 +1145,12 @@ def _get_time(self, entry, timescan, item=None, entryindex=None, caller=None):
                         self._update_suncalc(item, entry, entryindex, None)
                     compare_date = None if not next else next.date() - timedelta(days=1) if next_day else next.date()
                     if next and compare_date == dt.date():
-                        cond_istoday = next.date() == datetime.now().date()
+                        cond_istoday = next.date() == datetime.now(self._timezone).date()
                         if caller != 'dry_run' and (
                             not self._items[item]['interpolation'].get('perday') or cond_istoday
                         ):
                             self._itpl[item][next.timestamp() * 1000.0] = value
-                        if next - timedelta(seconds=1) > datetime.now().replace(tzinfo=self._timezone):
+                        if next - timedelta(seconds=1) > datetime.now(self._timezone):
                             self.logger.debug(f'{item}: Return from rrule {timescan}: {next}, value {value}.')
                             return next, value, None
                         else:
@@ -1162,7 +1162,7 @@ def _get_time(self, entry, timescan, item=None, entryindex=None, caller=None):
                     datetime.combine(today, datetime.min.time()).replace(tzinfo=self._timezone), time, timescan
                 )
                 cond_future = next > datetime.now(self._timezone)
-                cond_istoday = next.date() == datetime.now().date()
+                cond_istoday = next.date() == datetime.now(self._timezone).date()
                 if cond_future:
                     self.logger.debug(f'{item}: Result parsing time today (sun) {time}: {next}')
                     if entryindex is not None:
@@ -1191,7 +1191,7 @@ def _get_time(self, entry, timescan, item=None, entryindex=None, caller=None):
                 next = self._series_get_time(entry, timescan)
                 if next is None:
                     return None, None, False
-                cond_istoday = next.date() == datetime.now().date()
+                cond_istoday = next.date() == datetime.now(self._timezone).date()
                 if caller != 'dry_run' and (not self._items[item]['interpolation'].get('perday') or cond_istoday):
                     self._itpl[item][next.timestamp() * 1000.0] = value
                     self.logger.debug(f'{item}: Include {timescan} of series: {next} for interpolation.')
@@ -1295,9 +1295,7 @@ def _series_calculate(self, item, caller=None, source=None):
                         starttime = datetime.strptime(mydict['series']['timeSeriesMin'], '%H:%M')
                     else:
                         mytime = self._sun(
-                            datetime.now().replace(hour=0, minute=0, second=0).astimezone(self._timezone),
-                            seriesstart,
-                            'next',
+                            datetime.now(self._timezone).replace(hour=0, minute=0, second=0), seriesstart, 'next'
                         )
                         starttime = f'{mytime.hour:02d}:{mytime.minute:02d}'
                         starttime = datetime.strptime(starttime, '%H:%M')
@@ -1309,9 +1307,7 @@ def _series_calculate(self, item, caller=None, source=None):
 
                     if seriesend is not None and 'sun' in seriesend:
                         mytime = self._sun(
-                            datetime.now().replace(hour=0, minute=0, second=0).astimezone(self._timezone),
-                            seriesend,
-                            'next',
+                            datetime.now(self._timezone).replace(hour=0, minute=0, second=0), seriesend, 'next'
                         )
                         endtime = f'{mytime.hour:02d}:{mytime.minute:02d}'
                         endtime = datetime.strptime(endtime, '%H:%M')
@@ -1346,7 +1342,8 @@ def _series_calculate(self, item, caller=None, source=None):
                     rrule = rrulestr(
                         mydict['rrule'] + ';COUNT=7',
                         dtstart=datetime.combine(
-                            datetime.now(), parser.parse(str(starttime.hour) + ':' + str(starttime.minute)).time()
+                            datetime.now(self._timezone),
+                            parser.parse(str(starttime.hour) + ':' + str(starttime.minute)).time(),
                         ),
                     )
                     mynewlist = []
@@ -1365,7 +1362,7 @@ def _series_calculate(self, item, caller=None, source=None):
                         else:
                             seriesstart = mydict['series']['timeSeriesMin']
                             mytime = self._sun(
-                                day.replace(hour=0, minute=0, second=0).astimezone(self._timezone), seriesstart, 'next'
+                                day.replace(hour=0, minute=0, second=0, tzinfo=self._timezone), seriesstart, 'next'
                             )
                             starttime = f'{mytime.hour:02d}:{mytime.minute:02d}'
                             starttime = datetime.strptime(starttime, '%H:%M')
@@ -1417,9 +1414,10 @@ def _series_calculate(self, item, caller=None, source=None):
                                 count = 0
                                 seriestarttime = None
                                 actday = mydays[time.weekday()]
-                            if time.time() < datetime.now().time() and time.date() <= datetime.now().date():
+                            now_local = datetime.now(self._timezone).replace(tzinfo=None)
+                            if time.time() < now_local.time() and time.date() <= now_local.date():
                                 continue
-                            if time >= datetime.now() + timedelta(days=7):
+                            if time >= now_local + timedelta(days=7):
                                 continue
                             if seriestarttime is None:
                                 seriestarttime = time
@@ -1470,7 +1468,7 @@ def _get_sun4week(self, item, caller=None):
         """
         dayrule = rrulestr(
             'FREQ=WEEKLY;BYDAY=MO,TU,WE,TH,FR,SA,SU' + ';COUNT=7',
-            dtstart=datetime.now().replace(hour=0, minute=0, second=0),
+            dtstart=datetime.now(self._timezone).replace(hour=0, minute=0, second=0),
         )
         self.logger.debug(f'Get sun4week for item {item} called by {caller}')
         mynewdict = {'sunrise': {}, 'sunset': {}}
@@ -1525,17 +1523,13 @@ def _series_get_time(self, mydict, timescan=''):
         if 'sun' not in mydict['series']['timeSeriesMin']:
             starttime = datetime.strptime(mydict['series']['timeSeriesMin'], '%H:%M')
         else:
-            mytime = self._sun(
-                datetime.now().replace(hour=0, minute=0, second=0).astimezone(self._timezone), seriesstart, 'next'
-            )
+            mytime = self._sun(datetime.now(self._timezone).replace(hour=0, minute=0, second=0), seriesstart, 'next')
             starttime = f'{mytime.hour:02d}:{mytime.minute:02d}'
             starttime = datetime.strptime(starttime, '%H:%M')
 
         if daycount is None and seriesend is not None:
             if 'sun' in seriesend:
-                mytime = self._sun(
-                    datetime.now().replace(hour=0, minute=0, second=0).astimezone(self._timezone), seriesend, 'next'
-                )
+                mytime = self._sun(datetime.now(self._timezone).replace(hour=0, minute=0, second=0), seriesend, 'next')
                 seriesend = f'{mytime.hour:02d}:{mytime.minute:02d}'
                 endtime = datetime.strptime(seriesend, '%H:%M')
             else:
@@ -1576,7 +1570,7 @@ def _series_get_time(self, mydict, timescan=''):
         rrule = rrulestr(
             actrrule,
             dtstart=datetime.combine(
-                datetime.now() - timedelta(days=7),
+                datetime.now(self._timezone) - timedelta(days=7),
                 parser.parse(str(starttime.hour) + ':' + str(starttime.minute)).time(),
             ),
         )
@@ -1589,7 +1583,7 @@ def _series_get_time(self, mydict, timescan=''):
                 mylist[timestamp] = 'x'
                 mycount += 1
 
-        now = datetime.now()
+        now = datetime.now(self._timezone).replace(tzinfo=None)
         mylist[now] = 'now'
         mysortedlist = sorted(mylist)
         myindex = mysortedlist.index(now)
@@ -1601,7 +1595,7 @@ def _series_get_time(self, mydict, timescan=''):
         # Get correct "sun" for this Day
         if 'sun' in mydict['series']['timeSeriesMin'] and returnvalue is not None:
             mytime = self._sun(
-                returnvalue.replace(hour=0, minute=0, second=0).astimezone(self._timezone),
+                returnvalue.replace(hour=0, minute=0, second=0, tzinfo=self._timezone),
                 mydict['series']['timeSeriesMin'],
                 'next',
             )

uzsu/tests/__init__.py

@@ -0,0 +1,67 @@
+import os
+import unittest
+
+from tests import common
+from tests.mock.core import MockSmartHome
+
+from lib.orb import Orb
+from plugins.uzsu import UZSU
+
+BERLIN_LON = 13.4050
+BERLIN_LAT = 52.5200
+BERLIN_ELEV = 34
+
+
+class TestUZSUBase(unittest.TestCase):
+    """Base class for UZSU plugin tests.
+
+    Mirrors plugins/database/tests/base.py: builds a MockSmartHome with a
+    fixed set of test items, injects plugin.yaml defaults directly (since
+    the full parameter-loading infrastructure isn't available in test
+    environments), and constructs the plugin.
+    """
+
+    def plugin(self, parameters=None, tz='Europe/Berlin'):
+        self.sh = MockSmartHome()
+        self.sh.shtime.set_tz(tz)
+        # UZSU relies on sh.sun/sh.moon (set up by bin/smarthome.py in a real
+        # instance) for sunrise/sunset-bound entries; MockSmartHome doesn't
+        # provide these, so build them the same way lib/smarthome.py does.
+        self.sh.sun = Orb('sun', BERLIN_LON, BERLIN_LAT, BERLIN_ELEV)
+        self.sh.moon = Orb('moon', BERLIN_LON, BERLIN_LAT, BERLIN_ELEV)
+        self.sh.with_items_from(os.path.join(os.path.dirname(__file__), 'test_items.yaml'))
+
+        UZSU._parameters = {
+            'remove_duplicates': True,
+            'ignore_once_entries': False,
+            'suncalculation_cron': '0 0 * *',
+            'interpolation_interval': 5,
+            'interpolation_type': 'none',
+            'backintime': 0,
+            'interpolation_precision': 2,
+        }
+        if parameters:
+            UZSU._parameters.update(parameters)
+
+        plugin = UZSU(self.sh)
+        for item in self.sh.return_items():
+            callback = plugin.parse_item(item)
+            if callback is not None:
+                plugin._items_callback = callback
+        # run() does the real startup bookkeeping (self._series[item] = {},
+        # self._lastvalues[item] = None, initial scheduling) that
+        # bin/smarthome.py normally triggers once all plugins are loaded.
+        plugin.run()
+        return plugin
+
+    def plugin_with_entry(self, item_path, entry, active=True, rrule='FREQ=DAILY', **plugin_kwargs):
+        """Build a plugin and seed one uzsu list entry directly on
+        plugin._items[item]. Bypasses item()'s value-change callback (not
+        wired up in this mock setup) and writes the internal state the
+        scheduling methods (_get_time, _schedule, ...) actually consume."""
+        plugin = self.plugin(**plugin_kwargs)
+        item = self.sh.return_item(item_path)
+        full_entry = {'active': True, 'rrule': rrule, **entry}
+        plugin._items[item]['active'] = active
+        plugin._items[item]['list'] = [full_entry]
+        return plugin

uzsu/tests/base.py

@@ -0,0 +1,30 @@
+from plugins.uzsu.tests.base import TestUZSUBase
+
+
+class TestUZSUBasic(TestUZSUBase):
+    def test_plugin_constructs_and_parses_items(self):
+        plugin = self.plugin()
+        self.assertIn(self.sh.return_item('main.temp.uzsu'), plugin._items)
+        self.assertIn(self.sh.return_item('main.lamp.uzsu'), plugin._items)
+
+    def test_parsed_item_has_default_structure(self):
+        plugin = self.plugin()
+        item = self.sh.return_item('main.temp.uzsu')
+        self.assertEqual(plugin._items[item]['active'], False)
+        self.assertEqual(plugin._items[item]['list'], [])
+
+    def test_activate_sets_active_flag(self):
+        plugin = self.plugin()
+        item = self.sh.return_item('main.temp.uzsu')
+        plugin._items[item]['list'] = [{'value': 21.5, 'active': True, 'time': '07:00', 'rrule': 'FREQ=DAILY'}]
+        self.assertTrue(plugin.activate(True, item))
+        self.assertTrue(plugin._items[item]['active'])
+        self.assertFalse(plugin.activate(False, item))
+        self.assertFalse(plugin._items[item]['active'])
+
+    def test_get_type_returns_target_item_type(self):
+        plugin = self.plugin()
+        num_item = self.sh.return_item('main.temp.uzsu')
+        bool_item = self.sh.return_item('main.lamp.uzsu')
+        self.assertEqual(plugin._get_type(num_item), 'num')
+        self.assertEqual(plugin._get_type(bool_item), 'bool')

uzsu/tests/test_basic.py

@@ -0,0 +1,15 @@
+main:
+
+    temp:
+        type: num
+
+        uzsu:
+            type: dict
+            uzsu_item: ..
+
+    lamp:
+        type: bool
+
+        uzsu:
+            type: dict
+            uzsu_item: ..

uzsu/tests/test_items.yaml

@@ -0,0 +1,84 @@
+import datetime
+
+from tests import common
+
+from plugins.uzsu.tests.base import TestUZSUBase
+
+
+class TestPlainTimeEntries(TestUZSUBase):
+    """Coverage for _get_time() with a plain HH:MM entry.
+
+    These exercise the datetime.now(self._timezone) fix in _get_time(): the
+    rrule search seed and the final scheduled time must consistently be in
+    shng's configured timezone.
+
+    NOTE: a true red/green regression test (proving the OS-tz vs configured-tz
+    bug specifically) isn't practical here without monkeypatching
+    datetime.datetime.now() itself - the bug only manifests when the OS
+    timezone and configured timezone disagree about *which calendar day it
+    currently is* (i.e. right at local midnight). force_os_tz alone doesn't
+    shift "now" far enough to hit that window. These tests instead confirm
+    the now-fixed code produces correct, consistently-tz-aware results.
+    """
+
+    def setUp(self):
+        self.plugin = self.plugin_with_entry('main.temp.uzsu', {'value': 21.5, 'time': '07:00'})
+        self.item = self.sh.return_item('main.temp.uzsu')
+
+    def test_next_time_is_aware_in_configured_tz(self):
+        entry = self.plugin._items[self.item]['list'][0]
+        nxt, value, _ = self.plugin._get_time(entry, 'next', self.item, 0, 'test')
+        self.assertEqual(nxt.hour, 7)
+        self.assertEqual(nxt.minute, 0)
+        self.assertEqual(nxt.tzinfo, self.plugin._timezone)
+        self.assertEqual(value, 21.5)
+
+    def test_next_time_unaffected_by_os_tz(self):
+        entry = self.plugin._items[self.item]['list'][0]
+        with common.force_os_tz('Pacific/Honolulu'):
+            nxt, value, _ = self.plugin._get_time(entry, 'next', self.item, 0, 'test')
+        self.assertEqual(nxt.hour, 7)
+        self.assertEqual(nxt.utcoffset(), datetime.timedelta(hours=2))  # CEST in June
+
+
+class TestSunBoundEntries(TestUZSUBase):
+    """Coverage for _get_time()/_sun() with sunrise/sunset entries — exercises
+    the Orb tz fixes (lib/orb.py) through the uzsu integration path."""
+
+    def setUp(self):
+        self.plugin = self.plugin_with_entry('main.temp.uzsu', {'value': 18.0, 'time': 'sunset'})
+        self.item = self.sh.return_item('main.temp.uzsu')
+
+    def test_sunset_entry_returns_aware_datetime_in_configured_tz(self):
+        entry = self.plugin._items[self.item]['list'][0]
+        nxt, value, _ = self.plugin._get_time(entry, 'next', self.item, 0, 'test')
+        self.assertEqual(nxt.tzinfo, self.plugin._timezone)
+        self.assertEqual(value, 18.0)
+
+    def test_get_sun4week_populates_sunrise_and_sunset_for_all_days(self):
+        self.plugin._get_sun4week(self.item)
+        suncalc = self.plugin._items[self.item]['SunCalculated']
+        self.assertEqual(set(suncalc.keys()), {'sunrise', 'sunset'})
+        self.assertEqual(len(suncalc['sunrise']), 7)
+        self.assertEqual(len(suncalc['sunset']), 7)
+        for day_str in suncalc['sunrise'].values():
+            hour, minute = day_str.split(':')
+            self.assertTrue(0 <= int(hour) <= 23)
+            self.assertTrue(0 <= int(minute) <= 59)
+
+
+class TestInterpolation(TestUZSUBase):
+    def test_interpolation_set_and_get(self):
+        plugin = self.plugin()
+        item = self.sh.return_item('main.temp.uzsu')
+        result = plugin.interpolation('linear', interval=10, item=item)
+        self.assertEqual(result['type'], 'linear')
+        self.assertEqual(result['interval'], 10)
+        self.assertEqual(plugin.interpolation(item=item), result)
+
+    def test_invalid_interpolation_type_rejected(self):
+        plugin = self.plugin()
+        item = self.sh.return_item('main.temp.uzsu')
+        before = plugin.interpolation(item=item)
+        result = plugin.interpolation('exponential', item=item)
+        self.assertEqual(result, before)