run_tests.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tests/run_tests.py
 
"""
Test Runner — oc_diagnostics
=============================
 
Purpose
-------
Execute the full ``oc_diagnostics`` test suite from the command line by
importing and running all individual per-module test files via a lightweight
PASS/FAIL/SKIP reporting infrastructure.
 
Responsibilities
----------------
- Import test functions from each per-module test file and run them via the
  internal ``_run()`` wrapper, which catches ``AssertionError`` and
  unexpected exceptions and records PASS / FAIL / SKIP accordingly.
- Handle Dash-optional test modules cleanly: attempt import and skip with a
  clear message when Dash is absent.
- Invoke ``instrumentation_selftest_cli`` as a subprocess when Dash is
  available; treat a non-zero exit code as a test failure.
- Collect PASS/SKIP/FAIL per test, print a structured summary, and exit
  with code ``0`` on full success or ``1`` on any failure.
 
Diagnostics
-----------
Domain: TEST
Levels:
    L2 — lifecycle
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- Must exit ``0`` only when all non-skipped tests pass.
- Must exit ``1`` when any test fails or raises an unexpected exception.
- Core tests must run without Dash, FastAPI, or any optional dependency.
- Must not start a Dash server or produce interactive UI.
- Must not modify global oc_diagnostics state permanently; any state
  changes made during tests must be restored before the test exits.
 
Modules covered by this test suite
-----------------------------------
- unified_debug
- logging_handler
- fastapi_middleware
- id_instrumentation
- instrumentation_middleware
- ui_event_logger
- instrumentation_selftest_cli
- instrumentation_selftest_app
"""
 
import os
import subprocess
import sys
 
# ---------------------------------------------------------------------------
# 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)
 
# Ensure this tests/ directory is importable so the individual test modules
# can be imported by name (e.g. import test_unified_debug).
_TESTS_DIR = os.path.dirname(os.path.abspath(__file__))
if _TESTS_DIR not in sys.path:
    sys.path.insert(0, _TESTS_DIR)
 
 
# ============================================================
# Helpers
# ============================================================
 
PASS = "PASS"
FAIL = "FAIL"
SKIP = "SKIP"
 
_results: list[tuple[str, str, str]] = []  # (status, name, detail)
 
 
def _record(status: str, name: str, detail: str = ""):
    _results.append((status, name, detail))
    label = f"  {status:<5} {name}"
    if detail:
        label += f": {detail}"
    print(label)
 
 
def _run(name: str, fn):
    try:
        fn()
        _record(PASS, name)
    except AssertionError as exc:
        _record(FAIL, name, str(exc))
    except Exception as exc:
        _record(FAIL, name, f"{type(exc).__name__}: {exc}")
 
 
# ============================================================
# CLI self-test (Dash-optional subprocess)
# ============================================================
 
def _run_cli_selftest():
    """Run instrumentation_selftest_cli as a subprocess."""
    try:
        import dash  # noqa — check Dash is available before attempting subprocess
    except ImportError:
        _record(SKIP, "instrumentation_selftest_cli", "Dash not installed")
        return
 
    result = subprocess.run(
        [sys.executable, "-m", "oc_diagnostics.instrumentation_selftest_cli"],
        cwd=_REPO_ROOT,
        capture_output=True,
        text=True,
        timeout=60,
    )
 
    if result.returncode == 0:
        _record(PASS, "instrumentation_selftest_cli")
    else:
        detail = (result.stdout + result.stderr).strip().splitlines()
        last_lines = "\n    ".join(detail[-5:]) if detail else "(no output)"
        _record(
            FAIL,
            "instrumentation_selftest_cli",
            f"exit {result.returncode}\n    {last_lines}",
        )
 
 
# ============================================================
# MAIN
# ============================================================
 
