New group logging

unit_tests/test_logger_interface_grouping.py

@@ -0,0 +1,109 @@
+"""Verify scoped log grouping without replacing AppDaemon API methods."""
+import os
+import pickle
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+TOOLBOX = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(TOOLBOX))
+from logger_interface import LoggerInterface
+import smartcondition
+
+
+def load_smart_object():
+    source = (TOOLBOX / 'smartobject.py').read_text()
+    import ast
+    tree = ast.parse(source)
+    smart_object = next(node for node in tree.body
+                        if isinstance(node, ast.ClassDef) and node.name == 'SmartObject')
+    class FakeHass:
+        def log(self, message, level='INFO', log=None):
+            self.rows.append((level, message, log))
+    namespace = {'hass': SimpleNamespace(Hass=FakeHass),
+                 'LoggerInterface': LoggerInterface, 'VirtualSensors': object,
+                 'ParsingException': type('ParsingException', (Exception,), {}),
+                 'pickle': pickle, 'os': os}
+    exec(compile(ast.Module(body=[smart_object], type_ignores=[]), str(TOOLBOX / 'smartobject.py'), 'exec'), namespace)
+    return namespace['SmartObject']
+
+
+class FakeApp(LoggerInterface):
+    name = 'fake_app'
+
+    def __init__(self):
+        self.rows = []
+        super().__init__(self)
+
+    def log(self, message, level='INFO', log=None):
+        self.rows.append((level, message, log))
+
+    def initialize(self):
+        self.log_info('LoggerInterface initialization message')
+        self.log('Direct app log stays readable')
+
+
+class LoggerGroupingTests(unittest.TestCase):
+    def test_initialize_is_not_implicitly_wrapped_or_tagged(self):
+        app = FakeApp()
+        original_initialize = app.initialize.__func__
+        original_log = app.log.__func__
+        app.initialize()
+        self.assertIs(app.initialize.__func__, original_initialize)
+        self.assertIs(app.log.__func__, original_log)
+        self.assertEqual(app.rows[0][1], 'LoggerInterface initialization message')
+        self.assertEqual(app.rows[1][1], 'Direct app log stays readable')
+
+    def test_smart_object_groups_its_own_lifecycle(self):
+        SmartObject = load_smart_object()
+        app = SmartObject.__new__(SmartObject)
+        app.rows = []
+        app.name = 'test_smart_object'
+        app.args = {}
+        with tempfile.TemporaryDirectory() as folder:
+            app.AD = SimpleNamespace(app_dir=folder)
+            app.get_ad_api = lambda: app
+            app.on_initialize_smart_object = lambda: app.log_info('Subclass initialization hook')
+            app.initialize()
+        self.assertIn('Initialization started [group=initialization:', app.rows[0][1])
+        group_id = app.rows[0][1].split(':', 1)[1].rstrip(']')
+        self.assertIn(f'[group=initialization:{group_id}]', app.rows[1][1])
+        self.assertIn(f'[group=initialization:{group_id}]', app.rows[2][1])
+        self.assertEqual(app.rows[-1][1], f'Initialization completed [group=initialization:{group_id}]')
+
+    def test_smartcondition_uses_smartobject_api_scope_without_logger_interface_argument(self):
+        app = FakeApp()
+        evaluator = smartcondition.Evaluator.__new__(smartcondition.Evaluator)
+        evaluator._Evaluator__logger_interface = None
+        evaluator._Evaluator__appdaemon_api = app
+        evaluator._Evaluator__trace_name = 'smart_conditions'
+        with evaluator._Evaluator__log_group():
+            app.log_info('Callback trigger by sensor.example(0 => 1)')
+            app.log_info('[smart_conditions][Evaluating] Triggered by condition')
+        marker_a = app.rows[0][1].split('[group=')[1].rstrip(']')
+        marker_b = app.rows[1][1].split('[group=')[1].rstrip(']')
+        self.assertEqual(marker_a, marker_b)
+        self.assertTrue(marker_a.startswith('smartcondition:'))
+
+    def test_nested_scopes_inherit_outer_group_and_reset_after_exception(self):
+        app = FakeApp()
+        with app.log_group('value_selector', 'sensor.a') as outer:
+            app.log_info('candidate rejected')
+            with app.log_group('smartcondition', 'condition-a') as inner:
+                self.assertIs(inner, outer)
+                app.log_info('condition result')
+        app.log_info('outside group')
+        self.assertIn(f'[group=value_selector:{outer["id"]}]', app.rows[0][1])
+        self.assertIn(f'[group=value_selector:{outer["id"]}]', app.rows[1][1])
+        self.assertEqual(app.rows[2][1], 'outside group')
+        with self.assertRaises(RuntimeError):
+            with app.log_group('smartcondition'):
+                raise RuntimeError('expected')
+        app.log_info('after failure')
+        self.assertEqual(app.rows[3][1], 'after failure')
+
+
+if __name__ == '__main__':
+    unittest.main()