"""Applying a validated overlay: settings, profiles, llm, mcp."""
import pytest
import navi.config as config_mod
from navi.overlay import apply as apply_mod
from navi.overlay.apply import (
apply_profiles_list,
apply_profiles_registry,
apply_settings_live,
build_settings,
overlay_llm_raw,
overlay_mcp_configs,
refresh,
)
from navi.overlay.fetch import Fetched
from navi.overlay.state import get_state, reset_state, status_summary
from navi.settings_model import Settings
@pytest.fixture(autouse=True)
def _fresh_state():
reset_state()
yield
reset_state()
@pytest.fixture()
def base(tmp_path) -> Settings:
return Settings(
_env_file=None,
hive_config_cache_file=str(tmp_path / "cache.json"),
)
@pytest.fixture()
def state(base):
state = get_state()
state.base_settings = base
state.enabled = True
return state
# ── settings ────────────────────────────────────────────────────────────────
def test_build_settings_overlays_only_named_fields(base):
overlaid = build_settings(base, {"llm_complete_timeout": 99})
assert overlaid.llm_complete_timeout == 99
assert overlaid.navi_port == base.navi_port
assert overlaid is not base
def test_build_settings_without_patch_returns_the_same_object(base):
assert build_settings(base, {}) is base
def test_hot_setting_is_written_into_the_live_object(state):
hot, restart = apply_settings_live({"gnauth_client_secret": "rotated"})
assert hot == ["gnauth_client_secret"]
assert restart == []
assert config_mod.settings.gnauth_client_secret == "rotated"
def test_non_hot_setting_requires_a_restart(state):
before = config_mod.settings.openai_api_key
hot, restart = apply_settings_live({"openai_api_key": "sk-new"})
assert hot == []
assert restart == ["openai_api_key"]
# Not applied behind the operator's back.
assert config_mod.settings.openai_api_key == before
def test_removing_a_field_hands_it_back_to_the_local_value(state, base):
state.merged = {"settings": {"llm_complete_timeout": 99}}
apply_settings_live({"llm_complete_timeout": 99})
assert config_mod.settings.llm_complete_timeout == 99
state.merged = {"settings": {"llm_complete_timeout": 99}}
hot, _ = apply_settings_live({})
assert hot == ["llm_complete_timeout"]
assert config_mod.settings.llm_complete_timeout == base.llm_complete_timeout
def test_unchanged_hot_field_is_not_reported(state):
apply_settings_live({"gnauth_client_secret": "same"})
hot, _ = apply_settings_live({"gnauth_client_secret": "same"})
assert hot == []
# ── profiles ────────────────────────────────────────────────────────────────
def test_apply_profiles_list_overlays_the_named_profile(state):
from navi.profiles import ALL_PROFILES
state.merged = {"profiles": {"assistant": {"temperature": 0.11}}}
profiles = apply_profiles_list(ALL_PROFILES)
by_id = {p.id: p for p in profiles}
assert by_id["assistant"].temperature == 0.11
others = [p for p in profiles if p.id != "assistant"]
assert others # the untouched profiles are still all there
def test_apply_profiles_list_without_patch_returns_input(state):
from navi.profiles import ALL_PROFILES
state.merged = {}
assert apply_profiles_list(ALL_PROFILES) is ALL_PROFILES
def test_registry_rebuild_is_idempotent_and_reversible(state):
from navi.core.registry import ProfileRegistry
from navi.profiles import ALL_PROFILES
registry = ProfileRegistry()
for profile in ALL_PROFILES:
registry.register(profile)
original = registry.get("assistant").temperature
state.merged = {"profiles": {"assistant": {"temperature": 0.05}}}
apply_profiles_registry(registry)
assert registry.get("assistant").temperature == 0.05
state.merged = {}
apply_profiles_registry(registry)
assert registry.get("assistant").temperature == original
def test_registry_leaves_unknown_runtime_profiles_alone(state):
from navi.core.registry import ProfileRegistry
from navi.profiles.base import AgentProfile
registry = ProfileRegistry()
custom = AgentProfile(id="made-at-runtime", name="X", description="", system_prompt="")
registry.register(custom)
apply_profiles_registry(registry)
assert registry.get("made-at-runtime") is custom
# ── llm / mcp ───────────────────────────────────────────────────────────────
def test_overlay_llm_raw_merges_and_keeps_local_values(state):
state.merged = {"llm": {"targets": {"home": {"host": "http://new"}}}}
raw = {"targets": {"home": {"host": "http://old"}, "cloud": {"host": "https://c"}}}
merged = overlay_llm_raw(raw)
assert merged["targets"]["home"]["host"] == "http://new"
assert merged["targets"]["cloud"]["host"] == "https://c"
def test_overlay_llm_raw_without_patch_is_a_passthrough(state):
state.merged = {}
raw = {"targets": {}}
assert overlay_llm_raw(raw) is raw
assert overlay_llm_raw(None) is None
def test_overlay_mcp_configs_merges_and_deletes(state):
from navi.mcp.config import McpServerConfig
state.merged = {
"mcp_servers": {"srv": {"command": "new"}, "gone": None}
}
configs = {
"srv": McpServerConfig(transport="stdio", command="old"),
"gone": McpServerConfig(transport="stdio", command="x"),
"kept": McpServerConfig(transport="stdio", command="y"),
}
merged = overlay_mcp_configs(configs)
assert merged["srv"].command == "new"
assert "gone" not in merged
assert merged["kept"].command == "y"
# ── refresh ─────────────────────────────────────────────────────────────────
def test_refresh_applies_and_records(state, base, monkeypatch):
fetched = Fetched(
version=7,
layers=[{"name": "global", "config": {"settings": {"llm_complete_timeout": 33}}}],
hash="sha256:x",
)
monkeypatch.setattr(apply_mod, "fetch_layers", lambda b: fetched)
result = refresh("test")
assert result["status"] == "applied"
assert config_mod.settings.llm_complete_timeout == 33
assert state.version == 7
assert state.hash == "sha256:x"
assert state.layers == ["global"]
assert state.merged["settings"] == {"llm_complete_timeout": 33}
def test_refresh_rejects_and_keeps_the_previous_config(state, monkeypatch):
monkeypatch.setattr(
apply_mod,
"fetch_layers",
lambda b: Fetched(
version=1,
layers=[{"name": "g", "config": {"settings": {"llm_complete_timeout": 33}}}],
hash="sha256:good",
),
)
refresh("first")
good = config_mod.settings.llm_complete_timeout
monkeypatch.setattr(
apply_mod,
"fetch_layers",
lambda b: Fetched(
version=2,
layers=[{"name": "g", "config": {"settings": {"llm_complete_timeout": "soon"}}}],
hash="sha256:bad",
),
)
result = refresh("second")
assert result["status"] == "rejected"
assert config_mod.settings.llm_complete_timeout == good
assert state.version == 1
assert state.errors
def test_refresh_is_a_noop_when_the_channel_is_disabled(state):
state.enabled = False
assert refresh("test")["status"] == "disabled"
def test_refresh_survives_an_unreachable_hive(state, monkeypatch):
def boom(base_settings):
raise RuntimeError("no route to host")
monkeypatch.setattr(apply_mod, "fetch_layers", boom)
result = refresh("test")
assert result["status"] == "unreachable"
assert state.last_error and "no route" in state.last_error
# The reason reaches /health and the hive, not only the state object.
assert "no route" in (status_summary()["error"] or "")
# A later success clears it, so /health stops reporting an old outage.
monkeypatch.setattr(
apply_mod,
"fetch_layers",
lambda b: Fetched(
version=1,
layers=[{"name": "g", "config": {"settings": {"llm_complete_timeout": 21}}}],
hash="sha256:ok",
),
)
refresh("retry")
assert state.last_error is None
def _fake_registry(monkeypatch, backend):
import navi.api.deps as deps
class Registry:
def get(self, name):
return backend if name == "ollama" else None
monkeypatch.setattr(deps, "get_backend_registry", lambda: Registry())
def _llm_only_fetch(monkeypatch, interval: int):
monkeypatch.setattr(
apply_mod,
"fetch_layers",
lambda b: Fetched(
version=2,
layers=[{"name": "g", "config": {"llm": {"probe": {"interval_sec": interval}}}}],
hash="sha256:llm",
),
)
def test_an_llm_change_is_staged_when_the_backend_cannot_swap(state, monkeypatch):
_fake_registry(monkeypatch, object())
_llm_only_fetch(monkeypatch, 30)
result = refresh("test")
assert result["status"] == "staged"
assert "llm" in result["restart_required"]
# The document is remembered, so a later start picks it up from the cache.
assert state.merged["llm"]["probe"]["interval_sec"] == 30
def test_an_llm_change_applies_live_when_the_backend_can_swap(state, monkeypatch):
seen = {}
class Backend:
def reconfigure(self, config):
seen["config"] = config
_fake_registry(monkeypatch, Backend())
_llm_only_fetch(monkeypatch, 15)
result = refresh("test")
assert result["status"] == "applied"
assert "llm" in result["hot_applied"]
assert result["restart_required"] == []
assert seen["config"].probe_interval_sec == 15
def test_a_settings_only_refresh_does_not_stage_llm(state, monkeypatch):
monkeypatch.setattr(
apply_mod,
"fetch_layers",
lambda b: Fetched(
version=3,
layers=[
{"name": "g", "config": {"settings": {"llm_complete_timeout": 42}}}
],
hash="sha256:s",
),
)
result = refresh("test")
assert "llm" not in result["restart_required"]
assert "mcp_servers" not in result["restart_required"]
def test_refresh_staged_when_a_restart_is_needed(state, monkeypatch):
monkeypatch.setattr(
apply_mod,
"fetch_layers",
lambda b: Fetched(
version=1,
layers=[{"name": "g", "config": {"settings": {"openai_api_key": "sk-new"}}}],
hash="sha256:x",
),
)
result = refresh("test")
assert result["status"] == "staged"
assert "openai_api_key" in state.restart_required