def main():
    print("\noc_diagnostics — Test Suite")
    print("=" * 60)
 
    # Suppress the one-shot pre-init RuntimeWarning.  Tests deliberately omit
    # init() to avoid installing stream interceptors or the exception hook.
    import oc_diagnostics.unified_debug as _ud_runner
    _ud_runner._init_warned = True
 
    # ------------------------------------------------------------------
    # Core tests — no optional dependencies required
    # ------------------------------------------------------------------
    print("\n[Core tests — unified_debug]")
    import test_unified_debug as _tud
    _run("config_loading",              _tud.test_config_loading)
    _run("config_source_populated",     _tud.test_config_source_populated)
    _run("dbg_silent_no_exception",     _tud.test_dbg_silent_no_exception)
    _run("dbg_level_filtering",         _tud.test_dbg_level_filtering)
    _run("dbg_override_flag",           _tud.test_dbg_override_flag)
    _run("dbg_before_init_warning",     _tud.test_dbg_before_init_warning)
    _run("adbg_does_not_raise",         _tud.test_adbg_does_not_raise)
    _run("init_idempotent",             _tud.test_init_idempotent)
    _run("stream_interceptor_interface",_tud.test_stream_interceptor_interface)
    _run("debug_levels_dict_mutable",   _tud.test_debug_levels_dict_mutable)
    _run("set_debug_level_valid",       _tud.test_set_debug_level_valid)
    _run("set_debug_level_invalid",     _tud.test_set_debug_level_invalid)
    _run("set_global_setting_valid",    _tud.test_set_global_setting_valid)
    _run("set_global_setting_invalid",  _tud.test_set_global_setting_invalid)
    _run("json_output_mode",            _tud.test_json_output_mode)
    _run("dbg_timed_context_manager",   _tud.test_dbg_timed)
    _run("dbg_scope_enter_exit",        _tud.test_dbg_scope)
    _run("dbg_profile_decorator",       _tud.test_dbg_profile)
    _run("prod_mode_env_var",           _tud.test_prod_mode_env_var)
    _run("prod_mode_via_config",        _tud.test_prod_mode_via_config)
    _run("log_rotation_count",          _tud.test_log_rotation_count)
    _run("reload_stale_key_removed",    _tud.test_reload_stale_key_removed)
    _run("io_degradation_oserror",      _tud.test_io_degradation_oserror)
 
    # ------------------------------------------------------------------
    # Logging handler tests
    # ------------------------------------------------------------------
    print("\n[Logging handler tests]")
    import test_logging_handler as _tlh
    _run("handler_emit_info",       _tlh.test_handler_emit_info)
    _run("handler_level_mapping",   _tlh.test_handler_level_mapping)
    _run("handler_custom_domain",   _tlh.test_handler_custom_domain)
    _run("handler_never_raises",    _tlh.test_handler_never_raises)
 
    # ------------------------------------------------------------------
    # FastAPI middleware tests — pure ASGI contract, no optional deps
    # ------------------------------------------------------------------
    print("\n[FastAPI middleware tests — pure ASGI contract]")
    import test_fastapi_middleware as _tfm
    _run("fastapi_http_scope_status",     _tfm.test_http_scope_status_passthrough)
    _run("fastapi_websocket_passthrough", _tfm.test_websocket_passthrough)
    _run("fastapi_lifespan_passthrough",  _tfm.test_lifespan_scope_passthrough)
    _run("fastapi_exception_reraise",     _tfm.test_exception_reraise)
    _run("fastapi_http_header_redaction", _tfm.test_http_header_redaction)
    _run("fastapi_elapsed_time_in_log",   _tfm.test_elapsed_time_in_log)
 
    # ------------------------------------------------------------------
    # Dash tests — skipped automatically when Dash is not installed
    # ------------------------------------------------------------------
    print("\n[Dash tests — skipped automatically when Dash is not installed]")
    try:
        import dash  # noqa — availability check only
        _dash_available = True
    except ImportError:
        _dash_available = False
 
    if not _dash_available:
        _record(SKIP, "dash_tests", "Dash not installed — skipping all Dash tests")
    else:
        import test_id_instrumentation as _tid
        _run("structural_id_protection",       _tid.test_structural_id_protection)
        _run("instrument_ids_basic",           _tid.test_instrument_ids_basic)
        _run("dict_id_not_rewritten",          _tid.test_dict_id_not_rewritten)
        _run("no_id_component_unchanged",      _tid.test_no_id_component_unchanged)
        _run("max_depth_limit",                _tid.test_max_depth_limit)
        _run("nested_children_traversal",      _tid.test_nested_children_traversal)
        _run("additional_structural_prefixes", _tid.test_additional_structural_prefixes)
        _run("additional_structural_exact",    _tid.test_additional_structural_exact)
 
        import test_instrumentation_middleware as _tim
        _run("apply_instrumentation_no_op",    _tim.test_apply_instrumentation_no_op)
        _run("apply_list_return",              _tim.test_apply_list_return)
        _run("apply_tuple_return",             _tim.test_apply_tuple_return)
        _run("apply_single_component",         _tim.test_apply_single_component)
        _run("no_double_wrapping",             _tim.test_no_double_wrapping)
 
        import test_ui_event_logger as _tuel
        _run("ui_event_logger_import",         _tuel.test_ui_event_logger_import)
        _run("required_store_id_constant",     _tuel.test_required_store_id_constant)
        _run("ui_logger_custom_props",         _tuel.test_ui_logger_custom_props)
 
        _run_cli_selftest()
 
    # ------------------------------------------------------------------
    # Summary
    # ------------------------------------------------------------------
    passed = sum(1 for s, _, _ in _results if s == PASS)
    failed = sum(1 for s, _, _ in _results if s == FAIL)
    skipped = sum(1 for s, _, _ in _results if s == SKIP)
    total = len(_results)
 
    print("\n" + "=" * 60)
    print(
        f"Results: {passed} passed, {failed} failed, {skipped} skipped ({total} total)"
    )
 
    if failed:
        print("OVERALL: FAILED")
        print("=" * 60)
        sys.exit(1)
    else:
        print("OVERALL: PASSED")
        print("=" * 60)
        sys.exit(0)
 
 
if __name__ == "__main__":
    main()