Coverage for src / competitive_verifier / models / result.py: 97%
118 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-10-06 03:56 +0900
1import datetime
2import enum
3import pathlib
4from logging import getLogger
5from typing import TYPE_CHECKING, Any, overload
7from pydantic import BaseModel, Field, field_validator
9from competitive_verifier.log import GitHubMessageParams
10from competitive_verifier.util import to_relative
12from .path import ForcePosixPath
13from .result_status import JudgeStatus, ResultStatus
15if TYPE_CHECKING:
16 from _typeshed import StrPath
18logger = getLogger(__name__)
21class TestcaseResult(BaseModel):
22 name: str = Field(
23 description="The name of test case.",
24 )
25 """The name of test case.
26 """
28 status: JudgeStatus = Field(
29 description="The result status of the test case.",
30 )
31 """The result status of the test case.
32 """
34 elapsed: float = Field(
35 description="Number of seconds elapsed for the test case.",
36 )
37 """Number of seconds elapsed for the test case.
38 """
40 memory: float | None = Field(
41 default=None,
42 description="The size of memory used in megabytes.",
43 )
44 """The size of memory used in megabytes.
45 """
48class VerificationResult(BaseModel):
49 verification_name: str | None = Field(
50 default=None,
51 description="The name of verification.",
52 )
53 """The name of verification.
54 """
55 status: ResultStatus = Field(
56 description="The result status of verification.",
57 )
58 """The result status of verification.
59 """
61 elapsed: float = Field(
62 description="Total number of seconds elapsed for all test cases.",
63 )
64 """Total number of seconds elapsed for all test cases.
65 """
67 slowest: float | None = Field(
68 default=None,
69 description="Maximum number of seconds elapsed for each test cases.",
70 )
71 """Maximum number of seconds elapsed for each test cases.
72 """
74 heaviest: float | None = Field(
75 default=None,
76 description="Maximum size of memory used in megabytes.",
77 )
78 """Maximum size of memory used in megabytes.
79 """
81 testcases: list[TestcaseResult] | None = Field(
82 default=None,
83 description="The results of each test case.",
84 )
85 """The results of each test case.
86 """
88 last_execution_time: datetime.datetime = Field(
89 default_factory=lambda: datetime.datetime.now(datetime.timezone.utc),
90 description="The time at which the last validation was performed.",
91 )
92 """The time at which the last validation was performed.
93 """
95 @field_validator("status", mode="before")
96 @classmethod
97 def verification_list(cls, v: Any) -> Any: # noqa: ANN401
98 return v.lower() if isinstance(v, str) else v
100 def need_reverifying(self, base_time: datetime.datetime) -> bool:
101 if self.status != ResultStatus.SUCCESS:
102 return True
104 return self.last_execution_time < base_time
107class NeedVerification(enum.Enum):
108 """Why a previous ``FileResult`` can't be reused; falsy when it can."""
110 NO = "unchanged since the previous verification"
111 NO_RESULT = "no previous result"
112 NOT_SUCCESS = "the previous verification didn't succeed"
113 MODIFIED = "modified after the previous verification"
114 NO_CONTENT_HASH = "the previous result has no content hash"
115 CONTENT_CHANGED = "content hash changed"
116 NO_TESTDATA_HASH = "the previous result has no test data hash"
117 TESTDATA_CHANGED = "test data hash changed"
119 def __bool__(self) -> bool:
120 return self is not NeedVerification.NO
123class FileResult(BaseModel):
124 verifications: list[VerificationResult] = Field(
125 default_factory=list[VerificationResult],
126 description="The results of each verification.",
127 )
128 """The results of each verification.
129 """
131 content_hash: str | None = Field(
132 default=None,
133 description="Digest of the file and its transitive dependencies"
134 " at verification time.",
135 )
136 """Digest of the file and its transitive dependencies at verification time.
137 """
139 testdata_hash: str | None = Field(
140 default=None,
141 description="Digest of the test data used by the verifications,"
142 " or null if it is never expected to change or isn't downloaded yet.",
143 )
144 """Digest of the test data used by the verifications,
145 or null if it is never expected to change or isn't downloaded yet.
146 """
148 newest: bool = Field(
149 default=True,
150 description="Whether the verification was performed on the most recent run.",
151 )
152 """Whether the verification was performed on the most recent run.
153 """
155 @overload
156 def need_verification(
157 self,
158 *,
159 base_time: datetime.datetime,
160 testdata_hash: str | None = None,
161 ) -> NeedVerification: ...
163 @overload
164 def need_verification(
165 self,
166 *,
167 content_hash: str | None,
168 testdata_hash: str | None,
169 ) -> NeedVerification: ...
171 def need_verification(
172 self,
173 *,
174 base_time: datetime.datetime | None = None,
175 content_hash: str | None = None,
176 testdata_hash: str | None = None,
177 ) -> NeedVerification:
178 """Whether the file or its test data has changed since this result.
180 Pass ``base_time`` to compare against the file's modification time,
181 or ``content_hash`` (and ``testdata_hash``, when the test data can be
182 identified) to compare against the recorded hashes.
183 """
184 if len(self.verifications) == 0:
185 return NeedVerification.NO_RESULT
187 if testdata_hash is not None:
188 if self.testdata_hash is None:
189 return NeedVerification.NO_TESTDATA_HASH
190 if self.testdata_hash != testdata_hash:
191 return NeedVerification.TESTDATA_CHANGED
193 if base_time is not None:
194 return self._need_verification_by_time(base_time)
195 return self._need_verification_by_hash(content_hash)
197 def _need_verification_by_time(
198 self, base_time: datetime.datetime
199 ) -> NeedVerification:
200 if any(r.status != ResultStatus.SUCCESS for r in self.verifications):
201 return NeedVerification.NOT_SUCCESS
202 if any(r.last_execution_time < base_time for r in self.verifications):
203 return NeedVerification.MODIFIED
204 return NeedVerification.NO
206 def _need_verification_by_hash(self, content_hash: str | None) -> NeedVerification:
207 if self.content_hash is None:
208 return NeedVerification.NO_CONTENT_HASH
209 if self.content_hash != content_hash:
210 return NeedVerification.CONTENT_CHANGED
211 if not self.is_success(allow_skip=False):
212 return NeedVerification.NOT_SUCCESS
213 return NeedVerification.NO
215 def is_success(self, *, allow_skip: bool) -> bool:
216 if allow_skip:
217 return all(r.status != ResultStatus.FAILURE for r in self.verifications)
218 return all(r.status == ResultStatus.SUCCESS for r in self.verifications)
221class VerifyCommandResult(BaseModel):
222 total_seconds: float = Field(
223 description="Total number of seconds elapsed for all verification.",
224 )
225 """Total number of seconds elapsed for all verification.
226 """
228 files: dict[ForcePosixPath, FileResult] = Field(
229 default_factory=dict[ForcePosixPath, FileResult],
230 description="The files to be verified.",
231 )
232 """The files to be verified.
233 """
235 @classmethod
236 def parse_file_relative(cls, path: "StrPath") -> "VerifyCommandResult":
237 impl = cls.model_validate_json(pathlib.Path(path).read_bytes())
238 new_files: dict[pathlib.Path, FileResult] = {}
239 for p, f in impl.files.items():
240 rp = to_relative(p)
241 if not rp:
242 logger.warning(
243 "Files in other directories are not subject to verification: %s",
244 p,
245 extra={"github": GitHubMessageParams()},
246 )
247 continue
248 new_files[rp] = f
250 impl.files = new_files
251 return impl
253 def merge(self, other: "VerifyCommandResult") -> "VerifyCommandResult":
254 d = self.files.copy()
255 for k, r in other.files.items():
256 cur = d.get(k)
257 if r.newest or (cur is None) or (not cur.newest):
258 d[k] = r
259 return VerifyCommandResult(
260 total_seconds=self.total_seconds + other.total_seconds,
261 files=d,
262 )
264 def is_success(self, *, allow_skip: bool = True) -> bool:
265 return all(f.is_success(allow_skip=allow_skip) for f in self.files.values())