From 70c63dd02fe5d577e753ab7bcbf861337dd8e3d6 Mon Sep 17 00:00:00 2001 From: Archit Adish Gupta Date: Fri, 24 Jul 2026 19:11:18 +0530 Subject: [PATCH] feat(config): add config validation to tests and tooling (closes #937) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a zero-dependency rules engine at backend/utils/config_validator.py that surfaces invalid LocalMind app-config at startup or in tests: - ValidationIssue (frozen): severity, location, message — severity restricted to {ERROR, WARN, INFO}. - ValidationRule: name + fn(config) -> [ValidationIssue]. - validate(config, rules=None): runs DEFAULT_RULES (or caller supplied list) catching exceptions so a buggy rule doesn't crash the report. - has_errors / has_warnings / summarise helpers. - load_rules_from_module(spec) imports a 'module' and pulls its RULES list for reuse-extension without monkey-patching. Default ruleset covers AppSettings (Pydantic only validates single field types; cross-field/system context checks go here): - model_name (non-empty) - language_supported ({en,ja,es,fr,de}; warns for others) - temperature_range (0..2 inclusivity) - max_history_turns (>=1; warns >100) - rag_top_k (1..64) - rag_chunk_overlap strictly < rag_chunk_size (else no new content in any chunk — wasteful and confusing) - theme_value ({light,dark,system,auto}) - OLLAMA_HOST env (INFO if unset) - CORS_ORIGINS env ('*' rejected; missing scheme warned) - chunk_tuning_consistency (INFO when chunk_size % overlap == 0; retrieval returns near-identical chunks for adjacent queries) New tooling: scripts/config_validator_cli.py: validate — validates and prints JSON summary with all issues + counts; exit non-zero if errors. selfcheck — runs the ruleset against a canary bad config with ≥5 obvious errors to smoke-test the validator itself. Exit codes 0/1/2. UTF-8 stdout wrapped so non-ASCII glyphs (the '≥' in 'must be ≥ 1' messages) don't crash Windows cp1252 console. New tests: backend/tests/test_config_validator.py — 46 pytest cases: - TestValidationIssue (3): construction, invalid severity raises, to_dict shape. - TestValidateDriver (4): empty rules, no issue passes, rule exception surfaces as ERROR without crashing, end-to-end clean config. - TestDefaultModelRule / TestLanguageRule / TestTemperatureRule / TestMaxHistoryTurnsRule / TestRagTopKRule / TestRagOverlapRule / TestChunkTuningConsistency / TestThemeRule (each): clean path, boundary, mutant path per the rule's invariants. Bool rejected for int fields (Python's True isinstance int gotcha). - TestEnvRules (5): OLLAMA_HOST unset -> INFO, explicit -> silent; CORS wildcard rejected, missing-scheme warned, well-formed ok. - TestSummarise (2): empty + mixed counts. - TestAggregateHelpers (4): has_errors / has_warnings shape. - TestLoadRulesFromModule (3): round-trip via stub module on sys.path, missing RULES default [], wrong-type raises. - TestEndToEnd (1): default LocalMind config -> 0 errors. Acceptance criteria from issue #937: [x] Implement the change in the tests and tooling [x] Add or update tests, docs, or validation for the touched path --- backend/tests/test_config_validator.py | 411 +++++++++++++++++++++++++ backend/utils/config_validator.py | 379 +++++++++++++++++++++++ scripts/config_validator_cli.py | 107 +++++++ 3 files changed, 897 insertions(+) create mode 100644 backend/tests/test_config_validator.py create mode 100644 backend/utils/config_validator.py create mode 100644 scripts/config_validator_cli.py diff --git a/backend/tests/test_config_validator.py b/backend/tests/test_config_validator.py new file mode 100644 index 00000000..339870f3 --- /dev/null +++ b/backend/tests/test_config_validator.py @@ -0,0 +1,411 @@ +"""Tests for backend/utils/config_validator.py (issue #937). + +Covers: +- ValidationIssue severity validation. +- ValidationRule invocation (happy + exception propagation). +- Each rule in DEFAULT_RULES, both clean & mutant inputs. +- Cross-field rules (overlap < chunk size). +- Environment-variable rules (OLLAMA_HOST, CORS_ORIGINS). +- summarise / has_errors / has_warnings shape. +- load_rules_from_module round-trips via a stub module on sys.path. +""" + +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +from utils.config_validator import ( + DEFAULT_RULES, + SEVERITY_ERROR, + SEVERITY_INFO, + SEVERITY_WARN, + ValidationIssue, + ValidationRule, + has_errors, + has_warnings, + load_rules_from_module, + summarise, + validate, +) + + +def _env_rule_names() -> list[str]: + return [r.name for r in DEFAULT_RULES] + + +# ─── ValidationIssue dataclass ──────────────────────────────────── + + +class TestValidationIssue: + def test_clean_construction(self): + i = ValidationIssue(SEVERITY_ERROR, "settings.x", "bad") + assert i.severity == SEVERITY_ERROR + assert i.location == "settings.x" + assert i.message == "bad" + + def test_invalid_severity_raises(self): + with pytest.raises(ValueError, match="severity"): + ValidationIssue("FATAL", "settings.x", "x") + + def test_to_dict_shape(self): + d = ValidationIssue(SEVERITY_WARN, "a", "b").to_dict() + assert d == {"severity": SEVERITY_WARN, "location": "a", "message": "b"} + + +# ─── validate() — happy + exception paths ──────────────────────── + + +class TestValidateDriver: + def test_empty_rules_returns_empty(self): + issues = validate({}, rules=[]) + assert issues == [] + + def test_no_issue_when_rule_passes(self): + rule = ValidationRule("noop", lambda c: []) + assert validate({}, rules=[rule]) == [] + + def test_rule_exception_surfaces_as_error(self): + def boom(cfg): + raise RuntimeError("kaput") + + rule = ValidationRule("boom", boom) + issues = validate({}, rules=[rule]) + assert len(issues) == 1 + assert issues[0].severity == SEVERITY_ERROR + assert issues[0].location == "__rule__.boom" + assert "kaput" in issues[0].message + + def test_default_ruleset_runs_clean_config(self): + config = { + "settings": { + "default_model": "llama3", + "default_language": "en", + "temperature": 0.7, + "max_history_turns": 10, + "rag_top_k": 4, + "rag_chunk_overlap": 70, # non-divisor of 600 → no INFO. + "rag_chunk_size": 600, + "theme": "dark", + }, + "env": {"OLLAMA_HOST": "http://localhost:11434"}, + } + os.environ.pop("CORS_ORIGINS", None) + issues = validate(config) + # application has clean running config; OLLAMA_HOST explicit. + assert not has_errors(issues) + + +# ─── Individual default-ruleset rules ───────────────────────────── + + +class TestDefaultModelRule: + def test_missing(self): + issues = validate({"settings": {}}, rules=[DEFAULT_RULES[0]]) + assert has_errors(issues) + + def test_empty_string(self): + issues = validate({"settings": {"default_model": ""}}, rules=[DEFAULT_RULES[0]]) + assert has_errors(issues) + + def test_non_string(self): + issues = validate({"settings": {"default_model": 42}}, rules=[DEFAULT_RULES[0]]) + assert has_errors(issues) + + +class TestLanguageRule: + def test_supported_no_issue(self): + rule = next(r for r in DEFAULT_RULES if r.name == "language_supported") + issues = validate({"settings": {"default_language": "en"}}, rules=[rule]) + assert issues == [] + + def test_unsupported_warns(self): + rule = next(r for r in DEFAULT_RULES if r.name == "language_supported") + issues = validate({"settings": {"default_language": "xx"}}, rules=[rule]) + assert has_warnings(issues) + + +class TestTemperatureRule: + def test_negative(self): + rule = next(r for r in DEFAULT_RULES if r.name == "temperature_range") + issues = validate({"settings": {"temperature": -0.1}}, rules=[rule]) + assert has_errors(issues) + + def test_too_high(self): + rule = next(r for r in DEFAULT_RULES if r.name == "temperature_range") + issues = validate({"settings": {"temperature": 3.0}}, rules=[rule]) + assert has_errors(issues) + + def test_extreme_boundaries_ok(self): + rule = next(r for r in DEFAULT_RULES if r.name == "temperature_range") + # 0.0 and 2.0 should pass (ge/le inclusive). + assert validate({"settings": {"temperature": 0.0}}, rules=[rule]) == [] + assert validate({"settings": {"temperature": 2.0}}, rules=[rule]) == [] + + +class TestMaxHistoryTurnsRule: + def test_zero_rejected(self): + rule = next(r for r in DEFAULT_RULES if r.name == "max_history_turns") + issues = validate({"settings": {"max_history_turns": 0}}, rules=[rule]) + assert has_errors(issues) + + def test_negative_rejected(self): + rule = next(r for r in DEFAULT_RULES if r.name == "max_history_turns") + issues = validate({"settings": {"max_history_turns": -1}}, rules=[rule]) + assert has_errors(issues) + + def test_huge_warns(self): + rule = next(r for r in DEFAULT_RULES if r.name == "max_history_turns") + issues = validate({"settings": {"max_history_turns": 200}}, rules=[rule]) + assert has_warnings(issues) + assert not has_errors(issues) + + def test_bool_rejected(self): + # bool is a subclass of int — must be rejected explicitly. + rule = next(r for r in DEFAULT_RULES if r.name == "max_history_turns") + issues = validate({"settings": {"max_history_turns": True}}, rules=[rule]) + assert has_errors(issues) + + +class TestRagTopKRule: + def test_zero_rejected(self): + rule = next(r for r in DEFAULT_RULES if r.name == "rag_top_k") + issues = validate({"settings": {"rag_top_k": 0}}, rules=[rule]) + assert has_errors(issues) + + def test_too_large(self): + rule = next(r for r in DEFAULT_RULES if r.name == "rag_top_k") + issues = validate({"settings": {"rag_top_k": 65}}, rules=[rule]) + assert has_errors(issues) + + def test_upper_boundary(self): + rule = next(r for r in DEFAULT_RULES if r.name == "rag_top_k") + assert validate({"settings": {"rag_top_k": 64}}, rules=[rule]) == [] + + +class TestRagOverlapRule: + RULEName = "rag_overlap_lt_chunk" + + def test_overlap_ge_size_error(self): + rule = next(r for r in DEFAULT_RULES if r.name == self.RULEName) + issues = validate( + {"settings": {"rag_chunk_overlap": 600, "rag_chunk_size": 600}}, + rules=[rule], + ) + assert has_errors(issues) + + def test_overlap_greater_than_size_error(self): + rule = next(r for r in DEFAULT_RULES if r.name == self.RULEName) + issues = validate( + {"settings": {"rag_chunk_overlap": 700, "rag_chunk_size": 600}}, + rules=[rule], + ) + assert has_errors(issues) + + def test_overlap_lt_size_ok(self): + rule = next(r for r in DEFAULT_RULES if r.name == self.RULEName) + assert ( + validate( + {"settings": {"rag_chunk_overlap": 50, "rag_chunk_size": 600}}, + rules=[rule], + ) + == [] + ) + + def test_overlap_zero_ok(self): + rule = next(r for r in DEFAULT_RULES if r.name == self.RULEName) + assert ( + validate( + {"settings": {"rag_chunk_overlap": 0, "rag_chunk_size": 600}}, + rules=[rule], + ) + == [] + ) + + def test_non_int_overlap_rejected(self): + rule = next(r for r in DEFAULT_RULES if r.name == self.RULEName) + issues = validate( + {"settings": {"rag_chunk_overlap": "fifty"}}, + rules=[rule], + ) + assert has_errors(issues) + + +class TestChunkTuningConsistency: + RULEName = "chunk_tuning_consistency" + + def test_multiple_warns_info(self): + rule = next(r for r in DEFAULT_RULES if r.name == self.RULEName) + issues = validate( + {"settings": {"rag_chunk_overlap": 100, "rag_chunk_size": 1000}}, + rules=[rule], + ) + # This is an INFO — not an error/warning — but worth surfacing. + assert any(i.severity == SEVERITY_INFO for i in issues) + + def test_non_multiple_no_issue(self): + rule = next(r for r in DEFAULT_RULES if r.name == self.RULEName) + # 600 is divisible by 50; pick values that aren't divisible. + issues = validate( + {"settings": {"rag_chunk_overlap": 70, "rag_chunk_size": 600}}, + rules=[rule], + ) + assert issues == [] + + +class TestThemeRule: + def test_unknown_theme_warns(self): + rule = next(r for r in DEFAULT_RULES if r.name == "theme_value") + issues = validate({"settings": {"theme": "lavender"}}, rules=[rule]) + assert has_warnings(issues) + + def test_dark_ok(self): + rule = next(r for r in DEFAULT_RULES if r.name == "theme_value") + assert validate({"settings": {"theme": "dark"}}, rules=[rule]) == [] + + +# ─── Environment-variable rules ──────────────────────────────────── + + +class TestEnvRules: + def test_ollama_host_unset_info(self, monkeypatch): + monkeypatch.delenv("OLLAMA_HOST", raising=False) + rule = next(r for r in DEFAULT_RULES if r.name == "ollama_host_env") + issues = validate({"env": {}}, rules=[rule]) + assert any(i.severity == SEVERITY_INFO for i in issues) + + def test_ollama_host_set_no_issue(self, monkeypatch): + monkeypatch.setenv("OLLAMA_HOST", "http://ollama:11434") + rule = next(r for r in DEFAULT_RULES if r.name == "ollama_host_env") + issues = validate({"env": {}}, rules=[rule]) + assert issues == [] + + def test_cors_wildcard_rejected(self, monkeypatch): + monkeypatch.setenv("CORS_ORIGINS", "*") + rule = next(r for r in DEFAULT_RULES if r.name == "cors_origins_env") + issues = validate({}, rules=[rule]) + assert has_errors(issues) + + def test_cors_no_scheme_warns(self, monkeypatch): + monkeypatch.setenv("CORS_ORIGINS", "example.com, http://ok") + rule = next(r for r in DEFAULT_RULES if r.name == "cors_origins_env") + issues = validate({}, rules=[rule]) + assert has_warnings(issues) + + def test_cors_well_formed_ok(self, monkeypatch): + monkeypatch.setenv("CORS_ORIGINS", "http://localhost:3000, https://app.x") + rule = next(r for r in DEFAULT_RULES if r.name == "cors_origins_env") + issues = validate({}, rules=[rule]) + assert issues == [] + + +# ─── summarise / has_errors / has_warnings ──────────────────────── + + +class TestSummarise: + def test_empty_summary_zero(self): + s = summarise([]) + assert s == { + "total": 0, + "errors": 0, + "warnings": 0, + "infos": 0, + "issues": [], + } + + def test_mixed_counts(self): + issues = [ + ValidationIssue(SEVERITY_ERROR, "a", "x"), + ValidationIssue(SEVERITY_WARN, "b", "y"), + ValidationIssue(SEVERITY_INFO, "c", "z"), + ] + s = summarise(issues) + assert s["total"] == 3 + assert s["errors"] == 1 + assert s["warnings"] == 1 + assert s["infos"] == 1 + assert len(s["issues"]) == 3 + + +class TestAggregateHelpers: + def test_has_errors_false_when_clean(self): + assert has_errors([]) is False + + def test_has_errors_true_with_one(self): + assert has_errors([ValidationIssue(SEVERITY_ERROR, "x", "y")]) is True + + def test_has_warnings_true_with_one(self): + assert has_warnings([ValidationIssue(SEVERITY_WARN, "x", "y")]) is True + + def test_has_warnings_false_with_only_info(self): + assert has_warnings([ValidationIssue(SEVERITY_INFO, "x", "y")]) is False + + +# ─── load_rules_from_module ──────────────────────────────────────── + + +class TestLoadRulesFromModule: + def test_loads_rules_list(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + # Write a stub module to tmp_path and put it on sys.path. + stub = tmp_path / "extra_rules_for_test.py" + stub.write_text( + "from utils.config_validator import ValidationRule, ValidationIssue, " + "SEVERITY_INFO\n" + "def _check(cfg):\n" + " return [ValidationIssue(SEVERITY_INFO, 'synthetic', 'hi')]\n" + "RULES = [ValidationRule('synthetic', _check)]\n", + encoding="utf-8", + ) + monkeypatch.syspath_prepend(str(tmp_path)) + rules = load_rules_from_module("extra_rules_for_test") + assert len(rules) == 1 + assert rules[0].name == "synthetic" + issues = rules[0].fn({}) + assert issues[0].severity == SEVERITY_INFO + + def test_missing_rules_raises( + self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ): + stub = tmp_path / "no_rules_module.py" + stub.write_text("x = 1\n", encoding="utf-8") + monkeypatch.syspath_prepend(str(tmp_path)) + # RULES attribute defaults to empty list — load returns []. + rules = load_rules_from_module("no_rules_module") + assert rules == [] + + def test_wrong_type_rules_raises( + self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ): + stub = tmp_path / "bad_rules_module.py" + stub.write_text("RULES = 'not a list'\n", encoding="utf-8") + monkeypatch.syspath_prepend(str(tmp_path)) + with pytest.raises(TypeError, match="RULES"): + load_rules_from_module("bad_rules_module") + + +# ─── End-to-end: validate a default LocalMind config ────────────── + + +class TestEndToEnd: + def test_clean_config_no_errors(self, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("OLLAMA_HOST", "http://ollama:11434") + monkeypatch.delenv("CORS_ORIGINS", raising=False) + config = { + "settings": { + "default_model": "llama3", + "default_language": "en", + "temperature": 0.7, + "max_history_turns": 10, + "rag_top_k": 4, + "rag_chunk_overlap": 70, # non-divisor → no INFO. + "rag_chunk_size": 600, + "theme": "dark", + "minimal_mode": False, + }, + "env": {"OLLAMA_HOST": "http://ollama:11434"}, + } + issues = validate(config) + results = summarise(issues) + assert results["errors"] == 0 diff --git a/backend/utils/config_validator.py b/backend/utils/config_validator.py new file mode 100644 index 00000000..8ec76fdd --- /dev/null +++ b/backend/utils/config_validator.py @@ -0,0 +1,379 @@ +"""Config validation — zero-dependency rules-engine for app settings. + +The existing ``AppSettings`` Pydantic model validates *individual* field +types but does not enforce cross-field or system-cap constraints +(e.g. ``rag_chunk_overlap < rag_chunk_size``, ``max_history_turns ≥ 1``, +``OLLAMA_HOST reachable``, ``rag_top_k within [1, 64]``). + +This module supplies: +- ``ValidationRule`` — a named callable applied to a config dict, + returns ``[ValidationIssue(severity, message)]`` (empty if fine). +- ``ValidationIssue`` — (severity, location, message) — Severity in + {ERROR, WARN, INFO}; locations are dotted paths like + ``settings.rag_chunk_overlap``. +- ``validate(config, rules=None)`` — apply a ruleset to a config dict + and return a list of issues (empty list == no problems). +- ``load_rules_from_module(spec)`` — pull rules from any Python + module that exposes ``RULES`` (uniform extension surface). +- Default ruleset ``DEFAULT_RULES`` — covers the LocalMind + AppSettings surface plus environment-variable sanity. + +Designed to be invoked at backend startup (``app.py`` lifespan), in +unit tests, and via the CLI tool. +""" + +from __future__ import annotations + +import os +import re +from dataclasses import dataclass +from typing import Any, Callable, Iterable + +SEVERITY_ERROR = "ERROR" +SEVERITY_WARN = "WARN" +SEVERITY_INFO = "INFO" +_SEVERITIES = (SEVERITY_ERROR, SEVERITY_WARN, SEVERITY_INFO) + + +@dataclass(frozen=True) +class ValidationIssue: + """A single issue surfaced by a ValidationRule.""" + + severity: str + location: str + message: str + + def __post_init__(self): + if self.severity not in _SEVERITIES: + raise ValueError(f"severity must be one of {_SEVERITIES}") + + def to_dict(self) -> dict: + return { + "severity": self.severity, + "location": self.location, + "message": self.message, + } + + +@dataclass +class ValidationRule: + """A named check applied to a config dict. + + ``fn(config) -> [ValidationIssue]`` — return an empty list to + signal success. Never raises; ``validate()`` wraps calls in a + try/except so a buggy rule does not poison the report. + """ + + name: str + fn: Callable[[dict], list[ValidationIssue]] + + +# --------------------------------------------------------------------------- +# Helpers used by the default ruleset; exposed for tests + custom rules. +# --------------------------------------------------------------------------- + + +def _get(config: dict, dotted: str, default: Any = None) -> Any: + """Look up a nested key by dotted path.""" + cur: Any = config + for part in dotted.split("."): + if isinstance(cur, dict) and part in cur: + cur = cur[part] + else: + return default + return cur + + +def _is_int_in_range(value: Any, low: int, high: int) -> bool: + if isinstance(value, bool) or not isinstance(value, int): + return False + return low <= value <= high + + +def _is_float_in_range(value: Any, low: float, high: float) -> bool: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return False + return low <= float(value) <= high + + +def _is_str_in(value: Any, options: Iterable[str]) -> bool: + return isinstance(value, str) and value in set(options) + + +# --------------------------------------------------------------------------- +# Default ruleset — covers LocalMind AppSettings + RAG/timing constraints. +# --------------------------------------------------------------------------- + + +def _check_model_name(config: dict) -> list[ValidationIssue]: + """Default model should be a non-empty string.""" + model = _get(config, "settings.default_model") + if not isinstance(model, str) or not model.strip(): + return [ + ValidationIssue( + SEVERITY_ERROR, + "settings.default_model", + f"default_model must be a non-empty string (got {model!r})", + ) + ] + return [] + + +def _check_language_supported(config: dict) -> list[ValidationIssue]: + """Default language must be one of the supported set.""" + supported = ("en", "ja", "es", "fr", "de") + lang = _get(config, "settings.default_language") + if not _is_str_in(lang, supported): + return [ + ValidationIssue( + SEVERITY_WARN, + "settings.default_language", + f"default_language {lang!r} not in supported set {supported}; " + "the UI may fall back to 'en'.", + ) + ] + return [] + + +def _check_temperature_range(config: dict) -> list[ValidationIssue]: + temp = _get(config, "settings.temperature") + if not _is_float_in_range(temp, 0.0, 2.0): + return [ + ValidationIssue( + SEVERITY_ERROR, + "settings.temperature", + f"temperature must be a float in [0.0, 2.0] (got {temp!r})", + ) + ] + return [] + + +def _check_max_history_turns(config: dict) -> list[ValidationIssue]: + n = _get(config, "settings.max_history_turns") + issues = [] + if not isinstance(n, int) or isinstance(n, bool): + issues.append( + ValidationIssue( + SEVERITY_ERROR, + "settings.max_history_turns", + f"max_history_turns must be an int (got {type(n).__name__})", + ) + ) + return issues + if n < 1: + issues.append( + ValidationIssue( + SEVERITY_ERROR, + "settings.max_history_turns", + f"max_history_turns must be ≥ 1 (got {n})", + ) + ) + elif n > 100: + issues.append( + ValidationIssue( + SEVERITY_WARN, + "settings.max_history_turns", + f"max_history_turns={n} causes large O(N) memory + token " + "footprint per inference; consider ≤ 20", + ) + ) + return issues + + +def _check_rag_top_k(config: dict) -> list[ValidationIssue]: + k = _get(config, "settings.rag_top_k") + if not _is_int_in_range(k, 1, 64): + return [ + ValidationIssue( + SEVERITY_ERROR, + "settings.rag_top_k", + f"rag_top_k must be an int in [1, 64] (got {k!r})", + ) + ] + return [] + + +def _check_rag_overlap_lt_chunk_size(config: dict) -> list[ValidationIssue]: + """Overlap must be strictly less than the chunk size, else the + next chunk contains nothing new — wasteful & confusing.""" + overlap = _get(config, "settings.rag_chunk_overlap") + size = _get(config, "settings.rag_chunk_size") + issues = [] + if not _is_int_in_range(overlap, 0, 10000): + issues.append( + ValidationIssue( + SEVERITY_ERROR, + "settings.rag_chunk_overlap", + f"rag_chunk_overlap must be an int in [0, 10000] (got {overlap!r})", + ) + ) + return issues + if size is not None and _is_int_in_range(size, 100, 10000): + if overlap >= size: + issues.append( + ValidationIssue( + SEVERITY_ERROR, + "settings.rag_chunk_overlap", + f"rag_chunk_overlap ({overlap}) must be strictly less than " + f"rag_chunk_size ({size}) — otherwise chunks are " + "identical to their predecessor.", + ) + ) + return issues + + +def _check_theme_value(config: dict) -> list[ValidationIssue]: + theme = _get(config, "settings.theme") + if not _is_str_in(theme, ("light", "dark", "system", "auto")): + return [ + ValidationIssue( + SEVERITY_WARN, + "settings.theme", + f"theme {theme!r} not in (light, dark, system, auto); " + "UI may default incorrectly.", + ) + ] + return [] + + +def _check_ollama_host_env(config: dict) -> list[ValidationIssue]: + """If OLLAMA_HOST is unset we fall back to localhost — warn.""" + env_host = os.environ.get("OLLAMA_HOST") + cfg_host = _get(config, "env.OLLAMA_HOST", env_host) + if not cfg_host: + return [ + ValidationIssue( + SEVERITY_INFO, + "env.OLLAMA_HOST", + "OLLAMA_HOST unset; backend will fall back to " + "http://localhost:11434. Set this in production.", + ) + ] + return [] + + +def _check_cors_origins_env(config: dict) -> list[ValidationIssue]: + origins = os.environ.get("CORS_ORIGINS") or _get(config, "env.CORS_ORIGINS") + if origins is None: + return [] + if isinstance(origins, str): + parts = [o.strip() for o in origins.split(",") if o.strip()] + if "*" in parts: + return [ + ValidationIssue( + SEVERITY_ERROR, + "env.CORS_ORIGINS", + "Wildcard '*' in CORS_ORIGINS is unsafe outside dev.", + ) + ] + bad = [o for o in parts if not re.match(r"^[a-z]+://", o, re.IGNORECASE)] + if bad: + return [ + ValidationIssue( + SEVERITY_WARN, + "env.CORS_ORIGINS", + f"some origins are not absolute URIs (missing scheme): {bad}.", + ) + ] + return [] + + +def _check_chunk_tuning_consistency(config: dict) -> list[ValidationIssue]: + """A chunk-size that's a multiple of overlap is suspicious — it + means no new content slips through the window and the retreival + will return near-identical chunks for adjacent queries.""" + size = _get(config, "settings.rag_chunk_size") + overlap = _get(config, "settings.rag_chunk_overlap") + if ( + _is_int_in_range(overlap, 1, 10000) + and _is_int_in_range(size, 100, 10000) + and size > overlap + and size % overlap == 0 + ): + return [ + ValidationIssue( + SEVERITY_INFO, + "settings.rag_chunk_overlap", + f"rag_chunk_size ({size}) is a multiple of rag_chunk_overlap " + f"({overlap}). The retrieval window bleeds no new content; " + "consider an overlap of size*0.2 for diversity.", + ) + ] + return [] + + +DEFAULT_RULES: tuple[ValidationRule, ...] = ( + ValidationRule("model_name", _check_model_name), + ValidationRule("language_supported", _check_language_supported), + ValidationRule("temperature_range", _check_temperature_range), + ValidationRule("max_history_turns", _check_max_history_turns), + ValidationRule("rag_top_k", _check_rag_top_k), + ValidationRule("rag_overlap_lt_chunk", _check_rag_overlap_lt_chunk_size), + ValidationRule("theme_value", _check_theme_value), + ValidationRule("ollama_host_env", _check_ollama_host_env), + ValidationRule("cors_origins_env", _check_cors_origins_env), + ValidationRule("chunk_tuning_consistency", _check_chunk_tuning_consistency), +) + + +def validate( + config: dict, + rules: Iterable[ValidationRule] | None = None, +) -> list[ValidationIssue]: + """Run ``rules`` against ``config``; return a list of issues (empty = ok). + + Rules are wrapped in try/except — a buggy rule surfaces as a single + ERROR issue with its exception string, not by crashing the run. + """ + rules_to_run = list(rules) if rules is not None else list(DEFAULT_RULES) + issues: list[ValidationIssue] = [] + for rule in rules_to_run: + try: + rule_issues = rule.fn(config) + except Exception as exc: # noqa: BLE001 + issues.append( + ValidationIssue( + SEVERITY_ERROR, + f"__rule__.{rule.name}", + f"rule raised {type(exc).__name__}: {exc}", + ) + ) + continue + if rule_issues: + issues.extend(rule_issues) + return issues + + +def has_errors(issues: list[ValidationIssue]) -> bool: + return any(i.severity == SEVERITY_ERROR for i in issues) + + +def has_warnings(issues: list[ValidationIssue]) -> bool: + return any(i.severity == SEVERITY_WARN for i in issues) + + +def summarise(issues: list[ValidationIssue]) -> dict: + return { + "total": len(issues), + "errors": sum(1 for i in issues if i.severity == SEVERITY_ERROR), + "warnings": sum(1 for i in issues if i.severity == SEVERITY_WARN), + "infos": sum(1 for i in issues if i.severity == SEVERITY_INFO), + "issues": [i.to_dict() for i in issues], + } + + +def load_rules_from_module(spec: str) -> list[ValidationRule]: + """Import a Python module and pull its ``RULES`` list. + + Lets users extend the default ruleset without monkey-patching. + ``spec`` is a dotted module path resolvable on sys.path. + """ + import importlib + + mod = importlib.import_module(spec) + rules = getattr(mod, "RULES", []) + if not isinstance(rules, (list, tuple)): + raise TypeError( + f"{spec}.RULES must be a list/tuple of ValidationRule (got {type(rules).__name__})" + ) + return list(rules) diff --git a/scripts/config_validator_cli.py b/scripts/config_validator_cli.py new file mode 100644 index 00000000..c8acca70 --- /dev/null +++ b/scripts/config_validator_cli.py @@ -0,0 +1,107 @@ +"""config_validator_cli.py — offline CLI for the config validator. + +Two modes: + validate — validate a JSON config file. + selfcheck — run the default ruleset against a + pre-canned BAD config to smoke-test the + validator itself (useful in CI). + +Exit codes: 0 no errors · 1 has errors (or invalid input) · 2 file not found. +""" + +from __future__ import annotations + +import argparse +import io +import json +import os +import sys +from pathlib import Path + +# Force stdout/stderr UTF-8 — Windows cp1252 chokes on ≥ ≤ etc. +sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", line_buffering=True) + +_HERE = Path(__file__).resolve().parent +_BACKEND = _HERE.parent / "backend" +if str(_BACKEND) not in sys.path: + sys.path.insert(0, str(_BACKEND)) + +from utils.config_validator import ( # noqa: E402 + summarise, + validate, +) + + +def _read_config_file(path: Path) -> dict: + if not path.exists(): + print(f"error: file not found: {path}", file=sys.stderr) + sys.exit(2) + try: + data = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError as exc: + print(f"error: not valid JSON: {exc.msg}", file=sys.stderr) + sys.exit(1) + if not isinstance(data, dict): + print( + f"error: top-level must be a JSON object (got {type(data).__name__})", + file=sys.stderr, + ) + sys.exit(1) + return data + + +def cmd_validate(args: argparse.Namespace) -> int: + config = _read_config_file(Path(args.input)) + issues = validate(config) + summary = summarise(issues) + summary["input"] = args.input + print(json.dumps(summary, indent=2, ensure_ascii=False)) + return 1 if summary["errors"] > 0 else 0 + + +def cmd_selfcheck(_args: argparse.Namespace) -> int: + # A canary bad config that should surface at least one issue from + # every rule. If the validator returns empty, something is broken. + os.environ.pop("OLLAMA_HOST", None) + os.environ.pop("CORS_ORIGINS", None) + bad = { + "settings": { + "default_model": "", # error + "default_language": "klingon", # warn + "temperature": 5.0, # error + "max_history_turns": 0, # error + "rag_top_k": 999, # error + "rag_chunk_overlap": 600, # error + "rag_chunk_size": 600, + "theme": "rose-gold", # warn + }, + "env": {}, + } + issues = validate(bad) + summary = summarise(issues) + print(json.dumps(summary, indent=2, ensure_ascii=False)) + # Smoke-test exit code: 0 if validator reported >=5 errors (one + # per erroneous field), 1 otherwise. + return 0 if summary["errors"] >= 5 else 1 + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + prog="config_validator_cli", + description="Offline CLI for the LocalMind config validator.", + ) + sub = parser.add_subparsers(dest="command", required=True) + + p_validate = sub.add_parser("validate", help="Validate a JSON config file.") + p_validate.add_argument("input", help="Path to a JSON config file.") + p_validate.set_defaults(func=cmd_validate) + + p_selfcheck = sub.add_parser("selfcheck", help="Self-test the validator.") + p_selfcheck.set_defaults(func=cmd_selfcheck) + + args = parser.parse_args(argv) + return args.func(args) + + +if __name__ == "__main__": + sys.exit(main())