Coverage for src / competitive_verifier / inout / main.py: 100%

1import pathlib 

2from argparse import ArgumentParser 

3from collections import Counter 

4from collections.abc import Iterable 

5from functools import reduce 

6from logging import getLogger 

7from typing import Literal, TypeVar 

8 

9from pydantic import Field 

10 

11from competitive_verifier.arg import ( 

12 ResultJsonArguments, 

13 VerboseArguments, 

14 WriteSummaryArguments, 

15) 

16from competitive_verifier.models import ( 

17 ResultStatus, 

18 VerificationInput, 

19 VerifyCommandResult, 

20) 

21 

22logger = getLogger(__name__) 

23 

24T = TypeVar("T", VerificationInput, VerifyCommandResult) 

25 

26 

27def merge(results: Iterable[T]) -> T: 

28 return reduce(lambda a, b: a.merge(b), results) 

29 

30 

31class MergeInput(VerboseArguments): 

32 subcommand: Literal["merge-input"] = Field( 

33 default="merge-input", 

34 description="Merge verify_files.json`", 

35 ) 

36 verify_files_json: list[pathlib.Path] 

37 

38 @classmethod 

39 def add_parser(cls, parser: ArgumentParser): 

40 super().add_parser(parser) 

41 parser.add_argument( 

42 "verify_files_json", 

43 nargs="+", 

44 help="verify_files.json files", 

45 type=pathlib.Path, 

46 ) 

47 

48 def _run(self) -> bool: 

49 result = merge( 

50 map(VerificationInput.parse_file_relative, self.verify_files_json) 

51 ) 

52 print(result.model_dump_json(exclude_none=True)) 

53 return True 

54 

55 

56class MergeResult(WriteSummaryArguments, ResultJsonArguments, VerboseArguments): 

57 subcommand: Literal["merge-result"] = Field( 

58 default="merge-result", 

59 description="Merge result of `verify`", 

60 ) 

61 

62 def merge(self) -> VerifyCommandResult: 

63 return merge(map(VerifyCommandResult.parse_file_relative, self.result_json)) 

64 

65 def _run(self) -> bool: 

66 result = self.merge() 

67 self.write_result(result) 

68 

69 print(result.model_dump_json(exclude_none=True)) 

70 return True 

71 

72 

73class Check(ResultJsonArguments, VerboseArguments): 

74 subcommand: Literal["check"] = Field( 

75 default="check", 

76 description="Check result of `verify`", 

77 ) 

78 

79 def _run(self) -> bool: 

80 result = merge(map(VerifyCommandResult.parse_file_relative, self.result_json)) 

81 

82 counter = Counter( 

83 r.status for fr in result.files.values() for r in fr.verifications 

84 ) 

85 

86 for st in ResultStatus: 

87 print(f"{st.value}: {counter.get(st, 0)}") 

88 

89 if counter[ResultStatus.FAILURE] > 0: 

90 if self.verbose: 

91 logger.error("Failure test count: %d", counter[ResultStatus.FAILURE]) 

92 return False 

93 return True