Chris CoutinhoandClaude Opus 4.8
cc2ce6e853
- _DEFAULTS keys for NEXTCLOUD_OIDC_TOKEN_TYPE / NEXTCLOUD_OIDC_SCOPES were registered as oidc_* (uppercasing to OIDC_*), so dynaconf (ignore_unknown_envvars) never read the NEXTCLOUD_-prefixed env vars and the fields stayed at their defaults. Prefix the keys to match _field_map; add a regression test. - Add gte=1 validator for HEALTH_READY_REFRESH_INTERVAL and a 1..65535 range validator for PORT. - Tie ReadinessCache.ttl_seconds to 2x the configured refresh interval so is_stale() stays meaningful when the interval is tuned. - Raise the refresh-loop exception log from DEBUG to WARNING. - Make health_ready a sync handler (no awaits); dedupe the localhost fallback into _DEFAULT_MCP_SERVER_URL; use pytest.approx for the float default. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
65 lines
2.0 KiB
Python
65 lines
2.0 KiB
Python
"""Unit tests for the readiness dependency-health cache (Deck #302).
|
|
|
|
The readiness probe reads this snapshot without performing any I/O; these tests
|
|
pin the small amount of logic it relies on (update/snapshot/staleness).
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from nextcloud_mcp_server.observability.readiness import (
|
|
DependencyStatus,
|
|
ReadinessCache,
|
|
)
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
def test_dependency_status_defaults():
|
|
status = DependencyStatus(name="nextcloud")
|
|
assert status.healthy is None # not yet checked
|
|
assert status.detail == "pending"
|
|
assert status.checked_at == pytest.approx(0.0)
|
|
|
|
|
|
def test_update_and_snapshot_round_trip():
|
|
cache = ReadinessCache()
|
|
cache.update("nextcloud_reachable", True, "ok", now=100.0)
|
|
cache.update("qdrant", False, "error: status 503", now=100.0)
|
|
|
|
snap = cache.snapshot()
|
|
assert snap["nextcloud_reachable"].healthy is True
|
|
assert snap["nextcloud_reachable"].detail == "ok"
|
|
assert snap["qdrant"].healthy is False
|
|
assert snap["qdrant"].detail == "error: status 503"
|
|
|
|
|
|
def test_snapshot_is_a_copy():
|
|
cache = ReadinessCache()
|
|
cache.update("nextcloud_reachable", True, "ok", now=1.0)
|
|
snap = cache.snapshot()
|
|
# Mutating the returned mapping must not affect the cache.
|
|
snap.clear()
|
|
assert "nextcloud_reachable" in cache.snapshot()
|
|
|
|
|
|
def test_is_stale_when_empty():
|
|
assert ReadinessCache(ttl_seconds=30.0).is_stale(now=0.0) is True
|
|
|
|
|
|
def test_is_stale_respects_ttl():
|
|
cache = ReadinessCache(ttl_seconds=30.0)
|
|
cache.update("nextcloud_reachable", True, "ok", now=100.0)
|
|
|
|
# Within the TTL window the snapshot is fresh.
|
|
assert cache.is_stale(now=120.0) is False
|
|
# At/after the TTL boundary it is stale.
|
|
assert cache.is_stale(now=130.0) is True
|
|
|
|
|
|
def test_is_stale_if_any_entry_is_old():
|
|
cache = ReadinessCache(ttl_seconds=30.0)
|
|
cache.update("nextcloud_reachable", True, "ok", now=100.0)
|
|
cache.update("qdrant", True, "ok", now=200.0)
|
|
# nextcloud_reachable is well past the TTL even though qdrant is fresh.
|
|
assert cache.is_stale(now=205.0) is True
|