Coverage for src / competitive_verifier / models / verification.py: 98%
113 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-10-04 07:40 +0900
1from abc import ABC, abstractmethod
2from typing import Annotated, Literal, Protocol
4from pydantic import BaseModel, Field
6from .path import ForcePosixPath
7from .problem import TestCaseProvider
8from .result import VerificationResult
9from .result_status import ResultStatus
10from .shell import ShellCommand, ShellCommandLike
13class VerifcationTimeoutError(Exception):
14 pass
17class VerificationParams(Protocol):
18 default_tle: float | None
19 default_mle: float | None
22class BaseVerification(BaseModel, ABC):
23 name: str | None = None
25 @abstractmethod
26 def run(
27 self,
28 params: VerificationParams | None = None,
29 *,
30 deadline: float = float("inf"),
31 ) -> ResultStatus | VerificationResult: ...
33 @abstractmethod
34 def run_compile_command(
35 self,
36 params: VerificationParams | None = None,
37 ) -> bool: ...
39 @property
40 def is_lightweight(self) -> bool:
41 """The verification is lightweight."""
42 return False
44 def is_testdata_cached(self) -> bool:
45 """Whether the test data is present locally (best effort).
47 Must not download anything (see ``TestCaseProvider.is_testdata_cached``).
48 """
49 return True
51 def sync_testdata(self) -> None:
52 """Fetch the latest upstream test data identity so ``testdata_hash`` reflects it.
54 No-op unless the test data can change upstream (see
55 ``TestCaseProvider.sync_testdata``).
56 """
58 def testdata_hash(self) -> str | None:
59 """Digest identifying the verification's test data present in the local cache.
61 Must not download anything. ``None`` if the test data is never expected to
62 change, or it isn't downloaded yet (see ``TestCaseProvider.testdata_hash``).
63 """
64 return None
67class ConstVerification(BaseVerification):
68 type: Literal["const"] = "const"
69 status: ResultStatus = Field(description="The pre-defined result.")
70 """The pre-defined result.
71 """
73 @property
74 def is_lightweight(self) -> bool:
75 return True
77 def run(
78 self,
79 params: VerificationParams | None = None,
80 *,
81 deadline: float = float("inf"),
82 ) -> ResultStatus:
83 return self.status
85 def run_compile_command(
86 self,
87 params: VerificationParams | None = None,
88 ) -> bool:
89 return True
92class CommandVerification(BaseVerification):
93 type: Literal["command"] = "command"
95 command: ShellCommandLike = Field(description="The shell command for verification.")
96 """The shell command for verification.
97 """
98 compile: ShellCommandLike | None = Field(
99 default=None,
100 description="The shell command for compile.",
101 )
102 """The shell command for compile.
103 """
105 tempdir: ForcePosixPath | None = Field(
106 default=None,
107 description="The temporary directory for running verification.",
108 )
109 """The temporary directory for running verification.
110 """
112 def run(
113 self,
114 params: VerificationParams | None = None,
115 *,
116 deadline: float = float("inf"),
117 ) -> ResultStatus:
118 if self.tempdir:
119 self.tempdir.mkdir(parents=True, exist_ok=True)
120 c = ShellCommand.parse_command_like(self.command)
121 if c.exec_command(text=True).returncode == 0:
122 return ResultStatus.SUCCESS
123 return ResultStatus.FAILURE
125 def run_compile_command(
126 self,
127 params: VerificationParams | None = None,
128 ) -> bool:
129 if self.compile:
130 if self.tempdir:
131 self.tempdir.mkdir(parents=True, exist_ok=True)
132 c = ShellCommand.parse_command_like(self.compile)
133 return c.exec_command(text=True).returncode == 0
134 return True
137class BaseProblemVerification(BaseVerification, ABC):
138 command: ShellCommandLike = Field(description="The shell command for verification.")
139 """The shell command for verification.
140 """
141 compile: ShellCommandLike | None = Field(
142 default=None,
143 description="The shell command for compile.",
144 )
145 """The shell command for compile.
146 """
148 error: float | None = Field(
149 default=None,
150 examples=[1e-9],
151 description="The absolute or relative error to be considered as correct.",
152 )
153 """The absolute or relative error to be considered as correct.
154 """
155 tle: float | None = Field(
156 default=None,
157 examples=[10],
158 description="The TLE time in seconds.",
159 )
160 """The TLE time in seconds.
161 """
162 mle: float | None = Field(
163 default=None,
164 examples=[64],
165 description="The MLE memory size in megabytes.",
166 )
167 """The MLE memory size in megabytes.
168 """
170 @abstractmethod
171 def _problem(self) -> TestCaseProvider | None: ...
173 def is_testdata_cached(self) -> bool:
174 problem = self._problem()
175 return problem is None or problem.is_testdata_cached()
177 def sync_testdata(self) -> None:
178 problem = self._problem()
179 if problem is not None: 179 ↛ exitline 179 didn't return from function 'sync_testdata' because the condition on line 179 was always true
180 problem.sync_testdata()
182 def testdata_hash(self) -> str | None:
183 problem = self._problem()
184 if problem is None: 184 ↛ 185line 184 didn't jump to line 185 because the condition on line 184 was never true
185 return None
186 return problem.testdata_hash()
188 def run(
189 self,
190 params: VerificationParams | None = None,
191 *,
192 deadline: float = float("inf"),
193 ) -> VerificationResult | ResultStatus:
194 from competitive_verifier import oj # noqa: PLC0415
196 if not params:
197 raise ValueError("ProblemVerification.run requires VerificationParams")
199 problem = self._problem()
200 if not problem:
201 return ResultStatus.FAILURE
203 c = ShellCommand.parse_command_like(self.command)
204 result = oj.test(
205 problem=problem,
206 command=c.command,
207 env=c.env,
208 tle=self.tle or params.default_tle,
209 error=self.error,
210 mle=self.mle or params.default_mle,
211 deadline=deadline,
212 )
213 result.verification_name = self.name
214 return result
216 def run_compile_command(
217 self,
218 params: VerificationParams | None = None,
219 ) -> bool:
220 if self.compile:
221 c = ShellCommand.parse_command_like(self.compile)
222 return c.exec_command(text=True).returncode == 0
223 return True
226class ProblemVerification(BaseProblemVerification):
227 type: Literal["problem"] = "problem"
229 problem: str = Field(
230 description="The URL of problem.",
231 )
232 """
233 problem: URL of problem
234 """
236 def _problem(self) -> TestCaseProvider | None:
237 # circular dependency
238 from competitive_verifier.oj import problem_from_url # noqa: PLC0415
240 return problem_from_url(self.problem)
243class LocalProblemVerification(BaseProblemVerification):
244 type: Literal["local"] = "local"
246 input: ForcePosixPath = Field(
247 description="The file path of testcases.",
248 )
249 """
250 input: file path of testcases
251 """
253 tempdir: ForcePosixPath | None = Field(
254 default=None,
255 description="The temporary directory for running verification.",
256 )
257 """The temporary directory for running verification.
258 """
260 def _problem(self) -> TestCaseProvider | None:
261 # circular dependency
262 from competitive_verifier.oj import LocalProblem # noqa: PLC0415
264 return LocalProblem(self.input)
266 def run_compile_command(self, params: VerificationParams | None = None) -> bool:
267 if self.tempdir is not None:
268 self.tempdir.mkdir(parents=True, exist_ok=True)
270 return super().run_compile_command(params)
273Verification = Annotated[
274 ConstVerification
275 | CommandVerification
276 | ProblemVerification
277 | LocalProblemVerification,
278 Field(discriminator="type"),
279]