DEVX-129: feat: test isolation pytest plugin, shift-left quality gates, dep upgrades

tests/unit/test_check_test_isolation.py

@@ -0,0 +1,777 @@
+"""Unit tests for devx.tools.check_test_isolation."""
+
+from __future__ import annotations
+
+import textwrap
+from pathlib import Path
+
+from click.testing import CliRunner
+
+from devx.tools.check_test_isolation import (
+    HELPER_INTERNAL_CALLS,
+    KNOWN_SUBPROCESS_HELPERS,
+    analyze_file,
+    analyze_test_files,
+    cli,
+    find_test_files,
+)
+
+
+def _write_test_file(tmp_path: Path, content: str) -> Path:
+    """Write content to a test file and return the path."""
+    file = tmp_path / "test_example.py"
+    file.write_text(textwrap.dedent(content))
+    return file
+
+
+class TestFindTestFiles:
+    def test_finds_test_files_in_directory(self, tmp_path: Path) -> None:
+        (tmp_path / "test_foo.py").touch()
+        (tmp_path / "test_bar.py").touch()
+        (tmp_path / "helper.py").touch()
+        result = find_test_files(tmp_path)
+        assert len(result) == 2
+        assert all(f.name.startswith("test_") for f in result)
+
+    def test_single_file(self, tmp_path: Path) -> None:
+        file = tmp_path / "test_single.py"
+        file.touch()
+        result = find_test_files(file)
+        assert result == [file]
+
+    def test_non_python_file(self, tmp_path: Path) -> None:
+        file = tmp_path / "test_readme.md"
+        file.touch()
+        result = find_test_files(file)
+        assert result == []
+
+
+class TestAnalyzeFile:
+    def test_clean_file_no_violations(self, tmp_path: Path) -> None:
+        file = _write_test_file(
+            tmp_path,
+            """
+            from unittest.mock import patch, MagicMock
+
+            class TestExample:
+                @patch("mymodule.subprocess.run")
+                def test_with_patch(self, mock_run: MagicMock) -> None:
+                    mymodule.do_thing()
+        """,
+        )
+        violations = analyze_file(file)
+        assert violations == []
+
+    def test_unpatched_subprocess_run(self, tmp_path: Path) -> None:
+        file = _write_test_file(
+            tmp_path,
+            """
+            import subprocess
+
+            class TestExample:
+                def test_direct_subprocess(self) -> None:
+                    subprocess.run(["echo", "hi"])
+        """,
+        )
+        violations = analyze_file(file)
+        assert len(violations) == 1
+        assert violations[0].category == "unpatched-subprocess"
+        assert "subprocess.run" in violations[0].message
+
+    def test_patched_subprocess_no_violation(self, tmp_path: Path) -> None:
+        file = _write_test_file(
+            tmp_path,
+            """
+            from unittest.mock import patch, MagicMock
+            import subprocess
+
+            class TestExample:
+                @patch("subprocess.run")
+                def test_patched(self, mock_run: MagicMock) -> None:
+                    subprocess.run(["echo", "hi"])
+        """,
+        )
+        violations = analyze_file(file)
+        assert violations == []
+
+    def test_unpatched_time_sleep(self, tmp_path: Path) -> None:
+        file = _write_test_file(
+            tmp_path,
+            """
+            import time
+
+            class TestExample:
+                def test_with_sleep(self) -> None:
+                    time.sleep(5)
+        """,
+        )
+        violations = analyze_file(file)
+        assert len(violations) == 1
+        assert violations[0].category == "unpatched-sleep"
+
+    def test_patched_time_sleep_no_violation(self, tmp_path: Path) -> None:
+        file = _write_test_file(
+            tmp_path,
+            """
+            from unittest.mock import patch, MagicMock
+            import time
+
+            class TestExample:
+                @patch("time.sleep")
+                def test_patched_sleep(self, mock_sleep: MagicMock) -> None:
+                    time.sleep(5)
+        """,
+        )
+        violations = analyze_file(file)
+        assert violations == []
+
+    def test_unpatched_known_helper(self, tmp_path: Path) -> None:
+        file = _write_test_file(
+            tmp_path,
+            """
+            from mymodule import update_doc_versions
+
+            class TestExample:
+                def test_calls_helper(self) -> None:
+                    update_doc_versions("1.0.0")
+        """,
+        )
+        violations = analyze_file(file)
+        assert len(violations) == 1
+        assert violations[0].category == "unpatched-helper"
+        assert "update_doc_versions" in violations[0].message
+
+    def test_patched_helper_no_violation(self, tmp_path: Path) -> None:
+        file = _write_test_file(
+            tmp_path,
+            """
+            from unittest.mock import patch, MagicMock
+            from mymodule import update_doc_versions
+
+            class TestExample:
+                @patch("mymodule.update_doc_versions")
+                def test_patched_helper(self, mock: MagicMock) -> None:
+                    update_doc_versions("1.0.0")
+        """,
+        )
+        violations = analyze_file(file)
+        assert violations == []
+
+    def test_helper_safe_when_subprocess_patched(self, tmp_path: Path) -> None:
+        """update_doc_versions is safe if subprocess.run is patched."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            from unittest.mock import patch, MagicMock
+            from mymodule import update_doc_versions
+
+            class TestExample:
+                @patch("subprocess.run")
+                def test_subprocess_patched(self, mock: MagicMock) -> None:
+                    update_doc_versions("1.0.0")
+        """,
+        )
+        violations = analyze_file(file)
+        assert violations == []
+
+    def test_helper_safe_when_internal_dep_patched(self, tmp_path: Path) -> None:
+        """run_tests is safe if run_cmd is patched (run_tests calls run_cmd)."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            from unittest.mock import patch, MagicMock
+            from mymodule import run_tests
+
+            class TestExample:
+                @patch("mymodule.run_cmd")
+                def test_run_cmd_patched(self, mock: MagicMock) -> None:
+                    run_tests()
+        """,
+        )
+        violations = analyze_file(file)
+        assert violations == []
+
+    def test_excessive_iterations(self, tmp_path: Path) -> None:
+        file = _write_test_file(
+            tmp_path,
+            """
+            class TestExample:
+                def test_many_iterations(self) -> None:
+                    for _ in range(500):
+                        assert True
+        """,
+        )
+        violations = analyze_file(file)
+        assert len(violations) == 1
+        assert violations[0].category == "excessive-iterations"
+        assert "500" in violations[0].message
+
+    def test_acceptable_iterations(self, tmp_path: Path) -> None:
+        file = _write_test_file(
+            tmp_path,
+            """
+            class TestExample:
+                def test_few_iterations(self) -> None:
+                    for _ in range(50):
+                        assert True
+        """,
+        )
+        violations = analyze_file(file)
+        assert violations == []
+
+    def test_range_with_start_stop(self, tmp_path: Path) -> None:
+        """range(0, 500) should also be flagged."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            class TestExample:
+                def test_range_start_stop(self) -> None:
+                    for _ in range(0, 500):
+                        assert True
+        """,
+        )
+        violations = analyze_file(file)
+        assert len(violations) == 1
+        assert violations[0].category == "excessive-iterations"
+
+    def test_subprocess_check_output(self, tmp_path: Path) -> None:
+        """subprocess.check_output should also be flagged."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            import subprocess
+            class TestExample:
+                def test_check_output(self) -> None:
+                    result = subprocess.check_output(["echo", "hi"])
+        """,
+        )
+        violations = analyze_file(file)
+        assert len(violations) == 1
+        assert violations[0].category == "unpatched-subprocess"
+
+    def test_subprocess_popen(self, tmp_path: Path) -> None:
+        """subprocess.Popen should also be flagged."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            import subprocess
+            class TestExample:
+                def test_popen(self) -> None:
+                    p = subprocess.Popen(["echo", "hi"])
+        """,
+        )
+        violations = analyze_file(file)
+        assert len(violations) == 1
+        assert violations[0].category == "unpatched-subprocess"
+
+    def test_attribute_style_patch(self, tmp_path: Path) -> None:
+        """mock.patch.object style should be recognized."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            from unittest.mock import mock
+            import subprocess
+            class TestExample:
+                @mock.patch("subprocess.run")
+                def test_attr_patch(self, mock_run) -> None:
+                    subprocess.run(["echo"])
+        """,
+        )
+        violations = analyze_file(file)
+        assert violations == []
+
+    def test_subprocess_check_call(self, tmp_path: Path) -> None:
+        """subprocess.check_call should also be flagged."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            import subprocess
+            class TestExample:
+                def test_check_call(self) -> None:
+                    subprocess.check_call(["echo", "hi"])
+        """,
+        )
+        violations = analyze_file(file)
+        assert len(violations) == 1
+        assert violations[0].category == "unpatched-subprocess"
+
+    def test_subprocess_call(self, tmp_path: Path) -> None:
+        """subprocess.call should also be flagged."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            import subprocess
+            class TestExample:
+                def test_call(self) -> None:
+                    subprocess.call(["echo", "hi"])
+        """,
+        )
+        violations = analyze_file(file)
+        assert len(violations) == 1
+        assert violations[0].category == "unpatched-subprocess"
+
+    def test_non_subprocess_attribute_not_flagged(self, tmp_path: Path) -> None:
+        """subprocess.something_else should not be flagged."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            import subprocess
+            class TestExample:
+                def test_other(self) -> None:
+                    x = subprocess.PIPE
+        """,
+        )
+        violations = analyze_file(file)
+        assert violations == []
+
+    def test_async_test_function(self, tmp_path: Path) -> None:
+        """Async test functions should be analyzed too."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            import subprocess
+            class TestExample:
+                async def test_async(self) -> None:
+                    subprocess.run(["echo"])
+        """,
+        )
+        violations = analyze_file(file)
+        assert len(violations) == 1
+        assert violations[0].category == "unpatched-subprocess"
+
+    def test_call_with_no_name(self, tmp_path: Path) -> None:
+        """Calls with complex expressions (e.g. lambda) should not crash."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            class TestExample:
+                def test_lambda_call(self) -> None:
+                    (lambda: None)()
+        """,
+        )
+        violations = analyze_file(file)
+        assert violations == []
+
+    def test_range_with_no_args(self, tmp_path: Path) -> None:
+        """range() with no args should not crash."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            class TestExample:
+                def test_empty_range(self) -> None:
+                    for _ in range():
+                        pass
+        """,
+        )
+        violations = analyze_file(file)
+        assert violations == []
+
+    def test_range_with_non_constant_stop(self, tmp_path: Path) -> None:
+        """range(0, variable) should not be flagged (can't determine count)."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            class TestExample:
+                def test_variable_range(self) -> None:
+                    n = 100
+                    for _ in range(0, n):
+                        pass
+        """,
+        )
+        violations = analyze_file(file)
+        assert violations == []
+
+    def test_range_with_non_constant_start(self, tmp_path: Path) -> None:
+        """range(variable, 500) should be flagged with stop value."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            class TestExample:
+                def test_variable_start(self) -> None:
+                    s = 0
+                    for _ in range(s, 500):
+                        pass
+        """,
+        )
+        violations = analyze_file(file)
+        assert len(violations) == 1
+        assert violations[0].category == "excessive-iterations"
+
+    def test_for_loop_with_non_range_call(self, tmp_path: Path) -> None:
+        """for loop with a non-range call should not crash."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            class TestExample:
+                def test_iter_func(self) -> None:
+                    for _ in list([1, 2, 3]):
+                        pass
+        """,
+        )
+        violations = analyze_file(file)
+        assert violations == []
+
+    def test_for_loop_with_list(self, tmp_path: Path) -> None:
+        """for loop with a list literal should not crash."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            class TestExample:
+                def test_iter_list(self) -> None:
+                    for _ in [1, 2, 3]:
+                        pass
+        """,
+        )
+        violations = analyze_file(file)
+        assert violations == []
+
+    def test_range_with_single_non_int_arg(self, tmp_path: Path) -> None:
+        """range(variable) should not crash or flag."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            class TestExample:
+                def test_range_var(self) -> None:
+                    n = 50
+                    for _ in range(n):
+                        pass
+        """,
+        )
+        violations = analyze_file(file)
+        assert violations == []
+
+    def test_range_with_three_args(self, tmp_path: Path) -> None:
+        """range(0, 500, 1) should be flagged (3 args, stop=500)."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            class TestExample:
+                def test_range_step(self) -> None:
+                    for _ in range(0, 500, 1):
+                        pass
+        """,
+        )
+        violations = analyze_file(file)
+        assert len(violations) == 1
+        assert violations[0].category == "excessive-iterations"
+
+    def test_non_test_function_not_analyzed(self, tmp_path: Path) -> None:
+        """Non-test functions should not be analyzed."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            import subprocess
+
+            def helper_function() -> None:
+                subprocess.run(["echo", "hi"])
+
+            class TestExample:
+                def test_uses_helper(self) -> None:
+                    helper_function()
+        """,
+        )
+        violations = analyze_file(file)
+        # helper_function is not a test, so no violation for its subprocess call
+        # test_uses_helper calls helper_function, not subprocess directly
+        assert violations == []
+
+    def test_class_level_patch_satisfies_check(self, tmp_path: Path) -> None:
+        """@patch on the class should satisfy the check for all methods."""
+        file = _write_test_file(
+            tmp_path,
+            """
+            from unittest.mock import patch, MagicMock
+            import subprocess
+
+            @patch("subprocess.run")
+            class TestExample:
+                def test_method_a(self, mock: MagicMock) -> None:
+                    subprocess.run(["echo", "a"])
+
+                def test_method_b(self, mock: MagicMock) -> None:
+                    subprocess.run(["echo", "b"])
+        """,
+        )
+        violations = analyze_file(file)
+        assert violations == []
+
+    def test_syntax_error_returns_violation(self, tmp_path: Path) -> None:
+        file = tmp_path / "test_broken.py"
+        file.write_text("def test(:\n  pass\n")
+        violations = analyze_file(file)
+        assert len(violations) == 1
+        assert violations[0].category == "syntax-error"
+
+
+class TestAnalyzeTestFiles:
+    def test_multiple_files(self, tmp_path: Path) -> None:
+        _write_test_file(
+            tmp_path,
+            """
+            import subprocess
+            class TestA:
+                def test_a(self) -> None:
+                    subprocess.run(["echo"])
+        """,
+        )
+        file2 = tmp_path / "test_other.py"
+        file2.write_text(
+            textwrap.dedent("""
+            import time
+            class TestB:
+                def test_b(self) -> None:
+                    time.sleep(1)
+        """)
+        )
+        violations = analyze_test_files(tmp_path)
+        assert len(violations) == 2
+        categories = {v.category for v in violations}
+        assert "unpatched-subprocess" in categories
+        assert "unpatched-sleep" in categories
+
+    def test_category_filter(self, tmp_path: Path) -> None:
+        _write_test_file(
+            tmp_path,
+            """
+            import subprocess
+            class TestA:
+                def test_a(self) -> None:
+                    subprocess.run(["echo"])
+        """,
+        )
+        file2 = tmp_path / "test_other.py"
+        file2.write_text(
+            textwrap.dedent("""
+            import time
+            class TestB:
+                def test_b(self) -> None:
+                    time.sleep(1)
+        """)
+        )
+        violations = analyze_test_files(tmp_path, categories={"unpatched-sleep"})
+        assert len(violations) == 1
+        assert violations[0].category == "unpatched-sleep"
+
+
+class TestKnownHelpers:
+    def test_all_helpers_have_internal_calls(self) -> None:
+        """Every known helper should have its internal calls documented."""
+        for helper in KNOWN_SUBPROCESS_HELPERS:
+            assert helper in HELPER_INTERNAL_CALLS, f"Missing HELPER_INTERNAL_CALLS entry for {helper}"
+
+    def test_run_tests_internal_calls_include_run_cmd(self) -> None:
+        assert "run_cmd" in HELPER_INTERNAL_CALLS["run_tests"]
+
+    def test_update_doc_versions_internal_calls_include_subprocess(self) -> None:
+        assert "subprocess" in HELPER_INTERNAL_CALLS["update_doc_versions"]
+
+
+class TestCli:
+    """Tests for the standalone CLI interface."""
+
+    def test_clean_directory_exits_zero(self, tmp_path: Path) -> None:
+        _write_test_file(
+            tmp_path,
+            """
+            from unittest.mock import patch, MagicMock
+            class TestExample:
+                @patch("subprocess.run")
+                def test_ok(self, mock: MagicMock) -> None:
+                    pass
+        """,
+        )
+        runner = CliRunner()
+        result = runner.invoke(cli, ["--test-path", str(tmp_path)])
+        assert result.exit_code == 0
+        assert "no violations" in result.output
+
+    def test_violations_exit_nonzero(self, tmp_path: Path) -> None:
+        _write_test_file(
+            tmp_path,
+            """
+            import subprocess
+            class TestExample:
+                def test_bad(self) -> None:
+                    subprocess.run(["echo"])
+        """,
+        )
+        runner = CliRunner()
+        result = runner.invoke(cli, ["--test-path", str(tmp_path)])
+        assert result.exit_code == 1
+        assert "FAILED" in result.output
+        assert "unpatched-subprocess" in result.output
+
+    def test_strict_flag(self, tmp_path: Path) -> None:
+        _write_test_file(
+            tmp_path,
+            """
+            import subprocess
+            class TestExample:
+                def test_bad(self) -> None:
+                    subprocess.run(["echo"])
+        """,
+        )
+        runner = CliRunner()
+        result = runner.invoke(cli, ["--test-path", str(tmp_path), "--strict"])
+        assert result.exit_code == 1
+
+    def test_category_filter(self, tmp_path: Path) -> None:
+        _write_test_file(
+            tmp_path,
+            """
+            import subprocess, time
+            class TestExample:
+                def test_bad(self) -> None:
+                    subprocess.run(["echo"])
+                    time.sleep(1)
+        """,
+        )
+        runner = CliRunner()
+        result = runner.invoke(cli, ["--test-path", str(tmp_path), "--categories", "unpatched-sleep"])
+        assert result.exit_code == 1
+        assert "unpatched-sleep" in result.output
+        assert "unpatched-subprocess" not in result.output
+
+    def test_max_loop_iterations_option(self, tmp_path: Path) -> None:
+        _write_test_file(
+            tmp_path,
+            """
+            class TestExample:
+                def test_loop(self) -> None:
+                    for _ in range(10):
+                        assert True
+        """,
+        )
+        runner = CliRunner()
+        # With max=5, 10 iterations is a violation
+        result = runner.invoke(cli, ["--test-path", str(tmp_path), "--max-loop-iterations", "5"])
+        assert result.exit_code == 1
+        assert "excessive-iterations" in result.output
+
+    def test_no_test_files(self, tmp_path: Path) -> None:
+        runner = CliRunner()
+        result = runner.invoke(cli, ["--test-path", str(tmp_path)])
+        assert result.exit_code == 0
+        assert "no violations" in result.output
+
+    def test_strict_clean_directory_exits_zero(self, tmp_path: Path) -> None:
+        """Strict mode with no violations should still exit 0."""
+        _write_test_file(
+            tmp_path,
+            """
+            from unittest.mock import patch, MagicMock
+            class TestExample:
+                @patch("subprocess.run")
+                def test_ok(self, mock: MagicMock) -> None:
+                    pass
+        """,
+        )
+        runner = CliRunner()
+        result = runner.invoke(cli, ["--test-path", str(tmp_path), "--strict"])
+        assert result.exit_code == 0
+
+
+class TestPytestPlugin:
+    """Tests for the pytest plugin hooks.
+
+    These hooks are marked with pragma: no cover because they're loaded
+    by pytest before coverage instrumentation starts. We test them via
+    direct calls to verify correctness.
+    """
+
+    def test_pytest_addoption_registers_options(self) -> None:
+        """Verify that pytest_addoption registers the expected options."""
+        from unittest.mock import MagicMock
+
+        from devx.tools.check_test_isolation import pytest_addoption
+
+        parser = MagicMock()
+        pytest_addoption(parser)
+
+        addoption_calls = parser.addoption.call_args_list
+        assert len(addoption_calls) >= 3
+
+    def test_pytest_collection_finish_noop_when_disabled(self) -> None:
+        """Plugin should skip analysis when --no-test-isolation is set."""
+        from unittest.mock import MagicMock
+
+        from devx.tools.check_test_isolation import pytest_collection_finish
+
+        session = MagicMock()
+        session.config.getoption.side_effect = lambda opt: opt == "--no-test-isolation"
+        pytest_collection_finish(session)
+
+    def test_pytest_collection_finish_no_violations(self) -> None:
+        """Plugin should not emit warnings when there are no violations."""
+        from unittest.mock import MagicMock
+
+        from devx.tools.check_test_isolation import pytest_collection_finish
+
+        session = MagicMock()
+        session.config.getoption.side_effect = lambda opt: False
+        session.items = []
+        pytest_collection_finish(session)
+
+    def test_pytest_collection_finish_with_violation(self, tmp_path: Path) -> None:
+        """Plugin should emit warnings when violations are found."""
+        import warnings
+        from unittest.mock import MagicMock
+
+        from devx.tools.check_test_isolation import pytest_collection_finish
+
+        test_file = _write_test_file(
+            tmp_path,
+            """
+            import subprocess
+            class TestExample:
+                def test_bad(self) -> None:
+                    subprocess.run(["echo"])
+        """,
+        )
+
+        session = MagicMock()
+        session.config.getoption.side_effect = lambda opt: False
+        item = MagicMock()
+        item.fspath = str(test_file)
+        session.items = [item]
+
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
+            pytest_collection_finish(session)
+
+        assert len(w) >= 1
+        assert any("Test isolation violation" in str(warning.message) for warning in w)
+
+    def test_pytest_collection_finish_strict_mode(self, tmp_path: Path) -> None:
+        """Plugin should emit warnings and print summary in strict mode."""
+        import warnings
+        from unittest.mock import MagicMock
+
+        from devx.tools.check_test_isolation import pytest_collection_finish
+
+        test_file = _write_test_file(
+            tmp_path,
+            """
+            import subprocess
+            class TestExample:
+                def test_bad(self) -> None:
+                    subprocess.run(["echo"])
+        """,
+        )
+
+        session = MagicMock()
+        session.config.getoption.side_effect = lambda opt: {
+            "--no-test-isolation": False,
+            "--strict-test-isolation": True,
+            "--test-isolation-max-loop": 100,
+        }.get(opt, False)
+        item = MagicMock()
+        item.fspath = str(test_file)
+        session.items = [item]
+
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
+            pytest_collection_finish(session)
+
+        assert len(w) >= 1
+        assert any("Test isolation violation" in str(warning.message) for warning in w)