test_unified_debug.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tests/test_unified_debug.py
 
"""
Regression Tests — unified_debug
=================================
 
Purpose
-------
Verify the full public and internal API surface of ``oc_diagnostics.unified_debug``,
covering configuration loading, the ``_dbg`` debug logger, async logging, runtime
overrides, profiling helpers, production mode, log rotation, and I/O fallback.
 
Responsibilities
----------------
- Test config loading returns the expected 4-tuple structure.
- Test ``_dbg()`` domain-level filtering and ``override=True`` bypass.
- Test ``_adbg()`` completes without raising.
- Test ``init()`` idempotency and the pre-init RuntimeWarning.
- Test runtime API: ``set_debug_level()``, ``set_global_setting()``.
- Test profiling helpers: ``dbg_timed()``, ``dbg_scope()``, ``dbg_profile()``.
- Test JSONL output mode and ``_StreamInterceptor`` interface.
- Test production mode activation via env var and via config file.
- Test log-file count-based rotation via ``_prune_old_log_files()``.
- Test config hot-reload stale-key pruning via ``_reload_config()``.
- Test I/O degradation fallback on ``OSError`` (FS-25).
 
Diagnostics
-----------
Domain: TEST
Levels:
    L2 — lifecycle
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- Must not permanently modify oc_diagnostics global state.
- Must restore all patched globals and mocked internals in finally blocks.
- Must not start servers, threads, or require network access.
- Must run deterministically across repeated invocations.
"""
 
import asyncio
import io
import json
import os
import re
import subprocess
import sys
import tempfile
import time
import unittest.mock as mock
import warnings
 
# ---------------------------------------------------------------------------
# Ensure oc_diagnostics is importable when run from the project root or the
# tests/ subdirectory.
# ---------------------------------------------------------------------------
_REPO_ROOT = os.path.normpath(os.path.join(os.path.dirname(__file__), ".."))
if _REPO_ROOT not in sys.path:
    sys.path.insert(0, _REPO_ROOT)
 
import oc_diagnostics.unified_debug as _ud
from oc_diagnostics.unified_debug import (
    _load_debug_config,
    _StreamInterceptor,
    _prune_old_log_files,
)
 
 
# ============================================================
# Shared helper — mirrors run_tests.py for _reload_config tests
# ============================================================
 
def _build_mock_settings():
    """Return a minimal settings dict compatible with _reload_config()."""
    return {
        "instrumentation_enabled": _ud.INSTRUMENTATION_ENABLED,
        "silent_mode": _ud.SILENT_MODE,
        "structural_id_prefixes": list(_ud.ADDITIONAL_STRUCTURAL_PREFIXES),
        "structural_id_exact": list(_ud.ADDITIONAL_STRUCTURAL_EXACT),
        "enable_timestamps": _ud.ENABLE_TIMESTAMPS,
        "enable_colors": _ud.ENABLE_COLORS,
        "include_filename": _ud.INCLUDE_FILENAME,
        "output_mode": _ud.OUTPUT_MODE,
        "max_log_files": _ud.MAX_LOG_FILES,
        "max_log_size_mb": (
            float(_ud.MAX_LOG_SIZE_BYTES) / (1024 * 1024)
            if _ud.MAX_LOG_SIZE_BYTES else 0.0
        ),
        "instrumentation_allow_all": _ud.INSTRUMENTATION_ALL,
        "ui_event_properties": list(_ud.UI_EVENT_PROPERTIES),
        "http_redact_headers": list(_ud.HTTP_REDACT_HEADERS),
        "mode": "",
    }
 
 
# ============================================================
# CONFIG LOADING
# ============================================================
 
def test_config_loading():
    """_load_debug_config() returns a 4-tuple with expected types."""
    result = _load_debug_config()
    assert isinstance(result, tuple) and len(result) == 4, (
        f"Expected 4-tuple, got {type(result)} len={len(result) if isinstance(result, tuple) else 'N/A'}"
    )
    levels, colors, settings, source = result
    assert isinstance(levels, dict), f"levels must be dict, got {type(levels)}"
    assert isinstance(colors, dict), f"colors must be dict, got {type(colors)}"
    assert isinstance(settings, dict), f"settings must be dict, got {type(settings)}"
    assert isinstance(source, str) and source, "config_source must be a non-empty string"
 
 
def test_config_source_populated():
    """CONFIG_SOURCE is a non-empty string after module load."""
    assert isinstance(_ud.CONFIG_SOURCE, str) and _ud.CONFIG_SOURCE, (
        f"Expected non-empty string, got: {_ud.CONFIG_SOURCE!r}"
    )
 
 
# ============================================================
# _dbg CORE BEHAVIOUR
# ============================================================
 
def test_dbg_silent_no_exception():
    """_dbg() with SILENT_MODE=True must not raise."""
    old = _ud.SILENT_MODE
    _ud.SILENT_MODE = True
    try:
        _ud._dbg("TEST", "test_dbg_silent", "should not crash", level=2)
    finally:
        _ud.SILENT_MODE = old
 
 
def test_dbg_level_filtering():
    """_dbg() message is suppressed when the call level exceeds the domain level."""
    captured = []
    old_emit = _ud._emit
    old_silent = _ud.SILENT_MODE
    old_level = _ud.DEBUG_LEVELS.get("_FILTER_TEST_")
    old_init = _ud._initialized
    old_warned = _ud._init_warned
 
    def _capture(line, level, record=None):
        captured.append(line)
 
    _ud._emit = _capture
    _ud.SILENT_MODE = False
    _ud.DEBUG_LEVELS["_FILTER_TEST_"] = 3  # L1–L3 allowed
    _ud._initialized = True               # suppress pre-init warning
    try:
        # Level 5 > domain level 3 → filtered
        _ud._dbg("_FILTER_TEST_", "test_fn", "should-not-appear", level=5)
        assert not captured, (
            f"Level-5 message should be filtered at domain level 3; got: {captured!r}"
        )
 
        # Level 2 ≤ domain level 3 → emitted
        _ud._dbg("_FILTER_TEST_", "test_fn", "should-appear", level=2)
        assert any("should-appear" in l for l in captured), (
            f"Level-2 message should pass at domain level 3; got: {captured!r}"
        )
    finally:
        _ud._emit = old_emit
        _ud.SILENT_MODE = old_silent
        _ud._initialized = old_init
        _ud._init_warned = old_warned
        if old_level is None:
            _ud.DEBUG_LEVELS.pop("_FILTER_TEST_", None)
        else:
            _ud.DEBUG_LEVELS["_FILTER_TEST_"] = old_level
 
 
