Coverage for src / competitive_verifier / models / result.py: 96%

1import datetime 

2import enum 

3import pathlib 

4from logging import getLogger 

5from typing import TYPE_CHECKING, Any 

6 

7from pydantic import BaseModel, Field, field_validator 

8 

9from competitive_verifier.log import GitHubMessageParams 

10from competitive_verifier.util import to_relative 

11 

12from .path import ForcePosixPath 

13from .result_status import JudgeStatus, ResultStatus 

14 

15if TYPE_CHECKING: 

16 from _typeshed import StrPath 

17 

18logger = getLogger(__name__) 

19 

20 

21class TestcaseResult(BaseModel): 

22 name: str = Field( 

23 description="The name of test case.", 

24 ) 

25 """The name of test case. 

26 """ 

27 

28 status: JudgeStatus = Field( 

29 description="The result status of the test case.", 

30 ) 

31 """The result status of the test case. 

32 """ 

33 

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 """ 

39 

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 """ 

46 

47 

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 """ 

60 

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 """ 

66 

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 """ 

73 

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 """ 

80 

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 """ 

87 

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 """ 

94 

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 

99 

100 def need_reverifying(self, base_time: datetime.datetime) -> bool: 

101 if self.status != ResultStatus.SUCCESS: 

102 return True 

103 

104 return self.last_execution_time < base_time 

105 

106 

107class NeedVerification(enum.Enum): 

108 """Why a previous ``FileResult`` can't be reused; falsy when it can.""" 

109 

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 

117 def __bool__(self) -> bool: 

118 return self is not NeedVerification.NO 

119 

120 

121class FileResult(BaseModel): 

122 verifications: list[VerificationResult] = Field( 

123 default_factory=list[VerificationResult], 

124 description="The results of each verification.", 

125 ) 

126 """The results of each verification. 

127 """ 

128 

129 content_hash: str | None = Field( 

130 default=None, 

131 description="Digest of the file, its transitive dependencies," 

132 " and verification settings at verification time.", 

133 ) 

134 """Digest of the sources and verification settings at verification time. 

135 """ 

136 

137 newest: bool = Field( 

138 default=True, 

139 description="Whether the verification was performed on the most recent run.", 

140 ) 

141 """Whether the verification was performed on the most recent run. 

142 """ 

143 

144 def need_verification( 

145 self, 

146 base_time: datetime.datetime | None = None, 

147 *, 

148 content_hash: str | None = None, 

149 ) -> bool: 

150 """Whether the file needs verification. 

151 

152 Pass ``base_time`` to compare against the file's modification time, 

153 or ``content_hash`` to compare against the recorded content hash. 

154 """ 

155 return bool( 

156 self.get_verification_reason(base_time=base_time, content_hash=content_hash) 

157 ) 

158 

159 def get_verification_reason( 

160 self, 

161 *, 

162 base_time: datetime.datetime | None = None, 

163 content_hash: str | None = None, 

164 ) -> NeedVerification: 

165 """Return why the file needs verification, or ``NeedVerification.NO``.""" 

166 if len(self.verifications) == 0: 

167 return NeedVerification.NO_RESULT 

168 if base_time is not None: 

169 return self._need_verification_by_time(base_time) 

170 return self._need_verification_by_hash(content_hash) 

171 

172 def _need_verification_by_time( 

173 self, base_time: datetime.datetime 

174 ) -> NeedVerification: 

175 if any(r.status != ResultStatus.SUCCESS for r in self.verifications): 

176 return NeedVerification.NOT_SUCCESS 

177 if any(r.last_execution_time < base_time for r in self.verifications): 

178 return NeedVerification.MODIFIED 

179 return NeedVerification.NO 

180 

181 def _need_verification_by_hash(self, content_hash: str | None) -> NeedVerification: 

182 if self.content_hash is None: 

183 return NeedVerification.NO_CONTENT_HASH 

184 if self.content_hash != content_hash: 

185 return NeedVerification.CONTENT_CHANGED 

186 if not self.is_success(allow_skip=False): 

187 return NeedVerification.NOT_SUCCESS 

188 return NeedVerification.NO 

189 

190 def is_success(self, *, allow_skip: bool) -> bool: 

191 if allow_skip: 

192 return all(r.status != ResultStatus.FAILURE for r in self.verifications) 

193 return all(r.status == ResultStatus.SUCCESS for r in self.verifications) 

194 

195 

196class VerifyCommandResult(BaseModel): 

197 total_seconds: float = Field( 

198 description="Total number of seconds elapsed for all verification.", 

199 ) 

200 """Total number of seconds elapsed for all verification. 

201 """ 

202 

203 files: dict[ForcePosixPath, FileResult] = Field( 

204 default_factory=dict[ForcePosixPath, FileResult], 

205 description="The files to be verified.", 

206 ) 

207 """The files to be verified. 

208 """ 

209 

210 @classmethod 

211 def parse_file_relative(cls, path: "StrPath") -> "VerifyCommandResult": 

212 impl = cls.model_validate_json(pathlib.Path(path).read_bytes()) 

213 new_files: dict[pathlib.Path, FileResult] = {} 

214 for p, f in impl.files.items(): 

215 rp = to_relative(p) 

216 if not rp: 

217 logger.warning( 

218 "Files in other directories are not subject to verification: %s", 

219 p, 

220 extra={"github": GitHubMessageParams()}, 

221 ) 

222 continue 

223 new_files[rp] = f 

224 

225 impl.files = new_files 

226 return impl 

227 

228 def merge(self, other: "VerifyCommandResult") -> "VerifyCommandResult": 

229 d = self.files.copy() 

230 for k, r in other.files.items(): 

231 cur = d.get(k) 

232 if r.newest or (cur is None) or (not cur.newest): 

233 d[k] = r 

234 return VerifyCommandResult( 

235 total_seconds=self.total_seconds + other.total_seconds, 

236 files=d, 

237 ) 

238 

239 def is_success(self, *, allow_skip: bool = True) -> bool: 

240 return all(f.is_success(allow_skip=allow_skip) for f in self.files.values())