DEVX-134: feat: add I/O function isolation check and skip integration tests

src/devx/tools/check_test_isolation.py

@@ -30,7 +30,10 @@ Patterns detected:
    without patching it.
 3. **Unpatched known-subprocess-helpers** — functions known to spawn
    subprocesses (e.g. ``update_doc_versions``) called without patching.
-4. **Excessive iteration loops** — ``for _ in range(N)`` where N > 100.
+4. **Unpatched I/O functions** — functions known to do filesystem or
+   network I/O (e.g. ``get_pat``, ``load_secrets``, ``requests.get``)
+   called without patching.
+5. **Excessive iteration loops** — ``for _ in range(N)`` where N > 100.
 """
 
 from __future__ import annotations
@@ -57,6 +60,23 @@ KNOWN_SUBPROCESS_HELPERS: dict[str, str] = {
     "run_cmd": "calls subprocess.run for shell commands",
 }
 
+# Functions known to do filesystem or network I/O that should be mocked in tests.
+# Maps function name → description of what I/O it does.
+# If a test calls one of these without a corresponding @patch, it's a violation.
+KNOWN_IO_FUNCTIONS: dict[str, str] = {  # nosec B105 — descriptions, not passwords
+    "get_pat": "reads ZITADEL PAT from filesystem/env (ZitadelAuth._iter_sources)",
+    "load_secrets": "reads YAML config file from disk",
+    "get_customer_secret": "reads customer-specific config from disk",
+    "requests.get": "performs HTTP GET to a real server",
+    "requests.post": "performs HTTP POST to a real server",
+    "requests.put": "performs HTTP PUT to a real server",
+    "requests.patch": "performs HTTP PATCH to a real server",
+    "requests.delete": "performs HTTP DELETE to a real server",
+    "urlopen": "performs HTTP request to a real server",
+    "httpx.get": "performs HTTP GET to a real server",
+    "httpx.post": "performs HTTP POST to a real server",
+}
+
 # Transitive dependencies: if a helper calls another helper that is patched,
 # the call is safe.  Maps helper → set of function names it internally calls.
 # If ANY of these are in the test's patches, the helper call is safe.
@@ -126,6 +146,20 @@ def _is_test_function(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
     return node.name.startswith("test_")
 
 
+def _has_integration_marker(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
+    """Check if a test function has @pytest.mark.integration decorator."""
+    for decorator in node.decorator_list:
+        # @pytest.mark.integration  →  ast.Attribute(attr='integration')
+        if isinstance(decorator, ast.Attribute) and decorator.attr == "integration":
+            return True
+        # @pytest.mark.integration(...)  →  ast.Call(func=ast.Attribute(attr='integration'))
+        if isinstance(decorator, ast.Call):
+            func = decorator.func
+            if isinstance(func, ast.Attribute) and func.attr == "integration":
+                return True
+    return False
+
+
 def _get_called_name(node: ast.Call) -> str | None:
     func = node.func
     if isinstance(func, ast.Name):
@@ -208,6 +242,11 @@ class TestIsolationVisitor(ast.NodeVisitor):
             self.generic_visit(node)
             return
 
+        # Skip integration tests — they intentionally do real I/O
+        if _has_integration_marker(node):
+            self.generic_visit(node)
+            return
+
         patches = _extract_patch_targets(node)
         info = TestFunctionInfo(
             name=node.name,
@@ -299,6 +338,34 @@ class TestIsolationVisitor(ast.NodeVisitor):
                 )
             )
 
+        # Check 4: Known I/O functions (filesystem/network)
+        # Match by short name (e.g. "get_pat") or full name (e.g. "requests.get")
+        sn = short_name or ""
+        io_key = sn if sn in KNOWN_IO_FUNCTIONS else None
+        if io_key is None and full_name and full_name in KNOWN_IO_FUNCTIONS:
+            io_key = full_name
+        if io_key and not (
+            io_key in all_patches
+            or sn in all_patches
+            or any(io_key in p or sn in p for p in all_patches)
+            or any(p.endswith(f".{sn}") for p in all_patches)
+        ):
+            self.violations.append(
+                Violation(
+                    file=self.file_path,
+                    line=node.lineno,
+                    col=node.col_offset,
+                    category="unpatched-io",
+                    message=_(
+                        "{func} called in test '{test}' without @patch — "
+                        'this function {desc}. Add @patch("<module>.{func}").',
+                        func=io_key,
+                        test=self._current_function.name,
+                        desc=KNOWN_IO_FUNCTIONS[io_key],
+                    ),
+                )
+            )
+
         self.generic_visit(node)
 
     def visit_For(self, node: ast.For) -> None:
@@ -334,7 +401,13 @@ def find_test_files(test_path: Path) -> list[Path]:
 
 
 def analyze_file(file_path: Path, max_loop_iterations: int = DEFAULT_MAX_LOOP_ITERATIONS) -> list[Violation]:
-    """Analyze a single test file for isolation violations."""
+    """Analyze a single test file for isolation violations.
+
+    Files in ``integration/`` directories are skipped — integration tests
+    intentionally do real I/O (subprocess, network, filesystem).
+    """
+    if "integration" in file_path.parts:
+        return []
     try:
         source = file_path.read_text()
         tree = ast.parse(source, filename=str(file_path))

tests/unit/test_check_test_isolation.py

@@ -566,6 +566,147 @@ class TestKnownHelpers:
         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."""