def test_dbg_override_flag():
    """_dbg(..., override=True) always emits regardless of domain level."""
    captured = []
    old_emit = _ud._emit
    old_silent = _ud.SILENT_MODE
    old_level = _ud.DEBUG_LEVELS.get("_OVERRIDE_TEST_")
    old_init = _ud._initialized
 
    def _capture(line, level, record=None):
        captured.append(line)
 
    _ud._emit = _capture
    _ud.SILENT_MODE = False
    _ud.DEBUG_LEVELS["_OVERRIDE_TEST_"] = 0  # nothing allowed normally
    _ud._initialized = True
    try:
        _ud._dbg("_OVERRIDE_TEST_", "test_fn", "override-message", level=9, override=True)
        assert any("override-message" in l for l in captured), (
            f"override=True must bypass level filter; got: {captured!r}"
        )
    finally:
        _ud._emit = old_emit
        _ud.SILENT_MODE = old_silent
        _ud._initialized = old_init
        if old_level is None:
            _ud.DEBUG_LEVELS.pop("_OVERRIDE_TEST_", None)
        else:
            _ud.DEBUG_LEVELS["_OVERRIDE_TEST_"] = old_level
 
 
def test_dbg_before_init_warning():
    """_dbg() before init() emits a one-shot RuntimeWarning; second call does not repeat it."""
    old_init = _ud._initialized
    old_warned = _ud._init_warned
    old_silent = _ud.SILENT_MODE
    old_emit = _ud._emit
 
    captured_lines = []
 
    def _capture(line, level, record=None):
        captured_lines.append(line)
 
    _ud._initialized = False
    _ud._init_warned = False
    _ud.SILENT_MODE = False
    _ud._emit = _capture
    try:
        with warnings.catch_warnings(record=True) as w:
            warnings.simplefilter("always")
            _ud._dbg("GENERAL", "test_fn", "pre-init call", level=2, override=True)
 
        assert len(w) == 1, (
            f"Expected exactly 1 RuntimeWarning on first pre-init call, got {len(w)}: "
            f"{[str(x.message) for x in w]}"
        )
        assert issubclass(w[0].category, RuntimeWarning), (
            f"Expected RuntimeWarning, got {w[0].category}"
        )
        assert _ud._init_warned is True, (
            "_init_warned must be True after the first warning"
        )
 
        # Second call: no additional warning emitted
        with warnings.catch_warnings(record=True) as w2:
            warnings.simplefilter("always")
            _ud._dbg("GENERAL", "test_fn", "second call", level=2, override=True)
 
        assert len(w2) == 0, (
            f"Expected no second RuntimeWarning, got: {[str(x.message) for x in w2]}"
        )
    finally:
        _ud._initialized = old_init
        _ud._init_warned = old_warned
        _ud.SILENT_MODE = old_silent
        _ud._emit = old_emit
 
 
# ============================================================
# ASYNC LOGGING
# ============================================================
 
def test_adbg_does_not_raise():
    """_adbg() completes without raising, in silent mode for cleanliness."""
    old_silent = _ud.SILENT_MODE
    old_level = _ud.DEBUG_LEVELS.get("_ADBG_TEST_")
    _ud.SILENT_MODE = True
    _ud.DEBUG_LEVELS["_ADBG_TEST_"] = 9
    try:
        asyncio.run(_ud._adbg("_ADBG_TEST_", "test_fn", "async message", level=2))
    finally:
        _ud.SILENT_MODE = old_silent
        if old_level is None:
            _ud.DEBUG_LEVELS.pop("_ADBG_TEST_", None)
        else:
            _ud.DEBUG_LEVELS["_ADBG_TEST_"] = old_level
 
 
# ============================================================
# init() IDEMPOTENCY
# ============================================================
 
def test_init_idempotent():
    """init() can be called multiple times without error or side-effect duplication."""
    old_init = _ud._initialized
    old_stdout = sys.stdout
    old_stderr = sys.stderr
    old_excepthook = sys.excepthook
    _ud._initialized = False
    try:
        # Both calls must succeed without raising
        _ud.init(intercept_streams=False, intercept_exceptions=False)
        _ud.init(intercept_streams=False, intercept_exceptions=False)
        assert _ud._initialized is True, "init() must set _initialized=True"
    finally:
        _ud._initialized = old_init
        sys.stdout = old_stdout
        sys.stderr = old_stderr
        sys.excepthook = old_excepthook
 
 
# ============================================================
# _StreamInterceptor INTERFACE
# ============================================================
 
def test_stream_interceptor_interface():
    """_StreamInterceptor exposes the full io.TextIOBase-compatible interface."""
    si = _StreamInterceptor("stdout")
    assert si.writable() is True, "writable() must return True"
    assert si.readable() is False, "readable() must return False"
    assert si.seekable() is False, "seekable() must return False"
    assert isinstance(si.encoding, str), f"encoding must be str, got {type(si.encoding)}"
    assert isinstance(si.errors, str), f"errors must be str, got {type(si.errors)}"
    result = si.isatty()
    assert isinstance(result, bool), f"isatty() must return bool, got {type(result)}"
 
 
# ============================================================
# DEBUG_LEVELS MUTABILITY
# ============================================================
 
def test_debug_levels_dict_mutable():
    """DEBUG_LEVELS is a mutable dict and changes are reflected globally."""
    _save = _ud.DEBUG_LEVELS.get("_TEST_DOMAIN_")
    _ud.DEBUG_LEVELS["_TEST_DOMAIN_"] = 7
    assert _ud.DEBUG_LEVELS["_TEST_DOMAIN_"] == 7, "DEBUG_LEVELS mutation not reflected"
    if _save is None:
        del _ud.DEBUG_LEVELS["_TEST_DOMAIN_"]
    else:
        _ud.DEBUG_LEVELS["_TEST_DOMAIN_"] = _save
 
 
# ============================================================
# RUNTIME API (FS-03)
# ============================================================
 
def test_set_debug_level_valid():
    """set_debug_level() updates DEBUG_LEVELS at runtime."""
    old = _ud.DEBUG_LEVELS.get("_FS03_TEST_")
    _ud.set_debug_level("_FS03_TEST_", 5)
    try:
        assert _ud.DEBUG_LEVELS.get("_FS03_TEST_") == 5, (
            f"Expected 5, got {_ud.DEBUG_LEVELS.get('_FS03_TEST_')!r}"
        )
    finally:
        if old is None:
            _ud.DEBUG_LEVELS.pop("_FS03_TEST_", None)
        else:
            _ud.DEBUG_LEVELS["_FS03_TEST_"] = old
 
 
