test_component_base.py

# -----------------------------------------------------------------------------
# Predbat Home Battery System
# Copyright Trefor Southwell 2026 - All Rights Reserved
# This application maybe used for personal use only and not for commercial use
# -----------------------------------------------------------------------------
# fmt off
# pylint: disable=consider-using-f-string
# pylint: disable=line-too-long
# pylint: disable=attribute-defined-outside-init

"""
Tests for ComponentBase start method and backoff behavior
"""

import asyncio
from types import SimpleNamespace
from datetime import timezone
from unittest.mock import patch

from component_base import ComponentBase


# Save original sleep before any patching
_original_sleep = asyncio.sleep


# Fast sleep function for tests - sleeps 1/100th of the specified time
async def fast_sleep(delay, result=None):
    """Sleep for 1/100th of the specified time for faster tests"""
    await _original_sleep(delay / 100, result)


class MockBase:
    """Mock base object for testing ComponentBase"""

    def __init__(self):
        self.log_messages = []
        self.local_tz = timezone.utc
        self.prefix = "predbat"
        self.args = {}
        self.had_errors = False
        self.fatal_error = False

    def log(self, message):
        """Mock log function"""
        self.log_messages.append(message)
        print(message)

    def call_notify(self, message):
        """Mock notify method"""
        self.log_messages.append("Alert: " + message)


class TestComponent(ComponentBase):
    """Test component implementation"""

    def __init__(self, base, fail_until_attempt=0, return_true_on_run=True, **kwargs):
        """
        Args:
            fail_until_attempt: Number of run() calls that should return False before succeeding
            return_true_on_run: Whether run() should return True or False after fail_until_attempt
        """
        self.run_count = 0
        self.fail_until_attempt = fail_until_attempt
        self.return_true_on_run = return_true_on_run
        super().__init__(base, **kwargs)

    def initialize(self, **kwargs):
        """Initialize test component"""
        pass

    async def run(self, seconds, first):
        """Mock run method"""
        self.run_count += 1
        self.log(f"TestComponent: run() called (attempt {self.run_count}, seconds={seconds}, first={first})")

        # Fail for the first N attempts
        if self.run_count <= self.fail_until_attempt:
            return False

        return self.return_true_on_run


def test_component_base_not_calculating(my_predbat):
    """Test that components are not calculating unless they explicitly say otherwise."""
    component = TestComponent(MockBase())
    assert not component.is_calculating(), "ComponentBase should default to not calculating"
    print("PASS: ComponentBase defaults to not calculating")
    return False


def test_component_base_immediate_success(my_predbat):
    """Test component that succeeds on first run"""
    print("\n*** Test: ComponentBase immediate success ***")

    async def run_test():
        with patch("asyncio.sleep", side_effect=fast_sleep):
            base = MockBase()
            component = TestComponent(base, fail_until_attempt=0, return_true_on_run=True)

            # Start component in background
            task = asyncio.create_task(component.start())

            # Wait briefly for it to start (1.0 → 0.01s real via fast_sleep; must be
            # shorter than the component's 5s loop sleep → 0.05s real)
            await asyncio.sleep(1.0)

            # Check it started successfully
            assert component.api_started, "Component should have started"
            assert component.run_count == 1, f"Expected 1 run call, got {component.run_count}"

            # Stop component
            await component.stop()
            await task

            print("PASS: Component started immediately on first run")
        return False  # False = test passed (no failure)

    return asyncio.run(run_test())


def test_component_base_backoff_sequence(my_predbat):
    """Test component with backoff on failure"""
    print("\n*** Test: ComponentBase backoff sequence ***")

    async def run_test():
        with patch("asyncio.sleep", side_effect=fast_sleep):
            base = MockBase()
            component = TestComponent(base, fail_until_attempt=1, return_true_on_run=True)

            # Start component in background
            task = asyncio.create_task(component.start())

            # First run happens immediately (at second 0)
            await asyncio.sleep(1.0)
            assert component.run_count == 1, f"Expected 1 run after start, got {component.run_count}"
            assert not component.api_started, "Component should not have started yet (failed first attempt)"

            # Wait slightly longer - should still be 1 run (waiting for backoff)
            # Backoff = 60s real → 0.6s real via fast_sleep; we wait 0.5 → 0.005s real
            await asyncio.sleep(0.5)
            assert component.run_count == 1, f"Should still be 1 run (waiting for backoff), got {component.run_count}"

            # Stop component before the backoff completes
            await component.stop()
            await task

            print(f"PASS: Component backoff working (run_count={component.run_count})")
        return False  # False = test passed

    return asyncio.run(run_test())


