"""Local jsonschema validation (problems.md #13) — call_structured / call_multiturn. provider strict mode (LiteLLM ``response_format=json_schema strict=True``) 가 일부 provider 에서 schema enforcement 약화로 nested 필드 누락/enum 위반 통과될 수 있다. ``_validate_local_schema`` 가 ``jsonschema.validate`` 로 추가 검증하고 실패 시 ``SchemaValidationError`` raise → ``is_safety_related_error`` 가 True 로 분류하여 Tier 2/3 fallback 진행. 검증 시나리오: - default ON, valid payload → 통과 - default ON, invalid payload → SchemaValidationError + is_safety_related True - ENV ``LLM_LOCAL_SCHEMA_VALIDATE=false`` → no-op - empty/non-dict schema → no-op (legacy/free-form 보호) - 잘못된 schema → ValueError (fallback 분류 제외) - call_structured Tier 1 schema fail → Tier 2 sanitize 진입 → 통과 - call_multiturn structured 모드 schema fail → Tier 2 sanitize 진입 → 통과 - call_multiturn free-text 모드 → schema validate skip """ from __future__ import annotations import json from typing import Any, Dict from unittest.mock import MagicMock, patch import pytest # --------------------------------------------------------------------------- # 공용 헬퍼 # --------------------------------------------------------------------------- def _make_response(content: str) -> MagicMock: msg = MagicMock() msg.content = content choice = MagicMock() choice.message = msg resp = MagicMock() resp.choices = [choice] return resp def _ok_json(payload: Dict[str, Any]) -> MagicMock: return _make_response(json.dumps(payload)) @pytest.fixture def mock_router(): """``_get_router`` patch 하여 router.completion 을 직접 제어.""" # [2026-08-01 A5 후속] 봉합 지점이 `_get_router` 에서 `_get_router_binding` # 으로 옮겼다. Router 는 이제 **그것을 지은 슬롯**과 짝으로만 다뤄진다 — # 실패한 슬롯을 전역에서 다시 읽으면 stale Router 의 실패가 남의 슬롯 # 실패로 보고되기 때문이다. from app.modules.llm.llm_client import _RouterBinding with patch("app.modules.llm.llm_client._get_router_binding") as mock_get: router = MagicMock() mock_get.return_value = _RouterBinding(slot="primary", router=router) yield router @pytest.fixture(autouse=True) def _reset_validate_env(monkeypatch): monkeypatch.delenv("LLM_LOCAL_SCHEMA_VALIDATE", raising=False) _SCHEMA_REQUIRED_A = { "type": "object", "properties": {"a": {"type": "string"}}, "required": ["a"], "additionalProperties": False, } # --------------------------------------------------------------------------- # 1. _validate_local_schema 직접 단위 테스트 # --------------------------------------------------------------------------- class TestValidateLocalSchema: def test_default_on_passes_valid_payload(self): from app.modules.llm.llm_client import _validate_local_schema _validate_local_schema( {"a": "hello"}, _SCHEMA_REQUIRED_A, step="test", schema_name="resp", ) # no raise def test_default_on_raises_for_missing_required(self): from app.modules.llm.llm_client import _validate_local_schema from app.modules.llm.safety import ( SchemaValidationError, is_safety_related_error, ) with pytest.raises(SchemaValidationError) as excinfo: _validate_local_schema( {}, _SCHEMA_REQUIRED_A, step="test", schema_name="resp", ) assert "step=test" in str(excinfo.value) assert "schema=resp" in str(excinfo.value) # safety 분류로 fallback 진행 보장. assert is_safety_related_error(excinfo.value) is True def test_default_on_raises_for_wrong_type(self): from app.modules.llm.llm_client import _validate_local_schema from app.modules.llm.safety import SchemaValidationError with pytest.raises(SchemaValidationError) as excinfo: _validate_local_schema( {"a": 123}, _SCHEMA_REQUIRED_A, step="test", schema_name="resp", ) # 잘못된 path 노출. assert "path=a" in str(excinfo.value) def test_default_on_raises_for_extra_property(self): from app.modules.llm.llm_client import _validate_local_schema from app.modules.llm.safety import SchemaValidationError with pytest.raises(SchemaValidationError): _validate_local_schema( {"a": "hi", "extra": 1}, _SCHEMA_REQUIRED_A, step="test", schema_name="resp", ) def test_disabled_via_env_skips_validation(self, monkeypatch): from app.modules.llm.llm_client import _validate_local_schema monkeypatch.setenv("LLM_LOCAL_SCHEMA_VALIDATE", "false") _validate_local_schema( {}, _SCHEMA_REQUIRED_A, step="test", schema_name="resp", ) # no raise @pytest.mark.parametrize("falsy", ["false", "0", "no", "off"]) def test_disabled_env_truthy_parsing(self, monkeypatch, falsy): from app.modules.llm.llm_client import ( _is_local_schema_validate_enabled, _validate_local_schema, ) monkeypatch.setenv("LLM_LOCAL_SCHEMA_VALIDATE", falsy) assert _is_local_schema_validate_enabled() is False _validate_local_schema({}, _SCHEMA_REQUIRED_A, step="t", schema_name="r") @pytest.mark.parametrize("truthy", ["true", "1", "yes", "on", "TRUE"]) def test_enabled_env_truthy_parsing(self, monkeypatch, truthy): from app.modules.llm.llm_client import _is_local_schema_validate_enabled monkeypatch.setenv("LLM_LOCAL_SCHEMA_VALIDATE", truthy) assert _is_local_schema_validate_enabled() is True def test_unset_env_defaults_on(self, monkeypatch): from app.modules.llm.llm_client import _is_local_schema_validate_enabled monkeypatch.delenv("LLM_LOCAL_SCHEMA_VALIDATE", raising=False) assert _is_local_schema_validate_enabled() is True def test_empty_schema_no_op(self): from app.modules.llm.llm_client import _validate_local_schema _validate_local_schema({"any": "thing"}, {}, step="t", schema_name="r") _validate_local_schema(None, {}, step="t", schema_name="r") # 비-dict payload def test_non_dict_schema_no_op(self): from app.modules.llm.llm_client import _validate_local_schema _validate_local_schema({"a": 1}, None, step="t", schema_name="r") # type: ignore[arg-type] def test_invalid_schema_raises_invalid_schema_error(self): """잘못된 schema 자체 → InvalidSchemaError (ValueError 호환).""" from app.modules.llm.llm_client import _validate_local_schema from app.modules.llm.safety import InvalidSchemaError with pytest.raises(InvalidSchemaError) as excinfo: _validate_local_schema( {"a": 1}, {"type": "banana"}, step="t", schema_name="r", ) assert "Invalid jsonschema spec" in str(excinfo.value) # ValueError 호환성 보존 (caller 의 except ValueError: 에 잡혀야). assert isinstance(excinfo.value, ValueError) def test_invalid_schema_error_not_classified_as_safety(self): """InvalidSchemaError → fallback 분류 제외 (문제 #13 B2 회귀 가드). 메시지에 ``schema`` 키워드가 들어가도 클래스 이름 매칭이 우선이라 False. ``is_safety_related_error`` 가 True 반환하면 Tier 2/3 LLM 호출 3배 비용. """ from app.modules.llm.llm_client import _validate_local_schema from app.modules.llm.safety import InvalidSchemaError, is_safety_related_error with pytest.raises(InvalidSchemaError) as excinfo: _validate_local_schema({}, {"type": "banana"}, step="t", schema_name="r") assert is_safety_related_error(excinfo.value) is False # --------------------------------------------------------------------------- # 2. call_structured 통합 — Tier 진입/fallback 시나리오 # --------------------------------------------------------------------------- class TestCallStructuredSchemaValidation: def test_tier1_passes_when_schema_valid(self, mock_router): from app.modules.llm.llm_client import call_structured mock_router.completion.return_value = _ok_json({"a": "hello"}) result = call_structured( step="test", system_prompt="sys", user_prompt="user", response_schema=_SCHEMA_REQUIRED_A, ) assert result == {"a": "hello"} assert mock_router.completion.call_count == 1 def test_tier1_schema_fail_falls_back_to_tier2(self, mock_router): """Tier 1 응답이 schema 위반 → Tier 2 sanitize 진입 후 valid → 통과.""" from app.modules.llm.llm_client import call_structured mock_router.completion.side_effect = [ _ok_json({}), # Tier 1: schema fail (required missing) _ok_json({"a": "ok"}), # Tier 2: sanitize 후 valid ] result = call_structured( step="test", system_prompt="sys", user_prompt="user", response_schema=_SCHEMA_REQUIRED_A, ) assert result == {"a": "ok"} assert mock_router.completion.call_count == 2 def test_all_tiers_schema_fail_eventually_raises_or_returns( self, mock_router, ): """Tier 1/2 schema fail → Tier 3 (validate skip) 응답 그대로 반환. Tier 3 는 마지막 시도이므로 schema validation 실패해도 결과 반환 정책. 하지만 ``_do_call`` 안의 validate 는 모든 tier 에서 동작. Tier 3 가 invalid 결과를 반환하면 ``SchemaValidationError`` 가 caller 까지 전파됨. 본 테스트는 caller 가 받는 최종 결과 중 invalid 케이스 → SchemaValidationError. """ from app.modules.llm.llm_client import call_structured from app.modules.llm.safety import SchemaValidationError # 모든 tier 에서 invalid payload mock_router.completion.side_effect = [ _ok_json({}), _ok_json({}), _ok_json({}), ] with pytest.raises(SchemaValidationError): call_structured( step="test", system_prompt="sys", user_prompt="user", response_schema=_SCHEMA_REQUIRED_A, ) assert mock_router.completion.call_count == 3 def test_validate_disabled_skips_for_all_tiers(self, mock_router, monkeypatch): """ENV disabled 면 invalid 결과도 Tier 1 그대로 반환.""" from app.modules.llm.llm_client import call_structured monkeypatch.setenv("LLM_LOCAL_SCHEMA_VALIDATE", "false") mock_router.completion.return_value = _ok_json({}) result = call_structured( step="test", system_prompt="sys", user_prompt="user", response_schema=_SCHEMA_REQUIRED_A, ) assert result == {} assert mock_router.completion.call_count == 1 def test_enable_fallback_false_raises_immediately(self, mock_router): """enable_fallback=False + schema fail → 즉시 raise (Tier 2 진입 안 함).""" from app.modules.llm.llm_client import call_structured from app.modules.llm.safety import SchemaValidationError mock_router.completion.return_value = _ok_json({}) with pytest.raises(SchemaValidationError): call_structured( step="test", system_prompt="sys", user_prompt="user", response_schema=_SCHEMA_REQUIRED_A, enable_fallback=False, ) assert mock_router.completion.call_count == 1 # --------------------------------------------------------------------------- # 3. call_multiturn 통합 — structured / free-text 분기 # --------------------------------------------------------------------------- class TestCallMultiturnSchemaValidation: def test_structured_passes_when_schema_valid(self, mock_router): from app.modules.llm.llm_client import call_multiturn mock_router.completion.return_value = _ok_json({"a": "ok"}) result = call_multiturn( step="test", messages=[{"role": "system", "content": "s"}, {"role": "user", "content": "u"}], response_schema=_SCHEMA_REQUIRED_A, ) assert result == {"a": "ok"} assert mock_router.completion.call_count == 1 def test_structured_schema_fail_falls_back(self, mock_router): from app.modules.llm.llm_client import call_multiturn mock_router.completion.side_effect = [ _ok_json({}), # Tier 1: schema fail _ok_json({"a": "post"}), # Tier 2: valid ] result = call_multiturn( step="test", messages=[{"role": "system", "content": "s"}, {"role": "user", "content": "u"}], response_schema=_SCHEMA_REQUIRED_A, ) assert result == {"a": "post"} assert mock_router.completion.call_count == 2 def test_free_text_mode_skips_schema_validate(self, mock_router): """response_schema=None → schema validate skip, free text 그대로 반환.""" from app.modules.llm.llm_client import call_multiturn mock_router.completion.return_value = _make_response("just a text answer") result = call_multiturn( step="test", messages=[{"role": "system", "content": "s"}, {"role": "user", "content": "u"}], response_schema=None, ) assert result == "just a text answer" assert mock_router.completion.call_count == 1 def test_disabled_env_skips_for_multiturn(self, mock_router, monkeypatch): from app.modules.llm.llm_client import call_multiturn monkeypatch.setenv("LLM_LOCAL_SCHEMA_VALIDATE", "false") mock_router.completion.return_value = _ok_json({}) result = call_multiturn( step="test", messages=[{"role": "system", "content": "s"}], response_schema=_SCHEMA_REQUIRED_A, ) assert result == {} assert mock_router.completion.call_count == 1 # --------------------------------------------------------------------------- # 4. SchemaValidationError + is_safety_related_error 분류 # --------------------------------------------------------------------------- class TestSchemaValidationErrorClassification: def test_schema_validation_error_is_safety_related(self): from app.modules.llm.safety import ( SchemaValidationError, is_safety_related_error, ) exc = SchemaValidationError("any message") assert is_safety_related_error(exc) is True def test_schema_validation_error_inherits_runtime_error(self): """except RuntimeError 로 잡는 caller 가 깨지지 않게.""" from app.modules.llm.safety import SchemaValidationError assert issubclass(SchemaValidationError, RuntimeError) def test_distinct_from_empty_semantic_response_error(self): """두 예외 클래스는 별개 (의미적 빈 결과 vs shape drift).""" from app.modules.llm.safety import ( EmptySemanticResponseError, SchemaValidationError, ) assert SchemaValidationError is not EmptySemanticResponseError assert not issubclass(SchemaValidationError, EmptySemanticResponseError)