Coverage for sel_tools/code_evaluation/jobs/common.py: 91%

1"""Common evaluation job interface and utilities.""" 

2 

3import itertools 

4import subprocess 

5from abc import ABCMeta, abstractmethod 

6from pathlib import Path 

7 

8from sel_tools.code_evaluation.report import EvaluationResult 

9 

10 

11class EvaluationJob: 

12 """Interface for evaluation job. 

13 

14 Children implement _run 

15 """ 

16 

17 __metaclass__ = ABCMeta 

18 

19 def __init__(self, weight: int = 1) -> None: 

20 self.__weight: int = weight 

21 self._comment: str = "" 

22 self._message: str = "" 

23 

24 def run(self, repo_path: Path) -> list[EvaluationResult]: 

25 deps_results = [job.run(repo_path) for job in self.dependencies] 

26 print(f"\nRunning {self.name} on {repo_path}") 

27 job_result_score = self._run(repo_path) 

28 return [ 

29 *list(itertools.chain(*deps_results)), 

30 EvaluationResult(self.name, self.__weight * job_result_score, self.comment), 

31 ] 

32 

33 @property 

34 @abstractmethod 

35 def name(self) -> str: 

36 msg = "Don't call me, I'm abstract." 

37 raise NotImplementedError(msg) 

38 

39 @property 

40 def comment(self) -> str: 

41 return self._comment 

42 

43 @property 

44 def dependencies(self) -> list["EvaluationJob"]: 

45 return [] 

46 

47 @abstractmethod 

48 def _run(self, repo_path: Path) -> int: 

49 msg = "Don't call me, I'm abstract." 

50 raise NotImplementedError(msg) 

51 

52 

53def run_shell_command(command: str, cwd: Path) -> int: 

54 """Run shell command.""" 

55 try: 

56 subprocess.check_call(command, shell=True, cwd=cwd) 

57 except subprocess.CalledProcessError: 

58 return 0 

59 return 1 

60 

61 

62def run_shell_command_with_output(command: str, cwd: Path) -> tuple[int, str]: 

63 """Run shell command and get the output.""" 

64 try: 

65 data = subprocess.check_output(command, shell=True, cwd=cwd) 

66 except subprocess.CalledProcessError as ex: 

67 return 0, ex.output.decode("utf-8") 

68 return 1, data.decode("utf-8")