def test_component_base_stop_during_backoff(my_predbat):
    """Test that api_stop is respected during backoff period"""
    print("\n*** Test: ComponentBase respects api_stop during backoff ***")

    async def run_test():
        with patch("asyncio.sleep", side_effect=fast_sleep):
            base = MockBase()
            component = TestComponent(base, fail_until_attempt=10, return_true_on_run=True)

            # Start component in background
            task = asyncio.create_task(component.start())

            # Wait for first run
            await asyncio.sleep(1.0)
            assert component.run_count == 1, f"Expected 1 run, got {component.run_count}"
            assert not component.api_started, "Component should not have started yet"

            # Stop component during backoff period
            await component.stop()
            await task

            # Verify it stopped cleanly without waiting for the full backoff
            assert not component.api_started, "Component should not have started"
            print(f"PASS: Component stopped during backoff (run_count={component.run_count})")
        return False  # False = test passed

    return asyncio.run(run_test())


def test_component_base_normal_operation_after_start(my_predbat):
    """Test that component runs every 60 seconds after successful start"""
    print("\n*** Test: ComponentBase normal operation after start ***")

    async def run_test():
        with patch("asyncio.sleep", side_effect=fast_sleep):
            base = MockBase()
            component = TestComponent(base, fail_until_attempt=0, return_true_on_run=True)

            # Start component in background
            task = asyncio.create_task(component.start())

            # Wait for it to start
            await asyncio.sleep(1.0)
            assert component.api_started, "Component should have started"
            initial_run_count = component.run_count

            # Wait a bit more - should not run again immediately (only every 60 seconds)
            # Component loop sleep = 5s → 0.05s real; we wait 0.5 → 0.005s real
            await asyncio.sleep(0.5)
            assert component.run_count == initial_run_count, f"Should not run again immediately, expected {initial_run_count}, got {component.run_count}"

            # Stop component
            await component.stop()
            await task

            print(f"PASS: Component operates normally after start (run_count={component.run_count})")
        return False  # False = test passed

    return asyncio.run(run_test())


def test_component_base_exception_handling(my_predbat):
    """Test that exceptions during run() are handled with backoff"""
    print("\n*** Test: ComponentBase exception handling with backoff ***")

    class ExceptionComponent(ComponentBase):
        def __init__(self, base, fail_count=2):
            self.run_count = 0
            self.fail_count = fail_count
            super().__init__(base)

        def initialize(self, **kwargs):
            pass

        async def run(self, seconds, first):
            self.run_count += 1
            if self.run_count <= self.fail_count:
                raise Exception(f"Test exception {self.run_count}")
            return True

    async def run_test():
        with patch("asyncio.sleep", side_effect=fast_sleep):
            base = MockBase()
            component = ExceptionComponent(base, fail_count=1)

            # Start component in background
            task = asyncio.create_task(component.start())

            # Wait for first run
            await asyncio.sleep(1.0)
            assert component.run_count == 1, f"Expected 1 run, got {component.run_count}"
            assert not component.api_started, "Component should not have started due to exception"
            assert component.count_errors == 1, "Error count should be incremented"

            # Check error was logged
            error_logged = any("Error:" in msg for msg in base.log_messages)
            assert error_logged, "Exception should have been logged"

            # Stop component
            await component.stop()
            await task

            print(f"PASS: Component handles exceptions with backoff (run_count={component.run_count}, errors={component.count_errors})")
        return False  # False = test passed

    return asyncio.run(run_test())


