diff --git a/app/core/settings_override/__init__.py b/app/core/settings_override/__init__.py index 3b75e7833..eae1f0d39 100644 --- a/app/core/settings_override/__init__.py +++ b/app/core/settings_override/__init__.py @@ -27,6 +27,7 @@ "SettingClassEnum", "SettingOverride", "SettingsOverrideManager", + "SnapshotChange", "build_snapshot", "coerce_field_value", "coerce_nested_field_value", @@ -59,6 +60,7 @@ refresh_all, RefreshCallback, settings_override_refresher, + SnapshotChange, start_refresh_task, ) from app.core.settings_override.manager import SettingsOverrideManager diff --git a/app/core/settings_override/lifecycle.py b/app/core/settings_override/lifecycle.py index 46ba6a7a3..993a202b9 100644 --- a/app/core/settings_override/lifecycle.py +++ b/app/core/settings_override/lifecycle.py @@ -22,6 +22,7 @@ "ProxyEntry", "ProxyRegistry", "RefreshCallback", + "SnapshotChange", "fire_change_callbacks", "publish_snapshot", "refresh_all", @@ -62,11 +63,31 @@ def _drain_cancelled_seed_task(task: asyncio.Task) -> None: task.exception() +class SnapshotChange(NamedTuple): + """Represent the override snapshots on either side of a republish. + + A snapshot holds active overrides only -- never an effective view -- so a + key may be absent from ``previous``, ``current``, or both. When a key is + absent the prior (or new) effective value is the YAML/env one reachable + through the proxy's wrapped instance. + + :param previous: The snapshot in effect before the republish. + :type previous: Mapping[str, object] + :param current: The snapshot now in effect. + :type current: Mapping[str, object] + """ + + previous: Mapping[str, object] + current: Mapping[str, object] + + #: A rebind callback fired when a watched ``(setting_class, key)`` override -#: changes value between refresh cycles. The callback receives the new effective -#: snapshot mapping for its setting class; any exception it raises is caught and -#: logged by :func:`refresh_all` so one failing callback cannot break the cycle. -RefreshCallback = Callable[[Mapping[str, object]], Awaitable[None]] +#: changes value between refresh cycles. The callback receives a +#: :class:`SnapshotChange` carrying the override snapshots on either side of +#: the republish (overrides-only; a key may be absent from either side). Any +#: exception it raises is caught and logged by :func:`fire_change_callbacks` +#: so one failing callback cannot break the cycle. +RefreshCallback = Callable[[SnapshotChange], Awaitable[None]] CallbackRegistry = dict[tuple[SettingClassEnum, str], RefreshCallback] @@ -124,8 +145,15 @@ async def fire_change_callbacks( Compares ``previous`` against ``current`` and, for each ``(setting_class, key)`` whose value differs and has a registered callback, awaits the - callback. Each callback runs inside its own ``try/except`` so one failure - neither aborts the cycle nor blocks the remaining callbacks. + callback with a :class:`SnapshotChange` carrying both mappings. Each + callback runs inside its own ``try/except`` so one failure neither aborts + the cycle nor blocks the remaining callbacks. + + Both mappings are override snapshots only -- never an effective view -- so a + key may be absent from either side (for example on the override-delete path + the changed key is gone from ``current``). Callbacks that need the previous + effective value when the key is absent recover it through the proxy's + wrapped YAML/env instance. Shared by the background refresher (:func:`refresh_all`, diffing the snapshot it just rebuilt) and the settings-API PATCH/DELETE handlers (diffing the @@ -137,11 +165,12 @@ async def fire_change_callbacks( :type callbacks: CallbackRegistry :param setting_class: The class whose snapshot was just republished. :type setting_class: SettingClassEnum - :param previous: The snapshot in effect before the republish. + :param previous: The override snapshot in effect before the republish. :type previous: Mapping[str, object] - :param current: The snapshot now in effect. + :param current: The override snapshot now in effect. :type current: Mapping[str, object] """ + change = SnapshotChange(previous, current) for key in previous.keys() | current.keys(): if previous.get(key) == current.get(key): continue @@ -149,7 +178,7 @@ async def fire_change_callbacks( if callback is None: continue try: - await callback(current) + await callback(change) except Exception: logger.exception( "Rebind callback for %s.%s failed; keeping previous binding", diff --git a/app/sep/main.py b/app/sep/main.py index 16efda515..1dedc37da 100644 --- a/app/sep/main.py +++ b/app/sep/main.py @@ -16,7 +16,7 @@ """Define SEP routes.""" import logging.config -from collections.abc import AsyncGenerator, Callable, Mapping +from collections.abc import AsyncGenerator from contextlib import asynccontextmanager from copy import deepcopy from traceback import format_exception @@ -28,7 +28,7 @@ from fastapi.exceptions import RequestValidationError from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, Response from fastapi.security import OAuth2PasswordRequestForm -from pydantic import HttpUrl, ValidationError +from pydantic import ValidationError from starlette.staticfiles import StaticFiles from app import __summary__, __version__ @@ -45,8 +45,10 @@ from app.core.settings_override.lifecycle import ( RefreshCallback, settings_override_refresher, + SnapshotChange, ) from app.core.settings_override.models import SettingClassEnum +from app.core.settings_override.proxy import OverridableSettingsProxy from app.core.utils import run_pydantic_type_validator from app.core.utils.fields import URIPath from app.inventory.config import inventory_settings @@ -126,7 +128,8 @@ async def sep_startup() -> None: def _make_remote_api_rebinder( app: FastAPI, name: str, - endpoint_getter: Callable[[], HttpUrl], + proxy: OverridableSettingsProxy, + key: str, **ssl: Any, ) -> RefreshCallback: """Build a rebind callback for an ``app.state`` RemoteAPI endpoint override. @@ -136,17 +139,21 @@ def _make_remote_api_rebinder( the new endpoint and the old one closed. Under the combined ``app.main:app`` no ``app.state`` client exists -- ``get_*_client`` falls back to the registry-cached ``get_remote_api`` per request, which already key-misses to - the new HOT endpoint -- so the callback only evicts any stale client left on - the new endpoint. + the new HOT endpoint -- so the callback evicts the ordered de-duplicated set + of previous-and-current endpoints (covering endpoint moves as well as + same-endpoint credential/SSL changes). When ``key`` is absent from + ``change.previous`` (override created), the prior effective value is the + YAML/env one from the proxy's wrapped instance. :param app: The FastAPI application whose ``state`` holds the client. :type app: FastAPI :param name: The ``app.state`` attribute name (``inventory_api`` / ``tasks_api``). :type name: str - :param endpoint_getter: A zero-argument callable returning the current - (override-aware) endpoint. - :type endpoint_getter: Callable[[], HttpUrl] + :param proxy: The overridable settings proxy that owns the endpoint field. + :type proxy: OverridableSettingsProxy + :param key: The top-level snapshot key for the endpoint field. + :type key: str :param ssl: SSL keyword arguments forwarded to :class:`RemoteAPI` (not HOT, captured once at wiring time). :type ssl: Any @@ -154,11 +161,17 @@ def _make_remote_api_rebinder( :rtype: RefreshCallback """ - async def _rebind(_: Mapping[str, object]) -> None: - new_endpoint = endpoint_getter() + async def _rebind(change: SnapshotChange) -> None: + new_endpoint = getattr(proxy, key) old = getattr(app.state, name, None) if old is None: - await settings.invalidate_client(str(new_endpoint)) + previous_endpoint = change.previous.get(key) + if previous_endpoint is None: + previous_endpoint = getattr(proxy._resolve(), key) # noqa: SLF001 + for endpoint in dict.fromkeys( + str(ep) for ep in (previous_endpoint, new_endpoint) if ep is not None + ): + await settings.invalidate_client(endpoint) return try: new_api = await RemoteAPI(endpoint=new_endpoint, **ssl).open() @@ -171,7 +184,7 @@ async def _rebind(_: Mapping[str, object]) -> None: return _rebind -async def _apply_logging_dictconfig(_: Mapping[str, object]) -> None: +async def _apply_logging_dictconfig(_: SnapshotChange) -> None: """Re-apply ``logging.config.dictConfig`` after a global ``LOGGING`` override. ``LOGGING`` is a HOT field, but ``LOGGING_CONFIG`` (the dict handed to @@ -183,8 +196,8 @@ async def _apply_logging_dictconfig(_: Mapping[str, object]) -> None: process without a restart. Failures are logged and swallowed: a malformed config must not take the process down mid-request. - :param _: The new effective ``Settings`` snapshot mapping (unused -- the level - is re-read from the proxy). + :param _: The override snapshots on either side of the republish (unused -- + the level is re-read from the proxy). """ try: config = deepcopy(settings.LOGGING_CONFIG) @@ -195,7 +208,7 @@ async def _apply_logging_dictconfig(_: Mapping[str, object]) -> None: logger.exception("Failed to re-apply logging config after LOGGING override") -async def _reseed_system_periodic_tasks(_: Mapping[str, object]) -> None: +async def _reseed_system_periodic_tasks(_: SnapshotChange) -> None: """Re-seed the SEP beat schedule after a hot interval override. Wired for both ``SnippetsSettings.SYNC_INTERVAL`` (``sep__sync_snippets``) and @@ -211,8 +224,8 @@ async def _reseed_system_periodic_tasks(_: Mapping[str, object]) -> None: Updating the ``IntervalSchedule`` bumps ``PeriodicTaskChanged.last_update``, so Celery beat reloads the schedule on its next scheduler tick without a restart. - :param _: The new effective settings snapshot mapping (unused -- the interval is - re-read from the proxy by the task-set builder). + :param _: The override snapshots on either side of the republish (unused -- the + interval is re-read from the proxy by the task-set builder). """ await init_periodic_tasks_db(get_system_periodic_tasks(), "sep__") @@ -255,7 +268,8 @@ async def sep_overrides_lifespan(app: FastAPI) -> AsyncGenerator[None, None]: ): _make_remote_api_rebinder( app, "inventory_api", - lambda: sep_settings.INVENTORY_ENDPOINT, + sep_settings, + "INVENTORY_ENDPOINT", ssl_cafile=settings.SSL_CAFILE, ssl_keyfile=inventory_settings.SSL_KEYFILE, ssl_certfile=inventory_settings.SSL_CERTFILE, @@ -263,7 +277,8 @@ async def sep_overrides_lifespan(app: FastAPI) -> AsyncGenerator[None, None]: (SettingClassEnum.SEP_SETTINGS, "TASKS_ENDPOINT"): _make_remote_api_rebinder( app, "tasks_api", - lambda: sep_settings.TASKS_ENDPOINT, + sep_settings, + "TASKS_ENDPOINT", ssl_cafile=settings.SSL_CAFILE, ssl_keyfile=tasks_settings.SSL_KEYFILE, ssl_certfile=tasks_settings.SSL_CERTFILE, diff --git a/app/sep/settings_override.py b/app/sep/settings_override.py index 37dd51eec..d10f177b5 100644 --- a/app/sep/settings_override.py +++ b/app/sep/settings_override.py @@ -25,20 +25,20 @@ ``celery.tasks``. """ -from collections.abc import Mapping -from typing import Any +from typing import Any, cast from celery.signals import worker_process_init, worker_process_shutdown from sqlmodel.ext.asyncio.session import AsyncSession from app.celery import celery from app.core.alerts.config import alert_settings, AlertSettings -from app.core.config import Settings, settings +from app.core.config import PMMSettings, Settings, settings from app.core.settings_override.lifecycle import ( CallbackRegistry, ProxyEntry, ProxyRegistry, publish_snapshot, + SnapshotChange, ) from app.core.settings_override.models import SettingClassEnum from app.core.settings_override.worker import WorkerRefresher @@ -119,21 +119,28 @@ async def republish_sep_settings_snapshot(session: AsyncSession) -> None: await publish_snapshot(sep_settings, session, SEPSettings) -async def invalidate_pmm_clients(_: Mapping[str, object]) -> None: - """Evict the cached PMM client on the current endpoint after a ``PMM`` override. +async def invalidate_pmm_clients(change: SnapshotChange) -> None: + """Evict cached PMM clients on the previous and current endpoints after a ``PMM`` override. - A same-endpoint change (credentials, SSL) evicts the now-stale client so the - next :class:`PMMSyncer` key-misses to a fresh one via its ``default_factory`` - PMM read. Known limitation: an endpoint change leaves the client keyed by the - old endpoint cached until ``close_all`` closes it at shutdown; syncers key on - the new endpoint, so nothing reads it in the meantime. + Evicts the ordered de-duplicated set of previous-and-current endpoints so a + same-endpoint change (credentials, SSL) collapses to a single eviction, while + an endpoint change also drops the client keyed by the endpoint no longer in + use. The next :class:`PMMSyncer` key-misses to a fresh client via its + ``default_factory`` PMM read. When ``PMM`` is absent from ``change.previous`` + (override created), the prior effective value is the YAML/env one from the + proxy's wrapped instance. - :param _: The new effective ``Settings`` snapshot mapping (unused -- the - current PMM endpoint is re-read from the proxy). + :param change: The override snapshots on either side of the republish. """ - endpoint = settings.PMM.endpoint - if endpoint is not None: - await settings.invalidate_client(str(endpoint)) + previous_pmm = cast(PMMSettings | None, change.previous.get("PMM")) + if previous_pmm is None: + previous_pmm = settings._resolve().PMM # noqa: SLF001 + for endpoint in dict.fromkeys( + str(pmm.endpoint) + for pmm in (previous_pmm, settings.PMM) + if pmm is not None and pmm.endpoint is not None + ): + await settings.invalidate_client(endpoint) #: The callbacks the worker refresher registers -- deliberately a strict subset diff --git a/app/tasks/main.py b/app/tasks/main.py index 8e8e70c50..a8bcc77e4 100644 --- a/app/tasks/main.py +++ b/app/tasks/main.py @@ -17,7 +17,7 @@ import json import logging.config -from collections.abc import AsyncGenerator, Mapping +from collections.abc import AsyncGenerator from contextlib import asynccontextmanager from typing import Any @@ -33,6 +33,7 @@ CallbackRegistry, ProxyEntry, settings_override_refresher, + SnapshotChange, ) from app.core.settings_override.models import SettingClassEnum from app.tasks.anonymizer.config import anonymizer_settings, AnonymizerSettings @@ -53,7 +54,7 @@ celery_logger = get_task_logger(__name__) -async def _reconcile_nomad(_: Mapping[str, object]) -> None: +async def _reconcile_nomad(_: SnapshotChange) -> None: """Rebind the live Nomad executor when its override changed. Registered as the ``(TASKS_SETTINGS, NOMAD)`` rebind callback by @@ -62,9 +63,9 @@ async def _reconcile_nomad(_: Mapping[str, object]) -> None: shutdown can find ``tasks_app.state.nomad_lifecycle`` already gone; the rebind is skipped in that window rather than raising a noisy callback error. - :param _: The new effective ``TasksSettings`` snapshot mapping. Unused -- the - holder reads the live ``NOMAD`` config itself when reconciling. - :type _: Mapping[str, object] + :param _: The override snapshots on either side of the republish. Unused -- + the holder reads the live ``NOMAD`` config itself when reconciling. + :type _: SnapshotChange """ holder = getattr(tasks_app.state, "nomad_lifecycle", None) if holder is not None: diff --git a/changelog.d/SEP-1734.fixed.md b/changelog.d/SEP-1734.fixed.md new file mode 100644 index 000000000..1c8f5659e --- /dev/null +++ b/changelog.d/SEP-1734.fixed.md @@ -0,0 +1 @@ +A runtime override that changes the PMM, Inventory, or Tasks endpoint now evicts the cached client on the endpoint no longer in use, instead of leaving it open until process shutdown. diff --git a/tests/app/core/settings_override/test_lifecycle.py b/tests/app/core/settings_override/test_lifecycle.py index 28ea50448..17961d1b2 100644 --- a/tests/app/core/settings_override/test_lifecycle.py +++ b/tests/app/core/settings_override/test_lifecycle.py @@ -29,8 +29,10 @@ from app.core.db.utils import get_async_session_maker_from_engine from app.core.settings_override.cache import build_snapshot from app.core.settings_override.lifecycle import ( + fire_change_callbacks, ProxyEntry, refresh_all, + SnapshotChange, start_refresh_task, ) from app.core.settings_override.manager import SettingsOverrideManager @@ -426,6 +428,32 @@ async def _callback(_: object) -> None: assert proxy.CONNECTIVITY_CHECK_DEFAULT is override_value +@pytest.mark.asyncio +async def test_fire_change_callbacks_delivers_snapshot_change_on_delete() -> None: + """Callbacks receive a SnapshotChange; on delete the key is absent from current.""" + previous = {"CONNECTIVITY_CHECK_DEFAULT": True} + current: dict[str, object] = {} + received: list[object] = [] + + async def _callback(change: object) -> None: + received.append(change) + + await fire_change_callbacks( + {_CALLBACK_KEY: _callback}, + SettingClassEnum.SEP_SETTINGS, + previous, + current, + ) + + assert len(received) == 1 + change = received[0] + assert isinstance(change, SnapshotChange) + assert change.previous == previous + assert change.current == current + assert "CONNECTIVITY_CHECK_DEFAULT" not in change.current + assert change.previous["CONNECTIVITY_CHECK_DEFAULT"] is True + + @pytest.mark.asyncio async def test_refresh_all_skips_callback_for_unchanged_key( session_maker: async_sessionmaker, diff --git a/tests/app/core/settings_override/test_worker.py b/tests/app/core/settings_override/test_worker.py index fd356c595..d8c6dc828 100644 --- a/tests/app/core/settings_override/test_worker.py +++ b/tests/app/core/settings_override/test_worker.py @@ -16,7 +16,6 @@ """Tests for the reusable prefork-child settings-override refresher handle.""" import asyncio -from collections.abc import Mapping from contextlib import suppress from datetime import timedelta @@ -30,6 +29,7 @@ CallbackRegistry, ProxyEntry, ProxyRegistry, + SnapshotChange, ) from app.core.settings_override.models import SettingClassEnum from app.core.settings_override.proxy import OverridableSettingsProxy @@ -45,8 +45,8 @@ INTERVAL = timedelta(seconds=30) -async def _noop_callback(_: Mapping[str, object]) -> None: - """Accept a snapshot and do nothing; a stand-in registry entry.""" +async def _noop_callback(_: SnapshotChange) -> None: + """Accept a snapshot change and do nothing; a stand-in registry entry.""" def _make_registry() -> ProxyRegistry: diff --git a/tests/app/sep/test_override_callbacks.py b/tests/app/sep/test_override_callbacks.py index 5e1d25750..dc80ee48c 100644 --- a/tests/app/sep/test_override_callbacks.py +++ b/tests/app/sep/test_override_callbacks.py @@ -19,6 +19,7 @@ import pytest from fastapi import FastAPI +from pydantic import SecretStr from pytest_mock import MockerFixture from sqlalchemy_celery_beat.models import Period @@ -26,6 +27,7 @@ from app.core.celery.models import IntervalSchedule from app.core.config import PMMSettings, Settings, settings from app.core.requests import RemoteAPI +from app.core.settings_override.lifecycle import SnapshotChange from app.core.settings_override.models import SettingClassEnum from app.sep.config import sep_settings from app.sep.main import ( @@ -36,26 +38,42 @@ from app.sep.snippets.config import snippets_settings +def _awaited_endpoints(invalidate: AsyncMock) -> list[str]: + """Return the endpoint arguments passed to ``invalidate_client``, in order.""" + return [call.args[0] for call in invalidate.await_args_list] + + @pytest.mark.asyncio -async def test_endpoint_rebinder_swaps_app_state_client() -> None: - """Assert the rebinder opens a client on the new endpoint and drains the old one.""" +async def test_endpoint_rebinder_swaps_app_state_client( + mocker: MockerFixture, +) -> None: + """Assert the rebinder opens a client on the new endpoint and drains the old one. + + The standalone ``sep_lifespan`` path owns ``app.state`` clients: it closes the + old one explicitly and must not fall through to registry eviction. + """ app = FastAPI() old = await RemoteAPI(endpoint="https://old-inv.example.org").open() app.state.inventory_api = old sep_settings._set_snapshot({"INVENTORY_ENDPOINT": "https://new-inv.example.org"}) + invalidate = mocker.patch.object(Settings, "invalidate_client", new=AsyncMock()) + new = None rebind = _make_remote_api_rebinder( - app, "inventory_api", lambda: sep_settings.INVENTORY_ENDPOINT + app, "inventory_api", sep_settings, "INVENTORY_ENDPOINT" ) - await rebind({}) - - new = app.state.inventory_api try: + await rebind(SnapshotChange({}, {})) + + new = app.state.inventory_api assert new is not old assert str(new.endpoint).startswith("https://new-inv.example.org") assert old._session is None # the previous client was drained + invalidate.assert_not_awaited() finally: - await new.close() + if new is not None: + await new.close() + sep_settings._set_snapshot({}) @pytest.mark.asyncio @@ -64,15 +82,93 @@ async def test_endpoint_rebinder_invalidates_when_no_app_state_client( ) -> None: """Assert the rebinder evicts the registry client when no ``app.state`` client exists.""" app = FastAPI() - sep_settings._set_snapshot({"INVENTORY_ENDPOINT": "https://new-inv.example.org"}) + endpoint = "https://new-inv.example.org" + sep_settings._set_snapshot({"INVENTORY_ENDPOINT": endpoint}) + invalidate = mocker.patch.object(Settings, "invalidate_client", new=AsyncMock()) + + rebind = _make_remote_api_rebinder( + app, "inventory_api", sep_settings, "INVENTORY_ENDPOINT" + ) + try: + # Same previous/current endpoint collapses to one eviction (credential-style). + await rebind( + SnapshotChange( + {"INVENTORY_ENDPOINT": endpoint}, + {"INVENTORY_ENDPOINT": endpoint}, + ) + ) + invalidate.assert_awaited_once_with(endpoint) + finally: + sep_settings._set_snapshot({}) + + +@pytest.mark.asyncio +async def test_endpoint_rebinder_created_evicts_base_and_new( + mocker: MockerFixture, +) -> None: + """An endpoint override create evicts the YAML/env base and the new endpoint.""" + app = FastAPI() + new_endpoint = "https://new-inv.example.org" + base_endpoint = str(sep_settings._resolve().INVENTORY_ENDPOINT) + sep_settings._set_snapshot({"INVENTORY_ENDPOINT": new_endpoint}) + invalidate = mocker.patch.object(Settings, "invalidate_client", new=AsyncMock()) + + rebind = _make_remote_api_rebinder( + app, "inventory_api", sep_settings, "INVENTORY_ENDPOINT" + ) + try: + await rebind(SnapshotChange({}, {"INVENTORY_ENDPOINT": new_endpoint})) + assert _awaited_endpoints(invalidate) == list( + dict.fromkeys([base_endpoint, new_endpoint]) + ) + finally: + sep_settings._set_snapshot({}) + + +@pytest.mark.asyncio +async def test_endpoint_rebinder_changed_evicts_previous_and_new( + mocker: MockerFixture, +) -> None: + """An endpoint override change evicts both the previous and the new endpoint.""" + app = FastAPI() + previous_endpoint = "https://old-inv.example.org" + new_endpoint = "https://new-inv.example.org" + sep_settings._set_snapshot({"INVENTORY_ENDPOINT": new_endpoint}) invalidate = mocker.patch.object(Settings, "invalidate_client", new=AsyncMock()) rebind = _make_remote_api_rebinder( - app, "inventory_api", lambda: sep_settings.INVENTORY_ENDPOINT + app, "inventory_api", sep_settings, "INVENTORY_ENDPOINT" ) - await rebind({}) + try: + await rebind( + SnapshotChange( + {"INVENTORY_ENDPOINT": previous_endpoint}, + {"INVENTORY_ENDPOINT": new_endpoint}, + ) + ) + assert _awaited_endpoints(invalidate) == [previous_endpoint, new_endpoint] + finally: + sep_settings._set_snapshot({}) + + +@pytest.mark.asyncio +async def test_endpoint_rebinder_deleted_evicts_previous_and_base( + mocker: MockerFixture, +) -> None: + """Deleting an endpoint override evicts the previous override and the YAML/env base.""" + app = FastAPI() + previous_endpoint = "https://old-inv.example.org" + base_endpoint = str(sep_settings._resolve().INVENTORY_ENDPOINT) + sep_settings._set_snapshot({}) + invalidate = mocker.patch.object(Settings, "invalidate_client", new=AsyncMock()) - invalidate.assert_awaited_once_with("https://new-inv.example.org") + rebind = _make_remote_api_rebinder( + app, "inventory_api", sep_settings, "INVENTORY_ENDPOINT" + ) + await rebind(SnapshotChange({"INVENTORY_ENDPOINT": previous_endpoint}, {})) + assert _awaited_endpoints(invalidate) == list( + dict.fromkeys([previous_endpoint, base_endpoint]) + ) @pytest.mark.asyncio @@ -80,12 +176,16 @@ async def test_invalidate_pmm_clients_evicts_current_pmm_endpoint( mocker: MockerFixture, ) -> None: """Assert the PMM callback evicts cached clients on the overridden PMM endpoint.""" - settings._set_snapshot({"PMM": PMMSettings(endpoint="https://new-pmm.example.org")}) + pmm = PMMSettings(endpoint="https://new-pmm.example.org") + settings._set_snapshot({"PMM": pmm}) invalidate = mocker.patch.object(Settings, "invalidate_client", new=AsyncMock()) - await invalidate_pmm_clients({}) - - invalidate.assert_awaited_once_with("https://new-pmm.example.org") + try: + # Same previous/current endpoint collapses to one eviction (credential-style). + await invalidate_pmm_clients(SnapshotChange({"PMM": pmm}, {"PMM": pmm})) + invalidate.assert_awaited_once_with("https://new-pmm.example.org") + finally: + settings._set_snapshot({}) @pytest.mark.asyncio @@ -93,12 +193,98 @@ async def test_invalidate_pmm_clients_noop_without_endpoint( mocker: MockerFixture, ) -> None: """Assert the PMM callback is a no-op when no PMM endpoint is configured.""" - settings._set_snapshot({"PMM": PMMSettings(endpoint=None)}) + pmm = PMMSettings(endpoint=None) + settings._set_snapshot({"PMM": pmm}) + invalidate = mocker.patch.object(Settings, "invalidate_client", new=AsyncMock()) + + try: + await invalidate_pmm_clients(SnapshotChange({"PMM": pmm}, {"PMM": pmm})) + invalidate.assert_not_awaited() + finally: + settings._set_snapshot({}) + + +@pytest.mark.asyncio +async def test_invalidate_pmm_clients_created_evicts_base_and_new( + mocker: MockerFixture, +) -> None: + """A PMM override create evicts the YAML/env base endpoint and the new one.""" + new_pmm = PMMSettings(endpoint="https://new-pmm.example.org") + base_endpoint = settings._resolve().PMM.endpoint + settings._set_snapshot({"PMM": new_pmm}) + invalidate = mocker.patch.object(Settings, "invalidate_client", new=AsyncMock()) + + try: + await invalidate_pmm_clients(SnapshotChange({}, {"PMM": new_pmm})) + expected = [] + if base_endpoint is not None: + expected.append(str(base_endpoint)) + expected.append("https://new-pmm.example.org") + assert _awaited_endpoints(invalidate) == list(dict.fromkeys(expected)) + finally: + settings._set_snapshot({}) + + +@pytest.mark.asyncio +async def test_invalidate_pmm_clients_changed_evicts_previous_and_new( + mocker: MockerFixture, +) -> None: + """A PMM endpoint override change evicts both the previous and the new endpoint.""" + previous_pmm = PMMSettings(endpoint="https://old-pmm.example.org") + new_pmm = PMMSettings(endpoint="https://new-pmm.example.org") + settings._set_snapshot({"PMM": new_pmm}) + invalidate = mocker.patch.object(Settings, "invalidate_client", new=AsyncMock()) + + try: + await invalidate_pmm_clients( + SnapshotChange({"PMM": previous_pmm}, {"PMM": new_pmm}) + ) + assert _awaited_endpoints(invalidate) == [ + "https://old-pmm.example.org", + "https://new-pmm.example.org", + ] + finally: + settings._set_snapshot({}) + + +@pytest.mark.asyncio +async def test_invalidate_pmm_clients_deleted_evicts_previous_and_base( + mocker: MockerFixture, +) -> None: + """Deleting a PMM override evicts the previous override and the YAML/env base.""" + previous_pmm = PMMSettings(endpoint="https://old-pmm.example.org") + base_endpoint = settings._resolve().PMM.endpoint + settings._set_snapshot({}) invalidate = mocker.patch.object(Settings, "invalidate_client", new=AsyncMock()) - await invalidate_pmm_clients({}) + await invalidate_pmm_clients(SnapshotChange({"PMM": previous_pmm}, {})) + expected = ["https://old-pmm.example.org"] + if base_endpoint is not None: + expected.append(str(base_endpoint)) + assert _awaited_endpoints(invalidate) == list(dict.fromkeys(expected)) - invalidate.assert_not_awaited() + +@pytest.mark.asyncio +async def test_invalidate_pmm_clients_same_endpoint_credential_change_evicts_once( + mocker: MockerFixture, +) -> None: + """A same-endpoint credential change collapses to a single eviction.""" + previous_pmm = PMMSettings( + endpoint="https://same-pmm.example.org", api_key=SecretStr("old-key") + ) + new_pmm = PMMSettings( + endpoint="https://same-pmm.example.org", api_key=SecretStr("new-key") + ) + settings._set_snapshot({"PMM": new_pmm}) + invalidate = mocker.patch.object(Settings, "invalidate_client", new=AsyncMock()) + + try: + await invalidate_pmm_clients( + SnapshotChange({"PMM": previous_pmm}, {"PMM": new_pmm}) + ) + invalidate.assert_awaited_once_with("https://same-pmm.example.org") + finally: + settings._set_snapshot({}) @pytest.mark.asyncio @@ -117,7 +303,7 @@ async def test_reseed_callback_reseeds_beat_with_live_interval( {"SYNC_INTERVAL": IntervalSchedule(every=15, period=Period.MINUTES)} ) try: - await _reseed_system_periodic_tasks({}) + await _reseed_system_periodic_tasks(SnapshotChange({}, {})) finally: snippets_settings._set_snapshot({}) @@ -165,7 +351,7 @@ async def test_apply_logging_dictconfig_reapplies_new_level( dict_config = mocker.patch("app.sep.main.logging.config.dictConfig") settings._set_snapshot({"LOGGING": "DEBUG"}) try: - await sep_main._apply_logging_dictconfig({}) + await sep_main._apply_logging_dictconfig(SnapshotChange({}, {})) finally: settings._set_snapshot({}) @@ -186,7 +372,7 @@ async def test_apply_logging_dictconfig_swallows_failure( settings._set_snapshot({"LOGGING": "DEBUG"}) try: # Must not raise. - await sep_main._apply_logging_dictconfig({}) + await sep_main._apply_logging_dictconfig(SnapshotChange({}, {})) finally: settings._set_snapshot({}) diff --git a/tests/app/tasks/test_main.py b/tests/app/tasks/test_main.py index 7768a9699..032fcdfd0 100644 --- a/tests/app/tasks/test_main.py +++ b/tests/app/tasks/test_main.py @@ -21,6 +21,7 @@ from fastapi import FastAPI, HTTPException, status from sqlalchemy.dialects.postgresql import JSON, JSONB +from app.core.settings_override.lifecycle import SnapshotChange from app.core.settings_override.models import SettingClassEnum from app.tasks.db.seed import verify_taskhistory_execution_request_is_jsonb from app.tasks.execution.exceptions import TaskDataNotFoundInExecutorError @@ -109,7 +110,7 @@ async def test_reconcile_nomad_rebinds_when_holder_present(): app_mock = MagicMock() app_mock.state.nomad_lifecycle = holder with patch("app.tasks.main.tasks_app", app_mock): - await _reconcile_nomad({}) + await _reconcile_nomad(SnapshotChange({}, {})) holder.reconcile.assert_awaited_once() @@ -125,7 +126,7 @@ async def test_reconcile_nomad_skips_when_holder_absent(): app_mock = MagicMock() app_mock.state.nomad_lifecycle = None with patch("app.tasks.main.tasks_app", app_mock): - await _reconcile_nomad({}) + await _reconcile_nomad(SnapshotChange({}, {})) def test_task_data_not_found_detail_base_exception_without_structured_fields():