Coverage for src / competitive_verifier / oj / languages / python.py: 90%

1# pyright: reportUnknownVariableType=false, reportUnknownMemberType=false 

2# Python Version: 3.x 

3import concurrent.futures 

4import functools 

5import os 

6import pathlib 

7import platform 

8import sys 

9from collections.abc import Sequence 

10from logging import getLogger 

11 

12import importlab.environment 

13import importlab.fs 

14import importlab.graph 

15 

16from competitive_verifier.models import ShellCommand 

17 

18from .base import Language, LanguageEnvironment 

19 

20logger = getLogger(__name__) 

21 

22 

23class PythonLanguageEnvironment(LanguageEnvironment): 

24 @property 

25 def name(self) -> str: 

26 return "Python" 

27 

28 def _python_path(self, *, basedir: pathlib.Path) -> str: 

29 python_path = os.getenv("PYTHONPATH") 

30 return ( 

31 basedir.resolve().as_posix() + os.pathsep + python_path 

32 if python_path 

33 else basedir.resolve().as_posix() 

34 ) 

35 

36 def get_compile_command( 

37 self, path: pathlib.Path, *, basedir: pathlib.Path, tempdir: pathlib.Path 

38 ) -> ShellCommand: 

39 return ShellCommand( 

40 command=["python", "-m", "py_compile", str(path)], 

41 env={"PYTHONPATH": self._python_path(basedir=basedir)}, 

42 ) 

43 

44 def get_execute_command( 

45 self, path: pathlib.Path, *, basedir: pathlib.Path, tempdir: pathlib.Path 

46 ) -> ShellCommand: 

47 return ShellCommand( 

48 command=["python", str(path)], 

49 env={"PYTHONPATH": self._python_path(basedir=basedir)}, 

50 ) 

51 

52 

53@functools.cache 

54def _python_list_depending_files( 

55 path: pathlib.Path, basedir: pathlib.Path 

56) -> list[pathlib.Path]: 

57 # compute the dependency graph of the `path` 

58 env = importlab.environment.Environment( 

59 importlab.fs.Path([importlab.fs.OSFileSystem(str(basedir.resolve()))]), 

60 (sys.version_info.major, sys.version_info.minor), 

61 ) 

62 try: 

63 executor = concurrent.futures.ThreadPoolExecutor() 

64 future = executor.submit( 

65 importlab.graph.ImportGraph.create, # pyright: ignore[reportUnknownArgumentType] 

66 env, 

67 [str(path)], 

68 True, 

69 ) 

70 

71 timeout = 5.0 if platform.uname().system == "Windows" else 1.0 

72 # 1.0 sec causes timeout on CI using Windows 

73 

74 res_graph = future.result(timeout=timeout) 

75 except concurrent.futures.TimeoutError as e: 

76 raise RuntimeError( 

77 f"Failed to analyze the dependency graph (timeout): {path}" 

78 ) from e 

79 try: 

80 node_deps_pairs: list[tuple[str, list[str]]] = res_graph.deps_list() 

81 except Exception as e: 

82 raise RuntimeError( 

83 f"Failed to analyze the dependency graph (circular imports?): {path}" 

84 ) from e 

85 logger.debug("the dependency graph of %s: %s", path, node_deps_pairs) 

86 

87 # collect Python files which are depended by the `path` and under `basedir` 

88 res_deps: list[pathlib.Path] = [] 

89 res_deps.append(path.resolve()) 

90 for node_, deps_ in node_deps_pairs: 90 ↛ 100line 90 didn't jump to line 100 because the loop on line 90 didn't complete

91 node = pathlib.Path(node_) 

92 deps = list(map(pathlib.Path, deps_)) 

93 if node.resolve() == path.resolve(): 93 ↛ 90line 93 didn't jump to line 90 because the condition on line 93 was always true

94 res_deps.extend( 

95 dep.resolve() 

96 for dep in deps 

97 if basedir.resolve() in dep.resolve().parents 

98 ) 

99 break 

100 return list(set(res_deps)) 

101 

102 

103class PythonLanguage(Language): 

104 def list_dependencies( 

105 self, path: pathlib.Path, *, basedir: pathlib.Path 

106 ) -> list[pathlib.Path]: 

107 return _python_list_depending_files(path.resolve(), basedir) 

108 

109 def list_environments( 

110 self, path: pathlib.Path, *, basedir: pathlib.Path 

111 ) -> Sequence[PythonLanguageEnvironment]: 

112 return [PythonLanguageEnvironment()]