Coverage for mongo/problem/test_case.py: 100%

1import abc 

2import zipfile 

3from typing import BinaryIO, Set, TYPE_CHECKING, List 

4from .exception import BadTestCase 

5if TYPE_CHECKING: 

6 from .. import Problem # pragma: no cover 

7 

8 

9class TestCaseRule(abc.ABC): 

10 

11 def __init__(self, problem: 'Problem'): 

12 self.problem = problem 

13 

14 # TODO: define generic validation error 

15 @abc.abstractmethod 

16 def validate(self, test_case: BinaryIO) -> bool: 

17 ''' 

18 Validate test case 

19 ''' 

20 raise NotImplementedError # pragma: no cover 

21 

22 

23class IncludeDirectory(TestCaseRule): 

24 

25 def __init__( 

26 self, 

27 problem: 'Problem', 

28 path: str, 

29 optional: bool = True, 

30 ): 

31 self.path = path 

32 self.optional = optional 

33 super().__init__(problem) 

34 

35 def validate(self, test_case: BinaryIO) -> bool: 

36 if test_case is None: 

37 raise BadTestCase('test case is None') 

38 path = zipfile.Path(test_case, at=self.path) 

39 

40 if not path.exists(): 

41 if self.optional: 

42 return True 

43 raise BadTestCase(f'directory {self.path} does not exist') 

44 

45 if path.is_file(): 

46 raise BadTestCase(f'{self.path} is not a directory') 

47 

48 return True 

49 

50 

51class SimpleIO(TestCaseRule): 

52 ''' 

53 Test cases that only contains single input and output file. 

54 ''' 

55 

56 def __init__(self, problem: 'Problem', excludes: List[str] = []): 

57 self.excludes = excludes 

58 super().__init__(problem) 

59 

60 def validate(self, test_case: BinaryIO) -> bool: 

61 # test case must not be None 

62 if test_case is None: 

63 raise BadTestCase('test case is None') 

64 got = {*zipfile.ZipFile(test_case).namelist()} 

65 for ex in self.excludes: 

66 if ex.endswith('/'): 

67 got = {g for g in got if not g.startswith(ex)} 

68 else: 

69 got.discard(ex) 

70 expected = self.expected_test_case_filenames() 

71 if got != expected: 

72 extra = list(got - expected) 

73 missing = list(expected - got) 

74 raise BadTestCase( 

75 f'I/O data not equal to meta provided: {extra=}, {missing=}') 

76 # reset 

77 test_case.seek(0) 

78 return True 

79 

80 def expected_test_case_filenames(self) -> Set[str]: 

81 excepted = set() 

82 for i, task in enumerate(self.problem.test_case.tasks): 

83 for j in range(task.case_count): 

84 excepted.add(f'{i:02d}{j:02d}.in') 

85 excepted.add(f'{i:02d}{j:02d}.out') 

86 return excepted 

87 

88 

89class ContextIO(TestCaseRule): 

90 ''' 

91 Test cases that contains multiple file for input/output. 

92 e.g. given a image, rotate and save it on disk. 

93 ''' 

94 

95 def validate(self, test_case_fp: BinaryIO) -> bool: 

96 if test_case_fp is None: 

97 raise BadTestCase('test case is None') 

98 

99 test_case_root = zipfile.Path(test_case_fp, at='test-case/') 

100 if not test_case_root.exists(): 

101 raise BadTestCase('test-case not found') 

102 if not test_case_root.is_dir(): 

103 raise BadTestCase('test-case is not a directory') 

104 

105 expected_dirs = self.expected_test_case_dirs() 

106 

107 for test_case in test_case_root.iterdir(): 

108 try: 

109 expected_dirs.remove(test_case.name) 

110 except KeyError: 

111 raise BadTestCase( 

112 f'extra test case directory found: {test_case.name}') 

113 self.validate_test_case_dir(test_case) 

114 

115 if len(expected_dirs): 

116 raise BadTestCase( 

117 f'missing test case directory: {", ".join(expected_dirs)}') 

118 

119 def validate_test_case_dir(self, test_case_dir: zipfile.Path): 

120 requireds = { 

121 'STDIN', 

122 'STDOUT', 

123 } 

124 

125 for r in requireds: 

126 if not (test_case_dir / r).exists(): 

127 raise BadTestCase(f'required file/dir not found: {r}') 

128 

129 dirs = { 

130 'in', 

131 'out', 

132 } 

133 for fp in test_case_dir.iterdir(): 

134 # files under in/out are allowed 

135 if fp.is_dir() and fp.name in dirs: 

136 continue 

137 # STDIN/STDOUT are allowed 

138 if fp.name in requireds: 

139 continue 

140 raise BadTestCase(f'files in unallowed path: {fp.name}') 

141 

142 def expected_test_case_dirs(self) -> Set[str]: 

143 excepted = set() 

144 for i, task in enumerate(self.problem.test_case.tasks): 

145 for j in range(task.case_count): 

146 excepted.add(f'{i:02d}{j:02d}') 

147 return excepted