Coverage for src / competitive_verifier / models / verification.py: 100%

1from abc import ABC, abstractmethod 

2from typing import Annotated, Literal, Protocol 

3 

4from pydantic import BaseModel, Field 

5 

6from .path import ForcePosixPath 

7from .problem import TestCaseProvider 

8from .result import VerificationResult 

9from .result_status import ResultStatus 

10from .shell import ShellCommand, ShellCommandLike 

11 

12 

13class VerifcationTimeoutError(Exception): 

14 pass 

15 

16 

17class VerificationParams(Protocol): 

18 default_tle: float | None 

19 default_mle: float | None 

20 

21 

22class BaseVerification(BaseModel, ABC): 

23 name: str | None = None 

24 

25 @abstractmethod 

26 def run( 

27 self, 

28 params: VerificationParams | None = None, 

29 *, 

30 deadline: float = float("inf"), 

31 ) -> ResultStatus | VerificationResult: ... 

32 

33 @abstractmethod 

34 def run_compile_command( 

35 self, 

36 params: VerificationParams | None = None, 

37 ) -> bool: ... 

38 

39 @property 

40 def is_lightweight(self) -> bool: 

41 """The verification is lightweight.""" 

42 return False 

43 

44 

45class ConstVerification(BaseVerification): 

46 type: Literal["const"] = "const" 

47 status: ResultStatus = Field(description="The pre-defined result.") 

48 """The pre-defined result. 

49 """ 

50 

51 @property 

52 def is_lightweight(self) -> bool: 

53 return True 

54 

55 def run( 

56 self, 

57 params: VerificationParams | None = None, 

58 *, 

59 deadline: float = float("inf"), 

60 ) -> ResultStatus: 

61 return self.status 

62 

63 def run_compile_command( 

64 self, 

65 params: VerificationParams | None = None, 

66 ) -> bool: 

67 return True 

68 

69 

70class CommandVerification(BaseVerification): 

71 type: Literal["command"] = "command" 

72 

73 command: ShellCommandLike = Field(description="The shell command for verification.") 

74 """The shell command for verification. 

75 """ 

76 compile: ShellCommandLike | None = Field( 

77 default=None, 

78 description="The shell command for compile.", 

79 ) 

80 """The shell command for compile. 

81 """ 

82 

83 tempdir: ForcePosixPath | None = Field( 

84 default=None, 

85 description="The temporary directory for running verification.", 

86 ) 

87 """The temporary directory for running verification. 

88 """ 

89 

90 def run( 

91 self, 

92 params: VerificationParams | None = None, 

93 *, 

94 deadline: float = float("inf"), 

95 ) -> ResultStatus: 

96 if self.tempdir: 

97 self.tempdir.mkdir(parents=True, exist_ok=True) 

98 c = ShellCommand.parse_command_like(self.command) 

99 if c.exec_command(text=True).returncode == 0: 

100 return ResultStatus.SUCCESS 

101 return ResultStatus.FAILURE 

102 

103 def run_compile_command( 

104 self, 

105 params: VerificationParams | None = None, 

106 ) -> bool: 

107 if self.compile: 

108 if self.tempdir: 

109 self.tempdir.mkdir(parents=True, exist_ok=True) 

110 c = ShellCommand.parse_command_like(self.compile) 

111 return c.exec_command(text=True).returncode == 0 

112 return True 

113 

114 

115class BaseProblemVerification(BaseVerification, ABC): 

116 command: ShellCommandLike = Field(description="The shell command for verification.") 

117 """The shell command for verification. 

118 """ 

119 compile: ShellCommandLike | None = Field( 

120 default=None, 

121 description="The shell command for compile.", 

122 ) 

123 """The shell command for compile. 

124 """ 

125 

126 error: float | None = Field( 

127 default=None, 

128 examples=[1e-9], 

129 description="The absolute or relative error to be considered as correct.", 

130 ) 

131 """The absolute or relative error to be considered as correct. 

132 """ 

133 tle: float | None = Field( 

134 default=None, 

135 examples=[10], 

136 description="The TLE time in seconds.", 

137 ) 

138 """The TLE time in seconds. 

139 """ 

140 mle: float | None = Field( 

141 default=None, 

142 examples=[64], 

143 description="The MLE memory size in megabytes.", 

144 ) 

145 """The MLE memory size in megabytes. 

146 """ 

147 

148 @abstractmethod 

149 def _problem(self) -> TestCaseProvider | None: ... 

150 

151 def run( 

152 self, 

153 params: VerificationParams | None = None, 

154 *, 

155 deadline: float = float("inf"), 

156 ) -> VerificationResult | ResultStatus: 

157 from competitive_verifier import oj # noqa: PLC0415 

158 

159 if not params: 

160 raise ValueError("ProblemVerification.run requires VerificationParams") 

161 

162 problem = self._problem() 

163 if not problem: 

164 return ResultStatus.FAILURE 

165 

166 c = ShellCommand.parse_command_like(self.command) 

167 result = oj.test( 

168 problem=problem, 

169 command=c.command, 

170 env=c.env, 

171 tle=self.tle or params.default_tle, 

172 error=self.error, 

173 mle=self.mle or params.default_mle, 

174 deadline=deadline, 

175 ) 

176 result.verification_name = self.name 

177 return result 

178 

179 def run_compile_command( 

180 self, 

181 params: VerificationParams | None = None, 

182 ) -> bool: 

183 if self.compile: 

184 c = ShellCommand.parse_command_like(self.compile) 

185 return c.exec_command(text=True).returncode == 0 

186 return True 

187 

188 

189class ProblemVerification(BaseProblemVerification): 

190 type: Literal["problem"] = "problem" 

191 

192 problem: str = Field( 

193 description="The URL of problem.", 

194 ) 

195 """ 

196 problem: URL of problem 

197 """ 

198 

199 def _problem(self) -> TestCaseProvider | None: 

200 # circular dependency 

201 from competitive_verifier.oj import problem_from_url # noqa: PLC0415 

202 

203 return problem_from_url(self.problem) 

204 

205 

206class LocalProblemVerification(BaseProblemVerification): 

207 type: Literal["local"] = "local" 

208 

209 input: ForcePosixPath = Field( 

210 description="The file path of testcases.", 

211 ) 

212 """ 

213 input: file path of testcases 

214 """ 

215 

216 tempdir: ForcePosixPath | None = Field( 

217 default=None, 

218 description="The temporary directory for running verification.", 

219 ) 

220 """The temporary directory for running verification. 

221 """ 

222 

223 def _problem(self) -> TestCaseProvider | None: 

224 # circular dependency 

225 from competitive_verifier.oj import LocalProblem # noqa: PLC0415 

226 

227 return LocalProblem(self.input) 

228 

229 def run_compile_command(self, params: VerificationParams | None = None) -> bool: 

230 if self.tempdir is not None: 

231 self.tempdir.mkdir(parents=True, exist_ok=True) 

232 

233 return super().run_compile_command(params) 

234 

235 

236Verification = Annotated[ 

237 ConstVerification 

238 | CommandVerification 

239 | ProblemVerification 

240 | LocalProblemVerification, 

241 Field(discriminator="type"), 

242]