Moved here utbot_python_runner

gradle.properties

@@ -92,6 +92,9 @@ commonsIOVersion=2.11.0
 javaxVersion=2.2
 jakartaVersion=3.1.0
 
+# Python support
+utbotMypyRunnerVersion=0.2.11
+
 # use latest Java 8 compaitable Spring and Spring Boot versions
 springVersion=5.3.28
 springBootVersion=2.7.13

utbot-python-types/.gitignore

@@ -0,0 +1,3 @@
+gradle.properties
+local_mypy_path
+dist/

utbot-python-types/build.gradle.kts

@@ -5,4 +5,34 @@ dependencies {
     implementation("com.squareup.moshi:moshi-kotlin:1.11.0")
     implementation("com.squareup.moshi:moshi-adapters:1.11.0")
     implementation(group = "io.github.microutils", name = "kotlin-logging", version = kotlinLoggingVersion)
+}
+
+val utbotMypyRunnerVersion = File(project.projectDir, "src/main/resources/utbot_mypy_runner_version").readText()
+val pipToken: String? by project
+val pythonInterpreter: String? by project
+val utbotMypyRunnerPath = File(project.projectDir, "src/main/python/utbot_mypy_runner")
+val localMypyPath = File(utbotMypyRunnerPath, "dist")
+val localMypyPathText = File(project.projectDir, "src/main/resources/local_mypy_path")
+
+
+val setMypyRunnerVersion = tasks.register<Exec>("setVersion") {
+    group = "python"
+    workingDir = utbotMypyRunnerPath
+    commandLine(pythonInterpreter!!, "-m", "poetry", "version", utbotMypyRunnerVersion)
+}
+
+val buildMypyRunner = tasks.register<Exec>("buildUtbotMypyRunner") {
+    dependsOn(setMypyRunnerVersion)
+    group = "python"
+    workingDir = utbotMypyRunnerPath
+    commandLine(pythonInterpreter!!, "-m", "poetry", "build")
+    localMypyPathText.writeText(localMypyPath.canonicalPath)
+    localMypyPathText.createNewFile()
+}
+
+tasks.register<Exec>("publishUtbotMypyRunner") {
+    dependsOn(buildMypyRunner)
+    group = "python"
+    workingDir = utbotMypyRunnerPath
+    commandLine(pythonInterpreter!!, "-m", "poetry", "publish", "-u", "__token__", "-p", pipToken!!)
 }

utbot-python-types/src/main/kotlin/org/utbot/python/utils/ProcessUtils.kt

@@ -11,7 +11,15 @@ data class CmdResult(
     val terminatedByTimeout: Boolean = false
 )
 