def test_set_debug_level_invalid():
    """set_debug_level() with an out-of-range value emits UserWarning and does not update."""
    old = _ud.DEBUG_LEVELS.get("_FS03_TEST2_")
    with warnings.catch_warnings(record=True) as w:
        warnings.simplefilter("always")
        _ud.set_debug_level("_FS03_TEST2_", 99)
    try:
        assert len(w) == 1, f"Expected 1 warning, got {len(w)}"
        assert issubclass(w[0].category, UserWarning), "Expected UserWarning"
        assert _ud.DEBUG_LEVELS.get("_FS03_TEST2_") is old, (
            "Invalid level must not update DEBUG_LEVELS"
        )
    finally:
        if "_FS03_TEST2_" in _ud.DEBUG_LEVELS and old is None:
            del _ud.DEBUG_LEVELS["_FS03_TEST2_"]
 
 
def test_set_global_setting_valid():
    """set_global_setting() updates SILENT_MODE at runtime."""
    old = _ud.SILENT_MODE
    _ud.set_global_setting("silent_mode", True)
    try:
        assert _ud.SILENT_MODE is True, f"Expected True, got {_ud.SILENT_MODE!r}"
    finally:
        _ud.SILENT_MODE = old
 
 
def test_set_global_setting_invalid():
    """set_global_setting() with an unknown key emits UserWarning."""
    with warnings.catch_warnings(record=True) as w:
        warnings.simplefilter("always")
        _ud.set_global_setting("nonexistent_key_xyz", 42)
    assert len(w) == 1, f"Expected 1 warning, got {len(w)}"
    assert issubclass(w[0].category, UserWarning), "Expected UserWarning"
 
 
# ============================================================
# JSONL OUTPUT MODE (FS-07)
# ============================================================
 
def test_json_output_mode():
    """OUTPUT_MODE='json' writes valid JSONL records to the log file."""
    import glob as _glob
    old_mode = _ud.OUTPUT_MODE
    old_log_dir = _ud.LOG_DIR
    old_log_file = _ud.LOG_FILE
    old_header = _ud._header_written
    old_silent = _ud.SILENT_MODE
    old_level = _ud.DEBUG_LEVELS.get("_FSJSON_")
    old_init = _ud._initialized
    with tempfile.TemporaryDirectory() as tmpdir:
        _ud.OUTPUT_MODE = "json"
        _ud.LOG_DIR = tmpdir
        _ud.LOG_FILE = None
        _ud._header_written = False
        _ud.SILENT_MODE = False
        _ud.DEBUG_LEVELS["_FSJSON_"] = 9
        _ud._initialized = True   # suppress pre-init RuntimeWarning
        try:
            _ud._dbg("_FSJSON_", "test_fn", "hello json", level=2, override=True)
        finally:
            _ud.OUTPUT_MODE = old_mode
            _ud.LOG_DIR = old_log_dir
            _ud.LOG_FILE = old_log_file
            _ud._header_written = old_header
            _ud.SILENT_MODE = old_silent
            _ud._initialized = old_init
            if old_level is None:
                _ud.DEBUG_LEVELS.pop("_FSJSON_", None)
            else:
                _ud.DEBUG_LEVELS["_FSJSON_"] = old_level
        files = _glob.glob(os.path.join(tmpdir, "*.txt"))
        assert files, "Expected at least one log file to be created"
        with open(files[0], encoding="utf-8") as f:
            raw_lines = f.readlines()
        json_lines = []
        for raw in raw_lines:
            stripped = raw.strip()
            if stripped.startswith("{"):
                try:
                    json_lines.append(json.loads(stripped))
                except json.JSONDecodeError:
                    pass
        assert json_lines, (
            f"Expected at least one valid JSONL record, got lines: {raw_lines!r}"
        )
        rec = json_lines[0]
        assert rec.get("domain") == "_FSJSON_", (
            f"Expected domain '_FSJSON_', got {rec!r}"
        )
        assert rec.get("msg") == "hello json", (
            f"Expected msg 'hello json', got {rec!r}"
        )
 
 
# ============================================================
# PROFILING HELPERS (FS-12, FS-19)
# ============================================================
 
def test_dbg_timed():
    """dbg_timed() context manager emits ENTER and EXIT lines with elapsed ms."""
    captured = []
    old_emit = _ud._emit
    old_silent = _ud.SILENT_MODE
    old_level = _ud.DEBUG_LEVELS.get("GENERAL")
    old_init = _ud._initialized
 
    def _capture(line, level, record=None):
        captured.append(line)
 
    _ud._emit = _capture
    _ud.SILENT_MODE = False
    _ud.DEBUG_LEVELS["GENERAL"] = 9
    _ud._initialized = True  # suppress pre-init RuntimeWarning
    try:
        with _ud.dbg_timed("GENERAL", "timing_test", label="myop", level=2):
            pass
    finally:
        _ud._emit = old_emit
        _ud.SILENT_MODE = old_silent
        _ud._initialized = old_init
        if old_level is None:
            _ud.DEBUG_LEVELS.pop("GENERAL", None)
        else:
            _ud.DEBUG_LEVELS["GENERAL"] = old_level
 
    enter_lines = [l for l in captured if "ENTER" in l]
    exit_lines = [l for l in captured if "EXIT" in l and "ms" in l]
    assert enter_lines, f"Expected ENTER line in output, got: {captured!r}"
    assert exit_lines, f"Expected EXIT line with 'ms' in output, got: {captured!r}"
 
 
def test_dbg_scope():
    """dbg_scope() emits ENTER and EXIT lines without timing information (FS-19)."""
    captured = []
    old_emit = _ud._emit
    old_silent = _ud.SILENT_MODE
    old_level = _ud.DEBUG_LEVELS.get("GENERAL")
    old_init = _ud._initialized
 
    def _capture(line, level, record=None):
        captured.append(line)
 
    _ud._emit = _capture
    _ud.SILENT_MODE = False
    _ud.DEBUG_LEVELS["GENERAL"] = 9
    _ud._initialized = True  # suppress pre-init RuntimeWarning
    try:
        with _ud.dbg_scope("GENERAL", "scope_test", label="myblock", level=2):
            pass
    finally:
        _ud._emit = old_emit
        _ud.SILENT_MODE = old_silent
        _ud._initialized = old_init
        if old_level is None:
            _ud.DEBUG_LEVELS.pop("GENERAL", None)
        else:
            _ud.DEBUG_LEVELS["GENERAL"] = old_level
 
    enter_lines = [l for l in captured if "ENTER" in l]
    exit_lines = [l for l in captured if "EXIT" in l]
    assert enter_lines, f"Expected ENTER line, got: {captured!r}"
    assert exit_lines, f"Expected EXIT line, got: {captured!r}"
    assert not any("ms" in l for l in exit_lines), (
        f"dbg_scope EXIT must not include timing info, got: {exit_lines!r}"
    )
 
 
