Public Access
1782 lines
63 KiB
Python
1782 lines
63 KiB
Python
"""Unit tests for devx.tools.check_test_isolation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import subprocess
|
|
import textwrap
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from click.testing import CliRunner
|
|
|
|
from devx.tools.check_test_isolation import (
|
|
HEAVY_MODULE_IMPORTS,
|
|
HELPER_INTERNAL_CALLS,
|
|
IO_INTERNAL_CALLS,
|
|
KNOWN_IO_FUNCTIONS,
|
|
KNOWN_SUBPROCESS_HELPERS,
|
|
CallGraph,
|
|
_extract_patch_targets,
|
|
_is_integration_test,
|
|
_load_test_isolation_config,
|
|
_SubprocessAudit,
|
|
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"
|
|
|
|
def test_heavy_module_import_at_module_level(self, tmp_path: Path) -> None:
|
|
file = _write_test_file(
|
|
tmp_path,
|
|
"""
|
|
import pandas
|
|
|
|
def test_foo() -> None:
|
|
assert True
|
|
""",
|
|
)
|
|
violations = analyze_file(file)
|
|
assert len(violations) == 1
|
|
assert violations[0].category == "heavy-module-import"
|
|
assert "pandas" in violations[0].message
|
|
|
|
def test_heavy_import_inside_function_ok(self, tmp_path: Path) -> None:
|
|
file = _write_test_file(
|
|
tmp_path,
|
|
"""
|
|
def test_foo() -> None:
|
|
import pandas
|
|
assert True
|
|
""",
|
|
)
|
|
violations = analyze_file(file)
|
|
assert violations == []
|
|
|
|
def test_heavy_import_from_at_module_level(self, tmp_path: Path) -> None:
|
|
file = _write_test_file(
|
|
tmp_path,
|
|
"""
|
|
from matplotlib import pyplot as plt
|
|
|
|
def test_foo() -> None:
|
|
assert True
|
|
""",
|
|
)
|
|
violations = analyze_file(file)
|
|
assert len(violations) == 1
|
|
assert violations[0].category == "heavy-module-import"
|
|
|
|
def test_reload_without_cleanup_odd_count(self, tmp_path: Path) -> None:
|
|
file = _write_test_file(
|
|
tmp_path,
|
|
"""
|
|
import importlib
|
|
import devx.config as cfg
|
|
|
|
def test_reload_no_cleanup() -> None:
|
|
importlib.reload(cfg)
|
|
assert cfg.TASK_PREFIX == "CUSTOM"
|
|
""",
|
|
)
|
|
violations = analyze_file(file)
|
|
reload_violations = [v for v in violations if v.category == "reload-without-cleanup"]
|
|
assert len(reload_violations) == 1
|
|
assert "1 time(s)" in reload_violations[0].message
|
|
|
|
def test_reload_with_cleanup_even_count_ok(self, tmp_path: Path) -> None:
|
|
file = _write_test_file(
|
|
tmp_path,
|
|
"""
|
|
import importlib
|
|
import devx.config as cfg
|
|
|
|
def test_reload_with_cleanup() -> None:
|
|
importlib.reload(cfg)
|
|
assert cfg.TASK_PREFIX == "CUSTOM"
|
|
importlib.reload(cfg)
|
|
""",
|
|
)
|
|
violations = analyze_file(file)
|
|
reload_violations = [v for v in violations if v.category == "reload-without-cleanup"]
|
|
assert reload_violations == []
|
|
|
|
def test_reload_attribute_access_detected(self, tmp_path: Path) -> None:
|
|
file = _write_test_file(
|
|
tmp_path,
|
|
"""
|
|
import importlib
|
|
import devx.config
|
|
|
|
def test_reload_attr() -> None:
|
|
importlib.reload(devx.config)
|
|
""",
|
|
)
|
|
violations = analyze_file(file)
|
|
reload_violations = [v for v in violations if v.category == "reload-without-cleanup"]
|
|
assert len(reload_violations) == 1
|
|
assert "config" in reload_violations[0].message
|
|
|
|
|
|
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 TestIOFunctionChecks:
|
|
"""Tests for unpatched I/O function detection."""
|
|
|
|
def test_unpatched_get_pat_violation(self, tmp_path: Path) -> None:
|
|
file = _write_test_file(
|
|
tmp_path,
|
|
"""
|
|
from mymodule import get_pat
|
|
|
|
class TestExample:
|
|
def test_calls_get_pat(self) -> None:
|
|
result = get_pat("staging")
|
|
""",
|
|
)
|
|
violations = analyze_file(file)
|
|
assert len(violations) == 1
|
|
assert violations[0].category == "unpatched-io"
|
|
assert "get_pat" in violations[0].message
|
|
|
|
def test_patched_get_pat_no_violation(self, tmp_path: Path) -> None:
|
|
file = _write_test_file(
|
|
tmp_path,
|
|
"""
|
|
from unittest.mock import patch
|
|
from mymodule import get_pat
|
|
|
|
class TestExample:
|
|
@patch("mymodule.get_pat", return_value="pat")
|
|
def test_patched(self, mock) -> None:
|
|
result = get_pat("staging")
|
|
""",
|
|
)
|
|
violations = analyze_file(file)
|
|
assert violations == []
|
|
|
|
def test_unpatched_load_secrets_violation(self, tmp_path: Path) -> None:
|
|
file = _write_test_file(
|
|
tmp_path,
|
|
"""
|
|
from mymodule import load_secrets
|
|
|
|
class TestExample:
|
|
def test_calls_load_secrets(self) -> None:
|
|
result = load_secrets("staging")
|
|
""",
|
|
)
|
|
violations = analyze_file(file)
|
|
assert len(violations) == 1
|
|
assert violations[0].category == "unpatched-io"
|
|
assert "load_secrets" in violations[0].message
|
|
|
|
def test_patched_load_secrets_no_violation(self, tmp_path: Path) -> None:
|
|
file = _write_test_file(
|
|
tmp_path,
|
|
"""
|
|
from unittest.mock import patch
|
|
from mymodule import load_secrets
|
|
|
|
class TestExample:
|
|
@patch("mymodule.load_secrets", return_value={})
|
|
def test_patched(self, mock) -> None:
|
|
result = load_secrets("staging")
|
|
""",
|
|
)
|
|
violations = analyze_file(file)
|
|
assert violations == []
|
|
|
|
def test_unpatched_requests_get_violation(self, tmp_path: Path) -> None:
|
|
file = _write_test_file(
|
|
tmp_path,
|
|
"""
|
|
import requests
|
|
|
|
class TestExample:
|
|
def test_calls_requests(self) -> None:
|
|
resp = requests.get("https://example.com")
|
|
""",
|
|
)
|
|
violations = analyze_file(file)
|
|
assert len(violations) == 1
|
|
assert violations[0].category == "unpatched-io"
|
|
assert "requests.get" in violations[0].message or "get" in violations[0].message
|
|
|
|
def test_integration_marker_skips_subprocess(self, tmp_path: Path) -> None:
|
|
"""@pytest.mark.integration tests should not be flagged for subprocess.run."""
|
|
file = _write_test_file(
|
|
tmp_path,
|
|
"""
|
|
import subprocess
|
|
import pytest
|
|
|
|
@pytest.mark.integration
|
|
def test_real_subprocess():
|
|
subprocess.run(["echo", "hello"])
|
|
""",
|
|
)
|
|
violations = analyze_file(file)
|
|
assert violations == []
|
|
|
|
def test_integration_marker_skips_sleep(self, tmp_path: Path) -> None:
|
|
"""@pytest.mark.integration tests should not be flagged for time.sleep."""
|
|
file = _write_test_file(
|
|
tmp_path,
|
|
"""
|
|
import time
|
|
import pytest
|
|
|
|
@pytest.mark.integration
|
|
def test_real_sleep():
|
|
time.sleep(1)
|
|
""",
|
|
)
|
|
violations = analyze_file(file)
|
|
assert violations == []
|
|
|
|
def test_integration_marker_with_args_skips(self, tmp_path: Path) -> None:
|
|
"""@pytest.mark.integration(...) with args should also be skipped."""
|
|
file = _write_test_file(
|
|
tmp_path,
|
|
"""
|
|
import subprocess
|
|
import pytest
|
|
|
|
@pytest.mark.integration(scope="module")
|
|
def test_real_subprocess():
|
|
subprocess.run(["echo", "hello"])
|
|
""",
|
|
)
|
|
violations = analyze_file(file)
|
|
assert violations == []
|
|
|
|
def test_integration_directory_skipped(self, tmp_path: Path) -> None:
|
|
"""Files in integration/ directories should be skipped entirely."""
|
|
integration_dir = tmp_path / "integration"
|
|
integration_dir.mkdir()
|
|
file = integration_dir / "test_real_io.py"
|
|
file.write_text("import subprocess\ndef test_real_subprocess():\n subprocess.run(['echo', 'hello'])\n")
|
|
violations = analyze_file(file)
|
|
assert violations == []
|
|
|
|
|
|
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_always_strict(self, tmp_path: Path) -> None:
|
|
"""CLI is always strict — no --strict flag needed."""
|
|
_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
|
|
|
|
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
|
|
|
|
|
|
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) >= 2
|
|
|
|
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 fail when hard violations are found (always strict)."""
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
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,
|
|
"--test-isolation-max-loop": 100,
|
|
}.get(opt, False)
|
|
item = MagicMock()
|
|
item.fspath = str(test_file)
|
|
session.items = [item]
|
|
|
|
with pytest.raises(pytest.fail.Exception, match="Test isolation"):
|
|
pytest_collection_finish(session)
|
|
|
|
def test_pytest_collection_finish_advisory_only(self, tmp_path: Path) -> None:
|
|
"""Transitive-subprocess advisories should warn, not fail."""
|
|
import warnings
|
|
from unittest.mock import MagicMock
|
|
|
|
from devx.tools.check_test_isolation import pytest_collection_finish
|
|
|
|
# Create a src/ directory with a module that calls subprocess.run
|
|
# so the call graph can detect transitive subprocess calls.
|
|
src_dir = tmp_path / "src" / "mypkg"
|
|
src_dir.mkdir(parents=True)
|
|
(src_dir / "__init__.py").write_text("")
|
|
(src_dir / "cli.py").write_text("import subprocess\ndef main():\n subprocess.run(['echo'])\n")
|
|
|
|
test_file = _write_test_file(
|
|
tmp_path,
|
|
"""
|
|
from click.testing import CliRunner
|
|
from mypkg.cli import main
|
|
class TestExample:
|
|
def test_advisory(self) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, [])
|
|
""",
|
|
)
|
|
|
|
session = MagicMock()
|
|
session.config.getoption.side_effect = lambda opt: {
|
|
"--no-test-isolation": False,
|
|
"--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)
|
|
|
|
# Should only emit advisory warnings, not fail
|
|
assert any("advisory" in str(warning.message).lower() for warning in w)
|
|
|
|
|
|
class TestSubprocessAudit:
|
|
"""Tests for the _SubprocessAudit runtime wrapper (lines 157-195)."""
|
|
|
|
def test_ensure_installed_wraps_subprocess(self) -> None:
|
|
audit = _SubprocessAudit()
|
|
original_run = subprocess.run
|
|
try:
|
|
audit._ensure_installed()
|
|
assert audit._installed is True
|
|
assert "run" in audit._originals
|
|
# The subprocess.run should now be a wrapper, not the original
|
|
assert subprocess.run is not original_run
|
|
# Calling _ensure_installed again is a no-op (cached return)
|
|
audit._ensure_installed()
|
|
finally:
|
|
# Restore originals
|
|
for name, orig in audit._originals.items():
|
|
setattr(subprocess, name, orig)
|
|
|
|
def test_make_wrapper_records_calls_when_active(self) -> None:
|
|
audit = _SubprocessAudit()
|
|
mock_original = MagicMock(return_value="result")
|
|
wrapper = audit._make_wrapper("run", mock_original)
|
|
audit.start_test()
|
|
result = wrapper(["echo", "hi"], capture_output=True)
|
|
calls = audit.stop_test()
|
|
assert result == "result"
|
|
run_calls = [c for c in calls if c[0] == "run"]
|
|
assert len(run_calls) == 1
|
|
assert "echo" in run_calls[0][1]
|
|
mock_original.assert_called_once_with(["echo", "hi"], capture_output=True)
|
|
|
|
def test_make_wrapper_records_list_cmd_truncation(self) -> None:
|
|
"""Long command lists should be truncated to first 4 elements."""
|
|
audit = _SubprocessAudit()
|
|
mock_original = MagicMock(return_value="result")
|
|
wrapper = audit._make_wrapper("run", mock_original)
|
|
audit.start_test()
|
|
wrapper(["echo", "1", "2", "3", "4", "5", "6"], capture_output=True)
|
|
calls = audit.stop_test()
|
|
run_calls = [c for c in calls if c[0] == "run"]
|
|
assert len(run_calls) == 1
|
|
assert "..." in run_calls[0][1]
|
|
|
|
def test_make_wrapper_records_string_cmd(self) -> None:
|
|
"""A string command (not list) should be recorded as-is."""
|
|
audit = _SubprocessAudit()
|
|
mock_original = MagicMock(return_value="result")
|
|
wrapper = audit._make_wrapper("run", mock_original)
|
|
audit.start_test()
|
|
wrapper("echo hi", shell=True, capture_output=True)
|
|
calls = audit.stop_test()
|
|
run_calls = [c for c in calls if c[0] == "run"]
|
|
assert len(run_calls) == 1
|
|
assert "echo hi" in run_calls[0][1]
|
|
|
|
def test_calls_not_recorded_when_inactive(self) -> None:
|
|
"""When audit is not active, calls should not be recorded."""
|
|
audit = _SubprocessAudit()
|
|
mock_original = MagicMock(return_value="result")
|
|
wrapper = audit._make_wrapper("run", mock_original)
|
|
# Don't call start_test — audit inactive
|
|
wrapper(["echo", "hi"], capture_output=True)
|
|
# stop_test returns empty since no calls recorded
|
|
calls = audit.stop_test()
|
|
assert not calls
|
|
mock_original.assert_called_once_with(["echo", "hi"], capture_output=True)
|
|
|
|
def test_start_then_stop_returns_calls(self) -> None:
|
|
"""start_test initializes calls list, stop_test returns and clears it."""
|
|
audit = _SubprocessAudit()
|
|
mock_original = MagicMock(return_value="result")
|
|
wrapper = audit._make_wrapper("run", mock_original)
|
|
audit.start_test()
|
|
wrapper(["echo"], capture_output=True)
|
|
calls = audit.stop_test()
|
|
assert len(calls) == 1
|
|
# After stop, calls is cleared (None or empty)
|
|
calls2 = audit.stop_test()
|
|
assert not calls2
|
|
|
|
def test_ensure_installed_skips_missing_funcs(self) -> None:
|
|
"""If a subprocess func is missing (None), it should be skipped (line 162)."""
|
|
audit = _SubprocessAudit()
|
|
saved = subprocess.check_output
|
|
try:
|
|
# Temporarily make check_output "missing" (None)
|
|
subprocess.check_output = None # type: ignore[assignment]
|
|
audit._ensure_installed()
|
|
# check_output should NOT be in originals (skipped)
|
|
assert "check_output" not in audit._originals
|
|
# run should still be wrapped
|
|
assert "run" in audit._originals
|
|
finally:
|
|
subprocess.check_output = saved # type: ignore[assignment]
|
|
for name, orig in audit._originals.items():
|
|
setattr(subprocess, name, orig)
|
|
|
|
|
|
class TestExtractPatchTargets:
|
|
"""Tests for _extract_patch_targets (lines 263-291)."""
|
|
|
|
def _parse_func(self, source: str) -> ast.FunctionDef:
|
|
tree = ast.parse(textwrap.dedent(source))
|
|
return tree.body[0] # type: ignore[return-value]
|
|
|
|
def test_patch_object_extracted(self) -> None:
|
|
"""patch.object(module, "name") should extract the short name."""
|
|
node = self._parse_func(
|
|
"""
|
|
def test_foo():
|
|
with patch.object(mymodule, "subprocess"):
|
|
mymodule.do_thing()
|
|
"""
|
|
)
|
|
targets = _extract_patch_targets(node)
|
|
assert "subprocess" in targets
|
|
|
|
def test_patch_object_with_module_alias(self) -> None:
|
|
"""patch.object with a module alias Name as first arg."""
|
|
node = self._parse_func(
|
|
"""
|
|
def test_foo():
|
|
with patch.object(subprocess, "run"):
|
|
subprocess.run(["echo"])
|
|
"""
|
|
)
|
|
targets = _extract_patch_targets(node)
|
|
assert "run" in targets
|
|
|
|
def test_with_patch_context_manager_extracted(self) -> None:
|
|
"""with patch("module.func") in function body should be extracted."""
|
|
node = self._parse_func(
|
|
"""
|
|
def test_foo():
|
|
with patch("mymodule.subprocess.run"):
|
|
mymodule.do_thing()
|
|
"""
|
|
)
|
|
targets = _extract_patch_targets(node)
|
|
assert "mymodule.subprocess.run" in targets
|
|
assert "run" in targets
|
|
|
|
def test_with_multiple_patch_context_managers(self) -> None:
|
|
"""with patch("a"), patch("b") should extract both."""
|
|
node = self._parse_func(
|
|
"""
|
|
def test_foo():
|
|
with patch("mod.a"), patch("mod.b"):
|
|
pass
|
|
"""
|
|
)
|
|
targets = _extract_patch_targets(node)
|
|
assert "mod.a" in targets
|
|
assert "mod.b" in targets
|
|
assert "a" in targets
|
|
assert "b" in targets
|
|
|
|
def test_patch_object_non_string_second_arg_ignored(self) -> None:
|
|
"""patch.object with non-string 2nd arg should not crash."""
|
|
node = self._parse_func(
|
|
"""
|
|
def test_foo():
|
|
with patch.object(mymodule, some_var):
|
|
pass
|
|
"""
|
|
)
|
|
targets = _extract_patch_targets(node)
|
|
assert targets == set()
|
|
|
|
|
|
class TestCallGraph:
|
|
"""Tests for CallGraph building (lines 409, 419-420, 449, 470)."""
|
|
|
|
def _make_src(self, tmp_path: Path, files: dict[str, str]) -> Path:
|
|
src = tmp_path / "src"
|
|
src.mkdir()
|
|
for rel, content in files.items():
|
|
f = src / rel
|
|
f.parent.mkdir(parents=True, exist_ok=True)
|
|
f.write_text(textwrap.dedent(content))
|
|
return src
|
|
|
|
def test_ensure_built_cached(self, tmp_path: Path) -> None:
|
|
"""_ensure_built should only build once (cached return)."""
|
|
src = self._make_src(tmp_path, {"pkg/__init__.py": "", "pkg/mod.py": "def foo():\n pass\n"})
|
|
cg = CallGraph(src)
|
|
cg._ensure_built()
|
|
assert cg._built is True
|
|
nodes_before = dict(cg._nodes)
|
|
# Second call should be a no-op
|
|
cg._ensure_built()
|
|
assert cg._nodes == nodes_before
|
|
|
|
def test_build_skips_syntax_error(self, tmp_path: Path) -> None:
|
|
"""Files with syntax errors should be skipped, not crash."""
|
|
src = self._make_src(
|
|
tmp_path,
|
|
{
|
|
"pkg/__init__.py": "",
|
|
"pkg/broken.py": "def test(:\n pass\n",
|
|
"pkg/good.py": "def foo():\n pass\n",
|
|
},
|
|
)
|
|
cg = CallGraph(src)
|
|
cg._ensure_built()
|
|
# good.py's foo should be registered, broken.py skipped
|
|
assert any("foo" in k for k in cg._nodes)
|
|
|
|
def test_build_skips_unicode_decode_error(self, tmp_path: Path) -> None:
|
|
"""Files with invalid UTF-8 should be skipped."""
|
|
src = tmp_path / "src"
|
|
src.mkdir()
|
|
(src / "pkg").mkdir()
|
|
(src / "pkg" / "__init__.py").write_text("")
|
|
(src / "pkg" / "binary.py").write_bytes(b"\xff\xfe\x00\xbad bytes")
|
|
(src / "pkg" / "good.py").write_text("def foo():\n pass\n")
|
|
cg = CallGraph(src)
|
|
cg._ensure_built()
|
|
assert any("foo" in k for k in cg._nodes)
|
|
|
|
def test_scan_node_skips_classdef(self, tmp_path: Path) -> None:
|
|
"""Methods inside classes should NOT be registered."""
|
|
src = self._make_src(
|
|
tmp_path,
|
|
{
|
|
"pkg/__init__.py": "",
|
|
"pkg/mod.py": """
|
|
class MyClass:
|
|
def my_method(self):
|
|
subprocess.run(["echo"])
|
|
def top_level():
|
|
pass
|
|
""",
|
|
},
|
|
)
|
|
cg = CallGraph(src)
|
|
cg._ensure_built()
|
|
# top_level should be registered
|
|
assert "pkg.mod.top_level" in cg._nodes
|
|
# my_method should NOT be registered (class body skipped)
|
|
assert "pkg.mod.my_method" not in cg._nodes
|
|
assert "my_method" not in cg._by_short
|
|
|
|
def test_register_function_records_io_calls(self, tmp_path: Path) -> None:
|
|
"""KNOWN_IO_FUNCTIONS calls should be recorded in io_calls."""
|
|
src = self._make_src(
|
|
tmp_path,
|
|
{
|
|
"pkg/__init__.py": "",
|
|
"pkg/mod.py": """
|
|
def foo():
|
|
get_pat("staging")
|
|
load_secrets("prod")
|
|
""",
|
|
},
|
|
)
|
|
cg = CallGraph(src)
|
|
cg._ensure_built()
|
|
node = cg._nodes["pkg.mod.foo"]
|
|
assert "get_pat" in node.io_calls
|
|
assert "load_secrets" in node.io_calls
|
|
|
|
def test_register_function_records_subprocess_calls(self, tmp_path: Path) -> None:
|
|
"""subprocess.run calls should be recorded in subprocess_calls."""
|
|
src = self._make_src(
|
|
tmp_path,
|
|
{
|
|
"pkg/__init__.py": "",
|
|
"pkg/mod.py": """
|
|
import subprocess
|
|
def foo():
|
|
subprocess.run(["echo"])
|
|
""",
|
|
},
|
|
)
|
|
cg = CallGraph(src)
|
|
cg._ensure_built()
|
|
node = cg._nodes["pkg.mod.foo"]
|
|
assert "subprocess.run" in node.subprocess_calls
|
|
|
|
|
|
class TestFindReachableDangerous:
|
|
"""Tests for find_reachable_dangerous (lines 516-580)."""
|
|
|
|
def _make_src(self, tmp_path: Path, files: dict[str, str]) -> Path:
|
|
src = tmp_path / "src"
|
|
src.mkdir()
|
|
for rel, content in files.items():
|
|
f = src / rel
|
|
f.parent.mkdir(parents=True, exist_ok=True)
|
|
f.write_text(textwrap.dedent(content))
|
|
return src
|
|
|
|
def test_import_map_resolution(self, tmp_path: Path) -> None:
|
|
"""import_map should resolve target to a precise full name."""
|
|
src = self._make_src(
|
|
tmp_path,
|
|
{
|
|
"pkg/__init__.py": "",
|
|
"pkg/mod.py": """
|
|
import subprocess
|
|
def main():
|
|
subprocess.run(["echo"])
|
|
""",
|
|
},
|
|
)
|
|
cg = CallGraph(src)
|
|
dangerous = cg.find_reachable_dangerous("main", set(), import_map={"main": "pkg.mod.main"})
|
|
assert len(dangerous) == 1
|
|
assert "subprocess" in dangerous[0][1]
|
|
|
|
def test_import_map_falls_back_to_short_name(self, tmp_path: Path) -> None:
|
|
"""If import_map value not in nodes, fall back to short name (line 516)."""
|
|
src = self._make_src(
|
|
tmp_path,
|
|
{
|
|
"pkg/__init__.py": "",
|
|
"pkg/mod.py": """
|
|
import subprocess
|
|
def main():
|
|
subprocess.run(["echo"])
|
|
""",
|
|
},
|
|
)
|
|
cg = CallGraph(src)
|
|
# import_map points to a non-existent full name → fallback to by_short
|
|
dangerous = cg.find_reachable_dangerous("main", set(), import_map={"main": "nonexistent.pkg.main"})
|
|
assert len(dangerous) == 1
|
|
|
|
def test_no_candidates_returns_empty(self, tmp_path: Path) -> None:
|
|
"""If no candidates found, return empty list (line 526)."""
|
|
src = self._make_src(tmp_path, {"pkg/__init__.py": ""})
|
|
cg = CallGraph(src)
|
|
dangerous = cg.find_reachable_dangerous("nonexistent", set())
|
|
assert dangerous == []
|
|
|
|
def test_fully_qualified_name_candidate(self, tmp_path: Path) -> None:
|
|
"""A fully-qualified target_name in nodes should be used directly (line 519)."""
|
|
src = self._make_src(
|
|
tmp_path,
|
|
{
|
|
"pkg/__init__.py": "",
|
|
"pkg/mod.py": """
|
|
import subprocess
|
|
def main():
|
|
subprocess.run(["echo"])
|
|
""",
|
|
},
|
|
)
|
|
cg = CallGraph(src)
|
|
dangerous = cg.find_reachable_dangerous("pkg.mod.main", set())
|
|
assert len(dangerous) == 1
|
|
|
|
def test_short_name_fallback(self, tmp_path: Path) -> None:
|
|
"""target_name not in nodes falls back to short name (line 522)."""
|
|
src = self._make_src(
|
|
tmp_path,
|
|
{
|
|
"pkg/__init__.py": "",
|
|
"pkg/mod.py": """
|
|
import subprocess
|
|
def main():
|
|
subprocess.run(["echo"])
|
|
""",
|
|
},
|
|
)
|
|
cg = CallGraph(src)
|
|
# "pkg.main" is not a full name in nodes, so it falls back to "main"
|
|
dangerous = cg.find_reachable_dangerous("pkg.main", set())
|
|
assert len(dangerous) == 1
|
|
|
|
def test_visited_prevents_infinite_loop(self, tmp_path: Path) -> None:
|
|
"""Visited set prevents infinite loops (line 535)."""
|
|
src = self._make_src(
|
|
tmp_path,
|
|
{
|
|
"pkg/__init__.py": "",
|
|
"pkg/mod.py": """
|
|
def a():
|
|
b()
|
|
def b():
|
|
a()
|
|
subprocess.run(["echo"])
|
|
""",
|
|
},
|
|
)
|
|
cg = CallGraph(src)
|
|
dangerous = cg.find_reachable_dangerous("a", set())
|
|
assert len(dangerous) == 1
|
|
|
|
def test_depth_limit_stops_traversal(self, tmp_path: Path) -> None:
|
|
"""max_depth should stop traversal (line 534)."""
|
|
src = self._make_src(
|
|
tmp_path,
|
|
{
|
|
"pkg/__init__.py": "",
|
|
"pkg/mod.py": """
|
|
def a():
|
|
b()
|
|
def b():
|
|
c()
|
|
def c():
|
|
subprocess.run(["echo"])
|
|
""",
|
|
},
|
|
)
|
|
cg = CallGraph(src)
|
|
# With max_depth=0, only the direct node is visited
|
|
dangerous = cg.find_reachable_dangerous("a", set(), max_depth=0)
|
|
assert dangerous == []
|
|
|
|
def test_node_not_found_continues(self, tmp_path: Path) -> None:
|
|
"""If a queued node isn't in _nodes, continue (line 540)."""
|
|
src = self._make_src(tmp_path, {"pkg/__init__.py": ""})
|
|
cg = CallGraph(src)
|
|
# Manually inject a candidate that doesn't exist in nodes
|
|
cg._by_short["ghost"] = ["pkg.mod.ghost"]
|
|
dangerous = cg.find_reachable_dangerous("ghost", set())
|
|
assert dangerous == []
|
|
|
|
def test_io_calls_checked(self, tmp_path: Path) -> None:
|
|
"""IO calls should be reported as dangerous (lines 551-554)."""
|
|
src = self._make_src(
|
|
tmp_path,
|
|
{
|
|
"pkg/__init__.py": "",
|
|
"pkg/mod.py": """
|
|
def main():
|
|
get_pat("staging")
|
|
""",
|
|
},
|
|
)
|
|
cg = CallGraph(src)
|
|
dangerous = cg.find_reachable_dangerous("main", set())
|
|
assert len(dangerous) == 1
|
|
assert "PAT" in dangerous[0][1] or "get_pat" in str(dangerous)
|
|
|
|
def test_io_calls_patched_skipped(self, tmp_path: Path) -> None:
|
|
"""Patched IO calls should not be reported."""
|
|
src = self._make_src(
|
|
tmp_path,
|
|
{
|
|
"pkg/__init__.py": "",
|
|
"pkg/mod.py": """
|
|
def main():
|
|
get_pat("staging")
|
|
""",
|
|
},
|
|
)
|
|
cg = CallGraph(src)
|
|
dangerous = cg.find_reachable_dangerous("main", {"get_pat"})
|
|
assert dangerous == []
|
|
|
|
def test_patched_helper_skipped_in_enqueue(self, tmp_path: Path) -> None:
|
|
"""A patched helper should not be enqueued (lines 559-560)."""
|
|
src = self._make_src(
|
|
tmp_path,
|
|
{
|
|
"pkg/__init__.py": "",
|
|
"pkg/mod.py": """
|
|
def main():
|
|
run_cmd()
|
|
def run_cmd():
|
|
subprocess.run(["echo"])
|
|
""",
|
|
},
|
|
)
|
|
cg = CallGraph(src)
|
|
# run_cmd is patched → should not traverse into it
|
|
dangerous = cg.find_reachable_dangerous("main", {"run_cmd"})
|
|
assert dangerous == []
|
|
|
|
def test_same_module_resolution(self, tmp_path: Path) -> None:
|
|
"""Calls within the same module should prefer same-module resolution (lines 566-567)."""
|
|
src = self._make_src(
|
|
tmp_path,
|
|
{
|
|
"pkg/__init__.py": "",
|
|
"pkg/mod.py": """
|
|
def main():
|
|
helper()
|
|
def helper():
|
|
subprocess.run(["echo"])
|
|
""",
|
|
},
|
|
)
|
|
cg = CallGraph(src)
|
|
dangerous = cg.find_reachable_dangerous("main", set())
|
|
assert len(dangerous) == 1
|
|
|
|
def test_short_name_single_match_resolution(self, tmp_path: Path) -> None:
|
|
"""A single global match by short name should be resolved (lines 571-572)."""
|
|
src = self._make_src(
|
|
tmp_path,
|
|
{
|
|
"pkg/__init__.py": "",
|
|
"pkg/mod.py": """
|
|
def main():
|
|
helper()
|
|
""",
|
|
"pkg/other.py": """
|
|
import subprocess
|
|
def helper():
|
|
subprocess.run(["echo"])
|
|
""",
|
|
},
|
|
)
|
|
cg = CallGraph(src)
|
|
dangerous = cg.find_reachable_dangerous("main", set())
|
|
assert len(dangerous) == 1
|
|
|
|
def test_is_patched_endswith(self, tmp_path: Path) -> None:
|
|
"""_is_patched should match patches ending with .short (line 580)."""
|
|
src = self._make_src(
|
|
tmp_path,
|
|
{
|
|
"pkg/__init__.py": "",
|
|
"pkg/mod.py": """
|
|
import subprocess
|
|
def main():
|
|
subprocess.run(["echo"])
|
|
""",
|
|
},
|
|
)
|
|
cg = CallGraph(src)
|
|
# "devx.ci.release.subprocess.run" ends with ".run"
|
|
dangerous = cg.find_reachable_dangerous("main", {"devx.ci.release.subprocess.run"})
|
|
assert dangerous == []
|
|
|
|
def test_is_patched_full_name_match(self) -> None:
|
|
"""_is_patched should match exact full name."""
|
|
assert CallGraph._is_patched("subprocess.run", "run", {"subprocess.run"}) is True
|
|
|
|
def test_is_patched_short_name_match(self) -> None:
|
|
"""_is_patched should match short name in patches."""
|
|
assert CallGraph._is_patched("subprocess.run", "run", {"run"}) is True
|
|
|
|
def test_is_patched_no_match(self) -> None:
|
|
"""_is_patched should return False when not patched."""
|
|
assert CallGraph._is_patched("subprocess.run", "run", {"other"}) is False
|
|
|
|
def test_is_patched_endswith_no_false_positive(self) -> None:
|
|
"""endswith should not match substrings (e.g. 'run' vs 'run_cmd')."""
|
|
assert CallGraph._is_patched("mod.run_cmd", "run_cmd", {"mod.run"}) is False
|
|
|
|
|
|
class TestVisitCallAttributeTarget:
|
|
"""Tests for visit_Call with ast.Attribute target (lines 851-862)."""
|
|
|
|
def test_invoke_with_module_func_attribute(self, tmp_path: Path) -> None:
|
|
"""runner.invoke(module.func) should resolve via import_map."""
|
|
src = tmp_path / "src"
|
|
src.mkdir()
|
|
(src / "pkg").mkdir()
|
|
(src / "pkg" / "__init__.py").write_text("")
|
|
(src / "pkg" / "cli.py").write_text("import subprocess\ndef main():\n subprocess.run(['echo'])\n")
|
|
|
|
test_file = tmp_path / "test_example.py"
|
|
test_file.write_text(
|
|
textwrap.dedent(
|
|
"""
|
|
from click.testing import CliRunner
|
|
import pkg.cli as cli_mod
|
|
class TestExample:
|
|
def test_invoke(self) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli_mod.main, [])
|
|
"""
|
|
)
|
|
)
|
|
cg = CallGraph(src)
|
|
violations = analyze_file(test_file, call_graph=cg)
|
|
transitive = [v for v in violations if v.category == "transitive-subprocess"]
|
|
assert len(transitive) == 1
|
|
|
|
def test_invoke_with_attribute_no_import_map(self, tmp_path: Path) -> None:
|
|
"""runner.invoke(mod.func) where mod not in import_map uses attr only (line 860)."""
|
|
src = tmp_path / "src"
|
|
src.mkdir()
|
|
(src / "pkg").mkdir()
|
|
(src / "pkg" / "__init__.py").write_text("")
|
|
(src / "pkg" / "cli.py").write_text("import subprocess\ndef main():\n subprocess.run(['echo'])\n")
|
|
|
|
test_file = tmp_path / "test_example.py"
|
|
test_file.write_text(
|
|
textwrap.dedent(
|
|
"""
|
|
from click.testing import CliRunner
|
|
class TestExample:
|
|
def test_invoke(self) -> None:
|
|
runner = CliRunner()
|
|
# unknown_mod not imported, so falls back to attr name
|
|
result = runner.invoke(unknown_mod.main, [])
|
|
"""
|
|
)
|
|
)
|
|
cg = CallGraph(src)
|
|
violations = analyze_file(test_file, call_graph=cg)
|
|
transitive = [v for v in violations if v.category == "transitive-subprocess"]
|
|
assert len(transitive) == 1
|
|
|
|
def test_invoke_with_attribute_non_name_value(self, tmp_path: Path) -> None:
|
|
"""runner.invoke(get_obj().func) — target.value is not a Name (line 862)."""
|
|
src = tmp_path / "src"
|
|
src.mkdir()
|
|
(src / "pkg").mkdir()
|
|
(src / "pkg" / "__init__.py").write_text("")
|
|
(src / "pkg" / "cli.py").write_text("import subprocess\ndef main():\n subprocess.run(['echo'])\n")
|
|
|
|
test_file = tmp_path / "test_example.py"
|
|
test_file.write_text(
|
|
textwrap.dedent(
|
|
"""
|
|
from click.testing import CliRunner
|
|
class TestExample:
|
|
def test_invoke(self) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(CliRunner().main, [])
|
|
"""
|
|
)
|
|
)
|
|
cg = CallGraph(src)
|
|
violations = analyze_file(test_file, call_graph=cg)
|
|
transitive = [v for v in violations if v.category == "transitive-subprocess"]
|
|
assert len(transitive) == 1
|
|
|
|
|
|
class TestIsIntegrationTest:
|
|
"""Tests for _is_integration_test (lines 1076-1080)."""
|
|
|
|
def test_marker_based_integration(self) -> None:
|
|
"""A test item with 'integration' in keywords should be detected."""
|
|
item = MagicMock()
|
|
item.keywords = {"integration", "test_foo"}
|
|
item.fspath = "tests/unit/test_foo.py"
|
|
assert _is_integration_test(item) is True
|
|
|
|
def test_path_based_integration(self) -> None:
|
|
"""A test item in an integration/ directory should be detected."""
|
|
item = MagicMock()
|
|
item.keywords = {"test_foo"}
|
|
item.fspath = "tests/integration/test_foo.py"
|
|
assert _is_integration_test(item) is True
|
|
|
|
def test_not_integration_test(self) -> None:
|
|
"""A regular test item should not be detected as integration."""
|
|
item = MagicMock()
|
|
item.keywords = {"test_foo"}
|
|
item.fspath = "tests/unit/test_foo.py"
|
|
assert _is_integration_test(item) is False
|
|
|
|
def test_no_keywords_attr(self) -> None:
|
|
"""An item without keywords attr should use fspath only."""
|
|
item = MagicMock()
|
|
item.keywords = {}
|
|
item.fspath = "tests/unit/test_foo.py"
|
|
assert _is_integration_test(item) is False
|
|
|
|
|
|
class TestLoadTestIsolationConfig:
|
|
"""Tests for _load_test_isolation_config — project-specific rule merging."""
|
|
|
|
def test_merges_io_functions(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Project-specific io_functions are added to KNOWN_IO_FUNCTIONS."""
|
|
pyproject = tmp_path / "pyproject.toml"
|
|
pyproject.write_text(
|
|
'[tool.devx.check_test_isolation]\nio_functions = { "my_custom_io" = "reads from disk" }\n'
|
|
)
|
|
monkeypatch.chdir(tmp_path)
|
|
_load_test_isolation_config()
|
|
assert "my_custom_io" in KNOWN_IO_FUNCTIONS
|
|
assert KNOWN_IO_FUNCTIONS["my_custom_io"] == "reads from disk"
|
|
|
|
def test_merges_subprocess_helpers(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Project-specific subprocess_helpers are added."""
|
|
pyproject = tmp_path / "pyproject.toml"
|
|
pyproject.write_text(
|
|
'[tool.devx.check_test_isolation]\nsubprocess_helpers = { "my_sp_helper" = "calls subprocess.run" }\n'
|
|
)
|
|
monkeypatch.chdir(tmp_path)
|
|
_load_test_isolation_config()
|
|
assert "my_sp_helper" in KNOWN_SUBPROCESS_HELPERS
|
|
|
|
def test_merges_helper_internal_calls(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Project-specific helper_internal_calls are merged."""
|
|
pyproject = tmp_path / "pyproject.toml"
|
|
pyproject.write_text(
|
|
'[tool.devx.check_test_isolation]\nhelper_internal_calls = { "my_helper" = ["subprocess", "run_cmd"] }\n'
|
|
)
|
|
monkeypatch.chdir(tmp_path)
|
|
_load_test_isolation_config()
|
|
assert "my_helper" in HELPER_INTERNAL_CALLS
|
|
assert HELPER_INTERNAL_CALLS["my_helper"] == {"subprocess", "run_cmd"}
|
|
|
|
def test_merges_io_internal_calls(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Project-specific io_internal_calls are merged."""
|
|
pyproject = tmp_path / "pyproject.toml"
|
|
pyproject.write_text(
|
|
'[tool.devx.check_test_isolation]\nio_internal_calls = { "my_io_func" = ["open", "yaml"] }\n'
|
|
)
|
|
monkeypatch.chdir(tmp_path)
|
|
_load_test_isolation_config()
|
|
assert "my_io_func" in IO_INTERNAL_CALLS
|
|
assert IO_INTERNAL_CALLS["my_io_func"] == {"open", "yaml"}
|
|
|
|
def test_merges_heavy_module_imports(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Project-specific heavy_module_imports are merged."""
|
|
pyproject = tmp_path / "pyproject.toml"
|
|
pyproject.write_text('[tool.devx.check_test_isolation]\nheavy_module_imports = { "mymodule" = 150.0 }\n')
|
|
monkeypatch.chdir(tmp_path)
|
|
_load_test_isolation_config()
|
|
assert "mymodule" in HEAVY_MODULE_IMPORTS
|
|
assert HEAVY_MODULE_IMPORTS["mymodule"] == 150.0
|
|
|
|
def test_no_config_section_is_noop(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Missing [tool.devx.check_test_isolation] section is a no-op."""
|
|
pyproject = tmp_path / "pyproject.toml"
|
|
pyproject.write_text('[tool.devx]\nother_key = "value"\n')
|
|
monkeypatch.chdir(tmp_path)
|
|
before_io = dict(KNOWN_IO_FUNCTIONS)
|
|
_load_test_isolation_config()
|
|
assert before_io == KNOWN_IO_FUNCTIONS
|
|
|
|
def test_no_pyproject_is_noop(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""No pyproject.toml at all is a no-op."""
|
|
monkeypatch.chdir(tmp_path)
|
|
before = dict(KNOWN_SUBPROCESS_HELPERS)
|
|
_load_test_isolation_config()
|
|
assert before == KNOWN_SUBPROCESS_HELPERS
|
|
|
|
def test_non_dict_config_is_noop(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""A non-dict check_test_isolation section is a no-op."""
|
|
pyproject = tmp_path / "pyproject.toml"
|
|
pyproject.write_text('[tool.devx]\ncheck_test_isolation = "not_a_dict"\n')
|
|
monkeypatch.chdir(tmp_path)
|
|
before = dict(HEAVY_MODULE_IMPORTS)
|
|
_load_test_isolation_config()
|
|
assert before == HEAVY_MODULE_IMPORTS
|
|
|
|
def test_invalid_entry_types_are_skipped(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Entries with wrong types (non-str values) are silently skipped."""
|
|
pyproject = tmp_path / "pyproject.toml"
|
|
pyproject.write_text(
|
|
"[tool.devx.check_test_isolation]\n"
|
|
'io_functions = { "good_func" = "desc", "bad_func" = 123 }\n'
|
|
'heavy_module_imports = { "good_mod" = 100.0, "bad_mod" = "fast" }\n'
|
|
)
|
|
monkeypatch.chdir(tmp_path)
|
|
_load_test_isolation_config()
|
|
assert "good_func" in KNOWN_IO_FUNCTIONS
|
|
assert "bad_func" not in KNOWN_IO_FUNCTIONS
|
|
assert "good_mod" in HEAVY_MODULE_IMPORTS
|
|
assert "bad_mod" not in HEAVY_MODULE_IMPORTS
|
|
|
|
def test_extends_without_replacing_defaults(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Project config adds to defaults without removing them."""
|
|
pyproject = tmp_path / "pyproject.toml"
|
|
pyproject.write_text('[tool.devx.check_test_isolation]\nio_functions = { "project_func" = "project I/O" }\n')
|
|
monkeypatch.chdir(tmp_path)
|
|
_load_test_isolation_config()
|
|
# Default entries still present
|
|
assert "get_pat" in KNOWN_IO_FUNCTIONS
|
|
# Project entry added
|
|
assert "project_func" in KNOWN_IO_FUNCTIONS
|