diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index 9a4fbd7e..8c827fdb 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -8,7 +8,7 @@ This is a Python based repository providing an application for controlling batte - Remove excessive whitespaces. - Follow PEP8 standards. Use autopep8 for that. - Check against pylint. Target score is like 9.0-9.5, if you can achieve 10, do it. -- Build against Python 3.11 primarily, ensure compatibility with versions 3.9 to 3.13. +- Build against Python 3.14 primarily, ensure compatibility with versions 3.11 to 3.14. - Run all tests with `run_tests.sh` . - Create new parameters in `config/batcontrol_config_dummy.yaml` . @@ -45,7 +45,7 @@ This is a Python based repository providing an application for controlling batte 5. Lay test scripts for verification and simple testing into the folder `scripts`. 6. Never commit content of `tmp`. 7. User documentation lives in the `docs/` folder and is published via MkDocs to https://mastr.github.io/batcontrol/ — add or update pages there and register new pages in `mkdocs.yml`. -8. Ensure compatibility with supported Python versions (3.9 to 3.13) +8. Ensure compatibility with supported Python versions (3.11 to 3.14) 9. A change that affects which inverter mode is chosen, or why (`core.py`, `logic/default.py`, `logic/next.py`), must stay traceable: add or extend a `DecisionRecord` with a plain-language explanation (`logic/decision_trace.py`, `logic/decision_records.py`). This feeds the MQTT diff --git a/.github/workflows/pylint.yml b/.github/workflows/pylint.yml index 9e949214..7bc4cf3a 100644 --- a/.github/workflows/pylint.yml +++ b/.github/workflows/pylint.yml @@ -21,7 +21,7 @@ jobs: contents: read strategy: matrix: - python-version: ["3.9", "3.10" , "3.11" , "3.12", "3.13"] + python-version: ["3.11" , "3.12", "3.13", "3.14"] steps: - name: Checkout Repository uses: actions/checkout@v7 diff --git a/.github/workflows/pytest.yml b/.github/workflows/pytest.yml index 0930fa08..7e77a8ee 100644 --- a/.github/workflows/pytest.yml +++ b/.github/workflows/pytest.yml @@ -14,7 +14,7 @@ jobs: strategy: fail-fast: false matrix: - python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"] + python-version: ["3.11", "3.12", "3.13", "3.14"] steps: - uses: actions/checkout@v7 diff --git a/CLAUDE.md b/CLAUDE.md index a79578e5..1ec2a38d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1,15 +1,15 @@ # CLAUDE.md — batcontrol batcontrol charges a home battery when grid prices are cheap and preserves it for expensive -hours, based on dynamic tariffs, solar forecast, and consumption forecast. Python 3.9-3.13 -(primary target: 3.11). Full contribution guidelines: +hours, based on dynamic tariffs, solar forecast, and consumption forecast. Python 3.11-3.14 +(primary target: 3.14). Full contribution guidelines: [`.github/copilot-instructions.md`](.github/copilot-instructions.md) ## Commands ```bash ./run_tests.sh # full suite + coverage (creates .venv via uv) -uv venv --python 3.13 --allow-existing # setup only +uv venv --python 3.14 --allow-existing # setup only uv pip install -e '.[test]' pylint autopep8 # lint tools are not part of the test extras uv run pytest tests/ -k # single test / subset uv run pylint src/batcontrol # target score >= 9.0 (10 if achievable) diff --git a/Dockerfile b/Dockerfile index 92472bb3..2f4b2be4 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,5 +1,5 @@ # Stage 1: Build Stage -FROM python:3.13-alpine AS builder +FROM python:3.14-alpine AS builder # Install uv for fast, reliable Python packaging COPY --from=ghcr.io/astral-sh/uv:0.11 /uv /usr/local/bin/uv @@ -16,7 +16,7 @@ COPY ./src ./src RUN uv build --wheel --out-dir /wheels # Stage 2: Build the final image -FROM python:3.13-alpine +FROM python:3.14-alpine ARG VERSION ARG GIT_SHA diff --git a/docs/getting-started/installation.md b/docs/getting-started/installation.md index 8ec99f69..53d71449 100644 --- a/docs/getting-started/installation.md +++ b/docs/getting-started/installation.md @@ -108,12 +108,13 @@ Add the [batcontrol_ha_addon](https://github.com/MaStr/batcontrol_ha_addon) repo ## Local Python -Requires Python 3.13 and [uv](https://docs.astral.sh/uv/). +Requires Python 3.11 or newer (3.11 - 3.14 are supported and tested in CI; the +Docker image ships Python 3.14) and [uv](https://docs.astral.sh/uv/). ```sh git clone https://github.com/MaStr/batcontrol.git cd batcontrol -uv venv --python 3.13 --allow-existing +uv venv --python 3.14 --allow-existing source .venv/bin/activate uv pip install . ``` @@ -170,7 +171,7 @@ For working on the batcontrol source code, use an editable install: ```sh git clone https://github.com/MaStr/batcontrol.git cd batcontrol -uv venv --python 3.13 --allow-existing +uv venv --python 3.14 --allow-existing source .venv/bin/activate uv pip install --editable '.[test]' ``` diff --git a/pyproject.toml b/pyproject.toml index 826500a5..ac84674d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,16 +6,23 @@ authors = [{name = "Stephan Mükusch", email = "muexxl@gmx.net"}] license = "MIT" license-files = ["LICENSE"] readme = "README.md" -requires-python = ">3.8, <3.15" +requires-python = ">=3.11, <3.15" dependencies = [ - "numpy>=1.24", - "pandas>=2.0", - "PyYAML>=6.0", + # Python 3.14 needs higher floors: the older releases have no cp314 wheels, + # so a resolver would otherwise try to build them from source. + "numpy>=1.24; python_version < '3.14'", + "numpy>=2.3.2; python_version >= '3.14'", + "pandas>=2.0; python_version < '3.14'", + "pandas>=2.3.3; python_version >= '3.14'", + "PyYAML>=6.0; python_version < '3.14'", + "PyYAML>=6.0.3; python_version >= '3.14'", "requests>=2.28", "paho-mqtt>=2.0", "packaging>=23.0", "cachetools>=5.0", - "websockets>=11.0", + # websockets.asyncio.client is used by the HomeAssistant providers and was + # added in websockets 13.0. + "websockets>=13.0", "schedule>=1.2.0", "pytz>=2024.2" ] @@ -55,7 +62,7 @@ test = [ ] docs = [ "mkdocs-material>=9.5,<10", - "mkdocs-llmstxt>=0.5.0,<1.0; python_version >= '3.10'", + "mkdocs-llmstxt>=0.5.0,<1.0", ] # Bump Version diff --git a/run_tests.sh b/run_tests.sh index d2c01e5e..fa9351d0 100755 --- a/run_tests.sh +++ b/run_tests.sh @@ -1,6 +1,7 @@ #!/bin/bash -uv venv --python 3.13 --allow-existing +# Override with e.g. PYTHON_VERSION=3.11 ./run_tests.sh +uv venv --python "${PYTHON_VERSION:-3.14}" --allow-existing # Install the package together with test dependencies from pyproject.toml uv pip install -e '.[test]' diff --git a/src/batcontrol/async_utils.py b/src/batcontrol/async_utils.py new file mode 100644 index 00000000..c13e4aad --- /dev/null +++ b/src/batcontrol/async_utils.py @@ -0,0 +1,94 @@ +""" Helpers for running asyncio coroutines from synchronous code. + +batcontrol is a synchronous application, but a few providers talk to +HomeAssistant over a WebSocket and are therefore implemented as coroutines. +Those providers need a bridge from sync code into asyncio. + +Historically that bridge was ``asyncio.get_event_loop()`` with a +``RuntimeError`` fallback that created a loop and never closed it. Up to +Python 3.13 ``get_event_loop()`` silently created a loop when none was set; +since Python 3.14 it raises ``RuntimeError`` instead, so the fallback ran on +every call and left a loop registered as the thread's current loop. + +The helpers here delegate the loop lifecycle to ``asyncio.Runner``, which +cancels leftover tasks, flushes async generators and closes the loop on exit. +""" + +import asyncio +import contextlib +import logging +from typing import Iterator + +logger = logging.getLogger(__name__) + + +def _assert_no_running_loop(context: str) -> None: + """Raise RuntimeError if the calling thread already runs an event loop. + + Args: + context: Name used in the error message, usually the caller. + + Raises: + RuntimeError: If an event loop is already running in this thread. + """ + try: + asyncio.get_running_loop() + except RuntimeError: + # Expected case: no running loop, so we are free to create one. + return + + raise RuntimeError( + f"{context} cannot be used from a running event loop; " + "await the coroutine directly instead" + ) + + +@contextlib.contextmanager +def managed_event_loop() -> Iterator[asyncio.AbstractEventLoop]: + """Provide a private event loop for several run_until_complete() calls. + + Use this when a single connection has to be reused across multiple + coroutine invocations. The loop is closed when the block exits, also on + error. + + Yields: + A fresh event loop, set as the current loop for this thread. + + Raises: + RuntimeError: If an event loop is already running in this thread. + """ + _assert_no_running_loop("managed_event_loop()") + + with asyncio.Runner() as runner: + loop = runner.get_loop() + # Register the loop so libraries that call get_event_loop() outside of + # a running loop still find it. Runner.close() resets this. + asyncio.set_event_loop(loop) + try: + yield loop + finally: + asyncio.set_event_loop(None) + + +def run_coroutine(coro): + """Run a single coroutine from synchronous code and return its result. + + Args: + coro: The coroutine to run. + + Returns: + Whatever the coroutine returns. + + Raises: + RuntimeError: If an event loop is already running in this thread. The + coroutine is closed in that case, so it does not trigger a + "coroutine was never awaited" warning. + """ + try: + _assert_no_running_loop("run_coroutine()") + except RuntimeError: + coro.close() + raise + + with managed_event_loop() as loop: + return loop.run_until_complete(coro) diff --git a/src/batcontrol/forecastconsumption/forecast_homeassistant.py b/src/batcontrol/forecastconsumption/forecast_homeassistant.py index 5564521f..176643a3 100644 --- a/src/batcontrol/forecastconsumption/forecast_homeassistant.py +++ b/src/batcontrol/forecastconsumption/forecast_homeassistant.py @@ -5,7 +5,6 @@ statistics for each hour to predict future consumption. """ -import asyncio import datetime import json import logging @@ -15,6 +14,8 @@ import numpy as np from cachetools import TTLCache from websockets.asyncio.client import connect + +from ..async_utils import managed_event_loop, run_coroutine from .baseclass import ForecastConsumptionBaseclass logger = logging.getLogger(__name__) @@ -171,14 +172,7 @@ def _check_sensor_unit(self) -> float: ValueError: If unit_of_measurement is neither Wh nor kWh RuntimeError: If sensor cannot be queried """ - try: - loop = asyncio.get_event_loop() - except RuntimeError: - # No event loop in current thread, create a new one - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - - return loop.run_until_complete(self._check_sensor_unit_async()) + return run_coroutine(self._check_sensor_unit_async()) async def _check_sensor_unit_async(self) -> float: """Async implementation of sensor unit check @@ -566,21 +560,12 @@ def _fetch_hourly_statistics( Returns: Float """ - # Run async function in event loop - # Use asyncio.run() for Python 3.10+ compatibility - try: - loop = asyncio.get_running_loop() - # If we're already in an async context, we can't use asyncio.run() - # This shouldn't happen in practice for this method - raise RuntimeError( - "Cannot call _fetch_hourly_statistics from async context") - except RuntimeError: - # No running loop - this is the expected case - # asyncio.run() creates a new event loop, runs the coroutine, and - # closes it - return asyncio.run( - self._fetch_hourly_statistics_async(start_time, end_time) - ) + # Runs on a private event loop that is closed again afterwards. + # run_coroutine() refuses to run from an async context, which this + # method must not be called from. + return run_coroutine( + self._fetch_hourly_statistics_async(start_time, end_time) + ) def _update_cache_with_statistics( self, @@ -634,54 +619,28 @@ def _get_reference_slots(self) -> Dict[int, int]: reference_slots[day_offset] = self.history_weights[idx] return reference_slots - def refresh_data_with_limit(self, hours: int) -> None: - """Refresh historical data with specified hour limit + # pylint: disable=too-many-locals,too-many-branches + def _collect_history_periods( + self, + loop, + missing_periods, + now: datetime.datetime, + reference_slots: Dict[int, int], + history_periods: Dict[int, float] + ) -> None: + """Fetch history data for all missing hour slots over one connection. + + Opens a single WebSocket connection, queries every missing hour slot + for every configured history day, and writes the weighted result into + history_periods. Args: - hours: Number of hours to refresh (typically up to 48) + loop: Event loop used to drive the coroutines. Owned by the caller. + missing_periods: Hour offsets that are not cached yet. + now: Reference timestamp, aligned to a full hour. + reference_slots: Mapping of history day offset to weight. + history_periods: Output dict, filled with hour offset to Wh. """ - logger.info("Refreshing consumption forecast data from HomeAssistant") - - now = datetime.datetime.now(tz=self.timezone) - - # always have the next 48 hours in the forecast - # Create a list of cache_keys to ensure they are present - cache_keys = [ - self._get_cache_key( - (now + datetime.timedelta(hours=h)).weekday(), - (now + datetime.timedelta(hours=h)).hour - ) - for h in range(hours) - ] - - # Create a list of missing history data periods - missing_periods = [] - for h in range(hours): - if cache_keys[h] not in self.consumption_cache: - missing_periods.append(h) - - if missing_periods: - logger.info( - "Collecting data for missing hours: %s", - missing_periods) - else: - logger.debug( - "All forecast hours present in cache, no refresh needed") - return - - # now as full hour - now = now.replace(minute=0, second=0, microsecond=0) - - reference_slots = self._get_reference_slots() - history_periods = {} # Dict mapping hour offset to consumption value - - # Connect to WebSocket once for all requests - try: - loop = asyncio.get_event_loop() - except RuntimeError: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - websocket = None try: @@ -765,6 +724,54 @@ def refresh_data_with_limit(self, hours: int) -> None: if websocket is not None: loop.run_until_complete(self._websocket_disconnect(websocket)) + def refresh_data_with_limit(self, hours: int) -> None: + """Refresh historical data with specified hour limit + + Args: + hours: Number of hours to refresh (typically up to 48) + """ + logger.info("Refreshing consumption forecast data from HomeAssistant") + + now = datetime.datetime.now(tz=self.timezone) + + # always have the next 48 hours in the forecast + # Create a list of cache_keys to ensure they are present + cache_keys = [ + self._get_cache_key( + (now + datetime.timedelta(hours=h)).weekday(), + (now + datetime.timedelta(hours=h)).hour + ) + for h in range(hours) + ] + + # Create a list of missing history data periods + missing_periods = [] + for h in range(hours): + if cache_keys[h] not in self.consumption_cache: + missing_periods.append(h) + + if missing_periods: + logger.info( + "Collecting data for missing hours: %s", + missing_periods) + else: + logger.debug( + "All forecast hours present in cache, no refresh needed") + return + + # now as full hour + now = now.replace(minute=0, second=0, microsecond=0) + + reference_slots = self._get_reference_slots() + history_periods = {} # Dict mapping hour offset to consumption value + + # Connect to WebSocket once for all requests. The loop is reused for + # every request on that connection and closed when the block exits. + with managed_event_loop() as loop: + self._collect_history_periods( + loop, missing_periods, now, reference_slots, history_periods + ) + if not history_periods: logger.error( "No statistics data could be fetched, forecast unavailable") diff --git a/src/batcontrol/forecastsolar/forecast_homeassistant_ml.py b/src/batcontrol/forecastsolar/forecast_homeassistant_ml.py index d65d6c1d..f794b853 100644 --- a/src/batcontrol/forecastsolar/forecast_homeassistant_ml.py +++ b/src/batcontrol/forecastsolar/forecast_homeassistant_ml.py @@ -9,7 +9,6 @@ """ -import asyncio import datetime import json import logging @@ -17,6 +16,8 @@ from websockets.asyncio.client import connect from websockets.exceptions import WebSocketException + +from ..async_utils import run_coroutine from .baseclass import ForecastSolarBaseclass, ProviderError logger = logging.getLogger(__name__) @@ -143,14 +144,7 @@ def _check_sensor_unit(self) -> float: ValueError: If unit_of_measurement is neither Wh nor kWh RuntimeError: If sensor cannot be queried """ - try: - loop = asyncio.get_event_loop() - except RuntimeError: - # No event loop in current thread, create a new one - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - - return loop.run_until_complete(self._check_sensor_unit_async()) + return run_coroutine(self._check_sensor_unit_async()) async def _check_sensor_unit_async(self) -> float: """Async implementation of sensor unit check @@ -328,13 +322,7 @@ def get_raw_data_from_provider(self, pvinstallation_name: str) -> dict: ProviderError: If WebSocket connection or API request fails """ try: - loop = asyncio.get_event_loop() - except RuntimeError: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - - try: - return loop.run_until_complete(self._fetch_entity_state_async()) + return run_coroutine(self._fetch_entity_state_async()) except (OSError, WebSocketException, RuntimeError) as e: logger.error( 'HomeAssistant WebSocket request failed for entity %s: %s', diff --git a/tests/batcontrol/forecastconsumption/test_homeassistant.py b/tests/batcontrol/forecastconsumption/test_homeassistant.py index 35f0dbc9..83cad693 100644 --- a/tests/batcontrol/forecastconsumption/test_homeassistant.py +++ b/tests/batcontrol/forecastconsumption/test_homeassistant.py @@ -1,10 +1,15 @@ """Tests for HomeAssistant consumption forecasting""" +import asyncio +import contextlib import datetime +import gc import json +import warnings from unittest.mock import patch, AsyncMock import pytz import pytest +from src.batcontrol.async_utils import managed_event_loop from src.batcontrol.forecastconsumption.forecast_homeassistant import ( ForecastConsumptionHomeAssistant ) @@ -1003,3 +1008,133 @@ def test_sensor_unit_no_default_performs_autodiscovery(self, timezone): mock_check.assert_called_once() assert forecaster.sensor_unit == 'auto' assert forecaster.unit_conversion_factor == 1.0 + + +class TestEventLoopHandling: + """Regression tests for the sync-to-asyncio bridge (Python 3.14). + + Up to Python 3.13 asyncio.get_event_loop() silently created a loop; since + 3.14 it raises RuntimeError. The provider used to work around that by + creating a loop it never closed, which stayed registered as the thread's + current loop. These tests pin down that the provider now owns and releases + its loops. + """ + + @staticmethod + def _current_loop(): + """Return the thread's current event loop, or None if there is none.""" + try: + return asyncio.get_event_loop() + except RuntimeError: + return None + + @pytest.fixture(autouse=True) + def _no_current_loop(self): + """Start each test without a current event loop, as the daemon does.""" + asyncio.set_event_loop(None) + yield + asyncio.set_event_loop(None) + + @patch('src.batcontrol.forecastconsumption.forecast_homeassistant.connect') + def test_check_sensor_unit_leaves_no_current_loop( + self, mock_connect, base_config): + """_check_sensor_unit must not leave a loop registered behind.""" + mock_websocket = AsyncMock() + mock_websocket.recv = AsyncMock(side_effect=[ + json.dumps({"type": "auth_required"}), + json.dumps({"type": "auth_ok"}), + json.dumps({ + "id": 1, + "type": "result", + "success": True, + "result": [{ + "entity_id": base_config['entity_id'], + "state": "123", + "attributes": {"unit_of_measurement": "kWh"} + }] + }) + ]) + mock_websocket.send = AsyncMock() + mock_websocket.close = AsyncMock() + + async def mock_connect_coro(*args, **kwargs): + return mock_websocket + + mock_connect.side_effect = mock_connect_coro + + forecaster = ForecastConsumptionHomeAssistant(**base_config) + + assert forecaster.unit_conversion_factor == 1000.0 + assert self._current_loop() is None + + @patch('src.batcontrol.forecastconsumption.forecast_homeassistant.connect') + def test_fetch_hourly_statistics_leaves_no_current_loop( + self, mock_connect, base_config, mock_unit_check): + """The private loop of _fetch_hourly_statistics must be released.""" + forecaster = ForecastConsumptionHomeAssistant(**base_config) + + async def raise_error(*args, **kwargs): + raise OSError("Connection refused") + + mock_connect.side_effect = raise_error + + start = datetime.datetime(2025, 10, 27, 0, 0, tzinfo=pytz.UTC) + end = datetime.datetime(2025, 10, 28, 0, 0, tzinfo=pytz.UTC) + + with pytest.raises(RuntimeError): + forecaster._fetch_hourly_statistics(start, end) + + assert self._current_loop() is None + + def test_fetch_hourly_statistics_rejects_async_context( + self, base_config, mock_unit_check): + """The async-context guard used to be dead code and never fired. + + It raised RuntimeError inside the very try block that caught + RuntimeError, so the call silently fell through to asyncio.run(). + """ + forecaster = ForecastConsumptionHomeAssistant(**base_config) + + start = datetime.datetime(2025, 10, 27, 0, 0, tzinfo=pytz.UTC) + end = datetime.datetime(2025, 10, 28, 0, 0, tzinfo=pytz.UTC) + + async def call_from_async_context(): + with pytest.raises(RuntimeError, match="running event loop"): + forecaster._fetch_hourly_statistics(start, end) + + # The old code let the coroutine fall through to asyncio.run(), which + # abandoned it and emitted "coroutine was never awaited". + with warnings.catch_warnings(): + warnings.simplefilter("error", RuntimeWarning) + asyncio.run(call_from_async_context()) + gc.collect() + + @patch('src.batcontrol.forecastconsumption.forecast_homeassistant.connect') + def test_refresh_data_closes_its_loop( + self, mock_connect, base_config, mock_unit_check): + """refresh_data_with_limit must close the loop it opened.""" + forecaster = ForecastConsumptionHomeAssistant(**base_config) + + async def raise_error(*args, **kwargs): + raise OSError("Connection refused") + + mock_connect.side_effect = raise_error + + captured = {} + real_managed_event_loop = managed_event_loop + + @contextlib.contextmanager + def spy(): + with real_managed_event_loop() as loop: + captured['loop'] = loop + yield loop + + with patch( + 'src.batcontrol.forecastconsumption.forecast_homeassistant' + '.managed_event_loop', + spy + ): + forecaster.refresh_data_with_limit(2) + + assert captured['loop'].is_closed() + assert self._current_loop() is None diff --git a/tests/batcontrol/test_async_utils.py b/tests/batcontrol/test_async_utils.py new file mode 100644 index 00000000..6fd138bd --- /dev/null +++ b/tests/batcontrol/test_async_utils.py @@ -0,0 +1,170 @@ +"""Tests for the sync-to-asyncio bridge in batcontrol.async_utils. + +These cover the behaviour that broke on Python 3.14, where +asyncio.get_event_loop() raises RuntimeError instead of silently creating a +loop. The helpers must own the loop lifecycle themselves and never leak a loop. +""" + +import asyncio +import gc +import warnings + +import pytest + +from batcontrol.async_utils import managed_event_loop, run_coroutine + + +def _current_loop(): + """Return the thread's current event loop, or None if there is none.""" + try: + return asyncio.get_event_loop() + except RuntimeError: + return None + + +def _open_loops(): + """Count event loop objects that are alive and not closed.""" + gc.collect() + return sum( + 1 for obj in gc.get_objects() + if isinstance(obj, asyncio.AbstractEventLoop) and not obj.is_closed() + ) + + +class TestRunCoroutine: + """run_coroutine() runs a coroutine from synchronous code.""" + + def test_returns_result(self): + async def work(): + await asyncio.sleep(0) + return 42 + + assert run_coroutine(work()) == 42 + + def test_propagates_exception(self): + async def work(): + raise ValueError("boom") + + with pytest.raises(ValueError, match="boom"): + run_coroutine(work()) + + def test_leaves_no_current_loop_behind(self): + async def work(): + return None + + run_coroutine(work()) + assert _current_loop() is None + + def test_does_not_leak_loops_across_calls(self): + """The 3.14 regression: one leaked loop per call.""" + async def work(): + return 1 + + run_coroutine(work()) # warm up, ignore one-off allocations + before = _open_loops() + for _ in range(5): + run_coroutine(work()) + assert _open_loops() <= before + + def test_closes_loop_even_when_coroutine_raises(self): + async def work(): + raise RuntimeError("inner") + + before = _open_loops() + with pytest.raises(RuntimeError, match="inner"): + run_coroutine(work()) + assert _open_loops() <= before + assert _current_loop() is None + + def test_cancels_leftover_tasks(self): + """A task the coroutine never awaited must not keep the loop busy.""" + leftover = {} + + async def work(): + async def forever(): + await asyncio.sleep(3600) + + leftover["task"] = asyncio.ensure_future(forever()) + await asyncio.sleep(0) + return "done" + + assert run_coroutine(work()) == "done" + assert leftover["task"].cancelled() or leftover["task"].done() + + +class TestRunCoroutineFromAsyncContext: + """run_coroutine() must refuse to nest inside a running loop.""" + + def test_raises_inside_running_loop(self): + async def inner(): + return 1 + + async def outer(): + with pytest.raises(RuntimeError, match="running event loop"): + run_coroutine(inner()) + + asyncio.run(outer()) + + def test_refusal_does_not_warn_about_unawaited_coroutine(self): + async def inner(): + return 1 + + async def outer(): + coro = inner() + with pytest.raises(RuntimeError): + run_coroutine(coro) + + with warnings.catch_warnings(): + warnings.simplefilter("error", RuntimeWarning) + asyncio.run(outer()) + gc.collect() + + +class TestManagedEventLoop: + """managed_event_loop() hands out a loop for repeated use.""" + + def test_loop_is_usable_for_several_calls(self): + async def work(value): + await asyncio.sleep(0) + return value * 2 + + with managed_event_loop() as loop: + results = [loop.run_until_complete(work(i)) for i in range(3)] + + assert results == [0, 2, 4] + + def test_loop_is_current_inside_block(self): + with managed_event_loop() as loop: + assert _current_loop() is loop + + def test_loop_is_closed_after_block(self): + with managed_event_loop() as loop: + pass + + assert loop.is_closed() + assert _current_loop() is None + + def test_loop_is_closed_when_block_raises(self): + with pytest.raises(ValueError, match="failed"): + with managed_event_loop() as loop: + raise ValueError("failed") + + assert loop.is_closed() + assert _current_loop() is None + + def test_does_not_leak_loops_across_blocks(self): + with managed_event_loop(): + pass + before = _open_loops() + for _ in range(5): + with managed_event_loop(): + pass + assert _open_loops() <= before + + def test_raises_inside_running_loop(self): + async def outer(): + with pytest.raises(RuntimeError, match="running event loop"): + with managed_event_loop(): + pass + + asyncio.run(outer()) diff --git a/tests/test_forecast_solar_homeassistant_ml.py b/tests/test_forecast_solar_homeassistant_ml.py index d850b644..560cb846 100644 --- a/tests/test_forecast_solar_homeassistant_ml.py +++ b/tests/test_forecast_solar_homeassistant_ml.py @@ -941,3 +941,119 @@ def test_check_sensor_unit_async_none_unit_defaults_to_wh(self, pv_installations if __name__ == "__main__": pytest.main([__file__, "-v"]) + + +class TestEventLoopHandling: + """Regression tests for the sync-to-asyncio bridge (Python 3.14). + + Up to Python 3.13 asyncio.get_event_loop() silently created a loop; since + 3.14 it raises RuntimeError. The provider used to work around that by + creating a loop it never closed, which stayed registered as the thread's + current loop. These tests pin down that the provider now releases its loop. + """ + + @staticmethod + def _current_loop(): + """Return the thread's current event loop, or None if there is none.""" + try: + return asyncio.get_event_loop() + except RuntimeError: + return None + + @pytest.fixture(autouse=True) + def _no_current_loop(self): + """Start each test without a current event loop, as the daemon does.""" + asyncio.set_event_loop(None) + yield + asyncio.set_event_loop(None) + + @staticmethod + def _provider(pv_installations, timezone, sensor_unit="kWh"): + """Build a provider without triggering the unit auto-detection.""" + return ForecastSolarHomeAssistantML( + pvinstallations=pv_installations, + timezone=timezone, + base_url="http://homeassistant.local:8123", + api_token="test_token", + entity_id="sensor.solar_forecast", + sensor_unit=sensor_unit + ) + + def test_check_sensor_unit_leaves_no_current_loop( + self, pv_installations, timezone): + """sensor_unit='auto' runs _check_sensor_unit during construction.""" + mock_ws = AsyncMock() + mock_ws.recv = AsyncMock(side_effect=[ + json.dumps({"type": "auth_required"}), + json.dumps({"type": "auth_ok"}), + json.dumps({ + "id": 1, + "type": "result", + "success": True, + "result": [{ + "entity_id": "sensor.solar_forecast", + "state": "123", + "attributes": {"unit_of_measurement": "kWh"} + }] + }) + ]) + mock_ws.send = AsyncMock() + mock_ws.close = AsyncMock() + + async def mock_connect_coro(*args, **kwargs): + return mock_ws + + with patch( + 'src.batcontrol.forecastsolar.forecast_homeassistant_ml.connect', + side_effect=mock_connect_coro + ): + provider = self._provider( + pv_installations, timezone, sensor_unit="auto") + + assert provider.unit_conversion_factor == 1000.0 + assert self._current_loop() is None + + def test_get_raw_data_leaves_no_current_loop_on_success( + self, pv_installations, timezone, ha_entity_state): + """A successful fetch must not leave a loop registered behind.""" + provider = self._provider(pv_installations, timezone) + + async def fake_fetch(): + return ha_entity_state + + with patch.object(provider, '_fetch_entity_state_async', fake_fetch): + result = provider.get_raw_data_from_provider( + pv_installations[0]['name']) + + assert result == ha_entity_state + assert self._current_loop() is None + + def test_get_raw_data_leaves_no_current_loop_on_failure( + self, pv_installations, timezone): + """A failing fetch must also release the loop.""" + provider = self._provider(pv_installations, timezone) + + async def raise_error(*args, **kwargs): + raise OSError("Connection refused") + + with patch( + 'src.batcontrol.forecastsolar.forecast_homeassistant_ml.connect', + side_effect=raise_error + ): + with pytest.raises(ProviderError): + provider.get_raw_data_from_provider( + pv_installations[0]['name']) + + assert self._current_loop() is None + + def test_get_raw_data_rejects_async_context( + self, pv_installations, timezone): + """Calling the sync wrapper from a running loop must fail loudly.""" + provider = self._provider(pv_installations, timezone) + + async def call_from_async_context(): + with pytest.raises(ProviderError, match="running event loop"): + provider.get_raw_data_from_provider( + pv_installations[0]['name']) + + asyncio.run(call_from_async_context())