def test_dbg_profile():
    """dbg_profile() decorator wraps a function and preserves its return value."""
    captured = []
    old_emit = _ud._emit
    old_silent = _ud.SILENT_MODE
    old_level = _ud.DEBUG_LEVELS.get("GENERAL")
    old_init = _ud._initialized
 
    def _capture(line, level, record=None):
        captured.append(line)
 
    _ud._emit = _capture
    _ud.SILENT_MODE = False
    _ud.DEBUG_LEVELS["GENERAL"] = 9
    _ud._initialized = True  # suppress pre-init RuntimeWarning
    try:
        @_ud.dbg_profile("GENERAL", label="profiled_fn", level=2)
        def _sample():
            return 42
 
        result = _sample()
    finally:
        _ud._emit = old_emit
        _ud.SILENT_MODE = old_silent
        _ud._initialized = old_init
        if old_level is None:
            _ud.DEBUG_LEVELS.pop("GENERAL", None)
        else:
            _ud.DEBUG_LEVELS["GENERAL"] = old_level
 
    assert result == 42, (
        f"Decorated function return value must be preserved, got {result!r}"
    )
    enter_lines = [l for l in captured if "ENTER" in l]
    exit_lines = [l for l in captured if "EXIT" in l and "ms" in l]
    assert enter_lines, f"Expected ENTER line in output, got: {captured!r}"
    assert exit_lines, f"Expected EXIT line with 'ms' in output, got: {captured!r}"
 
 
# ============================================================
# PRODUCTION MODE (FS-15)
# ============================================================
 
def test_prod_mode_env_var():
    """OC_DIAGNOSTICS_MODE=prod sets PROD_MODE=True at module load."""
    result = subprocess.run(
        [
            sys.executable, "-c",
            "import oc_diagnostics.unified_debug as _ud; "
            "assert _ud.PROD_MODE is True, f'PROD_MODE={_ud.PROD_MODE!r}'"
        ],
        env={**os.environ, "OC_DIAGNOSTICS_MODE": "prod"},
        capture_output=True,
        text=True,
        cwd=_REPO_ROOT,
    )
    assert result.returncode == 0, (
        f"PROD_MODE not set True by env var; stderr: {result.stderr.strip()!r}"
    )
 
 
def test_prod_mode_via_config():
    """mode='prod' in debug_config.json global_settings sets PROD_MODE=True."""
    config = {"global_settings": {"mode": "prod"}}
    tmp_path = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".json", delete=False, encoding="utf-8"
        ) as f:
            json.dump(config, f)
            tmp_path = f.name
 
        env = dict(os.environ)
        env.pop("OC_DIAGNOSTICS_MODE", None)
        env["OC_DIAGNOSTICS_CONFIG"] = tmp_path
 
        result = subprocess.run(
            [
                sys.executable, "-c",
                "import oc_diagnostics.unified_debug as _ud; "
                "assert _ud.PROD_MODE is True, f'PROD_MODE={_ud.PROD_MODE!r}'"
            ],
            env=env,
            capture_output=True,
            text=True,
            cwd=_REPO_ROOT,
        )
        assert result.returncode == 0, (
            f"PROD_MODE not set by config mode='prod'; stderr: {result.stderr.strip()!r}"
        )
    finally:
        if tmp_path and os.path.exists(tmp_path):
            os.unlink(tmp_path)
 
 
# ============================================================
# LOG ROTATION (FS-06)
# ============================================================
 
def test_log_rotation_count():
    """_prune_old_log_files() removes the oldest files when count exceeds max."""
    import glob as _glob
 
    with tempfile.TemporaryDirectory() as tmpdir:
        # Create 5 fake log files with slight mtime spacing
        for i in range(5):
            path = os.path.join(
                tmpdir, f"unified_debug_logs_202401{i:02d}_000000.txt"
            )
            with open(path, "w") as f:
                f.write(f"log {i}")
            # Ensure distinct mtimes (file system granularity)
            os.utime(path, (time.time() - (5 - i), time.time() - (5 - i)))
 
        _prune_old_log_files(tmpdir, 3)
 
        remaining = sorted(
            _glob.glob(os.path.join(tmpdir, "unified_debug_logs_*.txt"))
        )
        assert len(remaining) == 3, (
            f"Expected 3 files after pruning max_count=3, got {len(remaining)}: {remaining}"
        )
 
 
# ============================================================
# HOT-RELOAD (OI-29)
# ============================================================
 
def test_reload_stale_key_removed():
    """_reload_config() prunes stale keys; DEBUG_LEVELS is never empty."""
    stale_key = "_OI29_TEST_STALE"
    _ud.DEBUG_LEVELS[stale_key] = 7
    new_levels = {"DATA": 3, "UI": 2}
    with mock.patch.object(
        _ud, "_load_debug_config",
        return_value=(
            new_levels, _ud.COLOR_CODES, _build_mock_settings(), _ud.CONFIG_SOURCE
        ),
    ):
        _ud._reload_config()
    assert stale_key not in _ud.DEBUG_LEVELS, (
        f"Stale key '{stale_key}' must be removed after _reload_config()"
    )
    assert "DATA" in _ud.DEBUG_LEVELS, "New key 'DATA' must be present after reload"
    assert len(_ud.DEBUG_LEVELS) > 0, "DEBUG_LEVELS must never be empty after reload"
 
 
# ============================================================
# I/O DEGRADATION (FS-25)
# ============================================================
 
def test_io_degradation_oserror():
    """FS-25: _write_to_file() OSError triggers one-time warning and mode fallback."""
    old_mode = _ud.OUTPUT_MODE
    old_warned = _ud._io_fallback_warned
    old_stdout = _ud.ORIGINAL_STDOUT
    stdout_capture = io.StringIO()
    _ud.OUTPUT_MODE = "file"
    _ud._io_fallback_warned = False
    _ud.ORIGINAL_STDOUT = stdout_capture
    try:
        with mock.patch("builtins.open", side_effect=OSError("No space left on device")):
            _ud._write_to_file("test line")
        assert _ud.OUTPUT_MODE == "terminal", (
            "OUTPUT_MODE must switch to 'terminal' after OSError"
        )
        assert _ud._io_fallback_warned is True, (
            "_io_fallback_warned must be True after first OSError"
        )
        warning_text = stdout_capture.getvalue()
        assert "WARNING" in warning_text, (
            f"Expected 'WARNING' in stdout, got: {warning_text!r}"
        )
        # Second call: no further warning
        stdout_capture.truncate(0)
        stdout_capture.seek(0)
        _ud._write_to_file("second line")
        assert stdout_capture.getvalue() == "", (
            "No second warning should be emitted for subsequent failures"
        )
    finally:
        _ud.OUTPUT_MODE = old_mode
        _ud._io_fallback_warned = old_warned
        _ud.ORIGINAL_STDOUT = old_stdout
 
 
# ============================================================
# _dbg_assert (D-01)
# ============================================================
 
