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 VerificationTimeoutError(Exception): 

14 pass 

15 

16 

17# Deprecated typo alias kept for compatibility with previous releases. 

18VerifcationTimeoutError = VerificationTimeoutError 

19 

20 

21class VerificationParams(Protocol): 

22 default_tle: float | None 

23 default_mle: float | None 

24 

25 

26class BaseVerification(BaseModel, ABC): 

27 name: str | None = None 

28 

29 @abstractmethod 

30 def run( 

31 self, 

32 params: VerificationParams | None = None, 

33 *, 

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

35 ) -> ResultStatus | VerificationResult: ... 

36 

37 @abstractmethod 

38 def run_compile_command( 

39 self, 

40 params: VerificationParams | None = None, 

41 ) -> bool: ... 

42 

43 @property 

44 def is_lightweight(self) -> bool: 

45 """The verification is lightweight.""" 

46 return False 

47 

48 def is_testdata_cached(self) -> bool: 

49 """Whether the test data is present locally (best effort). 

50 

51 Must not download anything (see ``TestCaseProvider.is_testdata_cached``). 

52 """ 

53 return True 

54 

55 

56class ConstVerification(BaseVerification): 

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

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

59 """The pre-defined result. 

60 """ 

61 

62 @property 

63 def is_lightweight(self) -> bool: 

64 return True 

65 

66 def run( 

67 self, 

68 params: VerificationParams | None = None, 

69 *, 

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

71 ) -> ResultStatus: 

72 return self.status 

73 

74 def run_compile_command( 

75 self, 

76 params: VerificationParams | None = None, 

77 ) -> bool: 

78 return True 

79 

80 

81class CommandVerification(BaseVerification): 

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

83 

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

85 """The shell command for verification. 

86 """ 

87 compile: ShellCommandLike | None = Field( 

88 default=None, 

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

90 ) 

91 """The shell command for compile. 

92 """ 

93 

94 tempdir: ForcePosixPath | None = Field( 

95 default=None, 

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

97 ) 

98 """The temporary directory for running verification. 

99 """ 

100 

101 def run( 

102 self, 

103 params: VerificationParams | None = None, 

104 *, 

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

106 ) -> ResultStatus: 

107 if self.tempdir: 

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

109 c = ShellCommand.parse_command_like(self.command) 

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

111 return ResultStatus.SUCCESS 

112 return ResultStatus.FAILURE 

113 

114 def run_compile_command( 

115 self, 

116 params: VerificationParams | None = None, 

117 ) -> bool: 

118 if self.compile: 

119 if self.tempdir: 

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

121 c = ShellCommand.parse_command_like(self.compile) 

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

123 return True 

124 

125 

126class BaseProblemVerification(BaseVerification, ABC): 

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

128 """The shell command for verification. 

129 """ 

130 compile: ShellCommandLike | None = Field( 

131 default=None, 

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

133 ) 

134 """The shell command for compile. 

135 """ 

136 

137 error: float | None = Field( 

138 default=None, 

139 examples=[1e-9], 

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

141 ) 

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

143 """ 

144 tle: float | None = Field( 

145 default=None, 

146 examples=[10], 

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

148 ) 

149 """The TLE time in seconds. 

150 """ 

151 mle: float | None = Field( 

152 default=None, 

153 examples=[64], 

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

155 ) 

156 """The MLE memory size in megabytes. 

157 """ 

158 

159 @abstractmethod 

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

161 

162 def is_testdata_cached(self) -> bool: 

163 problem = self._problem() 

164 return problem is None or problem.is_testdata_cached() 

165 

166 def run( 

167 self, 

168 params: VerificationParams | None = None, 

169 *, 

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

171 ) -> VerificationResult | ResultStatus: 

172 from competitive_verifier import oj # noqa: PLC0415 

173 

174 if not params: 

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

176 

177 problem = self._problem() 

178 if not problem: 

179 return ResultStatus.FAILURE 

180 

181 c = ShellCommand.parse_command_like(self.command) 

182 result = oj.test( 

183 problem=problem, 

184 command=c.command, 

185 env=c.env, 

186 tle=self.tle or params.default_tle, 

187 error=self.error, 

188 mle=self.mle or params.default_mle, 

189 deadline=deadline, 

190 ) 

191 result.verification_name = self.name 

192 return result 

193 

194 def run_compile_command( 

195 self, 

196 params: VerificationParams | None = None, 

197 ) -> bool: 

198 if self.compile: 

199 c = ShellCommand.parse_command_like(self.compile) 

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

201 return True 

202 

203 

204class ProblemVerification(BaseProblemVerification): 

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

206 

207 problem: str = Field( 

208 description="The URL of problem.", 

209 ) 

210 """ 

211 problem: URL of problem 

212 """ 

213 

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

215 # circular dependency 

216 from competitive_verifier.oj import problem_from_url # noqa: PLC0415 

217 

218 return problem_from_url(self.problem) 

219 

220 

221class LocalProblemVerification(BaseProblemVerification): 

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

223 

224 input: ForcePosixPath = Field( 

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

226 ) 

227 """ 

228 input: file path of testcases 

229 """ 

230 

231 tempdir: ForcePosixPath | None = Field( 

232 default=None, 

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

234 ) 

235 """The temporary directory for running verification. 

236 """ 

237 

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

239 # circular dependency 

240 from competitive_verifier.oj import LocalProblem # noqa: PLC0415 

241 

242 return LocalProblem(self.input) 

243 

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

245 if self.tempdir is not None: 

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

247 

248 return super().run_compile_command(params) 

249 

250 

251Verification = Annotated[ 

252 ConstVerification 

253 | CommandVerification 

254 | ProblemVerification 

255 | LocalProblemVerification, 

256 Field(discriminator="type"), 

257]