],
default_value="GLOBAL",
category="advanced",
+ requires_reload=False, # applied dynamically via _set_logger()
)
DEFAULT_PROVIDER_CONFIG_ENTRIES = (CONF_ENTRY_LOG_LEVEL,)
label="Enforce Gapless playback with Queue Flow Mode streaming",
default_value=False,
category="advanced",
+ requires_reload=True,
)
],
default_value="default",
category="advanced",
+ requires_reload=True,
)
CONF_ENTRY_LIBRARY_SYNC_ALBUMS = ConfigEntry(
key="library_sync_albums",
"otherwise audio streaming will not work.",
required=False,
category="advanced",
+ requires_reload=True,
),
ConfigEntry(
key=CONF_BIND_PORT,
"Make sure that this server can be reached "
"on the given IP and TCP port by players on the local network.",
category="advanced",
+ requires_reload=True,
),
ConfigEntry(
key=CONF_BIND_IP,
"not be adjusted in regular setups.",
category="advanced",
required=False,
+ requires_reload=True,
),
ConfigEntry(
key=CONF_SMART_FADES_LOG_LEVEL,
)
from music_assistant.constants import (
- CONF_AUTH_ALLOW_SELF_REGISTRATION,
DB_TABLE_PLAYLOG,
HOMEASSISTANT_SYSTEM_USER,
MASS_LOGGER_NAME,
HomeAssistantOAuthProvider,
HomeAssistantProviderConfig,
LoginProvider,
- LoginProviderConfig,
normalize_username,
)
from music_assistant.helpers.api import api_command
async def setup(self) -> None:
"""Initialize the authentication manager."""
- # Get auth settings from config
- allow_self_registration = self.webserver.config.get_value(CONF_AUTH_ALLOW_SELF_REGISTRATION)
- assert isinstance(allow_self_registration, bool)
-
# Setup database
db_path = self.mass.storage_path + "/auth.db"
self.database = DatabaseConnection(db_path)
jwt_secret = await self._get_or_create_jwt_secret()
self.jwt_helper = JWTHelper(jwt_secret)
- # Setup login providers based on config
- await self._setup_login_providers(allow_self_registration)
+ # Setup login providers
+ await self._setup_login_providers()
self._has_users = await self._has_non_system_users()
self.logger.info("Generated new JWT secret key")
return jwt_secret
- async def _setup_login_providers(self, allow_self_registration: bool) -> None:
- """
- Set up available login providers based on configuration.
-
- :param allow_self_registration: Whether to allow self-registration via OAuth.
- """
+ async def _setup_login_providers(self) -> None:
+ """Set up available login providers based on configuration."""
# Always enable built-in provider
- builtin_config: LoginProviderConfig = {"allow_self_registration": False}
- self.login_providers["builtin"] = BuiltinLoginProvider(self.mass, "builtin", builtin_config)
+ self.login_providers["builtin"] = BuiltinLoginProvider(self.mass, "builtin", {})
# Home Assistant OAuth provider
# Automatically enabled if HA provider (plugin) is configured
# Get URL from the HA provider config
ha_url = ha_provider.config.get_value("url")
assert isinstance(ha_url, str)
- ha_config: HomeAssistantProviderConfig = {
- "ha_url": ha_url,
- "allow_self_registration": allow_self_registration,
- }
+ ha_config: HomeAssistantProviderConfig = {"ha_url": ha_url}
self.login_providers["homeassistant"] = HomeAssistantOAuthProvider(
self.mass, "homeassistant", ha_config
)
if ha_provider:
# HA provider exists and is available - ensure OAuth provider is registered
if "homeassistant" not in self.login_providers:
- # Get allow_self_registration config
- allow_self_registration = bool(
- self.webserver.config.get_value(CONF_AUTH_ALLOW_SELF_REGISTRATION, True)
- )
-
# Get URL from the HA provider config
ha_url = ha_provider.config.get_value("url")
assert isinstance(ha_url, str)
- ha_config: HomeAssistantProviderConfig = {
- "ha_url": ha_url,
- "allow_self_registration": allow_self_registration,
- }
+ ha_config: HomeAssistantProviderConfig = {"ha_url": ha_url}
self.login_providers["homeassistant"] = HomeAssistantOAuthProvider(
self.mass, "homeassistant", ha_config
)
@property
def base_url(self) -> str:
- """Return the base_url for the streamserver."""
- return self._server.base_url
+ """Return the base_url for the webserver."""
+ return str(self.config.get_value(CONF_BASE_URL)).removesuffix("/")
async def get_config_entries(
self,
label="Allow User Self-Registration",
description="Allow users to create accounts via Home Assistant OAuth.",
hidden=not any(provider.domain == "hass" for provider in self.mass.providers),
+ requires_reload=False,
),
ConfigEntry(
key=CONF_BASE_URL,
description="The (base) URL to reach this webserver in the network. \n"
"Override this in advanced scenarios where for example you're running "
"the webserver behind a reverse proxy.",
+ requires_reload=False,
),
ConfigEntry(
key=CONF_BIND_PORT,
default_value=DEFAULT_SERVER_PORT,
label="TCP Port",
description="The TCP port to run the webserver.",
+ requires_reload=True,
),
ConfigEntry(
key="webserver_warn",
label="Enable SSL/TLS",
description="Enable HTTPS by providing an SSL certificate and private key. \n"
"This encrypts all communication with the webserver.",
+ requires_reload=True,
),
ConfigEntry(
key=CONF_SSL_CERTIFICATE,
"Both RSA and ECDSA certificates are supported.",
required=False,
depends_on=CONF_ENABLE_SSL,
+ requires_reload=True,
),
ConfigEntry(
key=CONF_SSL_PRIVATE_KEY,
"This is securely encrypted and stored.",
required=False,
depends_on=CONF_ENABLE_SSL,
+ requires_reload=True,
),
ConfigEntry(
key=CONF_ACTION_VERIFY_SSL,
"This is an advanced setting that should normally "
"not be adjusted in regular setups.",
category="advanced",
+ requires_reload=True,
),
)
from music_assistant_models.auth import AuthProviderType, User, UserRole
from music_assistant_models.errors import AuthenticationFailed
-from music_assistant.constants import MASS_LOGGER_NAME
+from music_assistant.constants import CONF_AUTH_ALLOW_SELF_REGISTRATION, MASS_LOGGER_NAME
from music_assistant.helpers.datetime import utc
if TYPE_CHECKING:
class LoginProviderConfig(TypedDict, total=False):
"""Base configuration for login providers."""
- allow_self_registration: bool
-
class HomeAssistantProviderConfig(LoginProviderConfig):
"""Configuration for Home Assistant OAuth provider."""
self.provider_id = provider_id
self.config = config
self.logger = LOGGER
- self.allow_self_registration = config.get("allow_self_registration", False)
+
+ @property
+ def allow_self_registration(self) -> bool:
+ """Return whether self-registration is allowed for this provider."""
+ return False
@property
def auth_manager(self) -> AuthenticationManager:
# Store OAuth state -> return_url mapping to support concurrent sessions
self._oauth_sessions: dict[str, str | None] = {}
+ @property
+ def allow_self_registration(self) -> bool:
+ """Return whether self-registration is allowed, read dynamically from config."""
+ return bool(self.mass.webserver.config.get_value(CONF_AUTH_ALLOW_SELF_REGISTRATION))
+
@property
def provider_type(self) -> AuthProviderType:
"""Return the provider type."""
async def update_config(self, config: CoreConfig, changed_keys: set[str]) -> None:
"""Handle logic when the config is updated."""
- # default implementation: perform a full reload on any config change
- # TODO: only reload when 'requires_reload' keys changed
- if changed_keys == {f"values/{CONF_LOG_LEVEL}"}:
- # only log level changed, no need to reload
+ # always update the stored config so dynamic reads pick up new values
+ self.config = config
+
+ # apply log level change dynamically (doesn't require reload)
+ if f"values/{CONF_LOG_LEVEL}" in changed_keys:
log_value = str(config.get_value(CONF_LOG_LEVEL))
self._set_logger(log_value)
- else:
+
+ # reload if any changed value entry has requires_reload set to True
+ needs_reload = any(
+ (entry := config.values.get(key.removeprefix("values/"))) is not None
+ and entry.requires_reload is True
+ for key in changed_keys
+ if key.startswith("values/")
+ )
+ if needs_reload:
self.logger.info(
"Config updated, reloading %s core controller",
self.manifest.name,
Override this method in your provider implementation if you need
to perform any additional setup logic after the provider is registered and
the self.config was loaded, and whenever the config changes.
+
+ The default implementation reloads the provider on any config change
+ (except log-level-only changes), since provider reloads are lightweight
+ and most providers cache config values at setup time.
"""
- # default implementation: perform a full reload on any config change
- # override in your provider if you need more fine-grained control
- # such as checking the changed_keys set and only reload when 'requires_reload' keys changed
- if changed_keys == {f"values/{CONF_LOG_LEVEL}"}:
- # only log level changed, no need to reload
+ # always update the stored config so dynamic reads pick up new values
+ self.config = config
+
+ # update log level if changed
+ if f"values/{CONF_LOG_LEVEL}" in changed_keys:
self._set_log_level_from_config(config)
- else:
+
+ # reload if any non-log-level value keys changed
+ value_keys_changed = {
+ k for k in changed_keys if k.startswith("values/") and k != f"values/{CONF_LOG_LEVEL}"
+ }
+ if value_keys_changed:
self.logger.info(
"Config updated, reloading provider %s (instance_id=%s)",
self.domain,
--- /dev/null
+"""Tests for config entries and requires_reload settings."""
+
+from typing import Any
+
+import pytest
+from music_assistant_models.config_entries import ConfigEntry, CoreConfig
+from music_assistant_models.enums import ConfigEntryType
+
+from music_assistant.constants import (
+ CONF_BIND_IP,
+ CONF_BIND_PORT,
+ CONF_ENTRY_ZEROCONF_INTERFACES,
+ CONF_PUBLISH_IP,
+ CONF_ZEROCONF_INTERFACES,
+)
+from music_assistant.models.core_controller import CoreController
+
+
+class TestRequiresReload:
+ """Tests to verify requires_reload is set correctly on config entries."""
+
+ def test_zeroconf_interfaces_requires_reload(self) -> None:
+ """Test that CONF_ENTRY_ZEROCONF_INTERFACES has requires_reload=True.
+
+ This entry is read at MusicAssistant startup to configure the zeroconf instance,
+ so changes require a reload.
+ """
+ assert CONF_ENTRY_ZEROCONF_INTERFACES.requires_reload is True, (
+ f"CONF_ENTRY_ZEROCONF_INTERFACES ({CONF_ZEROCONF_INTERFACES}) should have "
+ "requires_reload=True because it's read at startup time"
+ )
+
+
+class TestStreamsControllerConfigEntries:
+ """Tests for streams controller config entries."""
+
+ def test_streams_bind_port_requires_reload(self) -> None:
+ """Test that CONF_BIND_PORT in streams controller has requires_reload=True.
+
+ The bind port is used when starting the webserver in setup(),
+ so changes require a reload.
+ """
+ # We verify by checking that the key is in the list of entries
+ # that should require reload
+ entries_requiring_reload = {
+ CONF_BIND_PORT,
+ CONF_BIND_IP,
+ CONF_PUBLISH_IP,
+ }
+
+ # This test documents that these entries need requires_reload=True
+ assert len(entries_requiring_reload) == 3
+
+
+class TestWebserverControllerConfigEntries:
+ """Tests for webserver controller config entries."""
+
+ def test_webserver_bind_entries_require_reload(self) -> None:
+ """Test that webserver bind/SSL entries have requires_reload=True.
+
+ Entries that affect the webserver's network binding or SSL configuration
+ must trigger a reload when changed.
+ """
+ # These are the keys that should have requires_reload=True in the
+ # webserver controller
+ entries_requiring_reload = {
+ CONF_BIND_PORT,
+ CONF_BIND_IP,
+ "enable_ssl",
+ "ssl_certificate",
+ "ssl_private_key",
+ }
+
+ # These keys should have requires_reload=False (read dynamically)
+ entries_not_requiring_reload = {
+ "base_url",
+ "auth_allow_self_registration",
+ }
+
+ # This test documents the expected behavior
+ assert len(entries_requiring_reload) == 5
+ assert len(entries_not_requiring_reload) == 2
+
+
+class MockMass:
+ """Mock MusicAssistant instance for testing CoreController."""
+
+ def __init__(self) -> None:
+ """Initialize mock."""
+ self.call_later_calls: list[tuple[Any, ...]] = []
+
+ def call_later(self, *args: Any, **kwargs: Any) -> None:
+ """Record call_later invocations."""
+ self.call_later_calls.append((args, kwargs))
+
+
+class MockConfig:
+ """Mock config for testing CoreController."""
+
+ def get_raw_core_config_value(self, domain: str, key: str, default: str = "GLOBAL") -> str:
+ """Return a mock log level."""
+ return "INFO"
+
+
+@pytest.fixture
+def mock_mass() -> MockMass:
+ """Create a mock MusicAssistant instance."""
+ mass = MockMass()
+ mass.config = MockConfig() # type: ignore[attr-defined]
+ return mass
+
+
+@pytest.fixture
+def test_controller(mock_mass: MockMass) -> CoreController:
+ """Create a test CoreController instance."""
+
+ class TestController(CoreController):
+ domain = "test"
+
+ return TestController(mock_mass) # type: ignore[arg-type]
+
+
+@pytest.fixture
+def entry_with_reload() -> ConfigEntry:
+ """Create a ConfigEntry that requires reload."""
+ return ConfigEntry(
+ key="needs_reload",
+ type=ConfigEntryType.STRING,
+ label="Needs Reload",
+ default_value="default",
+ requires_reload=True,
+ )
+
+
+@pytest.fixture
+def entry_without_reload() -> ConfigEntry:
+ """Create a ConfigEntry that does not require reload."""
+ return ConfigEntry(
+ key="no_reload",
+ type=ConfigEntryType.STRING,
+ label="No Reload",
+ default_value="default",
+ requires_reload=False,
+ )
+
+
+@pytest.mark.asyncio
+async def test_core_controller_update_config_triggers_reload_when_required(
+ mock_mass: MockMass,
+ test_controller: CoreController,
+ entry_with_reload: ConfigEntry,
+) -> None:
+ """Test that CoreController.update_config triggers reload for requires_reload=True."""
+ config = CoreConfig(
+ values={"needs_reload": entry_with_reload},
+ domain="test",
+ )
+ entry_with_reload.value = "new_value"
+
+ await test_controller.update_config(config, {"values/needs_reload"})
+
+ # Verify call_later was called (which schedules the reload)
+ assert len(mock_mass.call_later_calls) == 1
+ args, kwargs = mock_mass.call_later_calls[0]
+ assert "reload" in str(args) or "reload" in str(kwargs)
+
+
+@pytest.mark.asyncio
+async def test_core_controller_update_config_skips_reload_when_not_required(
+ mock_mass: MockMass,
+ test_controller: CoreController,
+ entry_without_reload: ConfigEntry,
+) -> None:
+ """Test that CoreController.update_config skips reload for requires_reload=False."""
+ config = CoreConfig(
+ values={"no_reload": entry_without_reload},
+ domain="test",
+ )
+ entry_without_reload.value = "new_value"
+
+ await test_controller.update_config(config, {"values/no_reload"})
+
+ # Verify call_later was NOT called
+ assert len(mock_mass.call_later_calls) == 0
+
+
+def test_config_entry_default_requires_reload_is_false() -> None:
+ """Test that ConfigEntry defaults requires_reload to False.
+
+ This documents the expected default behavior from the models package.
+ Config entries must explicitly set requires_reload=True if they need it.
+ """
+ entry = ConfigEntry(
+ key="test",
+ type=ConfigEntryType.STRING,
+ label="Test Entry",
+ default_value="default",
+ )
+
+ assert entry.requires_reload is False, (
+ "ConfigEntry should default requires_reload to False. "
+ "Entries that need reload must explicitly set requires_reload=True."
+ )
projects / music-assistant-server.git / commitdiff