mirror of
https://github.com/ipnet-mesh/meshcore-hub.git
synced 2026-10-31 04:02:03 +01:00
Louis King
c643c50378
Remove the upgrade-compatibility items scheduled for v0.20 in the v0.19 release notes: - the `meshcore-hub db migrate-to-postgres` command and its `common/db_migrate.py` module (with its dedicated tests) - the `database_backend` settings field and its legacy-backend validator (a leftover DATABASE_BACKEND is now silently ignored) - the `[postgres]` pip extra alias (Dockerfile installs `.` now) `tests/test_main.py` is rewritten as a minimal CLI smoke test to keep entry-point coverage. Docs updated: v0.20.0 upgrade section notes the removals (SQLite deployments must upgrade through v0.19 first), README and database.md pointers reworded, stale .env.example comment dropped.
378 lines
14 KiB
Python
378 lines
14 KiB
Python
"""Tests for configuration settings."""
|
|
|
|
import pytest
|
|
|
|
from meshcore_hub.common.config import (
|
|
CommonSettings,
|
|
CollectorSettings,
|
|
APISettings,
|
|
WebSettings,
|
|
)
|
|
|
|
|
|
class TestCommonSettings:
|
|
"""Tests for CommonSettings."""
|
|
|
|
def test_custom_data_home(self) -> None:
|
|
"""Test custom DATA_HOME setting."""
|
|
settings = CommonSettings(_env_file=None, data_home="/custom/data")
|
|
|
|
assert settings.data_home == "/custom/data"
|
|
|
|
def test_websocket_transport_settings(self) -> None:
|
|
"""Test MQTT websocket transport settings."""
|
|
settings = CommonSettings(
|
|
_env_file=None,
|
|
mqtt_transport="websockets",
|
|
mqtt_ws_path="/",
|
|
)
|
|
|
|
assert settings.mqtt_transport.value == "websockets"
|
|
assert settings.mqtt_ws_path == "/"
|
|
|
|
def test_oidc_role_names_shared_by_all_components(self) -> None:
|
|
"""OIDC role-name mapping lives on CommonSettings so the API tier
|
|
resolves the same names as the web tier (authorization, channel
|
|
visibility, role-filtered lists)."""
|
|
settings = CommonSettings(
|
|
_env_file=None,
|
|
oidc_role_admin="superadmin",
|
|
oidc_role_operator="moderator",
|
|
oidc_role_member="user",
|
|
oidc_role_test="sandbox",
|
|
)
|
|
|
|
assert settings.oidc_role_admin == "superadmin"
|
|
assert settings.oidc_role_operator == "moderator"
|
|
assert settings.oidc_role_member == "user"
|
|
assert settings.oidc_role_test == "sandbox"
|
|
|
|
# Every component inherits the mapping (previously WebSettings-only).
|
|
for cls in (CollectorSettings, APISettings, WebSettings):
|
|
inherited = cls(_env_file=None, oidc_role_operator="moderator")
|
|
assert inherited.oidc_role_operator == "moderator"
|
|
assert inherited.oidc_role_admin == "admin"
|
|
|
|
def test_oidc_roles_claim_stays_web_specific(self) -> None:
|
|
"""The IdP claim name is only parsed by the web tier."""
|
|
settings = APISettings(_env_file=None)
|
|
|
|
assert not hasattr(settings, "oidc_roles_claim")
|
|
assert WebSettings(_env_file=None).oidc_roles_claim == "roles"
|
|
|
|
|
|
class TestCollectorSettings:
|
|
"""Tests for CollectorSettings."""
|
|
|
|
def test_custom_data_home(self) -> None:
|
|
"""Test that custom data_home affects effective paths."""
|
|
settings = CollectorSettings(_env_file=None, data_home="/custom/data")
|
|
|
|
assert settings.collector_data_dir == "/custom/data/collector"
|
|
|
|
def test_explicit_database_url_overrides(self) -> None:
|
|
"""Test that explicit database_url overrides the components."""
|
|
settings = CollectorSettings(
|
|
_env_file=None, database_url="postgresql://user@host/db"
|
|
)
|
|
|
|
assert settings.database_url == "postgresql://user@host/db"
|
|
assert settings.effective_database_url == "postgresql://user@host/db"
|
|
|
|
def test_raw_packet_retention_defaults_to_2(self) -> None:
|
|
"""Unset raw_packet_retention_days defaults to 2, independent of global."""
|
|
settings = CollectorSettings(_env_file=None, data_retention_days=12)
|
|
|
|
assert settings.raw_packet_retention_days == 2
|
|
assert settings.effective_raw_packet_retention_days == 2
|
|
|
|
def test_raw_packet_retention_explicit_override(self) -> None:
|
|
"""An explicit raw_packet_retention_days wins over the global value."""
|
|
settings = CollectorSettings(
|
|
_env_file=None, data_retention_days=30, raw_packet_retention_days=3
|
|
)
|
|
|
|
assert settings.effective_raw_packet_retention_days == 3
|
|
|
|
def test_raw_packet_capture_disabled_by_default(self) -> None:
|
|
"""raw_packet_capture_enabled defaults to False."""
|
|
settings = CollectorSettings(_env_file=None)
|
|
|
|
assert settings.raw_packet_capture_enabled is False
|
|
|
|
def test_observer_lists_default_empty(self) -> None:
|
|
"""Observer allow/deny lists default to empty (accept-all)."""
|
|
settings = CollectorSettings(_env_file=None)
|
|
|
|
assert settings.observer_allowlist_keys == []
|
|
assert settings.observer_denylist_keys == []
|
|
|
|
def test_observer_allowlist_parsed_and_trimmed(self) -> None:
|
|
"""Comma-separated allowlist is split and blank entries dropped."""
|
|
settings = CollectorSettings(
|
|
_env_file=None, observer_allowlist="aaa, bbb ,, ,ccc"
|
|
)
|
|
|
|
assert settings.observer_allowlist_keys == ["aaa", "bbb", "ccc"]
|
|
|
|
def test_observer_denylist_parsed_and_trimmed(self) -> None:
|
|
"""Comma-separated denylist is split and blank entries dropped."""
|
|
settings = CollectorSettings(_env_file=None, observer_denylist=" key1 ,key2")
|
|
|
|
assert settings.observer_denylist_keys == ["key1", "key2"]
|
|
|
|
def test_explicit_seed_home_overrides(self) -> None:
|
|
"""Test that explicit seed_home overrides the default."""
|
|
settings = CollectorSettings(_env_file=None, seed_home="/seed/data")
|
|
|
|
assert settings.seed_home == "/seed/data"
|
|
assert settings.effective_seed_home == "/seed/data"
|
|
assert settings.node_tags_file == "/seed/data/node_tags.yaml"
|
|
|
|
def test_channel_refresh_interval_seconds(self) -> None:
|
|
"""Channel refresh interval defaults to 300."""
|
|
settings = CollectorSettings(_env_file=None)
|
|
|
|
assert settings.channel_refresh_interval_seconds == 300
|
|
|
|
def test_channel_refresh_interval_seconds_custom(self) -> None:
|
|
"""Channel refresh interval can be overridden."""
|
|
settings = CollectorSettings(
|
|
_env_file=None,
|
|
channel_refresh_interval_seconds=60,
|
|
)
|
|
|
|
assert settings.channel_refresh_interval_seconds == 60
|
|
|
|
def test_route_evaluator_interval_seconds_default(self) -> None:
|
|
"""Route evaluator interval defaults to 300."""
|
|
settings = CollectorSettings(_env_file=None)
|
|
|
|
assert settings.route_evaluator_interval_seconds == 300
|
|
|
|
def test_route_evaluator_interval_seconds_custom(self) -> None:
|
|
"""Route evaluator interval can be overridden (0 disables)."""
|
|
settings = CollectorSettings(
|
|
_env_file=None,
|
|
route_evaluator_interval_seconds=0,
|
|
)
|
|
|
|
assert settings.route_evaluator_interval_seconds == 0
|
|
|
|
def test_channels_file_path(self) -> None:
|
|
"""channels_file property resolves to seed_home/channels.yaml."""
|
|
settings = CollectorSettings(_env_file=None, seed_home="/seed/data")
|
|
|
|
assert settings.channels_file == "/seed/data/channels.yaml"
|
|
|
|
def test_channels_file_default(self) -> None:
|
|
"""channels_file uses default seed_home."""
|
|
settings = CollectorSettings(_env_file=None)
|
|
|
|
assert settings.channels_file.endswith("channels.yaml")
|
|
assert "seed" in settings.channels_file
|
|
|
|
|
|
class TestAPISettings:
|
|
"""Tests for APISettings."""
|
|
|
|
def test_custom_data_home(self) -> None:
|
|
"""Test that custom data_home is stored (seed/health/content paths)."""
|
|
settings = APISettings(_env_file=None, data_home="/custom/data")
|
|
|
|
assert settings.data_home == "/custom/data"
|
|
|
|
def test_explicit_database_url_overrides(self) -> None:
|
|
"""Test that explicit database_url overrides the components."""
|
|
settings = APISettings(_env_file=None, database_url="postgresql://user@host/db")
|
|
|
|
assert settings.database_url == "postgresql://user@host/db"
|
|
assert settings.effective_database_url == "postgresql://user@host/db"
|
|
|
|
|
|
class TestWebSettings:
|
|
"""Tests for WebSettings."""
|
|
|
|
def test_custom_data_home(self) -> None:
|
|
"""Test that custom data_home affects effective paths."""
|
|
settings = WebSettings(_env_file=None, data_home="/custom/data")
|
|
|
|
assert settings.web_data_dir == "/custom/data/web"
|
|
|
|
def test_network_announcement_default_none(self) -> None:
|
|
"""Test that network_announcement defaults to None."""
|
|
settings = WebSettings(_env_file=None)
|
|
|
|
assert settings.network_announcement is None
|
|
|
|
def test_system_announcement_default_none(self) -> None:
|
|
"""Test that system_announcement defaults to None."""
|
|
settings = WebSettings(_env_file=None)
|
|
|
|
assert settings.system_announcement is None
|
|
|
|
def test_system_maintenance_default_false(self) -> None:
|
|
"""Test that system_maintenance defaults to False."""
|
|
settings = WebSettings(_env_file=None)
|
|
|
|
assert settings.system_maintenance is False
|
|
|
|
def test_feature_channels_default_true(self) -> None:
|
|
"""Test that feature_channels defaults to True."""
|
|
settings = WebSettings(_env_file=None)
|
|
|
|
assert settings.feature_channels is True
|
|
|
|
def test_radio_config_defaults(self) -> None:
|
|
"""Test that radio config fields default to EU/UK Narrow values."""
|
|
settings = WebSettings(_env_file=None)
|
|
|
|
assert settings.network_radio_profile == "EU/UK Narrow"
|
|
assert settings.network_radio_frequency == 869.618
|
|
assert settings.network_radio_bandwidth == 62.5
|
|
assert settings.network_radio_spreading_factor == 8
|
|
assert settings.network_radio_coding_rate == 8
|
|
assert settings.network_radio_tx_power == 22.0
|
|
|
|
def test_radio_config_no_legacy_field(self) -> None:
|
|
"""Test that network_radio_config no longer exists."""
|
|
settings = WebSettings(_env_file=None)
|
|
assert not hasattr(settings, "network_radio_config")
|
|
|
|
def test_radio_config_custom_frequency(self) -> None:
|
|
"""Test that frequency can be set as a float."""
|
|
settings = WebSettings(
|
|
_env_file=None,
|
|
network_radio_frequency=915.0,
|
|
)
|
|
assert settings.network_radio_frequency == 915.0
|
|
|
|
def test_feature_channels_override(self) -> None:
|
|
"""Test that feature_channels can be disabled."""
|
|
settings = WebSettings(_env_file=None, feature_channels=False)
|
|
|
|
assert settings.feature_channels is False
|
|
|
|
def test_features_dict_includes_channels(self) -> None:
|
|
"""Test that features dict includes channels key."""
|
|
settings = WebSettings(_env_file=None)
|
|
features = settings.features
|
|
|
|
assert "channels" in features
|
|
assert features["channels"] is True
|
|
|
|
def test_features_dashboard_auto_disables(self) -> None:
|
|
"""Dashboard disables when nodes, ads, and messages all off."""
|
|
settings = WebSettings(
|
|
_env_file=None,
|
|
feature_dashboard=True,
|
|
feature_nodes=False,
|
|
feature_advertisements=False,
|
|
feature_messages=False,
|
|
)
|
|
assert settings.features["dashboard"] is False
|
|
|
|
def test_features_map_auto_disables_without_nodes(self) -> None:
|
|
"""Map disables when nodes feature is off."""
|
|
settings = WebSettings(
|
|
_env_file=None,
|
|
feature_map=True,
|
|
feature_nodes=False,
|
|
)
|
|
assert settings.features["map"] is False
|
|
|
|
def test_features_members_auto_disables_without_oidc(self) -> None:
|
|
"""Members disables when OIDC is not enabled."""
|
|
settings = WebSettings(
|
|
_env_file=None,
|
|
feature_members=True,
|
|
oidc_enabled=False,
|
|
)
|
|
assert settings.features["members"] is False
|
|
|
|
def test_features_all_enabled_by_default(self) -> None:
|
|
"""All features are enabled with default settings."""
|
|
settings = WebSettings(
|
|
_env_file=None,
|
|
oidc_enabled=True,
|
|
)
|
|
features = settings.features
|
|
assert features["dashboard"] is True
|
|
assert features["nodes"] is True
|
|
assert features["advertisements"] is True
|
|
assert features["messages"] is True
|
|
assert features["map"] is True
|
|
assert features["members"] is True
|
|
assert features["channels"] is True
|
|
assert features["pages"] is True
|
|
assert features["radio_config"] is True
|
|
|
|
def test_feature_radio_config_default_true(self) -> None:
|
|
"""Test that feature_radio_config defaults to True."""
|
|
settings = WebSettings(_env_file=None)
|
|
|
|
assert settings.feature_radio_config is True
|
|
|
|
def test_feature_radio_config_can_disable(self) -> None:
|
|
"""Test that feature_radio_config can be disabled."""
|
|
settings = WebSettings(_env_file=None, feature_radio_config=False)
|
|
|
|
assert settings.feature_radio_config is False
|
|
assert settings.features["radio_config"] is False
|
|
|
|
|
|
class TestDatabaseUrlResolution:
|
|
"""Tests for the PostgreSQL-only URL/schema resolution contract."""
|
|
|
|
def test_components_assemble_url_and_schema(self) -> None:
|
|
"""DATABASE_* components assemble a URL; the schema is always exposed."""
|
|
settings = APISettings(
|
|
_env_file=None,
|
|
database_host="pg",
|
|
database_password="pw",
|
|
)
|
|
|
|
assert settings.effective_database_url == (
|
|
"postgresql+psycopg2://meshcorehub:pw@pg:5432/meshcorehub"
|
|
)
|
|
assert settings.effective_database_schema == "meshcorehub"
|
|
|
|
def test_password_is_url_encoded(self) -> None:
|
|
"""Special characters in the password are percent-encoded."""
|
|
settings = APISettings(
|
|
_env_file=None,
|
|
database_host="pg",
|
|
database_password="s3cr3t/p@ss",
|
|
)
|
|
|
|
assert "s3cr3t%2Fp%40ss" in settings.effective_database_url
|
|
|
|
def test_schema_override_per_instance(self) -> None:
|
|
"""DATABASE_SCHEMA isolates instances sharing one database."""
|
|
settings = APISettings(
|
|
_env_file=None,
|
|
database_host="pg",
|
|
database_password="pw",
|
|
database_schema="stg",
|
|
)
|
|
|
|
assert settings.effective_database_schema == "stg"
|
|
|
|
def test_missing_required_vars_fails_fast(self) -> None:
|
|
"""No DATABASE_URL and no components raises rather than guessing."""
|
|
settings = APISettings(_env_file=None)
|
|
|
|
with pytest.raises(ValueError, match="PostgreSQL connection is not configured"):
|
|
_ = settings.effective_database_url
|
|
|
|
def test_explicit_url_overrides_components(self) -> None:
|
|
"""An explicit DATABASE_URL wins over the component vars."""
|
|
settings = CollectorSettings(
|
|
_env_file=None,
|
|
database_host="pg",
|
|
database_password="pw",
|
|
database_url="postgresql+psycopg2://u:p@h/db",
|
|
)
|
|
|
|
assert settings.effective_database_url == "postgresql+psycopg2://u:p@h/db"
|