fix(config): enable documentation mode only explicitly

ConfigEOS.documentation_mode() was true as soon as Sphinx was imported
anywhere in the process. A test importing Sphinx therefore switched all
later tests into documentation mode, and test_config_md_current failed
depending on test order. Documentation mode is now enabled only by the
documentation generators (_force_documentation_mode) or by the Sphinx
configuration, which sets sys._called_from_sphinx in docs/conf.py.

Declare _force_documentation_mode as a ClassVar. Without the annotation
pydantic turned it into a private attribute, and the class attribute was
a truthy descriptor object.

Behavior change: because of that descriptor, every running EOS was in
documentation mode, so the get_adapter/config/measurement/prediction/ems/
database/resource_registry() guards never raised "access before init" and
created the singletons lazily on first access. They now raise for every
access before initialization. The worker process of uvicorn's auto-reload
(server.reload, make run-dev) relied on that: the app's lifespan now sets up
the configuration and initializes the singletons that do not exist yet.

Leave runtime state out of the documentation: the Home Assistant entity ID
properties and OptimizationCommonSettings.keys return an empty list in
documentation mode, so entity IDs and solution keys of a running system
no longer reach the generated configuration docs.

Keep the tests isolated: test_doc sets documentation mode on the class
and test_docstringrst restores sys._called_from_sphinx after reading the
Sphinx configuration. Add tests for how documentation mode is enabled and
for the runtime values it leaves out.

