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 

49class ConstVerification(BaseVerification): 

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

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

52 """The pre-defined result. 

53 """ 

54 

55 @property 

56 def is_lightweight(self) -> bool: 

57 return True 

58 

59 def run( 

60 self, 

61 params: VerificationParams | None = None, 

62 *, 

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

64 ) -> ResultStatus: 

65 return self.status 

66 

67 def run_compile_command( 

68 self, 

69 params: VerificationParams | None = None, 

70 ) -> bool: 

71 return True 

72 

73 

74class CommandVerification(BaseVerification): 

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

76 

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

78 """The shell command for verification. 

79 """ 

80 compile: ShellCommandLike | None = Field( 

81 default=None, 

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

83 ) 

84 """The shell command for compile. 

85 """ 

86 

87 tempdir: ForcePosixPath | None = Field( 

88 default=None, 

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

90 ) 

91 """The temporary directory for running verification. 

92 """ 

93 

94 def run( 

95 self, 

96 params: VerificationParams | None = None, 

97 *, 

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

99 ) -> ResultStatus: 

100 if self.tempdir: 

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

102 c = ShellCommand.parse_command_like(self.command) 

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

104 return ResultStatus.SUCCESS 

105 return ResultStatus.FAILURE 

106 

107 def run_compile_command( 

108 self, 

109 params: VerificationParams | None = None, 

110 ) -> bool: 

111 if self.compile: 

112 if self.tempdir: 

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

114 c = ShellCommand.parse_command_like(self.compile) 

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

116 return True 

117 

118 

119class BaseProblemVerification(BaseVerification, ABC): 

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

121 """The shell command for verification. 

122 """ 

123 compile: ShellCommandLike | None = Field( 

124 default=None, 

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

126 ) 

127 """The shell command for compile. 

128 """ 

129 

130 error: float | None = Field( 

131 default=None, 

132 examples=[1e-9], 

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

134 ) 

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

136 """ 

137 tle: float | None = Field( 

138 default=None, 

139 examples=[10], 

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

141 ) 

142 """The TLE time in seconds. 

143 """ 

144 mle: float | None = Field( 

145 default=None, 

146 examples=[64], 

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

148 ) 

149 """The MLE memory size in megabytes. 

150 """ 

151 

152 @abstractmethod 

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

154 

155 def run( 

156 self, 

157 params: VerificationParams | None = None, 

158 *, 

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

160 ) -> VerificationResult | ResultStatus: 

161 from competitive_verifier import oj # noqa: PLC0415 

162 

163 if not params: 

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

165 

166 problem = self._problem() 

167 if not problem: 

168 return ResultStatus.FAILURE 

169 

170 c = ShellCommand.parse_command_like(self.command) 

171 result = oj.test( 

172 problem=problem, 

173 command=c.command, 

174 env=c.env, 

175 tle=self.tle or params.default_tle, 

176 error=self.error, 

177 mle=self.mle or params.default_mle, 

178 deadline=deadline, 

179 ) 

180 result.verification_name = self.name 

181 return result 

182 

183 def run_compile_command( 

184 self, 

185 params: VerificationParams | None = None, 

186 ) -> bool: 

187 if self.compile: 

188 c = ShellCommand.parse_command_like(self.compile) 

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

190 return True 

191 

192 

193class ProblemVerification(BaseProblemVerification): 

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

195 

196 problem: str = Field( 

197 description="The URL of problem.", 

198 ) 

199 """ 

200 problem: URL of problem 

201 """ 

202 

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

204 # circular dependency 

205 from competitive_verifier.oj import problem_from_url # noqa: PLC0415 

206 

207 return problem_from_url(self.problem) 

208 

209 

210class LocalProblemVerification(BaseProblemVerification): 

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

212 

213 input: ForcePosixPath = Field( 

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

215 ) 

216 """ 

217 input: file path of testcases 

218 """ 

219 

220 tempdir: ForcePosixPath | None = Field( 

221 default=None, 

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

223 ) 

224 """The temporary directory for running verification. 

225 """ 

226 

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

228 # circular dependency 

229 from competitive_verifier.oj import LocalProblem # noqa: PLC0415 

230 

231 return LocalProblem(self.input) 

232 

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

234 if self.tempdir is not None: 

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

236 

237 return super().run_compile_command(params) 

238 

239 

240Verification = Annotated[ 

241 ConstVerification 

242 | CommandVerification 

243 | ProblemVerification 

244 | LocalProblemVerification, 

245 Field(discriminator="type"), 

246]