repro_libcst_matcher_crash.py

#!/usr/bin/env python3

# Reproduction case for https://github.com/Instagram/LibCST/issues/848

import contextlib
import os
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Generator

import libcst
from libcst import BaseExpression, Arg, Call, Name, matchers as m
from libcst.codemod import (
    CodemodContext,
    parallel_exec_transform_with_prettyprint,
    VisitorBasedCodemodCommand,
)

class TestIssueCommand(VisitorBasedCodemodCommand):
    # NOTE: The failures induced by these methods are mutually exclusive.  To see
    # one, you may need to comment out the other.

    # This one seems to only be a problem when both `call_if_inside` and `leave`
    # are used.
    #
    # Stack traces look something like:
    #
    #   File "/Users/apsaltis/workspace-git/libcst/libcst/codemod/_cli.py", line 279, in _execute_transform
    #     output_tree = transformer.transform_module(input_tree)
    #                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    #   File "/Users/apsaltis/workspace-git/libcst/libcst/codemod/_command.py", line 71, in transform_module
    #     tree = super().transform_module(tree)
    #            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    #   File "/Users/apsaltis/workspace-git/libcst/libcst/codemod/_codemod.py", line 108, in transform_module
    #     return self.transform_module_impl(tree_with_metadata)
    #            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    #   File "/Users/apsaltis/workspace-git/libcst/libcst/codemod/_visitor.py", line 32, in transform_module_impl
    #     return tree.visit(self)
    #            ^^^^^^^^^^^^^^^^
    #   File "/Users/apsaltis/workspace-git/libcst/libcst/_nodes/module.py", line 90, in visit
    #     result = super(Module, self).visit(visitor)
    #              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

    ####
    #### A bunch more frames...
    ####

    #   File "/Users/apsaltis/workspace-git/libcst/libcst/_nodes/internal.py", line 177, in visit_sequence
    #     return tuple(visit_iterable(parent, fieldname, children, visitor))
    #            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    #   File "/Users/apsaltis/workspace-git/libcst/libcst/_nodes/internal.py", line 159, in visit_iterable
    #     new_child = child.visit(visitor)
    #                 ^^^^^^^^^^^^^^^^^^^^
    #   File "/Users/apsaltis/workspace-git/libcst/libcst/_nodes/base.py", line 237, in visit
    #     leave_result = visitor.on_leave(self, with_updated_children)
    #                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    #   File "/Users/apsaltis/workspace-git/libcst/libcst/matchers/_visitors.py", line 537, in on_leave
    #     if _should_allow_visit(self._matchers, leave_func) and isinstance(
    #        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    #   File "/Users/apsaltis/workspace-git/libcst/libcst/matchers/_visitors.py", line 436, in _should_allow_visit
    #     return _all_positive_matchers_true(
    #            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    #   File "/Users/apsaltis/workspace-git/libcst/libcst/matchers/_visitors.py", line 415, in _all_positive_matchers_true
    #     if all_matchers[matcher] is None:
    #        ~~~~~~~~~~~~^^^^^^^^^
    # KeyError: Call(func=Name(value='foo', lpar=DoNotCare(), rpar=DoNotCare(), metadata=DoNotCare()), args=DoNotCare(), lpar=DoNotCare(), rpar=DoNotCare(), whitespace_after_func=DoNotCare(), whitespace_before_args=DoNotCare(), metadata=DoNotCare())
    @m.call_if_inside(m.Call(func = m.Name("foo")))
    @m.leave(m.Arg(keyword = m.Name("end")))
    def fail_keyerror(self, original_node: Arg, updated_node: Arg):
        return updated_node
    
    # It seems that the `~` (DoesNotMatch) causes is crash here -- the crash
    # does not occur when it is removed.
    #
    # Stack traces look something like:
    #
    # Traceback (most recent call last):
    #   File "/opt/local/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/multiprocessing/process.py", line 314, in _bootstrap
    #     self.run()
    #   File "/opt/local/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/multiprocessing/process.py", line 108, in run
    #     self._target(*self._args, **self._kwargs)
    #   File "/opt/local/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/multiprocessing/pool.py", line 114, in worker
    #     task = get()
    #            ^^^^^
    #   File "/opt/local/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/multiprocessing/queues.py", line 367, in get
    #     return _ForkingPickler.loads(res)
    #            ^^^^^^^^^^^^^^^^^^^^^^^^^^
    #   File "/Users/$USER/workspace-git/libcst/libcst/matchers/_matcher_base.py", line 386, in __getattr__
    #     return getattr(self._matcher, key)
    #                    ^^^^^^^^^^^^^
    #   File "/Users/$USER/workspace-git/libcst/libcst/matchers/_matcher_base.py", line 386, in __getattr__
    #     return getattr(self._matcher, key)
    #                    ^^^^^^^^^^^^^
    #   File "/Users/$USER/workspace-git/libcst/libcst/matchers/_matcher_base.py", line 386, in __getattr__
    #     return getattr(self._matcher, key)
    #                    ^^^^^^^^^^^^^
    #   [Previous line repeated 989 more times]
    # RecursionError: maximum recursion depth exceeded
    # @m.leave(
    #     m.Call(
    #         func=m.Name("print"),
    #         args=[m.ZeroOrMore(~m.Arg(value=m.SimpleString('"some argument"')))],
    #     )
    # )
    # def fail_infinite_recursion(self, original_node: Call, updated_node: Call):
    #     return updated_node.with_changes(func=Name("pprint"))

@contextlib.contextmanager
def temp_workspace() -> Generator[Path, None, None]:
    cwd = os.getcwd()
    with TemporaryDirectory() as temp_dir:
        try:
            ws = Path(temp_dir).resolve()
            os.chdir(ws)
            yield ws
        finally:
            os.chdir(cwd)

if __name__ == "__main__":
    with temp_workspace() as tmp:
        files = []
        # Five files seems to be the magic number for failures
        for i in range(0, 5):
            example: Path = tmp / ("example" + str(i))
            example.write_text("""def foo(): print("some argument", end="\\n")""")
            files.append(example)

        # Run command
        command_instance = TestIssueCommand(CodemodContext())
        result = parallel_exec_transform_with_prettyprint(
            command_instance,
            files,
            hide_progress=True,
            # This all Works fine if jobs = 1
            #jobs = 1,
        )

        print(result)