Additional changes
- Generate openapi.json in documentation mode, as the config docs.
- Share the documentation mode check as coreabc.documentation_mode().
- Move the logging and server address setup of run_eos() to setup_eos_config().
- Test that EOS starts with server.reload.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Bobby Noelte
2026-11-09 12:40:15 +01:00
co-authored by Claude Opus 5.5
parent a0e57a18b7
commit 57132df44e
12 changed files with 293 additions and 29 deletions
+4
View File
@@ -12,6 +12,10 @@ PROJECT_ROOT = Path(__file__).parent.parent
SRC_DIR = PROJECT_ROOT / "src" SRC_DIR = PROJECT_ROOT / "src"
sys.path.insert(0, str(SRC_DIR)) sys.path.insert(0, str(SRC_DIR))
# Tell akkudoktoreos that it is imported for the documentation build
# (see ConfigEOS.documentation_mode()).
sys._called_from_sphinx = True # type: ignore[attr-defined]
from akkudoktoreos.core.version import __version__ from akkudoktoreos.core.version import __version__
# -- Project information ----------------------------------------------------- # -- Project information -----------------------------------------------------
+17 -9
View File
@@ -25,21 +25,29 @@ PROJECT_ROOT = Path(__file__).parent.parent
SRC_DIR = PROJECT_ROOT / "src" SRC_DIR = PROJECT_ROOT / "src"
sys.path.insert(0, str(SRC_DIR)) sys.path.insert(0, str(SRC_DIR))
from akkudoktoreos.config.config import ConfigEOS
from akkudoktoreos.core.coreabc import get_config from akkudoktoreos.core.coreabc import get_config
from akkudoktoreos.server.eos import app from akkudoktoreos.server.eos import app
def generate_openapi() -> dict: def generate_openapi() -> dict:
# Make minimal config to make the generation reproducable # Documentation mode: leave runtime state (e.g. optimization solution keys,
config_eos = get_config(init={ # Home Assistant entity IDs) out of the specification.
"with_init_settings": True, force_documentation_mode = ConfigEOS._force_documentation_mode
"with_env_settings": False, ConfigEOS._force_documentation_mode = True
"with_dotenv_settings": False, try:
"with_file_settings": False, # Make minimal config to make the generation reproducable
"with_file_secret_settings": False, config_eos = get_config(init={
}) "with_init_settings": True,
"with_env_settings": False,
"with_dotenv_settings": False,
"with_file_settings": False,
"with_file_secret_settings": False,
})
openapi_spec = app.openapi() openapi_spec = app.openapi()
finally:
ConfigEOS._force_documentation_mode = force_documentation_mode
config_schema = ( config_schema = (
openapi_spec openapi_spec
+10 -1
View File
@@ -10,7 +10,7 @@ from pydantic import Field, PrivateAttr, computed_field, field_validator
from akkudoktoreos.adapter.adapterabc import AdapterProvider from akkudoktoreos.adapter.adapterabc import AdapterProvider
from akkudoktoreos.config.configabc import SettingsBaseModel from akkudoktoreos.config.configabc import SettingsBaseModel
from akkudoktoreos.core.coreabc import get_adapter from akkudoktoreos.core.coreabc import documentation_mode, get_adapter
from akkudoktoreos.core.emplan import ( from akkudoktoreos.core.emplan import (
DDBCInstruction, DDBCInstruction,
FRBCInstruction, FRBCInstruction,
@@ -147,6 +147,9 @@ class HomeAssistantAdapterCommonSettings(SettingsBaseModel):
@property @property
def homeassistant_entity_ids(self) -> list[str]: def homeassistant_entity_ids(self) -> list[str]:
"""Entity IDs available at Home Assistant.""" """Entity IDs available at Home Assistant."""
if documentation_mode():
# Documentation shows no runtime state.
return []
try: try:
adapter_eos = get_adapter() adapter_eos = get_adapter()
provider = adapter_eos.provider_by_id("HomeAssistant") provider = adapter_eos.provider_by_id("HomeAssistant")
@@ -164,6 +167,9 @@ class HomeAssistantAdapterCommonSettings(SettingsBaseModel):
@property @property
def eos_solution_entity_ids(self) -> list[str]: def eos_solution_entity_ids(self) -> list[str]:
"""Entity IDs for optimization solution available at EOS.""" """Entity IDs for optimization solution available at EOS."""
if documentation_mode():
# Documentation shows no runtime state.
return []
try: try:
adapter_eos = get_adapter() adapter_eos = get_adapter()
provider = adapter_eos.provider_by_id("HomeAssistant") provider = adapter_eos.provider_by_id("HomeAssistant")
@@ -181,6 +187,9 @@ class HomeAssistantAdapterCommonSettings(SettingsBaseModel):
@property @property
def eos_device_instruction_entity_ids(self) -> list[str]: def eos_device_instruction_entity_ids(self) -> list[str]:
"""Entity IDs for energy management instructions available at EOS.""" """Entity IDs for energy management instructions available at EOS."""
if documentation_mode():
# Documentation shows no runtime state.
return []
try: try:
adapter_eos = get_adapter() adapter_eos = get_adapter()
provider = adapter_eos.provider_by_id("HomeAssistant") provider = adapter_eos.provider_by_id("HomeAssistant")
+12 -5
View File
@@ -466,7 +466,9 @@ class ConfigEOS(SingletonMixin, SettingsEOSDefaults):
_config_autosave: ClassVar[str] = "" _config_autosave: ClassVar[str] = ""
# Settings provided at runtime, e.g. by the REST interface. Highest priority after CLI. # Settings provided at runtime, e.g. by the REST interface. Highest priority after CLI.
_runtime_settings: ClassVar[dict[str, Any]] = {} _runtime_settings: ClassVar[dict[str, Any]] = {}
_force_documentation_mode = False # A ClassVar: without the annotation pydantic makes it a private attribute,
# and the class attribute would be a (truthy) descriptor object.
_force_documentation_mode: ClassVar[bool] = False
def __hash__(self) -> int: def __hash__(self) -> int:
# ConfigEOS is a singleton # ConfigEOS is a singleton
@@ -767,11 +769,16 @@ class ConfigEOS(SingletonMixin, SettingsEOSDefaults):
def documentation_mode(cls) -> bool: def documentation_mode(cls) -> bool:
"""Are we running in documentation mode. """Are we running in documentation mode.
Some checks may be relaxed to allow for proper documentation execution. Some checks may be relaxed to allow for proper documentation execution,
and runtime state is left out of the documentation.
Documentation mode is set explicitly: by the documentation generators
(``_force_documentation_mode``) or by the Sphinx configuration
(``sys._called_from_sphinx``). Merely having Sphinx imported, e.g. by a
test in the same process, does not enable it.
""" """
# Detect if Sphinx is importing this module is_sphinx = getattr(sys, "_called_from_sphinx", False)
is_sphinx = "sphinx" in sys.modules or getattr(sys, "_called_from_sphinx", False) return bool(cls._force_documentation_mode or is_sphinx)
return cls._force_documentation_mode or is_sphinx
def __init__(self, *args: Any, **kwargs: Any) -> None: def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Initializes the singleton ConfigEOS instance. """Initializes the singleton ConfigEOS instance.
+25
View File
@@ -106,6 +106,31 @@ class AdapterMixin:
return get_adapter() return get_adapter()
def documentation_mode() -> bool:
"""Return whether EOS runs in documentation mode.
Shortcut for ``ConfigEOS.documentation_mode()`` for modules that cannot
import ``ConfigEOS`` at module level. Documentation shows no runtime state,
e.g. entity IDs of a Home Assistant instance or keys of an optimization
solution.
Returns:
True if the documentation is being generated.
"""
from akkudoktoreos.config.config import ConfigEOS
return ConfigEOS.documentation_mode()
def config_initialized() -> bool:
"""Return whether the configuration singleton exists, without creating it.
Returns:
True if ``get_config(init=...)`` already created the configuration.
"""
return _config_eos is not None
def get_config(init: Union[bool, dict[str, bool]] = False) -> ConfigEOS: def get_config(init: Union[bool, dict[str, bool]] = False) -> ConfigEOS:
"""Retrieve the singleton EOS configuration instance. """Retrieve the singleton EOS configuration instance.
@@ -4,7 +4,7 @@ from typing import Any, Optional
from pydantic import Field, computed_field, model_validator from pydantic import Field, computed_field, model_validator
from akkudoktoreos.config.configabc import SettingsBaseModel from akkudoktoreos.config.configabc import SettingsBaseModel
from akkudoktoreos.core.coreabc import get_ems from akkudoktoreos.core.coreabc import documentation_mode, get_ems
from akkudoktoreos.core.pydantic import ( from akkudoktoreos.core.pydantic import (
PydanticBaseModel, PydanticBaseModel,
PydanticDateTimeDataFrame, PydanticDateTimeDataFrame,
@@ -92,6 +92,9 @@ class OptimizationCommonSettings(SettingsBaseModel):
@property @property
def keys(self) -> list[str]: def keys(self) -> list[str]:
"""The keys of the solution.""" """The keys of the solution."""
if documentation_mode():
# Documentation shows no runtime state, e.g. of a previous run.
return []
try: try:
ems_eos = get_ems() ems_eos = get_ems()
except Exception: except Exception:
+34 -10
View File
@@ -29,6 +29,7 @@ from loguru import logger
from akkudoktoreos.config.config import ConfigEOS, SettingsEOS from akkudoktoreos.config.config import ConfigEOS, SettingsEOS
from akkudoktoreos.core.cache import CacheFileStore, cache_clear, cache_load, cache_save from akkudoktoreos.core.cache import CacheFileStore, cache_clear, cache_load, cache_save
from akkudoktoreos.core.coreabc import ( from akkudoktoreos.core.coreabc import (
config_initialized,
get_config, get_config,
get_ems, get_ems,
get_measurement, get_measurement,
@@ -189,6 +190,13 @@ def config_eos_ready() -> bool:
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
"""Lifespan manager for the app.""" """Lifespan manager for the app."""
# On startup # On startup
if not config_initialized():
# The worker process of uvicorn's auto-reload (server.reload) imports the
# app afresh; run_eos() did not run in this process.
setup_eos_config(get_config(init=True))
# Create the singletons that do not exist yet; a no-op after run_eos().
singletons_init()
await load_eos_state() await load_eos_state()
# Prepare the Manager and all task that are handled by the manager # Prepare the Manager and all task that are handled by the manager
@@ -2498,19 +2506,18 @@ def redirect(request: Request, path: str) -> Union[HTMLResponse, RedirectRespons
return RedirectResponse(url="/docs", status_code=303) return RedirectResponse(url="/docs", status_code=303)
def run_eos() -> None: def setup_eos_config(config_eos: ConfigEOS) -> None:
"""Run the EOS server with the specified configurations. """Set up logging and the server address defaults from the configuration.
Starts the EOS server using the Uvicorn ASGI server. Logs an error and exits if Called once per process that serves the app: by ``run_eos()``, and by the
binding to the host and port fails. app's lifespan in the worker process of uvicorn's auto-reload.
Returns: Args:
None config_eos: The EOS configuration of this process.
Raises:
RuntimeError: If no server port is set after applying the defaults.
""" """
# get_config(init=True) creates the configuration, includes also ARGV args.
# This should not be done before nor later
config_eos = get_config(init=True)
# Set logging to what is in config # Set logging to what is in config
logger.remove() logger.remove()
logging_track_config(config_eos, "logging", None, None) logging_track_config(config_eos, "logging", None, None)
@@ -2532,6 +2539,23 @@ def run_eos() -> None:
if config_eos.server.eosdash_port is None: if config_eos.server.eosdash_port is None:
config_eos.set_nested_value("server/eosdash_port", config_eos.server.port + 1) config_eos.set_nested_value("server/eosdash_port", config_eos.server.port + 1)
def run_eos() -> None:
"""Run the EOS server with the specified configurations.
Starts the EOS server using the Uvicorn ASGI server. Logs an error and exits if
binding to the host and port fails.
Returns:
None
"""
# get_config(init=True) creates the configuration, includes also ARGV args.
# This should not be done before nor later
config_eos = get_config(init=True)
setup_eos_config(config_eos)
if config_eos.server.port is None: # make mypy happy
raise RuntimeError("server.port is None despite default setup")
# Switch data directories ownership to user # Switch data directories ownership to user
fix_data_directories_permissions(run_as_user=config_eos.server.run_as_user) fix_data_directories_permissions(run_as_user=config_eos.server.run_as_user)
+39
View File
@@ -885,3 +885,42 @@ class TestConfigProviderValidation:
for section, value in providers.items(): for section, value in providers.items():
assert getattr(config, section).provider == value["provider"] assert getattr(config, section).provider == value["provider"]
class TestDocumentationMode:
"""Documentation mode is enabled explicitly, never by an imported Sphinx."""
def test_imported_sphinx_does_not_enable_it(self, monkeypatch):
import sphinx # noqa: F401 # the module is in sys.modules now
monkeypatch.delattr("sys._called_from_sphinx", raising=False)
monkeypatch.setattr(ConfigEOS, "_force_documentation_mode", False)
assert not ConfigEOS.documentation_mode()
def test_sphinx_configuration_enables_it(self, monkeypatch):
monkeypatch.setattr("sys._called_from_sphinx", True, raising=False)
monkeypatch.setattr(ConfigEOS, "_force_documentation_mode", False)
assert ConfigEOS.documentation_mode()
def test_documentation_generators_enable_it(self, monkeypatch):
monkeypatch.delattr("sys._called_from_sphinx", raising=False)
monkeypatch.setattr(ConfigEOS, "_force_documentation_mode", True)
assert ConfigEOS.documentation_mode()
def test_optimization_solution_keys_are_left_out(self, monkeypatch):
"""Keys of a solution of a previous run do not reach the documentation."""
from akkudoktoreos.optimization import optimization
from akkudoktoreos.optimization.optimization import OptimizationCommonSettings
solution = Mock()
solution.solution.to_dataframe.return_value.columns.tolist.return_value = ["b", "a"]
ems = Mock()
ems.optimization_solution.return_value = solution
monkeypatch.setattr(optimization, "get_ems", lambda: ems)
monkeypatch.delattr("sys._called_from_sphinx", raising=False)
settings = OptimizationCommonSettings()
monkeypatch.setattr(ConfigEOS, "_force_documentation_mode", False)
assert settings.keys == ["a", "b"]
monkeypatch.setattr(ConfigEOS, "_force_documentation_mode", True)
assert settings.keys == []
+4 -2
View File
@@ -9,6 +9,8 @@ from unittest.mock import patch
import pytest import pytest
from akkudoktoreos.config.config import ConfigEOS
DIR_PROJECT_ROOT = Path(__file__).parent.parent DIR_PROJECT_ROOT = Path(__file__).parent.parent
DIR_TESTDATA = Path(__file__).parent / "testdata" DIR_TESTDATA = Path(__file__).parent / "testdata"
@@ -205,10 +207,10 @@ def test_config_md_current(config_eos, set_other_timezone):
# Create test files # Create test files
try: try:
config_eos._force_documentation_mode = True ConfigEOS._force_documentation_mode = True
config_md = generate_config_md.generate_config_md(tested[0], config_eos) config_md = generate_config_md.generate_config_md(tested[0], config_eos)
finally: finally:
config_eos._force_documentation_mode = False ConfigEOS._force_documentation_mode = False
# Check test files are the same as the expected files # Check test files are the same as the expected files
for i, expected_path in enumerate(expected): for i, expected_path in enumerate(expected):
+12 -1
View File
@@ -142,7 +142,18 @@ sphinx_conf = importlib.util.module_from_spec(spec)
sys.modules["sphinx_conf"] = sphinx_conf sys.modules["sphinx_conf"] = sphinx_conf
if spec.loader is None: if spec.loader is None:
raise AssertionError(f"Can not import sphinx_conf from {conf_path}") raise AssertionError(f"Can not import sphinx_conf from {conf_path}")
spec.loader.exec_module(sphinx_conf) # conf.py marks the process as a Sphinx build, which enables documentation mode
# of akkudoktoreos. Only its settings are read here: keep the other tests of this
# process out of documentation mode.
_called_from_sphinx = getattr(sys, "_called_from_sphinx", None)
try:
spec.loader.exec_module(sphinx_conf)
finally:
if _called_from_sphinx is None:
if hasattr(sys, "_called_from_sphinx"):
del sys._called_from_sphinx
else:
sys._called_from_sphinx = _called_from_sphinx # type: ignore[attr-defined]
# Build NapoleonConfig with all options # Build NapoleonConfig with all options
napoleon_config = NapoleonConfig( napoleon_config = NapoleonConfig(
+111
View File
@@ -2,6 +2,9 @@ import asyncio
import json import json
import os import os
import re import re
import socket
import subprocess
import sys
import time import time
from http import HTTPStatus from http import HTTPStatus
from pathlib import Path from pathlib import Path
@@ -395,6 +398,114 @@ class TestServerWithEnv:
) )
class TestServerReload:
"""EOS with uvicorn's auto-reload (``server.reload``, used by ``make run-dev``)."""
STARTUP_TIMEOUT_S = 90
@staticmethod
def free_port() -> int:
"""Return a free TCP port on the default host that EOS accepts (1024..49151).
Starts above the ports of the other server tests (8503, 8504, 8555).
"""
for port in range(8513, 8600):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
try:
sock.bind((get_default_host(), port))
except OSError:
continue
return port
pytest.fail("No free port for the EOS reload test in 8513..8599")
@staticmethod
def stop_process_tree(process: subprocess.Popen) -> None:
"""Stop a process and all of its descendants (reloader, worker)."""
try:
parent = psutil.Process(process.pid)
processes = [*parent.children(recursive=True), parent]
except psutil.NoSuchProcess:
return
for proc in processes:
try:
proc.terminate()
except psutil.NoSuchProcess:
pass
_, alive = psutil.wait_procs(processes, timeout=15)
for proc in alive:
try:
proc.kill()
except psutil.NoSuchProcess:
pass
process.wait(timeout=15)
def test_server_starts_with_reload(self, tmp_path: Path) -> None:
"""The reload worker initializes the singletons that run_eos() set up in the parent.
The worker process imports the app afresh. Without initialization in the
app's lifespan, the startup fails with 'access before init'.
"""
eos_dir = tmp_path / "eos"
eos_dir.mkdir()
# An empty config file, so EOS does not fall back to the user's config file.
(eos_dir / "EOS.config.json").write_text("{}", encoding="utf-8")
host = get_default_host()
port = self.free_port()
env = os.environ.copy()
env.update(
{
"EOS_DIR": str(eos_dir),
"EOS_CONFIG_DIR": str(eos_dir),
"PYTHONIOENCODING": "utf-8",
}
)
log_path = tmp_path / "eos.log"
with log_path.open("w", encoding="utf-8") as log_file:
# cwd: the reloader watches the working directory, not the repository.
process = subprocess.Popen(
[
sys.executable,
"-m",
"akkudoktoreos.server.eos",
"--host",
host,
"--port",
str(port),
"--reload",
"true",
"--startup_eosdash",
"false",
],
cwd=tmp_path,
env=env,
stdout=log_file,
stderr=subprocess.STDOUT,
)
try:
health: dict = {}
deadline = time.monotonic() + self.STARTUP_TIMEOUT_S
while time.monotonic() < deadline and process.poll() is None:
try:
response = requests.get(f"http://{host}:{port}/v1/health", timeout=2)
if response.status_code == HTTPStatus.OK:
health = response.json()
break
except requests.RequestException:
pass
time.sleep(1)
worker_pids = {
child.pid for child in psutil.Process(process.pid).children(recursive=True)
} if process.poll() is None else set()
finally:
self.stop_process_tree(process)
log = log_path.read_text(encoding="utf-8", errors="replace")
assert health.get("status") == "alive", f"EOS with --reload did not start:\n{log}"
assert "before init" not in log, log
# The app runs in the reload worker, a child of the started process.
assert health.get("pid") in worker_pids
class TestEosdashRedirect: class TestEosdashRedirect:
"""Redirects to EOSdash must target an address the client can reach. """Redirects to EOSdash must target an address the client can reach.
+21
View File
@@ -117,6 +117,27 @@ def test_homeassistant_settings_preserve_entity_lookup_and_unavailability(
assert getattr(settings, property_name) == [] assert getattr(settings, property_name) == []
@pytest.mark.parametrize(
"property_name",
["homeassistant_entity_ids", "eos_solution_entity_ids", "eos_device_instruction_entity_ids"],
)
def test_homeassistant_settings_leave_runtime_entities_out_of_documentation(
monkeypatch: pytest.MonkeyPatch, property_name: str
) -> None:
"""Entity IDs of a running system do not reach the documentation."""
provider = HomeAssistantAdapter()
lookup = MagicMock(return_value=["sensor.eos_example"])
monkeypatch.setattr(HomeAssistantAdapter, f"get_{property_name}", lookup)
adapter = MagicMock()
adapter.provider_by_id.return_value = provider
monkeypatch.setattr(homeassistant, "get_adapter", lambda: adapter)
monkeypatch.setattr(ConfigEOS, "_force_documentation_mode", True)
settings = HomeAssistantAdapterCommonSettings()
assert getattr(settings, property_name) == []
lookup.assert_not_called()
@pytest.mark.parametrize("operation", ["revert_settings", "list_backups"]) @pytest.mark.parametrize("operation", ["revert_settings", "list_backups"])
def test_backup_operations_report_an_uninitialized_path_as_an_invariant_failure( def test_backup_operations_report_an_uninitialized_path_as_an_invariant_failure(
config_eos: ConfigEOS, monkeypatch: pytest.MonkeyPatch, operation: str config_eos: ConfigEOS, monkeypatch: pytest.MonkeyPatch, operation: str