def test_dbg_assert_passes_silently():
    """_dbg_assert() with truthy condition does not raise or emit."""
    captured = []
    old_emit = _ud._emit
    old_init = _ud._initialized
    old_prod = _ud.PROD_MODE
    _ud._emit = lambda line, level, record=None: captured.append(line)
    _ud._initialized = True
    _ud.PROD_MODE = False
    try:
        _ud._dbg_assert("TEST", True, "should not fire", func="test_fn")
        assert not captured, f"Truthy condition should emit nothing, got: {captured!r}"
    finally:
        _ud._emit = old_emit
        _ud._initialized = old_init
        _ud.PROD_MODE = old_prod
 
 
def test_dbg_assert_raises_on_false():
    """_dbg_assert() with falsy condition emits L1 override + raises AssertionError."""
    captured = []
    old_emit = _ud._emit
    old_init = _ud._initialized
    old_prod = _ud.PROD_MODE
    _ud._emit = lambda line, level, record=None: captured.append(line)
    _ud._initialized = True
    _ud.PROD_MODE = False
    try:
        raised = False
        try:
            _ud._dbg_assert("TEST", False, "batch must not be empty", func="process")
        except AssertionError as e:
            raised = True
            assert "batch must not be empty" in str(e)
        assert raised, "_dbg_assert with False must raise AssertionError"
        assert any("ASSERT FAILED" in l for l in captured), (
            f"Expected 'ASSERT FAILED' in output, got: {captured!r}"
        )
    finally:
        _ud._emit = old_emit
        _ud._initialized = old_init
        _ud.PROD_MODE = old_prod
 
 
def test_dbg_assert_prod_mode_noop():
    """_dbg_assert() in PROD_MODE is a complete no-op — condition not evaluated."""
    old_prod = _ud.PROD_MODE
    _ud.PROD_MODE = True
    try:
        # Pass a falsy condition — must NOT raise in prod mode
        _ud._dbg_assert("TEST", False, "should be suppressed", func="test_fn")
    finally:
        _ud.PROD_MODE = old_prod
 
 
# ============================================================
# dbg_trace DECORATOR (D-02)
# ============================================================
 
def test_dbg_trace_start_end():
    """@dbg_trace emits START and END messages."""
    captured = []
    old_emit = _ud._emit
    old_init = _ud._initialized
    old_level = _ud.DEBUG_LEVELS.get("_TRACE_")
    _ud._emit = lambda line, level, record=None: captured.append(line)
    _ud._initialized = True
    _ud.DEBUG_LEVELS["_TRACE_"] = 9
    try:
        @_ud.dbg_trace("_TRACE_", level=2)
        def my_func():
            return 42
 
        result = my_func()
        assert result == 42, f"Return value must be preserved, got {result!r}"
        assert any("START" in l for l in captured), f"Expected START, got: {captured!r}"
        assert any("END" in l for l in captured), f"Expected END, got: {captured!r}"
    finally:
        _ud._emit = old_emit
        _ud._initialized = old_init
        if old_level is None:
            _ud.DEBUG_LEVELS.pop("_TRACE_", None)
        else:
            _ud.DEBUG_LEVELS["_TRACE_"] = old_level
 
 
def test_dbg_trace_exception():
    """@dbg_trace logs exception at L1 and re-raises."""
    captured = []
    old_emit = _ud._emit
    old_init = _ud._initialized
    old_level = _ud.DEBUG_LEVELS.get("_TRACE_")
    _ud._emit = lambda line, level, record=None: captured.append(line)
    _ud._initialized = True
    _ud.DEBUG_LEVELS["_TRACE_"] = 9
    try:
        @_ud.dbg_trace("_TRACE_")
        def failing_func():
            raise ValueError("test error")
 
        raised = False
        try:
            failing_func()
        except ValueError:
            raised = True
        assert raised, "Exception must be re-raised"
        assert any("EXCEPTION" in l and "ValueError" in l for l in captured), (
            f"Expected EXCEPTION log with ValueError, got: {captured!r}"
        )
    finally:
        _ud._emit = old_emit
        _ud._initialized = old_init
        if old_level is None:
            _ud.DEBUG_LEVELS.pop("_TRACE_", None)
        else:
            _ud.DEBUG_LEVELS["_TRACE_"] = old_level
 
 
def test_dbg_trace_preserves_name():
    """@dbg_trace preserves __name__ via functools.wraps."""
    @_ud.dbg_trace("TEST")
    def named_function():
        pass
 
    assert named_function.__name__ == "named_function", (
        f"Expected 'named_function', got {named_function.__name__!r}"
    )
 
 
# ============================================================
# dbg_metrics DECORATOR (D-03)
# ============================================================
 
def test_dbg_metrics_output():
    """@dbg_metrics emits structured JSON with duration_ms and memory_mb."""
    captured = []
    old_emit = _ud._emit
    old_init = _ud._initialized
    old_level = _ud.DEBUG_LEVELS.get("_METRICS_")
    _ud._emit = lambda line, level, record=None: captured.append(line)
    _ud._initialized = True
    _ud.DEBUG_LEVELS["_METRICS_"] = 9
    try:
        @_ud.dbg_metrics("_METRICS_", level=3)
        def compute():
            return sum(range(1000))
 
        result = compute()
        assert result == 499500, f"Return value must be preserved, got {result!r}"
        metrics_lines = [l for l in captured if "METRICS:" in l]
        assert metrics_lines, f"Expected METRICS line, got: {captured!r}"
        # Extract the JSON part
        json_str = metrics_lines[0].split("METRICS: ", 1)[1]
        # Strip ANSI codes
        json_str = _ud._ANSI_ESCAPE_RE.sub("", json_str)
        data = json.loads(json_str)
        assert "duration_ms" in data, f"Expected duration_ms key, got: {data!r}"
        assert "memory_mb" in data, f"Expected memory_mb key, got: {data!r}"
        assert isinstance(data["duration_ms"], (int, float)), "duration_ms must be numeric"
    finally:
        _ud._emit = old_emit
        _ud._initialized = old_init
        if old_level is None:
            _ud.DEBUG_LEVELS.pop("_METRICS_", None)
        else:
            _ud.DEBUG_LEVELS["_METRICS_"] = old_level
 
 
# ============================================================
# AI-TRACE (D-04)
# ============================================================
 
def test_ai_trace_console_suppressed():
    """AI-Trace messages (level='ai_trace') do not reach the console."""
    console_captured = []
    old_stdout = _ud.ORIGINAL_STDOUT
    old_mode = _ud.OUTPUT_MODE
    old_ai = _ud.AI_TRACE_ENABLED
    old_silent = _ud.SILENT_MODE
    old_init = _ud._initialized
 
    fake_stdout = io.StringIO()
    _ud.ORIGINAL_STDOUT = fake_stdout
    _ud.OUTPUT_MODE = "both"
    _ud.AI_TRACE_ENABLED = True
    _ud.SILENT_MODE = False
    _ud._initialized = True
    try:
        _ud._dbg("TEST", "test_fn", "ai trace msg", level="ai_trace")
        console_output = fake_stdout.getvalue()
        assert "ai trace msg" not in console_output, (
            f"AI-Trace must not reach console, got: {console_output!r}"
        )
    finally:
        _ud.ORIGINAL_STDOUT = old_stdout
        _ud.OUTPUT_MODE = old_mode
        _ud.AI_TRACE_ENABLED = old_ai
        _ud.SILENT_MODE = old_silent
        _ud._initialized = old_init
 
 
