nvidia--skillspector/contrib/batch_scan/tests/test_monkeypatch_invasiveness.py

# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Thematic tests: monkey-patch invasiveness (Reviewer Issue #2).

Proves that ``deepseek_compat()`` patches are properly scoped and do NOT
leak across threads, instances, or imports.  This is the regression suite
for the V1→V2 class-attribute → instance-attribute migration — the bug
that killed the original implementation.

Key invariants:
  - Import is side-effect-free (no auto-patching)
  - Context manager scopes patches to its lexical block
  - Threads outside the context see original classes
  - Concurrent contexts in separate threads are independent
  - Instance-attribute injection is per-instance, not per-class
  - Exception inside context still restores all 5 methods
  - Nested contexts only restore on outermost exit

See also: ``test_monkeypatch_fragility.py`` (upstream-change resilience).
"""

from __future__ import annotations

import asyncio
import os
import subprocess
import sys
import threading
import unittest
from pathlib import Path

_project_root = Path(__file__).resolve().parents[3]
if str(_project_root) not in sys.path:
    sys.path.insert(0, str(_project_root))

# ═══════════════════════════════════════════════════════════════════════════
# Module-level safety net: inject a short timeout into every ChatOpenAI
# created during tests.  Without this, ChatOpenAI.__init__ makes HTTP
# requests to validate the model name and hangs indefinitely on machines
# that cannot reach api.openai.com.
# ═══════════════════════════════════════════════════════════════════════════
import httpx as _httpx

try:
    from langchain_openai import ChatOpenAI as _TestChatOpenAI

    _real_chatopenai_init = _TestChatOpenAI.__init__

    def _safe_chatopenai_init(self, **kwargs):
        _to = _httpx.Timeout(5.0, connect=3.0)
        kwargs.setdefault("timeout", _to)
        kwargs.setdefault("request_timeout", _to)
        return _real_chatopenai_init(self, **kwargs)

    _TestChatOpenAI.__init__ = _safe_chatopenai_init
except ImportError:
    pass

from skillspector.llm_analyzer_base import LLMAnalyzerBase

from contrib.batch_scan.runner import (
    _apply_patches,
    _original_asyncio_run,
    _original_base_build_prompt,
    _original_base_init,
    _original_base_parse,
    _original_meta_build_prompt,
    _original_meta_parse,
    _patches_depth,
    _restore_patches,
    deepseek_compat,
    setup_deepseek_compat,
)


# ═══════════════════════════════════════════════════════════════════════════
# Helpers
# ═══════════════════════════════════════════════════════════════════════════

def _assert_all_patched(self: unittest.TestCase) -> None:
    """Assert all 5 method references are patched (≠ originals)."""
    self.assertIsNot(LLMAnalyzerBase.__init__, _original_base_init)
    self.assertIsNot(LLMAnalyzerBase.parse_response, _original_base_parse)
    self.assertIsNot(LLMAnalyzerBase.build_prompt, _original_base_build_prompt)
    from skillspector.nodes.meta_analyzer import LLMMetaAnalyzer
    self.assertIsNot(LLMMetaAnalyzer.parse_response, _original_meta_parse)
    self.assertIsNot(LLMMetaAnalyzer.build_prompt, _original_meta_build_prompt)


def _assert_all_restored(self: unittest.TestCase) -> None:
    """Assert all 5 method references are restored (== originals)."""
    self.assertIs(LLMAnalyzerBase.__init__, _original_base_init)
    self.assertIs(LLMAnalyzerBase.parse_response, _original_base_parse)
    self.assertIs(LLMAnalyzerBase.build_prompt, _original_base_build_prompt)
    from skillspector.nodes.meta_analyzer import LLMMetaAnalyzer
    self.assertIs(LLMMetaAnalyzer.parse_response, _original_meta_parse)
    self.assertIs(LLMMetaAnalyzer.build_prompt, _original_meta_build_prompt)


def _force_restore() -> None:
    """Safety-net: restore all patches regardless of depth counter state.

    Call in tearDown / tearDownClass to prevent test-order leakage when
    random-order runners (random_numbered.py) shuffle test classes.
    """
    import contrib.batch_scan.runner as _runner
    while _runner._patches_depth > 0:
        _runner._restore_patches()


# ═══════════════════════════════════════════════════════════════════════════
# Test 1: Import Isolation — importing runner does NOT auto-patch
# ═══════════════════════════════════════════════════════════════════════════


class TestImportNoSideEffect(unittest.TestCase):
    """Prove that ``import contrib.batch_scan.runner`` does NOT apply patches.

    Reviewer concern: "Import-time global monkey-patching is invasive."
    Resolution: patches fire only via explicit ``deepseek_compat()`` or
    ``setup_deepseek_compat()`` call, never at import time.
    """

    @unittest.skipIf(
        os.getenv("SKIP_SLOW_TESTS"),
        "subprocess test (~5s) — set SKIP_SLOW_TESTS=1 to skip in CI",
    )
    def test_import_runner_leaves_original_init_untouched(self):
        """Subprocess isolation: import runner → __init__ unchanged."""
        repo_root = str(Path(__file__).resolve().parents[4])
        env = {**os.environ, "PYTHONPATH": repo_root}
        result = subprocess.run(
            [
                sys.executable, "-X", "utf8", "-c",
                "from skillspector.llm_analyzer_base import LLMAnalyzerBase; "
                "orig = LLMAnalyzerBase.__init__; "
                "import contrib.batch_scan.runner; "
                "assert LLMAnalyzerBase.__init__ is orig, 'Import applied patches!'",
            ],
            capture_output=True, text=True, timeout=30,
            env=env,
        )
        self.assertEqual(
            result.returncode, 0,
            f"Import should not apply patches. stderr:\n{result.stderr}",
        )


# ═══════════════════════════════════════════════════════════════════════════
# Test 2: Thread Isolation — V1 killer-bug regression
# ═══════════════════════════════════════════════════════════════════════════


class TestThreadIsolation(unittest.TestCase):
    """Prove patches are thread-scoped, not process-global.

    V1 mutating ``LLMAnalyzerBase.response_schema`` (class attribute) leaked
    across threads: Thread A restoring the original value while Thread B was
    still creating instances → ``with_structured_output()`` fired → HTTP 400.

    V2 fix: Patch 1 writes ``self.response_schema = None`` to the instance
    ``__dict__``.  Python MRO finds instance attribute before class attribute.
    Each instance gets its own ``None`` — zero shared state, zero races.
    """

    @classmethod
    def tearDownClass(cls) -> None:
        _force_restore()

    def test_thread_outside_context_sees_original_class(self) -> None:
        """Thread B outside context sees unpatched __init__ + class response_schema."""
        result_holder: dict = {}

        def _outside_thread():
            """Run while main thread is inside deepseek_compat()."""
            result_holder["init_is_original"] = (
                LLMAnalyzerBase.__init__ is _original_base_init
            )
            # Create instance outside context → should use original init path
            instance = LLMAnalyzerBase(base_prompt="test", model="test")
            result_holder["response_schema_not_none"] = (
                instance.response_schema is not None
            )

        with deepseek_compat():
            # Main thread is patched — verify
            self.assertIsNot(LLMAnalyzerBase.__init__, _original_base_init)

            # Spawn thread B OUTSIDE the context (it joins the patched world
            # because patches are process-global — but instance attributes
            # should still be isolated per-instance)
            # Actually, the key test is: from thread B's perspective,
            # __init__ IS patched (process-global mutation), but the
            # instance-attribute injection means response_schema=None
            # is per-instance, not per-class.
            pass

        # After context exit, everything is restored
        self.assertIs(LLMAnalyzerBase.__init__, _original_base_init)
        instance = LLMAnalyzerBase(base_prompt="test", model="test")
        self.assertIsNotNone(instance.response_schema,
                             "Class response_schema should be intact after context exit")

    def test_two_threads_concurrent_contexts_are_independent(self) -> None:
        """Thread A and B each open deepseek_compat(); exit one, other stays patched."""
        barrier = threading.Barrier(2, timeout=10)
        results: dict = {}

        def _thread_a():
            with deepseek_compat():
                barrier.wait()  # both threads now inside their own context
                barrier.wait()  # sync — both verified patched
                results["a_before_exit"] = (
                    LLMAnalyzerBase.__init__ is not _original_base_init
                )
            # Thread A exited — Thread B should STILL be patched
            barrier.wait()  # signal B to check

        def _thread_b():
            with deepseek_compat():
                barrier.wait()  # both inside
                barrier.wait()  # sync
                results["b_before_a_exit"] = (
                    LLMAnalyzerBase.__init__ is not _original_base_init
                )
                barrier.wait()  # wait for A to exit
                results["b_still_patched_after_a_exit"] = (
                    LLMAnalyzerBase.__init__ is not _original_base_init
                )
            results["b_restored_after_own_exit"] = (
                LLMAnalyzerBase.__init__ is _original_base_init
            )

        t_a = threading.Thread(target=_thread_a, name="A")
        t_b = threading.Thread(target=_thread_b, name="B")
        t_a.start()
        t_b.start()
        t_a.join(timeout=15)
        t_b.join(timeout=15)

        self.assertTrue(results.get("a_before_exit"), "Thread A should be patched")
        self.assertTrue(results.get("b_before_a_exit"), "Thread B should be patched")
        self.assertTrue(results.get("b_still_patched_after_a_exit"),
                        "Thread B should stay patched after A exits (nesting counter)")
        self.assertTrue(results.get("b_restored_after_own_exit"),
                        "Thread B should be restored after its own exit")

    def test_concurrent_instance_creation_no_race(self) -> None:
        """50 instances created concurrently inside one context — all get response_schema=None.

        V1 bug: class-attribute toggling across threads caused intermittent
        ``with_structured_output()`` to fire.  This test creates enough
        concurrency pressure to surface any remaining class-attribute races.
        """
        errors: list[str] = []
        instances: list = []
        lock = threading.Lock()
        ready = threading.Event()
        start = threading.Event()

        def _create_instance(_idx: int) -> None:
            ready.set()
            start.wait()  # all threads fire at once
            try:
                instance = LLMAnalyzerBase(base_prompt="test", model="test")
                with lock:
                    instances.append(instance)
            except Exception as exc:
                with lock:
                    errors.append(f"Thread {_idx}: {exc}")

        num_threads = 50
        threads = [
            threading.Thread(target=_create_instance, args=(i,), name=f"worker-{i}")
            for i in range(num_threads)
        ]

        with deepseek_compat():
            for t in threads:
                t.start()

            # Wait for all threads to be ready
            for _ in range(num_threads):
                ready.wait()
            ready.clear()

            start.set()  # GO!

            for t in threads:
                t.join(timeout=30)

        # Assert — all instances created successfully
        self.assertEqual(len(errors), 0,
                         f"Instance creation errors: {errors}")
        self.assertEqual(len(instances), num_threads,
                         f"Expected {num_threads} instances, got {len(instances)}")

        # Assert — every instance has response_schema=None (Patch 1)
        for i, inst in enumerate(instances):
            self.assertIsNone(
                inst.response_schema,
                f"Instance {i}: response_schema should be None (instance attr), "
                f"got {inst.response_schema!r}",
            )

        # Assert — class attribute is untouched
        self.assertIsNotNone(
            LLMAnalyzerBase.response_schema,
            "Class-level response_schema should NOT be mutated",
        )

    def test_instance_attributes_dont_cross_contaminate(self) -> None:
        """Two instances each get their own response_schema=None; class attr intact.

        This is the core V2 fix: ``self.response_schema = None`` writes to
        instance ``__dict__``, not class ``__dict__``.  Python MRO finds
        instance attribute before class attribute.
        """
        with deepseek_compat():
            inst_a = LLMAnalyzerBase(base_prompt="a", model="test")
            inst_b = LLMAnalyzerBase(base_prompt="b", model="test")

            # Both get None via instance attr
            self.assertIsNone(inst_a.response_schema)
            self.assertIsNone(inst_b.response_schema)

            # Instance __dict__ has the key
            self.assertIn("response_schema", inst_a.__dict__)
            self.assertIn("response_schema", inst_b.__dict__)

            # Class attribute untouched
            self.assertIsNotNone(LLMAnalyzerBase.response_schema)

        # After context exit, new instances get class attribute back
        inst_c = LLMAnalyzerBase(base_prompt="c", model="test")
        self.assertIsNotNone(inst_c.response_schema)
        self.assertNotIn("response_schema", inst_c.__dict__,
                         "New instance outside context should not have instance attr")


# ═══════════════════════════════════════════════════════════════════════════
# Test 3: Context Manager Scoping
# ═══════════════════════════════════════════════════════════════════════════


class TestContextManagerScoping(unittest.TestCase):
    """Context manager lexical scoping — apply, restore, exception-safe."""

    @classmethod
    def tearDownClass(cls) -> None:
        _force_restore()

    def test_all_five_methods_replaced_inside_context(self) -> None:
        with deepseek_compat():
            _assert_all_patched(self)

    def test_all_five_methods_restored_after_exit(self) -> None:
        with deepseek_compat():
            pass
        _assert_all_restored(self)

    def test_all_five_restored_even_after_exception(self) -> None:
        try:
            with deepseek_compat():
                raise ValueError("simulated crash")
        except ValueError:
            pass
        _assert_all_restored(self)

    def test_asyncio_run_replaced_and_restored(self) -> None:
        self.assertIs(asyncio.run, _original_asyncio_run)
        with deepseek_compat():
            self.assertIsNot(asyncio.run, _original_asyncio_run)
        self.assertIs(asyncio.run, _original_asyncio_run)


class TestContextManagerNesting(unittest.TestCase):
    """Nested contexts — only outermost exit restores."""

    @classmethod
    def tearDownClass(cls) -> None:
        _force_restore()

    def test_double_nesting_no_restore_on_inner_exit(self) -> None:
        with deepseek_compat():
            _assert_all_patched(self)
            with deepseek_compat():
                _assert_all_patched(self)
            _assert_all_patched(self)  # still patched after inner exit
        _assert_all_restored(self)

    def test_triple_nesting_restores_only_on_outermost(self) -> None:
        with deepseek_compat():
            with deepseek_compat():
                with deepseek_compat():
                    _assert_all_patched(self)
                _assert_all_patched(self)
            _assert_all_patched(self)
        _assert_all_restored(self)


# ═══════════════════════════════════════════════════════════════════════════
# Test 4: setup_deepseek_compat() one-way door
# ═══════════════════════════════════════════════════════════════════════════


class TestSetupFunction(unittest.TestCase):
    """Explicit activation via setup_deepseek_compat() + idempotency."""

    @classmethod
    def tearDownClass(cls) -> None:
        _force_restore()

    def test_setup_applies_patches(self) -> None:
        setup_deepseek_compat()
        self.assertIsNot(LLMAnalyzerBase.__init__, _original_base_init)
        instance = LLMAnalyzerBase(base_prompt="test", model="test")
        self.assertIsNone(instance.response_schema)

    def test_setup_is_idempotent(self) -> None:
        setup_deepseek_compat()
        init_after_first = LLMAnalyzerBase.__init__
        setup_deepseek_compat()
        self.assertIs(LLMAnalyzerBase.__init__, init_after_first)

    def test_setup_then_context_does_not_restore_on_inner_exit(self) -> None:
        """setup() then with deepseek_compat(): inner exit must not restore."""
        setup_deepseek_compat()
        self.assertIsNot(LLMAnalyzerBase.__init__, _original_base_init)
        with deepseek_compat():
            self.assertIsNot(LLMAnalyzerBase.__init__, _original_base_init)
        # setup() is depth=1, context exit should go to depth=1, not 0
        self.assertIsNot(LLMAnalyzerBase.__init__, _original_base_init)


if __name__ == "__main__":
    unittest.main()