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

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.oj.verify.models import Language, LanguageEnvironment 

17 

18logger = getLogger(__name__) 

19 

20 

21class PythonLanguageEnvironment(LanguageEnvironment): 

22 @property 

23 def name(self) -> str: 

24 return "Python" 

25 

26 def get_execute_command( 

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

28 ) -> str: 

29 python_path = os.getenv("PYTHONPATH") 

30 python_path = ( 

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

32 if python_path 

33 else basedir.resolve().as_posix() 

34 ) 

35 return f"env PYTHONPATH={python_path} python {path}" 

36 

37 

38@functools.cache 

39def _python_list_depending_files( 

40 path: pathlib.Path, basedir: pathlib.Path 

41) -> list[pathlib.Path]: 

42 # compute the dependency graph of the `path` 

43 env = importlab.environment.Environment( 

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

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

46 ) 

47 try: 

48 executor = concurrent.futures.ThreadPoolExecutor() 

49 future = executor.submit( 

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

51 env, 

52 [str(path)], 

53 True, 

54 ) 

55 

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

57 # 1.0 sec causes timeout on CI using Windows 

58 

59 res_graph = future.result(timeout=timeout) 

60 except concurrent.futures.TimeoutError as e: 

61 raise RuntimeError( 

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

63 ) from e 

64 try: 

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

66 except Exception as e: 

67 raise RuntimeError( 

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

69 ) from e 

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

71 

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

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

74 res_deps.append(path.resolve()) 

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

76 node = pathlib.Path(node_) 

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

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

79 res_deps.extend( 

80 dep.resolve() 

81 for dep in deps 

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

83 ) 

84 break 

85 return list(set(res_deps)) 

86 

87 

88class PythonLanguage(Language): 

89 def list_dependencies( 

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

91 ) -> list[pathlib.Path]: 

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

93 

94 def list_environments( 

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

96 ) -> Sequence[PythonLanguageEnvironment]: 

97 return [PythonLanguageEnvironment()]