def test_ai_trace_file_written():
    """AI-Trace messages are written to the log file."""
    old_mode = _ud.OUTPUT_MODE
    old_ai = _ud.AI_TRACE_ENABLED
    old_log_dir = _ud.LOG_DIR
    old_log_file = _ud.LOG_FILE
    old_header = _ud._header_written
    old_silent = _ud.SILENT_MODE
    old_init = _ud._initialized
 
    with tempfile.TemporaryDirectory() as tmpdir:
        _ud.OUTPUT_MODE = "both"
        _ud.AI_TRACE_ENABLED = True
        _ud.LOG_DIR = tmpdir
        _ud.LOG_FILE = None
        _ud._header_written = False
        _ud.SILENT_MODE = False
        _ud._initialized = True
        try:
            _ud._dbg("TEST", "test_fn", "ai shadow message", level="ai_trace")
        finally:
            _ud.OUTPUT_MODE = old_mode
            _ud.AI_TRACE_ENABLED = old_ai
            _ud.LOG_DIR = old_log_dir
            _ud.LOG_FILE = old_log_file
            _ud._header_written = old_header
            _ud.SILENT_MODE = old_silent
            _ud._initialized = old_init
 
        import glob as _glob
        files = _glob.glob(os.path.join(tmpdir, "*.txt"))
        assert files, "Expected at least one log file"
        content = open(files[0], encoding="utf-8").read()
        assert "ai shadow message" in content, (
            f"AI-Trace message must be in log file, got: {content!r}"
        )
 
 
def test_ai_trace_disabled_discards():
    """AI-Trace messages are silently discarded when AI_TRACE_ENABLED is False."""
    captured = []
    old_emit = _ud._emit
    old_ai = _ud.AI_TRACE_ENABLED
    old_silent = _ud.SILENT_MODE
    _ud._emit = lambda line, level, record=None: captured.append(line)
    _ud.AI_TRACE_ENABLED = False
    _ud.SILENT_MODE = False
    try:
        _ud._dbg("TEST", "test_fn", "should vanish", level="ai_trace")
        assert not captured, f"Disabled AI-Trace must emit nothing, got: {captured!r}"
    finally:
        _ud._emit = old_emit
        _ud.AI_TRACE_ENABLED = old_ai
        _ud.SILENT_MODE = old_silent
 
 
# ============================================================
# VERSION CHECK (D-08)
# ============================================================
 
def test_version_is_2_0():
    """oc_diagnostics version must be 2.0.0."""
    import oc_diagnostics
    assert oc_diagnostics.__version__ == "2.0.0", (
        f"Expected version 2.0.0, got {oc_diagnostics.__version__!r}"
    )
 
 
# ============================================================
# CONFIG SCHEMA V2.0 — DEBUG_LEVELS key (D-07)
# ============================================================
 
def test_config_v2_debug_levels_key():
    """V2.0 config with 'DEBUG_LEVELS' key loads correctly."""
    config = {
        "_schema_version": "2.0",
        "DEBUG_LEVELS": {"DATA": {"level": 4, "color": "cyan"}, "UI": 2},
        "global_settings": {"mode": "dev"},
    }
    tmp_path = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".json", delete=False, encoding="utf-8"
        ) as f:
            json.dump(config, f)
            tmp_path = f.name
 
        env = dict(os.environ)
        env["OC_DIAGNOSTICS_CONFIG"] = tmp_path
        result = subprocess.run(
            [
                sys.executable, "-c",
                "import oc_diagnostics.unified_debug as _ud; "
                "assert _ud.DEBUG_LEVELS.get('DATA') == 4, f'DATA={_ud.DEBUG_LEVELS.get(\"DATA\")}'; "
                "assert _ud.DEBUG_LEVELS.get('UI') == 2, f'UI={_ud.DEBUG_LEVELS.get(\"UI\")}'"
            ],
            env=env,
            capture_output=True,
            text=True,
            cwd=_REPO_ROOT,
        )
        assert result.returncode == 0, (
            f"DEBUG_LEVELS key not loaded; stderr: {result.stderr.strip()!r}"
        )
    finally:
        if tmp_path and os.path.exists(tmp_path):
            os.unlink(tmp_path)
 
 
def test_config_v2_nested_log_rotation():
    """V2.0 nested log_rotation object is parsed correctly."""
    config = {
        "DEBUG_LEVELS": {"GENERAL": 2},
        "global_settings": {
            "log_rotation": {"max_files": 10, "max_size_mb": 2.5},
        },
    }
    tmp_path = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".json", delete=False, encoding="utf-8"
        ) as f:
            json.dump(config, f)
            tmp_path = f.name
 
        env = dict(os.environ)
        env["OC_DIAGNOSTICS_CONFIG"] = tmp_path
        result = subprocess.run(
            [
                sys.executable, "-c",
                "import oc_diagnostics.unified_debug as _ud; "
                "assert _ud.MAX_LOG_FILES == 10, f'max_files={_ud.MAX_LOG_FILES}'; "
                f"expected_bytes = int(2.5 * 1024 * 1024); "
                "assert _ud.MAX_LOG_SIZE_BYTES == expected_bytes, "
                "f'max_size={_ud.MAX_LOG_SIZE_BYTES}'"
            ],
            env=env,
            capture_output=True,
            text=True,
            cwd=_REPO_ROOT,
        )
        assert result.returncode == 0, (
            f"Nested log_rotation not parsed; stderr: {result.stderr.strip()!r}"
        )
    finally:
        if tmp_path and os.path.exists(tmp_path):
            os.unlink(tmp_path)
 
 
# ============================================================
# ENV VAR OVERRIDES (D-05, D-06)
# ============================================================
 