def test_component_base_run_timeout(my_predbat):
    """Test that a hung run() is detected, its stack is logged, and it is treated as a failure"""
    print("\n*** Test: ComponentBase run() timeout detection ***")

    class SlowComponent(ComponentBase):
        def __init__(self, base):
            self.run_count = 0
            super().__init__(base)
            self.run_timeout = 0.05  # 50 ms - fires before fast_sleep(10) completes (~100 ms real)

        def initialize(self, **kwargs):
            pass

        async def run(self, seconds, first):
            self.run_count += 1
            await asyncio.sleep(10)  # fast_sleep makes this ~100 ms real - longer than timeout
            return True

    async def run_test():
        with patch("asyncio.sleep", side_effect=fast_sleep):
            base = MockBase()
            component = SlowComponent(base)

            task = asyncio.create_task(component.start())

            # Wait long enough for: timeout to fire + error processing + one sleep(5) cycle
            await asyncio.sleep(2)  # ~20 ms real time via fast_sleep - enough for timeout + bookkeeping

            await component.stop()
            await task

            # Component should not have started - run() never returned True
            assert not component.api_started, "Component should not have started (run timed out)"

            # Error count must have been incremented
            assert component.count_errors > 0, f"Error count should be > 0, got {component.count_errors}"

            # A 'timeout' message must appear in the log
            timeout_logged = any("timeout" in msg.lower() for msg in base.log_messages)
            assert timeout_logged, "Timeout should have been logged. Messages:\n" + "\n".join(base.log_messages)

            # A traceback line should also have been logged (stack dump)
            traceback_logged = any("File" in msg for msg in base.log_messages)
            assert traceback_logged, "Stack trace should have been logged. Messages:\n" + "\n".join(base.log_messages)

            print(f"PASS: Timeout caught and stack-traced (error_count={component.count_errors})")
        return False  # False = test passed

    return asyncio.run(run_test())


def test_component_base_first_cleared_when_run_presets_api_started(my_predbat):
    """Regression: a component that sets api_started itself must still leave the startup path.

    The gateway's MQTT background loop sets self.api_started = True before run(first=True)
    returns. If start() only clears the `first` flag inside `if not self.api_started`, the
    flag stays True forever and start() keeps re-running the first=True startup path on
    backoff, never reaching the steady-state (first=False) housekeeping that publishes the
    plan. This verifies start() transitions to first=False regardless of who set api_started.
    """
    print("\n*** Test: ComponentBase clears first when run() pre-sets api_started ***")

    class PresetComponent(ComponentBase):
        def __init__(self, base):
            self.first_flags = []
            super().__init__(base)

        def initialize(self, **kwargs):
            pass

        async def run(self, seconds, first):
            self.first_flags.append(first)
            # Mimic a background task marking the component started before run() returns.
            self.api_started = True
            return True

    async def run_test():
        with patch("asyncio.sleep", side_effect=fast_sleep):
            base = MockBase()
            component = PresetComponent(base)

            task = asyncio.create_task(component.start())

            # Wait long enough (sped up 100x by fast_sleep → ~2s real) for the component
            # loop to advance past simulated seconds=60 so a steady-state run can occur.
            await asyncio.sleep(200)

            assert component.api_started, "Component should be started"

            await component.stop()
            await task

            assert component.first_flags, "run() should have been called"
            assert component.first_flags[0] is True, "First run should be first=True"
            assert any(f is False for f in component.first_flags), "Component must reach steady-state housekeeping (first=False); got first flags: {}".format(component.first_flags)
            assert component.first_flags.count(True) == 1, "Startup run() should happen exactly once, got {}".format(component.first_flags)

            print(f"PASS: first cleared despite self-set api_started (flags={component.first_flags})")
        return False  # False = test passed

    return asyncio.run(run_test())


def test_component_base_set_arg_auto(my_predbat):
    """
    Test ComponentBase.set_arg_auto() (issue #4494 follow-up, PR #4500 review): warns once when
    it overwrites a key the user had explicitly set in apps.yaml, otherwise behaves exactly like
    set_arg() - auto-discovery always wins either way, this only makes the override discoverable.
    """
    print("\n*** Test: ComponentBase.set_arg_auto warns once on apps.yaml override ***")

    base = MockBase()
    base.args_from_apps_yaml = {"battery_scaling": [0.9]}
    base.apps_yaml_override_warned = set()
    set_calls = {}
    base.set_arg = lambda arg, value: set_calls.__setitem__(arg, value)

    component = TestComponent(base)

    # apps.yaml had a different value - warn once, auto-discovered value still applied
    component.set_arg_auto("battery_scaling", ["sensor.predbat_battery_soh"])
    assert set_calls.get("battery_scaling") == ["sensor.predbat_battery_soh"], "Auto-discovered value should be applied"
    assert any("apps.yaml sets 'battery_scaling: [0.9]'" in msg for msg in base.log_messages), "Should warn about the override"

    # Second call for the same key must not repeat the warning
    component.set_arg_auto("battery_scaling", ["sensor.predbat_battery_soh"])
    warn_count = sum(1 for msg in base.log_messages if "apps.yaml sets 'battery_scaling" in msg)
    assert warn_count == 1, f"Warning should not repeat, got {warn_count}"

    # A key never present in apps.yaml at all - no warning, behaves like plain set_arg
    component.set_arg_auto("num_inverters", 1)
    assert set_calls.get("num_inverters") == 1, "Should still set the value for an unconfigured key"
    assert not any("num_inverters" in msg for msg in base.log_messages), "Should not warn for a key the user never configured"

    # Base with no args_from_apps_yaml snapshot at all (e.g. component created outside
    # PredBat.initialize(), as in most unit tests) must not raise, and must not warn
    bare_base = MockBase()
    bare_set_calls = {}
    bare_base.set_arg = lambda arg, value: bare_set_calls.__setitem__(arg, value)
    bare_component = TestComponent(bare_base)
    bare_component.set_arg_auto("battery_scaling", ["sensor.predbat_battery_soh"])
    assert bare_set_calls.get("battery_scaling") == ["sensor.predbat_battery_soh"], "Should still work without an apps_yaml snapshot"

    print("PASS: set_arg_auto warns once on a genuine override, stays silent otherwise, and is safe without a snapshot")
    return False


