""" Coverage for core.json_parsing.extract_json_from_response. Driven by the 2026-04-29 LIVE incident: Claude returns JSON wrapped in markdown fences (sometimes with leading/trailing prose). The previous parser only stripped a fence at the very start/end; prose-prefixed or prose-suffixed responses raised JSONDecodeError -> API_PARSE_ERROR -> lost trades on 6C and MNQ. Two assertion levels per case: 1) extracted text matches expectation 2) json.loads(extracted) round-trips when the inner JSON is valid Negative cases verify graceful fallback (no crash) — the caller is responsible for handling json.loads failures downstream. """ from __future__ import annotations import json import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from core.json_parsing import extract_json_from_response def _ok(label: str) -> None: print(f" ok {label}") PAYLOAD = '{"approved": true, "score": 0.82, "note": "OK"}' PARSED = {"approved": True, "score": 0.82, "note": "OK"} # ============================================================ # 1. EXTRACTION + JSON.LOADS HAPPY PATHS # ============================================================ def test_fence_with_json_tag(): text = f"```json\n{PAYLOAD}\n```" assert extract_json_from_response(text) == PAYLOAD assert json.loads(extract_json_from_response(text)) == PARSED _ok("fence ```json") def test_fence_no_language_tag(): text = f"```\n{PAYLOAD}\n```" assert extract_json_from_response(text) == PAYLOAD assert json.loads(extract_json_from_response(text)) == PARSED _ok("fence ``` (no language)") def test_fence_uppercase_json_tag(): text = f"```JSON\n{PAYLOAD}\n```" assert extract_json_from_response(text) == PAYLOAD assert json.loads(extract_json_from_response(text)) == PARSED _ok("fence ```JSON (case-insensitive)") def test_no_fence_raw_json(): text = PAYLOAD assert extract_json_from_response(text) == PAYLOAD assert json.loads(extract_json_from_response(text)) == PARSED _ok("raw JSON, no fence") def test_fence_with_leading_whitespace(): text = f" \n ```json\n{PAYLOAD}\n``` \n" assert extract_json_from_response(text) == PAYLOAD assert json.loads(extract_json_from_response(text)) == PARSED _ok("fence + leading/trailing whitespace") def test_fence_with_prose_prefix(): text = f"Ecco la risposta:\n```json\n{PAYLOAD}\n```" assert extract_json_from_response(text) == PAYLOAD assert json.loads(extract_json_from_response(text)) == PARSED _ok("fence + prose prefix") def test_fence_with_prose_suffix(): text = f"```json\n{PAYLOAD}\n```\n\nNote: this is the answer." assert extract_json_from_response(text) == PAYLOAD assert json.loads(extract_json_from_response(text)) == PARSED _ok("fence + prose suffix") def test_fence_with_prose_both_sides(): text = ( f"Ecco il mio output finale:\n" f"```json\n{PAYLOAD}\n```\n\n" f"Ho deciso così perché RSI è in zona pullback." ) assert extract_json_from_response(text) == PAYLOAD assert json.loads(extract_json_from_response(text)) == PARSED _ok("fence + prose both sides") def test_fence_single_line(): text = f"```json {PAYLOAD} ```" assert extract_json_from_response(text) == PAYLOAD assert json.loads(extract_json_from_response(text)) == PARSED _ok("fence single-line") # ============================================================ # 2. NEGATIVE / EDGE CASES — graceful, no crash # ============================================================ def test_empty_string_returns_empty(): assert extract_json_from_response("") == "" _ok("empty input -> empty output") def test_whitespace_only_returns_empty(): assert extract_json_from_response(" \n\n \t ") == "" _ok("whitespace-only input -> empty output") def test_none_input_returns_empty(): # Defensive: caller might pass None (e.g. resp.text is None) assert extract_json_from_response(None) == "" # type: ignore[arg-type] _ok("None input -> empty output (defensive)") def test_prose_only_no_json_returns_stripped(): """No fence, no braces -> last-resort raw stripped text.""" text = "Mi dispiace, non posso rispondere." out = extract_json_from_response(text) assert out == text # already stripped # Caller's json.loads will raise — that's correct behaviour try: json.loads(out) except json.JSONDecodeError: pass else: raise AssertionError("expected JSONDecodeError on prose-only input") _ok("prose only, no JSON -> stripped raw, json.loads raises") def test_malformed_json_inside_fence_extracts_but_loads_fails(): """Trailing comma -> extraction succeeds, json.loads is the gate.""" bad = '{"approved": true, "score": 0.82,}' text = f"```json\n{bad}\n```" assert extract_json_from_response(text) == bad try: json.loads(extract_json_from_response(text)) except json.JSONDecodeError: pass else: raise AssertionError("expected JSONDecodeError on trailing-comma JSON") _ok("malformed JSON inside fence -> extracted, json.loads raises") def test_missing_closing_fence_brace_fallback(): """ Truncated response (no closing ```): fence regex misses -> brace fallback recovers the object. Mirrors a max_tokens cutoff scenario. """ text = f"```json\n{PAYLOAD}\n some trailing prose without a fence" out = extract_json_from_response(text) # Brace fallback: from first '{' to last '}' assert out == PAYLOAD assert json.loads(out) == PARSED _ok("missing closing fence -> brace fallback recovers JSON") def test_brace_fallback_when_no_fence(): """No fence, prose around the JSON object.""" text = f"Sure thing! {PAYLOAD} done." out = extract_json_from_response(text) assert out == PAYLOAD assert json.loads(out) == PARSED _ok("no fence, prose around -> brace fallback") def test_markdown_chain_of_thought_no_json(): """ 2026-04-29 incident #2 (TF entry on 6A): AI returned a long markdown chain-of-thought response with zero JSON anywhere — the prompt was missing the "JSON only" guard that MR/bias both had. Parser must NOT crash. Caller's json.loads raises JSONDecodeError; Brain code maps that to API_PARSE_ERROR (not RuntimeError). """ text = ( "# ANALISI 6A — M5 SELL PULLBACK\n" "\n" "═══ STEP 1: QUALITÀ DELLO SCONTO ═══\n" "**RSI M5 = 45.5** (zona TF: 42–58)\n" "- Distanza da 58 (resistenza): +12.5 punti = **SCONTO PROFONDO**\n" "- Posizionamento: centro-basso\n" "\n" "═══ STEP 2: TIMING PULLBACK ═══\n" "RSI in decay (46.7 → 45.5): coerente.\n" "Bouncing=False (favorevole SELL).\n" "\n" "═══ STEP 3: CONTESTO MACRO ═══\n" "Trend H1 BEARISH, divergence assente.\n" ) # 1) Extraction must not crash out = extract_json_from_response(text) # 2) No fence regex match, no braces -> last-resort returns stripped raw assert out == text.strip() # 3) json.loads raises a parse error (caller catches it) try: json.loads(out) except json.JSONDecodeError: pass else: raise AssertionError("expected JSONDecodeError on prose-only response") _ok("markdown CoT, zero JSON -> graceful fail, json.loads raises") def test_real_incident_shape_6c(): """ Reconstructed shape from the 2026-04-29 LIVE incident on 6C. response_preview shows ```json\n{... — full payload likely had trailing prose or a stray fence. Verify the new parser is robust to a plausible reconstruction. """ text = ( '```json\n' '{\n' ' "step_1_qualita_sconto": "OTTIMO: RSI 46.7 in zona pullback",\n' ' "step_2_timing_pullback": "OTTIMO: Bouncing=True",\n' ' "approved": true,\n' ' "confidence": 0.78\n' '}\n' '```\n' '\n' 'Spiegazione: la setup è ottima per SELL pro-trend.' ) out = extract_json_from_response(text) parsed = json.loads(out) assert parsed["approved"] is True assert parsed["confidence"] == 0.78 _ok("real-incident shape (6C-style) parses end-to-end") # ============================================================ # MAIN # ============================================================ def main() -> int: print("test_ai_json_parsing.py") test_fence_with_json_tag() test_fence_no_language_tag() test_fence_uppercase_json_tag() test_no_fence_raw_json() test_fence_with_leading_whitespace() test_fence_with_prose_prefix() test_fence_with_prose_suffix() test_fence_with_prose_both_sides() test_fence_single_line() test_empty_string_returns_empty() test_whitespace_only_returns_empty() test_none_input_returns_empty() test_prose_only_no_json_returns_stripped() test_malformed_json_inside_fence_extracts_but_loads_fails() test_missing_closing_fence_brace_fallback() test_brace_fallback_when_no_fence() test_markdown_chain_of_thought_no_json() test_real_incident_shape_6c() print("ALL 18 TESTS PASSED") return 0 if __name__ == "__main__": sys.exit(main())