def test_env_var_level_override():
    """OC_DIAGNOSTICS_LEVEL env var overrides all domain levels."""
    config = {
        "DEBUG_LEVELS": {"DATA": 2, "UI": 3},
        "global_settings": {},
    }
    tmp_path = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".json", delete=False, encoding="utf-8"
        ) as f:
            json.dump(config, f)
            tmp_path = f.name
 
        env = dict(os.environ)
        env["OC_DIAGNOSTICS_CONFIG"] = tmp_path
        env["OC_DIAGNOSTICS_LEVEL"] = "7"
        result = subprocess.run(
            [
                sys.executable, "-c",
                "import oc_diagnostics.unified_debug as _ud; "
                "assert _ud.DEBUG_LEVELS.get('DATA') == 7, f'DATA={_ud.DEBUG_LEVELS.get(\"DATA\")}'; "
                "assert _ud.DEBUG_LEVELS.get('UI') == 7, f'UI={_ud.DEBUG_LEVELS.get(\"UI\")}'"
            ],
            env=env,
            capture_output=True,
            text=True,
            cwd=_REPO_ROOT,
        )
        assert result.returncode == 0, (
            f"OC_DIAGNOSTICS_LEVEL override failed; stderr: {result.stderr.strip()!r}"
        )
    finally:
        if tmp_path and os.path.exists(tmp_path):
            os.unlink(tmp_path)
 
 
def test_env_var_log_file_override():
    """OC_DIAGNOSTICS_LOG_FILE env var sets LOG_FILE at load time."""
    result = subprocess.run(
        [
            sys.executable, "-c",
            "import oc_diagnostics.unified_debug as _ud; "
            "assert _ud.LOG_FILE == '/tmp/test_log.txt', f'LOG_FILE={_ud.LOG_FILE!r}'"
        ],
        env={**os.environ, "OC_DIAGNOSTICS_LOG_FILE": "/tmp/test_log.txt"},
        capture_output=True,
        text=True,
        cwd=_REPO_ROOT,
    )
    assert result.returncode == 0, (
        f"OC_DIAGNOSTICS_LOG_FILE override failed; stderr: {result.stderr.strip()!r}"
    )
 
 
# ============================================================
# REDACTION PATTERNS (V2.0)
# ============================================================
 
def test_redaction_in_prod_mode():
    """Redaction patterns scrub matching key=value pairs in production mode."""
    old_prod = _ud.PROD_MODE
    old_patterns = _ud.REDACTION_PATTERNS
    _ud.PROD_MODE = True
    _ud.REDACTION_PATTERNS = [
        re.compile(r'(?i)(password)\s*[=:]\s*\S+'),
        re.compile(r'(?i)(api_key)\s*[=:]\s*\S+'),
    ]
    try:
        result = _ud._apply_redaction("user login password=secret123 api_key=abc-xyz")
        assert "secret123" not in result, f"password value not redacted: {result!r}"
        assert "abc-xyz" not in result, f"api_key value not redacted: {result!r}"
        assert "[REDACTED]" in result, f"Expected [REDACTED] in output: {result!r}"
    finally:
        _ud.PROD_MODE = old_prod
        _ud.REDACTION_PATTERNS = old_patterns
 
 
def test_redaction_disabled_in_dev_mode():
    """Redaction patterns are not applied when PROD_MODE is False."""
    old_prod = _ud.PROD_MODE
    old_patterns = _ud.REDACTION_PATTERNS
    _ud.PROD_MODE = False
    _ud.REDACTION_PATTERNS = [
        re.compile(r'(?i)(password)\s*[=:]\s*\S+'),
    ]
    try:
        text = "password=secret123"
        result = _ud._apply_redaction(text)
        assert result == text, f"Dev mode should not redact: {result!r}"
    finally:
        _ud.PROD_MODE = old_prod
        _ud.REDACTION_PATTERNS = old_patterns
 
 
def test_redaction_quoted_double():
    """Redaction handles double-quoted multi-word values (OI-401)."""
    old_prod = _ud.PROD_MODE
    old_patterns = _ud.REDACTION_PATTERNS
    _ud.PROD_MODE = True
    _ud.REDACTION_PATTERNS = [
        re.compile(r'(?i)(password)\s*[=:]\s*(?:"[^"]*"|\'[^\']*\'|\S+)'),
    ]
    try:
        result = _ud._apply_redaction('password: "my secret value"')
        assert "my secret value" not in result
        assert "[REDACTED]" in result
    finally:
        _ud.PROD_MODE = old_prod
        _ud.REDACTION_PATTERNS = old_patterns
 
 
def test_redaction_quoted_single():
    """Redaction handles single-quoted multi-word values (OI-401)."""
    old_prod = _ud.PROD_MODE
    old_patterns = _ud.REDACTION_PATTERNS
    _ud.PROD_MODE = True
    _ud.REDACTION_PATTERNS = [
        re.compile(r"(?i)(api_key)\s*[=:]\s*(?:\"[^\"]*\"|'[^']*'|\S+)"),
    ]
    try:
        result = _ud._apply_redaction("api_key='my secret key'")
        assert "my secret key" not in result
        assert "[REDACTED]" in result
    finally:
        _ud.PROD_MODE = old_prod
        _ud.REDACTION_PATTERNS = old_patterns
 
 
def test_redaction_bare_value_still_works():
    """Original bare-value redaction still works after regex update (OI-401)."""
    old_prod = _ud.PROD_MODE
    old_patterns = _ud.REDACTION_PATTERNS
    _ud.PROD_MODE = True
    _ud.REDACTION_PATTERNS = [
        re.compile(r'(?i)(password)\s*[=:]\s*(?:"[^"]*"|\'[^\']*\'|\S+)'),
    ]
    try:
        result = _ud._apply_redaction("password=secret123")
        assert "secret123" not in result
        assert "[REDACTED]" in result
    finally:
        _ud.PROD_MODE = old_prod
        _ud.REDACTION_PATTERNS = old_patterns
 
 
# ============================================================
# GLOBAL DEBUG LEVEL (V2.0)
# ============================================================
 
def test_global_debug_level_override():
    """global_debug_level in config overrides all domain levels."""
    config = {
        "DEBUG_LEVELS": {"DATA": 2, "UI": 3, "NET": 1},
        "global_settings": {"global_debug_level": 5},
    }
    tmp_path = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".json", delete=False, encoding="utf-8"
        ) as f:
            json.dump(config, f)
            tmp_path = f.name
 
        env = dict(os.environ)
        env["OC_DIAGNOSTICS_CONFIG"] = tmp_path
        env.pop("OC_DIAGNOSTICS_LEVEL", None)
        result = subprocess.run(
            [
                sys.executable, "-c",
                "import oc_diagnostics.unified_debug as _ud; "
                "assert all(v == 5 for v in _ud.DEBUG_LEVELS.values()), "
                "f'Not all levels are 5: {dict(_ud.DEBUG_LEVELS)}'"
            ],
            env=env,
            capture_output=True,
            text=True,
            cwd=_REPO_ROOT,
        )
        assert result.returncode == 0, (
            f"global_debug_level override failed; stderr: {result.stderr.strip()!r}"
        )
    finally:
        if tmp_path and os.path.exists(tmp_path):
            os.unlink(tmp_path)
 
 
# ============================================================
# V2.0 BUNDLED CONFIG SCHEMA VERSION
# ============================================================
 