def test_component_base_set_state_external(my_predbat):
    """
    Test ComponentBase.set_state_external() forwards to the HA interface with the attributes intact.

    Components use this (rather than set_state_wrapper) when auto-discovery has to change one of
    Predbat's own settings - only this path updates the matching CONFIG_ITEMS value, so writing the
    state alone would move the displayed entity without changing what the planner reads.
    """
    print("\n*** Test: ComponentBase.set_state_external forwards to the HA interface ***")

    calls = []

    async def capture(entity_id, state, attributes={}):
        """Record a forwarded external state write."""
        calls.append((entity_id, state, attributes))
        return "written"

    base = MockBase()
    base.ha_interface = SimpleNamespace(set_state_external=capture)
    component = TestComponent(base)

    result = asyncio.run(component.set_state_external("switch.predbat_inverter_hybrid", False))
    assert calls == [("switch.predbat_inverter_hybrid", False, {})], f"Unexpected forwarded call {calls}"
    assert result == "written", "The HA interface's return value should be passed back to the caller"

    asyncio.run(component.set_state_external("sensor.predbat_test", 42, {"unit_of_measurement": "W"}))
    assert calls[1] == ("sensor.predbat_test", 42, {"unit_of_measurement": "W"}), f"Attributes not forwarded: {calls[1]}"

    print("PASS: set_state_external forwards entity, state and attributes and returns the result")
    return False


def test_component_base_all(my_predbat):
    """Run all component_base tests"""
    tests = [
        ("not_calculating", test_component_base_not_calculating, "Component defaults to not calculating"),
        ("immediate_success", test_component_base_immediate_success, "Component starts immediately on first successful run"),
        ("backoff_sequence", test_component_base_backoff_sequence, "Component uses backoff on startup failures"),
        ("stop_during_backoff", test_component_base_stop_during_backoff, "Component respects api_stop during backoff"),
        ("normal_operation", test_component_base_normal_operation_after_start, "Component runs every 60s after start"),
        ("exception_handling", test_component_base_exception_handling, "Component handles exceptions with backoff"),
        ("run_timeout", test_component_base_run_timeout, "Hung run() triggers timeout, stack trace, and error count"),
        ("first_cleared_preset", test_component_base_first_cleared_when_run_presets_api_started, "first flag clears even when run() pre-sets api_started"),
        ("set_arg_auto", test_component_base_set_arg_auto, "set_arg_auto warns once on an apps.yaml override, silent otherwise"),
        ("set_state_external", test_component_base_set_state_external, "set_state_external forwards to the HA interface"),
    ]

    failed = []
    for name, test_func, description in tests:
        print(f"\n*** Running: {name} - {description} ***")
        try:
            result = test_func(my_predbat)
            if result:
                failed.append(name)
                print(f"FAILED: {name}")
        except Exception as e:
            failed.append(name)
            print(f"ERROR in {name}: {e}")

    if failed:
        print(f"\n*** {len(failed)} test(s) failed: {', '.join(failed)} ***")
        return True  # True = test failed
    else:
        print(f"\n*** All {len(tests)} component_base tests passed ***")
        return False  # False = test passed