Coverage for src / competitive_verifier / models / result.py: 97%
118 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-10-04 07:40 +0900
1import datetime
2import enum
3import pathlib
4from logging import getLogger
5from typing import TYPE_CHECKING, Any
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 def need_verification(
156 self,
157 *,
158 base_time: datetime.datetime | None = None,
159 content_hash: str | None = None,
160 testdata_hash: str | None = None,
161 ) -> NeedVerification:
162 """Whether the file or its test data has changed since this result.
164 Pass ``base_time`` to compare against the file's modification time,
165 or ``content_hash`` (and ``testdata_hash``, when the test data can be
166 identified) to compare against the recorded hashes.
167 """
168 if len(self.verifications) == 0:
169 return NeedVerification.NO_RESULT
170 if base_time is not None:
171 return self._need_verification_by_time(base_time)
172 return self._need_verification_by_hash(content_hash, testdata_hash)
174 def _need_verification_by_time(
175 self, base_time: datetime.datetime
176 ) -> NeedVerification:
177 if any(r.status != ResultStatus.SUCCESS for r in self.verifications):
178 return NeedVerification.NOT_SUCCESS
179 if any(r.last_execution_time < base_time for r in self.verifications):
180 return NeedVerification.MODIFIED
181 return NeedVerification.NO
183 def _need_verification_by_hash(
184 self, content_hash: str | None, testdata_hash: str | None
185 ) -> NeedVerification:
186 if self.content_hash is None:
187 return NeedVerification.NO_CONTENT_HASH
188 if self.content_hash != content_hash:
189 return NeedVerification.CONTENT_CHANGED
190 if testdata_hash is not None:
191 if self.testdata_hash is None:
192 return NeedVerification.NO_TESTDATA_HASH
193 if self.testdata_hash != testdata_hash:
194 return NeedVerification.TESTDATA_CHANGED
195 if not self.is_success(allow_skip=False):
196 return NeedVerification.NOT_SUCCESS
197 return NeedVerification.NO
199 def is_success(self, *, allow_skip: bool) -> bool:
200 if allow_skip:
201 return all(r.status != ResultStatus.FAILURE for r in self.verifications)
202 return all(r.status == ResultStatus.SUCCESS for r in self.verifications)
205class VerifyCommandResult(BaseModel):
206 total_seconds: float = Field(
207 description="Total number of seconds elapsed for all verification.",
208 )
209 """Total number of seconds elapsed for all verification.
210 """
212 files: dict[ForcePosixPath, FileResult] = Field(
213 default_factory=dict[ForcePosixPath, FileResult],
214 description="The files to be verified.",
215 )
216 """The files to be verified.
217 """
219 @classmethod
220 def parse_file_relative(cls, path: "StrPath") -> "VerifyCommandResult":
221 impl = cls.model_validate_json(pathlib.Path(path).read_bytes())
222 new_files: dict[pathlib.Path, FileResult] = {}
223 for p, f in impl.files.items():
224 rp = to_relative(p)
225 if not rp:
226 logger.warning(
227 "Files in other directories are not subject to verification: %s",
228 p,
229 extra={"github": GitHubMessageParams()},
230 )
231 continue
232 new_files[rp] = f
234 impl.files = new_files
235 return impl
237 def merge(self, other: "VerifyCommandResult") -> "VerifyCommandResult":
238 d = self.files.copy()
239 for k, r in other.files.items():
240 cur = d.get(k)
241 if r.newest or (cur is None) or (not cur.newest):
242 d[k] = r
243 return VerifyCommandResult(
244 total_seconds=self.total_seconds + other.total_seconds,
245 files=d,
246 )
248 def is_success(self, *, allow_skip: bool = True) -> bool:
249 return all(f.is_success(allow_skip=allow_skip) for f in self.files.values())