mkdir codemod fixes (#586)

* tmpfile codemod able to report unfixed

* refactor

* use original leading line and test in func scope

* lower cov to allow for one py version

* ensure class attrs are instance attrs

* clear mktmp set

Author: clavedeluna

pyproject.toml

@@ -114,7 +114,7 @@ extend-exclude = '''
 '''
 
 [coverage-threshold]
-line_coverage_min = 93.5
+line_coverage_min = 93
 [coverage-threshold.modules."src/core_codemods/"]
 # Detect if a codemod is missing unit or integration tests
 file_line_coverage_min = 50

src/codemodder/context.py

@@ -35,7 +35,6 @@
 
 
 class CodemodExecutionContext:
-    _results_by_codemod: dict[str, list[ChangeSet]] = {}
     _failures_by_codemod: dict[str, list[Path]] = {}
     _dependency_update_by_codemod: dict[str, PackageStore | None] = {}
     _unfixed_findings_by_codemod: dict[str, list[UnfixedFinding]] = {}
@@ -69,6 +68,8 @@ def __init__(
         self.verbose = verbose
         self._changesets_by_codemod: dict[str, list[ChangeSet]] = {}
         self._failures_by_codemod = {}
+        self._dependency_update_by_codemod = {}
+        self._unfixed_findings_by_codemod = {}
         self.dependencies = {}
         self.registry = registry
         self.repo_manager = repo_manager

src/core_codemods/tempfile_mktemp.py

@@ -3,12 +3,15 @@
 
 import libcst as cst
 from libcst import matchers
+from libcst.codemod import CodemodContext
 
 from codemodder.codemods.libcst_transformer import (
     LibcstResultTransformer,
     LibcstTransformerPipeline,
 )
 from codemodder.codemods.utils_mixin import NameAndAncestorResolutionMixin
+from codemodder.file_context import FileContext
+from codemodder.result import Result, same_line
 from codemodder.utils.utils import clean_simplestring
 from core_codemods.api import CoreCodemod, Metadata, Reference, ReviewGuidance
 
@@ -19,41 +22,83 @@ class TempfileMktempTransformer(
     change_description = "Replaces `tempfile.mktemp` with `tempfile.mkstemp`."
     _module_name = "tempfile"
 
-    def leave_SimpleStatementLine(self, original_node, updated_node):
+    def __init__(
+        self,
+        context: CodemodContext,
+        results: list[Result] | None,
+        file_context: FileContext,
+        _transformer: bool = False,
+    ):
+        self.mktemp_calls: set[cst.Call] = set()
+        super().__init__(context, results, file_context, _transformer)
+
+    def visit_Call(self, node: cst.Call) -> None:
+        if self._is_mktemp_call(node):
+            self.mktemp_calls.add(node)
+
+    def leave_SimpleStatementLine(
+        self,
+        original_node: cst.SimpleStatementLine,
+        updated_node: cst.SimpleStatementLine,
+    ) -> cst.SimpleStatementLine | cst.FlattenSentinel:
+        if not self.node_is_selected(original_node):
+            return updated_node
+
         match original_node:
             case cst.SimpleStatementLine(body=[bsstmt]):
-                return self.check_mktemp(original_node, bsstmt)
+                if maybe_tuple := self._is_assigned_to_mktemp(bsstmt):
+                    assign_name, call = maybe_tuple
+                    return self.report_and_change(
+                        call, assign_name, original_node.leading_lines
+                    )
+                if maybe_tuple := self._mktemp_is_sink(bsstmt):
+                    wrapper_func_name, call = maybe_tuple
+                    return self.report_and_change(
+                        call,
+                        wrapper_func_name,
+                        original_node.leading_lines,
+                        assignment=False,
+                    )
+
+        # If we get here it's because there is a mktemp call but we haven't fixed it yet.
+        for unfixed in self.mktemp_calls:
+            self.report_unfixed(unfixed, reason="Pixee does not yet support this fix.")
+        self.mktemp_calls.clear()
         return updated_node
 
-    def check_mktemp(
-        self, original_node: cst.SimpleStatementLine, bsstmt: cst.BaseSmallStatement
-    ) -> cst.SimpleStatementLine | cst.FlattenSentinel:
-        if maybe_tuple := self._is_assigned_to_mktemp(bsstmt):  # type: ignore
-            assign_name, call = maybe_tuple
-            return self.report_and_change(call, assign_name)
-        if maybe_tuple := self._mktemp_is_sink(bsstmt):
-            wrapper_func_name, call = maybe_tuple
-            return self.report_and_change(call, wrapper_func_name, assignment=False)
-        return original_node
+    def filter_by_result(self, node) -> bool:
+        match node:
+            case cst.SimpleStatementLine():
+                pos_to_match = self.node_position(node)
+                return self.results is None or any(
+                    self.match_location(pos_to_match, result)
+                    for result in self.results or []
+                )
+        return False
+
+    def match_location(self, pos, result):
+        return any(same_line(pos, location) for location in result.locations)
 
     def report_and_change(
-        self, node: cst.Call, name: cst.Name, assignment=True
+        self, node: cst.Call, name: cst.Name, leading_lines: tuple, assignment=True
     ) -> cst.FlattenSentinel:
+        self.mktemp_calls.clear()
         self.report_change(node)
         self.add_needed_import(self._module_name)
         self.remove_unused_import(node)
         with_block = (
             f"{name.value} = tf.name" if assignment else f"{name.value}(tf.name)"
         )
         new_stmt = dedent(
-            f"""
+            f"""\
         with tempfile.NamedTemporaryFile({self._make_args(node)}) as tf:
             {with_block}
         """
         ).rstrip()
+
         return cst.FlattenSentinel(
             [
-                cst.parse_statement(new_stmt),
+                cst.parse_statement(new_stmt).with_changes(leading_lines=leading_lines),
             ]
         )
 

tests/codemods/sonar/test_sonar_tempfile_mktemp.py

@@ -39,3 +39,43 @@ def test_simple(self, tmpdir):
             ]
         }
         self.run_and_assert(tmpdir, input_code, expected, results=json.dumps(issues))
+        unfixed = self.execution_context.get_unfixed_findings(self.codemod.id)
+        assert len(unfixed) == 0
+
+    def test_unfixed(self, tmpdir):
+        RULE_ID = "python:S5445"
+
+        input_code = (
+            expected
+        ) = """
+        import tempfile
+
+        tmp_file = open(tempfile.mktemp(), "w+")
+        tmp_file.write("text")
+        """
+
+        issues = {
+            "issues": [
+                {
+                    "rule": RULE_ID,
+                    "status": "OPEN",
+                    "component": "code.py",
+                    "textRange": {
+                        "startLine": 4,
+                        "endLine": 4,
+                        "startOffset": 15,
+                        "endOffset": 33,
+                    },
+                }
+            ]
+        }
+
+        self.run_and_assert(tmpdir, input_code, expected, results=json.dumps(issues))
+
+        unfixed = self.execution_context.get_unfixed_findings(self.codemod.id)
+        assert len(unfixed) == 1
+        assert unfixed[0].id == RULE_ID
+        assert unfixed[0].rule.id == RULE_ID
+        assert unfixed[0].lineNumber == 4
+        assert unfixed[0].reason == "Pixee does not yet support this fix."
+        assert unfixed[0].path == "code.py"

tests/codemods/test_tempfile_mktemp.py

@@ -48,7 +48,6 @@ def test_as_sink(self, tmpdir):
         with tempfile.NamedTemporaryFile(delete=False) as tf:
             print(tf.name)
         var = "hello"
-
         with tempfile.NamedTemporaryFile(delete=False) as tf:
             bool(tf.name)
         """
@@ -70,16 +69,12 @@ def test_import_with_arg(self, tmpdir):
 
         with tempfile.NamedTemporaryFile(suffix="suffix", delete=False) as tf:
             filename = tf.name
-        
         with tempfile.NamedTemporaryFile(suffix="suffix", prefix="prefix", delete=False) as tf:
             print(tf.name)
-        
         with tempfile.NamedTemporaryFile(suffix="suffix", prefix="prefix", dir="dir", delete=False) as tf:
             filename = tf.name
-        
         with tempfile.NamedTemporaryFile(suffix="suffix", prefix="prefix", delete=False) as tf:
             filename = tf.name
-        
         with tempfile.NamedTemporaryFile(suffix="suffix", prefix="prefix", dir="dir", delete=False) as tf:
             filename = tf.name
         var = "hello"
@@ -166,6 +161,37 @@ def test_open_and_write_no_change(self, tmpdir):
         """
         self.run_and_assert(tmpdir, input_code, input_code)
 
+    def test_exclude_line(self, tmpdir):
+        input_code = (
+            expected
+        ) = """
+        import tempfile
+        name = tempfile.mktemp()
+        """
+        lines_to_exclude = [3]
+        self.run_and_assert(
+            tmpdir,
+            input_code,
+            expected,
+            lines_to_exclude=lines_to_exclude,
+        )
+
+    def test_in_func_scope(self, tmpdir):
+        input_code = """
+        import tempfile
+
+        def make_file():
+            filename = tempfile.mktemp()
+        """
+        expected_output = """
+        import tempfile
+
+        def make_file():
+            with tempfile.NamedTemporaryFile(delete=False) as tf:
+                filename = tf.name
+        """
+        self.run_and_assert(tmpdir, input_code, expected_output)
+
     @pytest.mark.xfail(reason="Not currently supported")
     def test_as_str_concat(self, tmpdir):
         input_code = """