def test_bundled_config_schema_version():
    """Bundled debug_config.json uses V2.0 schema with DEBUG_LEVELS key."""
    config_path = os.path.join(
        os.path.dirname(_ud.__file__), "debug_config.json"
    )
    with open(config_path, encoding="utf-8") as f:
        config = json.load(f)
    assert config.get("_schema_version") == "2.0", (
        f"Expected schema version 2.0, got {config.get('_schema_version')!r}"
    )
    assert "DEBUG_LEVELS" in config, "Bundled config must use DEBUG_LEVELS key"
 
 
# ============================================================
# OI-407: production-mode _dbg() fail-safe (PROD_MAX_LEVEL)
# ============================================================
 
def _reset_prod_warnings():
    _ud._prod_downgrade_warned.clear()
 
 
def test_prod_mode_suppresses_high_level():
    """In PROD_MODE, _dbg() calls above PROD_MAX_LEVEL are suppressed (OI-407)."""
    captured = []
    old_emit = _ud._emit
    old_prod = _ud.PROD_MODE
    old_max = _ud.PROD_MAX_LEVEL
    old_levels = dict(_ud.DEBUG_LEVELS)
    _ud._emit = lambda line, level, record=None: captured.append((level, line))
    _ud.PROD_MODE = True
    _ud.PROD_MAX_LEVEL = 2
    _ud.DEBUG_LEVELS["GENERAL"] = 9  # ensure domain gate would allow L3
    _reset_prod_warnings()
    try:
        _ud._dbg("GENERAL", "test_fn", "verbose L3 detail", level=3)
        # Only the one-shot warning should be emitted — no user L3 line.
        assert len(captured) == 1
        assert "suppressed in production mode" in captured[0][1]
        assert "L3" in captured[0][1]
    finally:
        _ud._emit = old_emit
        _ud.PROD_MODE = old_prod
        _ud.PROD_MAX_LEVEL = old_max
        _ud.DEBUG_LEVELS.clear()
        _ud.DEBUG_LEVELS.update(old_levels)
        _reset_prod_warnings()
 
 
def test_prod_mode_allows_low_level():
    """L1/L2 calls in PROD_MODE are still emitted normally."""
    captured = []
    old_emit = _ud._emit
    old_prod = _ud.PROD_MODE
    old_max = _ud.PROD_MAX_LEVEL
    old_levels = dict(_ud.DEBUG_LEVELS)
    _ud._emit = lambda line, level, record=None: captured.append((level, line))
    _ud.PROD_MODE = True
    _ud.PROD_MAX_LEVEL = 2
    _ud.DEBUG_LEVELS["GENERAL"] = 2
    _reset_prod_warnings()
    try:
        _ud._dbg("GENERAL", "test_fn", "normal L2", level=2)
        assert len(captured) == 1
        assert "normal L2" in captured[0][1]
        assert "suppressed" not in captured[0][1]
    finally:
        _ud._emit = old_emit
        _ud.PROD_MODE = old_prod
        _ud.PROD_MAX_LEVEL = old_max
        _ud.DEBUG_LEVELS.clear()
        _ud.DEBUG_LEVELS.update(old_levels)
        _reset_prod_warnings()
 
 
def test_prod_mode_warning_one_shot():
    """Repeated L3 calls in PROD_MODE emit only one warning per (domain, level)."""
    captured = []
    old_emit = _ud._emit
    old_prod = _ud.PROD_MODE
    old_max = _ud.PROD_MAX_LEVEL
    old_levels = dict(_ud.DEBUG_LEVELS)
    _ud._emit = lambda line, level, record=None: captured.append(line)
    _ud.PROD_MODE = True
    _ud.PROD_MAX_LEVEL = 2
    _ud.DEBUG_LEVELS["GENERAL"] = 9
    _reset_prod_warnings()
    try:
        for _ in range(5):
            _ud._dbg("GENERAL", "f", "m", level=3)
        warn_lines = [c for c in captured if "suppressed" in c]
        assert len(warn_lines) == 1, f"expected 1 warning, got {len(warn_lines)}"
    finally:
        _ud._emit = old_emit
        _ud.PROD_MODE = old_prod
        _ud.PROD_MAX_LEVEL = old_max
        _ud.DEBUG_LEVELS.clear()
        _ud.DEBUG_LEVELS.update(old_levels)
        _reset_prod_warnings()
 
 
def test_prod_max_level_configurable():
    """Setting PROD_MAX_LEVEL=3 allows L3 but still blocks L4."""
    captured = []
    old_emit = _ud._emit
    old_prod = _ud.PROD_MODE
    old_max = _ud.PROD_MAX_LEVEL
    old_levels = dict(_ud.DEBUG_LEVELS)
    _ud._emit = lambda line, level, record=None: captured.append((level, line))
    _ud.PROD_MODE = True
    _ud.PROD_MAX_LEVEL = 3
    _ud.DEBUG_LEVELS["GENERAL"] = 9
    _reset_prod_warnings()
    try:
        _ud._dbg("GENERAL", "f", "l3 ok", level=3)
        _ud._dbg("GENERAL", "f", "l4 blocked", level=4)
        bodies = [c[1] for c in captured]
        assert any("l3 ok" in b for b in bodies)
        assert not any("l4 blocked" in b for b in bodies)
        assert any("suppressed" in b and "L4" in b for b in bodies)
    finally:
        _ud._emit = old_emit
        _ud.PROD_MODE = old_prod
        _ud.PROD_MAX_LEVEL = old_max
        _ud.DEBUG_LEVELS.clear()
        _ud.DEBUG_LEVELS.update(old_levels)
        _reset_prod_warnings()
 
 
def test_non_prod_mode_unaffected():
    """Outside PROD_MODE, PROD_MAX_LEVEL has no effect."""
    captured = []
    old_emit = _ud._emit
    old_prod = _ud.PROD_MODE
    old_max = _ud.PROD_MAX_LEVEL
    old_levels = dict(_ud.DEBUG_LEVELS)
    _ud._emit = lambda line, level, record=None: captured.append((level, line))
    _ud.PROD_MODE = False
    _ud.PROD_MAX_LEVEL = 1  # very restrictive
    _ud.DEBUG_LEVELS["GENERAL"] = 9
    _reset_prod_warnings()
    try:
        _ud._dbg("GENERAL", "f", "l3 allowed in dev", level=3)
        bodies = [c[1] for c in captured]
        assert any("l3 allowed in dev" in b for b in bodies)
        assert not any("suppressed" in b for b in bodies)
    finally:
        _ud._emit = old_emit
        _ud.PROD_MODE = old_prod
        _ud.PROD_MAX_LEVEL = old_max
        _ud.DEBUG_LEVELS.clear()
        _ud.DEBUG_LEVELS.update(old_levels)
        _reset_prod_warnings()