Private
Public Access
ed
3c839c910a
Phase 11.2. Adds the LEGITIMATE heuristic that recognizes the canonical data-oriented pattern: \ ry: ...; except: return Result(data=..., errors=[...])\ is the convention's canonical recovery pattern. Detection: - New _returns_result(stmts) helper on ExceptionVisitor - New step 0 in _classify_except (BEFORE BOUNDARY_CONVERSION check) - Classifies as INTERNAL_COMPLIANT with a hint that names the pattern The function-name-not-ending-in-_result is documented as a smell (rename to xxx_result for canonical naming), but the pattern itself is compliant. Tests: - 2 new tests in test_audit_exception_handling_heuristics.py: - test_result_returning_recovery_in_non_result_named_function_is_compliant - test_result_returning_recovery_in_result_named_function_is_compliant - Both pass; the 2 REJECTED tests (#22, #23) remain xfailed. Per conductor/tracks/result_migration_small_files_20260617/plan.md section 11.2.
398 lines
16 KiB
Python
398 lines
16 KiB
Python
"""Tests for the new heuristics added in scripts/audit_exception_handling.py.
|
|
|
|
Each test creates a fixture file with a specific pattern and runs the audit
|
|
script against it, verifying the pattern is classified correctly.
|
|
|
|
The new heuristics (added by result_migration_review_pass_20260617):
|
|
1. list.index with ValueError fallback to default index (INTERNAL_COMPLIANT)
|
|
2. dict lookup (KeyError) with default value (INTERNAL_COMPLIANT)
|
|
3. datetime.fromisoformat(s) with ValueError: None (INTERNAL_COMPLIANT)
|
|
4. Path.resolve(strict=True) with OSError/ValueError fallback (INTERNAL_COMPLIANT)
|
|
5. Path.relative_to with ValueError: pass (INTERNAL_COMPLIANT)
|
|
6. asyncio.get_running_loop() with RuntimeError: asyncio.run() (INTERNAL_COMPLIANT)
|
|
7. Narrow except (ImportError, AttributeError) + fallback stub (INTERNAL_COMPLIANT)
|
|
8. raise NotImplementedError() as entire function body (INTERNAL_PROGRAMMER_RAISE)
|
|
9. if None: raise ImportError() validation pattern (INTERNAL_PROGRAMMER_RAISE)
|
|
10. try/except (json.JSONDecodeError, KeyError) around JSON parse + print + return (INTERNAL_COMPLIANT)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import textwrap
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SCRIPT = ROOT / "scripts" / "audit_exception_handling.py"
|
|
|
|
|
|
def _run_audit_on_fixture(source: str) -> dict:
|
|
"""Run the audit script on a fixture and return the parsed JSON output."""
|
|
# Use a temp dir outside tests/artifacts/ to avoid the audit's
|
|
# default artifacts-exclusion filter.
|
|
import tempfile
|
|
tmpdir = Path(tempfile.mkdtemp(prefix="audit_fixture_"))
|
|
fixture = tmpdir / "audit_heuristic_fixture.py"
|
|
fixture.write_text(textwrap.dedent(source), encoding="utf-8")
|
|
try:
|
|
result = subprocess.run(
|
|
[sys.executable, str(SCRIPT), "--json", "--src", str(tmpdir), "--verbose"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
cwd=str(ROOT),
|
|
)
|
|
finally:
|
|
if fixture.exists():
|
|
fixture.unlink()
|
|
if tmpdir.exists():
|
|
tmpdir.rmdir()
|
|
if result.returncode not in (0, 1):
|
|
raise RuntimeError(f"audit failed: {result.stderr}")
|
|
return json.loads(result.stdout)
|
|
|
|
|
|
def _classifications_for_file(data: dict, filename_suffix: str) -> list[dict]:
|
|
"""Return all findings for files whose path ends with `filename_suffix`."""
|
|
return [
|
|
f
|
|
for file_info in data.get("files", [])
|
|
for f in file_info.get("findings", [])
|
|
if file_info["filename"].endswith(filename_suffix)
|
|
]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Heuristic 1: list.index with ValueError fallback to default index
|
|
# ---------------------------------------------------------------------------
|
|
def test_list_index_valueerror_fallback_is_compliant():
|
|
"""try: list.index(x); except ValueError: idx = N is compliant."""
|
|
src = '''
|
|
def func(items, target):
|
|
try:
|
|
idx = items.index(target)
|
|
except ValueError:
|
|
idx = 0
|
|
return idx
|
|
'''
|
|
data = _run_audit_on_fixture(src)
|
|
findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
|
|
excepts = [f for f in findings if f["kind"] == "EXCEPT"]
|
|
assert len(excepts) == 1, f"expected 1 except, got {len(excepts)}: {excepts}"
|
|
assert excepts[0]["category"] == "INTERNAL_COMPLIANT", (
|
|
f"list.index+ValueError fallback should be INTERNAL_COMPLIANT, got {excepts[0]['category']}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Heuristic 2: KeyError lookup with default value
|
|
# ---------------------------------------------------------------------------
|
|
def test_keyerror_lookup_with_default_is_compliant():
|
|
"""try: dict[x]; except KeyError: val = default is compliant."""
|
|
src = '''
|
|
def func(d, key):
|
|
try:
|
|
val = d[key]
|
|
except KeyError:
|
|
val = "default"
|
|
return val
|
|
'''
|
|
data = _run_audit_on_fixture(src)
|
|
findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
|
|
excepts = [f for f in findings if f["kind"] == "EXCEPT"]
|
|
assert len(excepts) == 1
|
|
assert excepts[0]["category"] == "INTERNAL_COMPLIANT", (
|
|
f"KeyError+default should be INTERNAL_COMPLIANT, got {excepts[0]['category']}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Heuristic 3: datetime.fromisoformat(s) with ValueError: None
|
|
# ---------------------------------------------------------------------------
|
|
def test_fromisoformat_valueerror_none_is_compliant():
|
|
"""try: datetime.fromisoformat(s); except ValueError: x = None is compliant."""
|
|
src = '''
|
|
import datetime
|
|
def func(s):
|
|
try:
|
|
d = datetime.datetime.fromisoformat(s)
|
|
except ValueError:
|
|
d = None
|
|
return d
|
|
'''
|
|
data = _run_audit_on_fixture(src)
|
|
findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
|
|
excepts = [f for f in findings if f["kind"] == "EXCEPT"]
|
|
assert len(excepts) == 1
|
|
assert excepts[0]["category"] == "INTERNAL_COMPLIANT", (
|
|
f"fromisoformat+ValueError:None should be INTERNAL_COMPLIANT, got {excepts[0]['category']}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Heuristic 4: Path.resolve(strict=True) with OSError/ValueError fallback
|
|
# ---------------------------------------------------------------------------
|
|
def test_path_resolve_strict_fallback_is_compliant():
|
|
"""try: Path(p).resolve(strict=True); except (OSError, ValueError): Path(p).resolve() is compliant."""
|
|
src = '''
|
|
from pathlib import Path
|
|
def func(p):
|
|
try:
|
|
rp = Path(p).resolve(strict=True)
|
|
except (OSError, ValueError):
|
|
rp = Path(p).resolve()
|
|
return rp
|
|
'''
|
|
data = _run_audit_on_fixture(src)
|
|
findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
|
|
excepts = [f for f in findings if f["kind"] == "EXCEPT"]
|
|
assert len(excepts) == 1
|
|
assert excepts[0]["category"] == "INTERNAL_COMPLIANT", (
|
|
f"Path.resolve(strict=True)+fallback should be INTERNAL_COMPLIANT, got {excepts[0]['category']}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Heuristic 5: Path.relative_to with ValueError: pass
|
|
# ---------------------------------------------------------------------------
|
|
def test_path_relative_to_valueerror_is_compliant():
|
|
"""try: rp.relative_to(base); except ValueError: pass is compliant (canonical subpath check)."""
|
|
src = '''
|
|
from pathlib import Path
|
|
def is_subpath(rp, base):
|
|
try:
|
|
rp.relative_to(base)
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
'''
|
|
data = _run_audit_on_fixture(src)
|
|
findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
|
|
excepts = [f for f in findings if f["kind"] == "EXCEPT"]
|
|
assert len(excepts) == 1
|
|
assert excepts[0]["category"] == "INTERNAL_COMPLIANT", (
|
|
f"Path.relative_to+ValueError should be INTERNAL_COMPLIANT, got {excepts[0]['category']}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Heuristic 6: asyncio.get_running_loop() with RuntimeError: asyncio.run()
|
|
# ---------------------------------------------------------------------------
|
|
def test_asyncio_get_running_loop_fallback_is_compliant():
|
|
"""try: get_running_loop(); except RuntimeError: asyncio.run() is compliant."""
|
|
src = '''
|
|
import asyncio
|
|
def bridge(coro):
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
return loop
|
|
except RuntimeError:
|
|
return asyncio.run(coro)
|
|
'''
|
|
data = _run_audit_on_fixture(src)
|
|
findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
|
|
excepts = [f for f in findings if f["kind"] == "EXCEPT"]
|
|
assert len(excepts) == 1
|
|
assert excepts[0]["category"] == "INTERNAL_COMPLIANT", (
|
|
f"get_running_loop+RuntimeError fallback should be INTERNAL_COMPLIANT, got {excepts[0]['category']}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Heuristic 7: narrow except (ImportError, AttributeError) + fallback stub
|
|
# ---------------------------------------------------------------------------
|
|
def test_import_attr_fallback_stub_is_compliant():
|
|
"""narrow except (ImportError, AttributeError) with fallback attribute assignment is compliant."""
|
|
src = '''
|
|
class Stub:
|
|
available = False
|
|
def get_thing():
|
|
try:
|
|
import real_module
|
|
return real_module
|
|
except (ImportError, ModuleNotFoundError, AttributeError):
|
|
return Stub()
|
|
'''
|
|
data = _run_audit_on_fixture(src)
|
|
findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
|
|
excepts = [f for f in findings if f["kind"] == "EXCEPT"]
|
|
assert len(excepts) == 1
|
|
assert excepts[0]["category"] == "INTERNAL_COMPLIANT", (
|
|
f"narrow except + fallback stub should be INTERNAL_COMPLIANT, got {excepts[0]['category']}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Heuristic 8: raise NotImplementedError() as entire function body
|
|
# ---------------------------------------------------------------------------
|
|
def test_raise_notimplemented_entire_body_is_programmer_error():
|
|
"""raise NotImplementedError() as the entire function body is INTERNAL_PROGRAMMER_RAISE."""
|
|
src = '''
|
|
class Base:
|
|
def abstract_method(self):
|
|
raise NotImplementedError()
|
|
'''
|
|
data = _run_audit_on_fixture(src)
|
|
findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
|
|
raises = [f for f in findings if f["kind"] == "RAISE"]
|
|
assert len(raises) == 1
|
|
assert raises[0]["category"] == "INTERNAL_PROGRAMMER_RAISE", (
|
|
f"raise NotImplementedError() in entire body should be INTERNAL_PROGRAMMER_RAISE, got {raises[0]['category']}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Heuristic 9: if None: raise ImportError() validation
|
|
# ---------------------------------------------------------------------------
|
|
def test_validation_raise_is_programmer_error():
|
|
"""if <var> is None: raise ImportError() is INTERNAL_PROGRAMMER_RAISE."""
|
|
src = '''
|
|
def func(dep):
|
|
if dep is None:
|
|
raise ImportError("dependency missing")
|
|
return dep
|
|
'''
|
|
data = _run_audit_on_fixture(src)
|
|
findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
|
|
raises = [f for f in findings if f["kind"] == "RAISE"]
|
|
assert len(raises) == 1
|
|
assert raises[0]["category"] == "INTERNAL_PROGRAMMER_RAISE", (
|
|
f"validation raise should be INTERNAL_PROGRAMMER_RAISE, got {raises[0]['category']}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Heuristic 10: try/except (json.JSONDecodeError, KeyError) + print + return
|
|
# ---------------------------------------------------------------------------
|
|
def test_json_parse_with_print_is_compliant():
|
|
"""try/except (json.JSONDecodeError, KeyError) around JSON parse with print is compliant."""
|
|
src = '''
|
|
import json
|
|
def parse(s):
|
|
try:
|
|
data = json.loads(s)
|
|
if not isinstance(data, list):
|
|
print("not a list")
|
|
return
|
|
return data
|
|
except json.JSONDecodeError as e:
|
|
print(f"json error: {e}")
|
|
except KeyError as e:
|
|
print(f"missing key: {e}")
|
|
'''
|
|
data = _run_audit_on_fixture(src)
|
|
findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
|
|
excepts = [f for f in findings if f["kind"] == "EXCEPT"]
|
|
assert len(excepts) == 2
|
|
for e in excepts:
|
|
assert e["category"] == "INTERNAL_COMPLIANT", (
|
|
f"json parse with print should be INTERNAL_COMPLIANT, got {e['category']}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Heuristic 22: Narrow except + return fallback value (REJECTED Phase 11)
|
|
# ---------------------------------------------------------------------------
|
|
@pytest.mark.xfail(reason="Heuristic #22 REVERTED in Phase 11 (laundering heuristic; full Result[T] migration required). See conductor/tracks/result_migration_small_files_20260617/plan.md §11.1.1.")
|
|
def test_narrow_except_returns_fallback_is_compliant():
|
|
"""REJECTED in Phase 11. Heuristic #22 classified narrow-catch + fallback as compliant, which is WRONG. The convention requires `Result[T]`; this test is preserved as xfail for traceability and to ensure the count of 11 test tiers is maintained."""
|
|
src = '''
|
|
def get_git_commit(git_dir):
|
|
try:
|
|
r = subprocess.run(["git", "rev-parse", "HEAD"], capture_output=True, text=True, cwd=git_dir, timeout=5)
|
|
return r.stdout.strip() if r.returncode == 0 else ""
|
|
except (OSError, subprocess.SubprocessError, subprocess.TimeoutExpired):
|
|
return ""
|
|
'''
|
|
data = _run_audit_on_fixture(src)
|
|
findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
|
|
excepts = [f for f in findings if f["kind"] == "EXCEPT"]
|
|
assert len(excepts) == 1
|
|
assert excepts[0]["category"] == "INTERNAL_COMPLIANT", (
|
|
f"narrow except returning fallback should be INTERNAL_COMPLIANT, got {excepts[0]['category']}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Heuristic 23: Narrow except + use error inline (REJECTED Phase 11)
|
|
# ---------------------------------------------------------------------------
|
|
@pytest.mark.xfail(reason="Heuristic #23 REVERTED in Phase 11 (laundering heuristic; full Result[T] migration required). See conductor/tracks/result_migration_small_files_20260617/plan.md §11.1.2.")
|
|
def test_narrow_except_uses_error_inline_is_compliant():
|
|
"""REJECTED in Phase 11. Heuristic #23 classified narrow-catch + use-error-inline as compliant, which is WRONG. The convention requires `Result[T]`; this test is preserved as xfail for traceability and to ensure the count of 11 test tiers is maintained."""
|
|
src = '''
|
|
def write_script(ps1_path, script):
|
|
try:
|
|
ps1_path.write_text(script, encoding="utf-8")
|
|
except (OSError, UnicodeEncodeError) as exc:
|
|
ps1_name = f"(write error: {exc})"
|
|
return ps1_name
|
|
'''
|
|
data = _run_audit_on_fixture(src)
|
|
findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
|
|
excepts = [f for f in findings if f["kind"] == "EXCEPT"]
|
|
assert len(excepts) == 1
|
|
assert excepts[0]["category"] == "INTERNAL_COMPLIANT", (
|
|
f"narrow except using error inline should be INTERNAL_COMPLIANT, got {excepts[0]['category']}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Heuristic A: Result-returning recovery in non-*_result function (Phase 11.2)
|
|
# ---------------------------------------------------------------------------
|
|
def test_result_returning_recovery_in_non_result_named_function_is_compliant():
|
|
"""try: ...; except SpecificError: return Result(data=..., errors=[ErrorInfo(...)]) is compliant.
|
|
|
|
The function returns a Result with errors= on failure (the canonical Result
|
|
recovery pattern). The convention requires Result[T] for try/except sites
|
|
that can fail; this pattern satisfies the requirement. The function name
|
|
not ending in '_result' is a smell (the function should be renamed to
|
|
'xxx_result') but the pattern itself is compliant.
|
|
This is the pattern used by src/hot_reloader.py:reload(),
|
|
src/warmup.py:on_complete/_record_success/_record_failure, and the
|
|
other 17 sites migrated in Phase 11.3.
|
|
"""
|
|
src = '''
|
|
from src.result_types import Result, ErrorInfo, ErrorKind
|
|
|
|
def reload(module_name):
|
|
try:
|
|
importlib.reload(sys.modules[module_name])
|
|
return Result(data=True)
|
|
except (ImportError, ModuleNotFoundError) as e:
|
|
return Result(data=False, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="hot_reloader.reload", original=e)])
|
|
'''
|
|
data = _run_audit_on_fixture(src)
|
|
findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
|
|
excepts = [f for f in findings if f["kind"] == "EXCEPT"]
|
|
assert len(excepts) == 1
|
|
assert excepts[0]["category"] == "INTERNAL_COMPLIANT", (
|
|
f"Result-returning recovery in non-*_result function should be INTERNAL_COMPLIANT, got {excepts[0]['category']}"
|
|
)
|
|
|
|
|
|
def test_result_returning_recovery_in_result_named_function_is_compliant():
|
|
"""Same pattern but with a function name ending in '_result' is also compliant (and ideal).
|
|
|
|
This is the canonical naming: functions that return Result should end in '_result'.
|
|
"""
|
|
src = '''
|
|
from src.result_types import Result, ErrorInfo, ErrorKind
|
|
|
|
def reload_result(module_name):
|
|
try:
|
|
importlib.reload(sys.modules[module_name])
|
|
return Result(data=True)
|
|
except (ImportError, ModuleNotFoundError) as e:
|
|
return Result(data=False, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="hot_reloader.reload_result", original=e)])
|
|
'''
|
|
data = _run_audit_on_fixture(src)
|
|
findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
|
|
excepts = [f for f in findings if f["kind"] == "EXCEPT"]
|
|
assert len(excepts) == 1
|
|
assert excepts[0]["category"] == "INTERNAL_COMPLIANT", (
|
|
f"Result-returning recovery in *_result function should be INTERNAL_COMPLIANT, got {excepts[0]['category']}"
|
|
)
|