-fun startProcess(command: List<String>): Process = ProcessBuilder(command).start()
+fun startProcess(
+    command: List<String>,
+    environmentVariables: Map<String, String> = emptyMap()
+): Process {
+    val pb = ProcessBuilder(command)
+    val env = pb.environment()
+    env += environmentVariables
+    return pb.start()
+}
 
 fun getResult(process: Process, timeout: Long? = null): CmdResult {
     if (timeout != null) {
@@ -36,7 +44,7 @@ fun getResult(process: Process, timeout: Long? = null): CmdResult {
     return CmdResult(stdout.trimIndent(), stderr, process.exitValue())
 }
 
-fun runCommand(command: List<String>, timeout: Long? = null): CmdResult {
-    val process = startProcess(command)
+fun runCommand(command: List<String>, timeout: Long? = null, environmentVariables: Map<String, String> = emptyMap()): CmdResult {
+    val process = startProcess(command, environmentVariables)
     return getResult(process, timeout)
 }

utbot-python-types/src/main/python/utbot_mypy_runner/pyproject.toml

@@ -0,0 +1,16 @@
+[tool.poetry]
+name = "utbot_mypy_runner"
+version = "0.2.12.dev1"
+description = ""
+authors = ["Ekaterina Tochilina <[email protected]>"]
+readme = "README.md"
+packages = [{include = "utbot_mypy_runner"}]
+
+[tool.poetry.dependencies]
+python = "^3.8"
+mypy = "1.0.0"
+
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

utbot-python-types/src/main/python/utbot_mypy_runner/utbot_mypy_runner/__main__.py

@@ -0,0 +1,54 @@
+import argparse
+import os
+
+import utbot_mypy_runner.mypy_main as mypy_main
+import utbot_mypy_runner.extract_annotations as extraction
+
+
+parser = argparse.ArgumentParser()
+parser.add_argument('--config', required=True)
+parser.add_argument('--sources', required=True, nargs='+')
+parser.add_argument('--modules', required=True, nargs='+')
+parser.add_argument('--annotations_out')
+parser.add_argument('--mypy_stdout')
+parser.add_argument('--mypy_stderr')
+parser.add_argument('--mypy_exit_status')
+parser.add_argument('--module_for_types')
+parser.add_argument('--indent', type=int)
+
+args = parser.parse_args()
+
+if len(args.sources) != len(args.modules):
+    print("Sources must correspond to modules")
+    exit(10)
+
+mypy_args = ["--config-file", args.config]
+for module_name in args.modules:
+    mypy_args += ["-m", module_name]
+
+stdout, stderr, exit_status, build_result = mypy_main.run(mypy_args)
+
+if args.mypy_stdout is not None:
+    with open(args.mypy_stdout, "w") as file:
+        file.write(stdout)
+    print("Wrote mypy stdout to", args.mypy_stdout)
+
+if args.mypy_stderr is not None:
+    with open(args.mypy_stderr, "w") as file:
+        file.write(stderr)
+    print("Wrote mypy stderr to", args.mypy_stderr)
+
+if args.mypy_exit_status is not None:
+    with open(args.mypy_exit_status, "w") as file:
+        file.write(str(exit_status))
+    print("Wrote mypy exit status to", args.mypy_exit_status)
+
+if args.annotations_out is not None:
+    if build_result is not None:
+        with open(args.annotations_out, "w") as file:
+            sources = [os.path.abspath(x) for x in args.sources]
+            file.write(extraction.get_result_from_mypy_build(build_result, sources, args.module_for_types, args.indent))
+        print("Extracted annotations and wrote to", args.annotations_out)
+    else:
+        print("For some reason BuildResult is None")
+        exit(11)

utbot-python-types/src/main/python/utbot_mypy_runner/utbot_mypy_runner/expression_traverser.py

@@ -0,0 +1,130 @@
+import typing as tp
+
+from mypy.nodes import *
+from mypy.traverser import *
+import mypy.types
+
+
+class MyTraverserVisitor(TraverserVisitor):
+    def __init__(self, types, processor: tp.Callable[[int, int, int, int, mypy.types.Type], None]):
+        self.types = types
+        self.processor = processor
+
+    def process_expression(self, o: Expression) -> None:
+        if o in self.types.keys() and not isinstance(self.types[o], mypy.types.AnyType) \
+            and o.end_line is not None and o.end_column is not None and o.line >= 0:
+            self.processor(o.line, o.column, o.end_line, o.end_column, self.types[o])
+
+    def visit_name_expr(self, o: NameExpr) -> None:
+        self.process_expression(o)
+        super().visit_name_expr(o)
+
+    def visit_member_expr(self, o: MemberExpr) -> None:
+        self.process_expression(o)
+        super().visit_member_expr(o)
+
+"""
+    def visit_yield_expr(self, o: YieldExpr) -> None:
+        self.process_expression(o)
+        super().visit_yield_expr(o)
+
+    def visit_call_expr(self, o: CallExpr) -> None:
+        self.process_expression(o)
+        super().visit_call_expr(o)
+
+    def visit_op_expr(self, o: OpExpr) -> None:
+        self.process_expression(o)
+        super().visit_op_expr(o)
+
+    def visit_comparison_expr(self, o: ComparisonExpr) -> None:
+        self.process_expression(o)
+        super().visit_comparison_expr(o)
+
+    def visit_slice_expr(self, o: SliceExpr) -> None:
+        self.process_expression(o)
+        super().visit_slice_expr(o)
+
+    def visit_cast_expr(self, o: CastExpr) -> None:
+        self.process_expression(o)
+        super().visit_cast_expr(o)
+
+    def visit_assert_type_expr(self, o: AssertTypeExpr) -> None:
+        self.process_expression(o)
+        super().visit_assert_type_expr(o)
+
+    def visit_reveal_expr(self, o: RevealExpr) -> None:
+        self.process_expression(o)
+        super().visit_reveal_expr(o)
+
+    def visit_assignment_expr(self, o: AssignmentExpr) -> None:
+        self.process_expression(o)
+        super().visit_assignment_expr(o)
+
+    def visit_unary_expr(self, o: UnaryExpr) -> None:
+        self.process_expression(o)
+        super().visit_unary_expr(o)
+
+    def visit_list_expr(self, o: ListExpr) -> None:
+        self.process_expression(o)
+        super().visit_list_expr(o)
+
+    def visit_tuple_expr(self, o: TupleExpr) -> None:
+        self.process_expression(o)
+        super().visit_tuple_expr(o)
+
+    def visit_dict_expr(self, o: DictExpr) -> None:
+        self.process_expression(o)
+        super().visit_dict_expr(o)
+
+    def visit_set_expr(self, o: SetExpr) -> None:
+        self.process_expression(o)
+        super().visit_set_expr(o)
+
+    def visit_index_expr(self, o: IndexExpr) -> None:
+        self.process_expression(o)
+        super().visit_index_expr(o)
+
+    def visit_generator_expr(self, o: GeneratorExpr) -> None:
+        self.process_expression(o)
+        super().visit_generator_expr(o)
+
+    def visit_dictionary_comprehension(self, o: DictionaryComprehension) -> None:
+        self.process_expression(o)
+        super().visit_dictionary_comprehension(o)
+
+    def visit_list_comprehension(self, o: ListComprehension) -> None:
+        self.process_expression(o)
+        super().visit_list_comprehension(o)
+
+    def visit_set_comprehension(self, o: SetComprehension) -> None:
+        self.process_expression(o)
+        super().visit_set_comprehension(o)
+
+    def visit_conditional_expr(self, o: ConditionalExpr) -> None: 
+        self.process_expression(o)
+        super().visit_conditional_expr(o)
+
+    def visit_type_application(self, o: TypeApplication) -> None:
+        self.process_expression(o)
+        super().visit_type_application(o)
+
+    def visit_lambda_expr(self, o: LambdaExpr) -> None:
+        self.process_expression(o)
+        super().visit_lambda_expr(o)
+
+    def visit_star_expr(self, o: StarExpr) -> None:
+        self.process_expression(o)
+        super().visit_star_expr(o)
+
+    def visit_backquote_expr(self, o: BackquoteExpr) -> None:
+        self.process_expression(o)
+        super().visit_backquote_expr(o)
+
+    def visit_await_expr(self, o: AwaitExpr) -> None:
+        self.process_expression(o)
+        super().visit_await_expr(o)
+
+    def visit_super_expr(self, o: SuperExpr) -> None:
+        self.process_expression(o)
+        super().visit_super_expr(o)
+"""

utbot-python-types/src/main/python/utbot_mypy_runner/utbot_mypy_runner/extract_annotations.py

@@ -0,0 +1,100 @@
+import json
+import typing as tp
+from collections import defaultdict
+
+import mypy.nodes
+import mypy.types
+
+import utbot_mypy_runner.mypy_main as mypy_main
+import utbot_mypy_runner.expression_traverser as expression_traverser
+import utbot_mypy_runner.names
+from utbot_mypy_runner.utils import get_borders
+from utbot_mypy_runner.nodes import *
+
+
+class ExpressionType:
+    def __init__(self, start_offset: int, end_offset: int, line: int, type_: Annotation):
+        self.start_offset = start_offset
+        self.end_offset = end_offset
+        self.line = line
+        self.type_ = type_
+
+    def encode(self):
+        return {
+            "startOffset": self.start_offset, 
+            "endOffset": self.end_offset, 
+            "line": self.line,
+            "type": self.type_.encode()
+        }
+
+
+def get_output_json(annotations: tp.Dict[str, tp.Dict[str, Definition]], 
+                    expression_types: tp.Dict[str, tp.List[ExpressionType]],
+                    names_dict: tp.Dict[str, tp.List[utbot_mypy_runner.names.Name]],
+                    indent: tp.Optional[int]):
+    node_storage_key = 'nodeStorage'
+    types_key = 'types'
+    definitions_key = 'definitions'
+    names_key = 'names'
+
+    result: tp.Dict[str, tp.Any] = {node_storage_key: {}, types_key: {}}
+    for key in annotation_node_dict:
+        result[node_storage_key][str(key)] = annotation_node_dict[key].encode()
+
+    result[definitions_key] = {}
+    for module in annotations.keys():
+        result[definitions_key][module] = {}
+        for name in annotations[module].keys():
+            result[definitions_key][module][name] = annotations[module][name].encode()
+
+    for module in expression_types.keys():
+        result[types_key][module] = [x.encode() for x in expression_types[module]]
+
+    result[names_key] = {}
+    for module in names_dict.keys():
+        result[names_key][module] = [x.encode() for x in names_dict[module]]
+
+    return json.dumps(result, indent=indent)
+
+
+def skip_node(node: mypy.nodes.SymbolTableNode) -> bool:
+
+    if isinstance(node.node, mypy.nodes.TypeInfo):
+        x = node.node
+        return x.is_named_tuple or (x.typeddict_type is not None) or x.is_newtype or x.is_intersection
+
+    return False
+
+
+def get_result_from_mypy_build(build_result: mypy_main.build.BuildResult, source_paths: tp.List[str],
+                               module_for_types: tp.Optional[str], indent=None) -> str:
+    annotation_dict: tp.Dict[str, tp.Dict[str, Definition]] = defaultdict(dict)
+    names_dict: tp.Dict[str, tp.List[utbot_mypy_runner.names.Name]] = utbot_mypy_runner.names.get_names(build_result)
+    for module in build_result.files.keys():
+        mypy_file: mypy.nodes.MypyFile = build_result.files[module]
+
+        for name in mypy_file.names.keys():
+            symbol_table_node = build_result.files[module].names[name]
+
+            if skip_node(symbol_table_node):
+                continue
+
+            only_types = mypy_file.path not in source_paths
+
+            definition = get_definition_from_symbol_node(symbol_table_node, Meta(module), only_types)
+            if definition is not None:
+                annotation_dict[module][name] = definition
+
+    expression_types: tp.Dict[str, tp.List[ExpressionType]] = defaultdict(list)
+    if module_for_types is not None:
+        mypy_file = build_result.files[module_for_types]
+        with open(mypy_file.path, "r") as file:
+            content = file.readlines()
+            processor = lambda line, col, end_line, end_col, type_: \
+                    expression_types[module_for_types].append( # TODO: proper Meta
+                        ExpressionType(*get_borders(line, col, end_line, end_col, content), line, get_annotation(type_, Meta(module_for_types)))
+                    )
+            traverser = expression_traverser.MyTraverserVisitor(build_result.types, processor)
+            traverser.visit_mypy_file(build_result.files[module_for_types])
+
+    return get_output_json(annotation_dict, expression_types, names_dict, indent)