Coverage for src / ocarina / dsl / invariants / assertions.py: 12.95%

1"""Invariants assertions. 

2 

3This module provides composable validation predicates for building validation chains. 

4Each invariant is a predicate function that raises an InvariantViolationError 

5when the condition is not met. 

6 

7Predicates can be composed using the validation chain DSL: 

8 validate(email, name="email") 

9 .assert_that(is_str) 

10 .assert_that(is_email) 

11 .execute() 

12 .raise_if_invalid() 

13 

14Higher-order predicates (those that return predicates) allow for parameterized 

15validation: 

16 validate(age, name="age") 

17 .assert_that(is_equal_to(18)) 

18 .otherwise(is_less_than_or_equal_to(65)) 

19 .execute() 

20 .raise_if_invalid() 

21""" 

22 

23import re 

24from datetime import UTC, datetime 

25from pathlib import Path 

26from typing import TYPE_CHECKING, Any 

27 

28from .errors import ( 

29 DuplicatesError, 

30 InvariantViolationError, 

31) 

32 

33if TYPE_CHECKING: 

34 from collections.abc import Callable, Iterable, Sized 

35 

36 from .validate import Predicate 

37 

38 

39def is_str(value: Any) -> None: # noqa: ANN401 

40 """Assert that value is a string. 

41 

42 Args: 

43 value: The value to check. 

44 

45 Raises: 

46 InvariantViolationError: If value is not a string. 

47 

48 Example: 

49 >>> is_str("hello") # OK 

50 >>> is_str(42) # Raises InvariantViolationError 

51 

52 """ 

53 if not isinstance(value, str): 

54 msg = "Expected value to be string." 

55 raise InvariantViolationError(msg) 

56 

57 

58def is_none(value: Any) -> None: # noqa: ANN401 

59 """Assert that value is None. 

60 

61 Args: 

62 value: The value to check. 

63 

64 Raises: 

65 InvariantViolationError: If value is not None. 

66 

67 Example: 

68 >>> is_none(None) # OK 

69 >>> is_none(42) # Raises InvariantViolationError 

70 

71 """ 

72 if value is not None: 

73 msg = f"Expected None, got {value!r}." 

74 raise InvariantViolationError(msg) 

75 

76 

77def is_not_none(value: Any) -> None: # noqa: ANN401 

78 """Assert that value is not None. 

79 

80 Args: 

81 value: The value to check. 

82 

83 Raises: 

84 InvariantViolationError: If value is None. 

85 

86 Example: 

87 >>> is_not_none(42) # OK 

88 >>> is_not_none(None) # Raises InvariantViolationError 

89 

90 """ 

91 if value is None: 

92 msg = "Expected value not to be None." 

93 raise InvariantViolationError(msg) 

94 

95 

96def is_equal_to(cmp: Any) -> Predicate[Any]: # noqa: ANN401 

97 """Create a predicate that asserts value equals the comparison value. 

98 

99 This is a higher-order function that returns a predicate configured 

100 with a specific comparison value. 

101 

102 Args: 

103 cmp: The value to compare against. 

104 

105 Returns: 

106 A predicate function that checks equality. 

107 

108 Raises: 

109 InvariantViolationError: If value != cmp (raised by returned predicate). 

110 

111 Example: 

112 >>> check_is_five = is_equal_to(5) 

113 >>> check_is_five(5) # OK 

114 >>> check_is_five(3) # Raises InvariantViolationError 

115 

116 """ 

117 

118 def unwrapped(value: Any) -> None: # noqa: ANN401 

119 if value != cmp: 

120 msg = f"{value} is not equal to {cmp}." 

121 raise InvariantViolationError(msg) 

122 

123 return unwrapped 

124 

125 

126def is_less_than_or_equal_to(cmp: float) -> Predicate[float]: 

127 """Create a predicate that asserts value <= comparison value. 

128 

129 Args: 

130 cmp: The upper bound (inclusive). 

131 

132 Returns: 

133 A predicate function that checks the bound. 

134 

135 Raises: 

136 InvariantViolationError: If value > cmp (raised by returned predicate). 

137 

138 Example: 

139 >>> check_max_100 = is_less_than_or_equal_to(100) 

140 >>> check_max_100(50) # OK 

141 >>> check_max_100(150) # Raises InvariantViolationError 

142 

143 """ 

144 

145 def unwrapped(value: float) -> None: 

146 if value > cmp: 

147 msg = f"{value} is not less than or equal to {cmp}." 

148 raise InvariantViolationError(msg) 

149 

150 return unwrapped 

151 

152 

153def is_positive(value: float) -> None: 

154 """Assert that value is positive (>= 0). 

155 

156 Args: 

157 value: The numeric value to check. 

158 

159 Raises: 

160 InvariantViolationError: If value < 0. 

161 

162 Example: 

163 >>> is_positive(42) # OK 

164 >>> is_positive(0) # OK 

165 >>> is_positive(-1) # Raises InvariantViolationError 

166 

167 """ 

168 if value < 0: 

169 msg = f"Expected positive number, got {value}." 

170 raise InvariantViolationError(msg) 

171 

172 

173def is_not_zero(value: float) -> None: 

174 """Assert that value is not zero. 

175 

176 Args: 

177 value: The numeric value to check. 

178 

179 Raises: 

180 InvariantViolationError: If value == 0. 

181 

182 Example: 

183 >>> is_not_zero(42) # OK 

184 >>> is_not_zero(-1) # OK 

185 >>> is_not_zero(0) # Raises InvariantViolationError 

186 

187 """ 

188 if value == 0: 

189 msg = "Value must not be zero." 

190 raise InvariantViolationError(msg) 

191 

192 

193def is_in(elements: Iterable[Any]) -> Predicate[Any]: 

194 """Create a predicate that asserts value is in the given collection. 

195 

196 Args: 

197 elements: The collection of allowed values. 

198 

199 Returns: 

200 A predicate function that checks membership. 

201 

202 Raises: 

203 InvariantViolationError: If value not in elements. 

204 

205 Example: 

206 >>> check_color = is_in(["red", "green", "blue"]) 

207 >>> check_color("red") # OK 

208 >>> check_color("yellow") # Raises InvariantViolationError 

209 

210 """ 

211 elms = tuple(elements) 

212 

213 def unwrapped(value: Any) -> None: # noqa: ANN401 

214 if value not in elms: 

215 pretty_elms = ", ".join(map(str, elms)) 

216 msg = f"Value {value!r} must be in: {pretty_elms}." 

217 raise InvariantViolationError(msg) 

218 

219 return unwrapped 

220 

221 

222def is_file(value: str | Path) -> None: 

223 """Assert that the path points to an existing file. 

224 

225 Args: 

226 value: A file path (string or Path object). 

227 

228 Raises: 

229 InvariantViolationError: If the path doesn't point to a file. 

230 

231 Example: 

232 >>> is_file("/path/to/existing/file.txt") # OK 

233 >>> is_file("/path/to/directory") # Raises InvariantViolationError 

234 

235 """ 

236 if not Path(value).is_file(): 

237 msg = f"'{value}' does not point to a file." 

238 raise InvariantViolationError(msg) 

239 

240 

241def is_dir(value: str | Path) -> None: 

242 """Assert that the path points to an existing directory. 

243 

244 Args: 

245 value: A directory path (string or Path object). 

246 

247 Raises: 

248 InvariantViolationError: If the path doesn't point to a directory. 

249 

250 Example: 

251 >>> is_dir("/path/to/directory") # OK 

252 >>> is_dir("/path/to/file.txt") # Raises InvariantViolationError 

253 

254 """ 

255 if not Path(value).is_dir(): 

256 msg = f"'{value}' does not point to a directory." 

257 raise InvariantViolationError(msg) 

258 

259 

260def is_iso_date_string(value: str) -> None: 

261 """Assert that the string is a valid ISO 8601 date string. 

262 

263 Args: 

264 value: The string to validate. 

265 

266 Raises: 

267 InvariantViolationError: If the string is not a valid ISO date. 

268 

269 Example: 

270 >>> is_iso_date_string("2025-12-18") # OK 

271 >>> is_iso_date_string("2025-12-18T10:30:00Z") # OK 

272 >>> is_iso_date_string("invalid") # Raises InvariantViolationError 

273 

274 """ 

275 try: 

276 datetime.fromisoformat(value) 

277 except Exception as exc: 

278 msg = f"'{value}' is not a valid ISO date string." 

279 raise InvariantViolationError(msg) from exc 

280 

281 

282def is_iso_utc_date_string(value: str) -> None: 

283 """Assert that the string is a valid ISO 8601 UTC date string. 

284 

285 The date must be in UTC timezone (not just a valid ISO date). 

286 

287 Args: 

288 value: The string to validate. 

289 

290 Raises: 

291 InvariantViolationError: If the string is not a valid UTC ISO date. 

292 

293 Example: 

294 >>> is_iso_utc_date_string("2025-12-18T10:30:00+00:00") # OK 

295 >>> is_iso_utc_date_string("2025-12-18T10:30:00Z") # OK 

296 >>> is_iso_utc_date_string("2025-12-18T10:30:00+02:00") # Not UTC, raises error 

297 >>> is_iso_utc_date_string("2025-12-18") # No timezone, raises error 

298 

299 """ 

300 try: 

301 dt = datetime.fromisoformat(value) 

302 except Exception as exc: 

303 msg = f"'{value}' is not a valid ISO date string." 

304 raise InvariantViolationError(msg) from exc 

305 

306 if dt.tzinfo != UTC: 

307 msg = f"'{value}' is not in UTC (tz={dt.tzinfo})." 

308 raise InvariantViolationError(msg) 

309 

310 

311def is_email(value: str) -> None: 

312 """Assert that the string is a valid email address (fast check).""" 

313 if " " in value: 

314 msg = f"'{value}' must not contain whitespace." 

315 raise InvariantViolationError(msg) 

316 

317 if value.count("@") != 1: 

318 msg = f"'{value}' must contain exactly one '@' character." 

319 raise InvariantViolationError(msg) 

320 

321 local_part, domain_part = value.split("@") 

322 if not local_part or not domain_part: 

323 msg = f"'{value}' must have non-empty local and domain parts." 

324 raise InvariantViolationError(msg) 

325 

326 if "." not in domain_part: 

327 msg = f"Domain part of '{value}' must contain at least one '.'." 

328 raise InvariantViolationError(msg) 

329 

330 

331def has_unique_elements( 

332 *, 

333 key: Callable[[Any], Any] | None = None, 

334): 

335 """Create a predicate that asserts all elements in a collection are unique. 

336 

337 Uses strict type checking: 1, 1.0, and True are considered different values. 

338 Supports unhashable types (lists, dicts, sets) by comparing them directly. 

339 

340 Args: 

341 key: Optional function to extract comparison key from each element. 

342 If None, elements are compared directly. 

343 

344 Returns: 

345 A predicate function that checks uniqueness. 

346 

347 Raises: 

348 DuplicatesError: If duplicates are found (raised by returned predicate). 

349 

350 Example: 

351 >>> check_unique = has_unique_elements() 

352 >>> check_unique([1, 2, 3]) # OK 

353 >>> check_unique([1, 2, 2, 3]) # Raises DuplicatesError 

354 >>> check_unique([[1, 2], [3, 4]]) # OK (unhashable types supported) 

355 

356 """ 

357 

358 def unwrapped(value: Iterable[Any]) -> None: 

359 items = list(value) 

360 key_fn = key or (lambda x: x) 

361 

362 seen: list[tuple[type, Any]] = [] 

363 duplicates = [] 

364 

365 for item in items: 

366 keyed = key_fn(item) 

367 typed_key = (type(keyed), keyed) 

368 

369 found = False 

370 for seen_key in seen: 

371 if typed_key[0] == seen_key[0] and typed_key[1] == seen_key[1]: 

372 found = True 

373 if keyed not in duplicates: 

374 duplicates.append(keyed) 

375 break 

376 

377 if not found: 

378 seen.append(typed_key) 

379 

380 if duplicates: 

381 raise DuplicatesError(duplicates) 

382 

383 return unwrapped 

384 

385 

386def is_empty(value: Sized) -> None: 

387 """Assert that the collection is empty. 

388 

389 Args: 

390 value: Any collection with a length (list, dict, string, etc.). 

391 

392 Raises: 

393 InvariantViolationError: If the collection is not empty. 

394 

395 Example: 

396 >>> is_empty([]) # OK 

397 >>> is_empty("") # OK 

398 >>> is_empty([1, 2, 3]) # Raises InvariantViolationError 

399 

400 """ 

401 if len(value) != 0: 

402 msg = "Value must be empty." 

403 raise InvariantViolationError(msg) 

404 

405 

406def is_truthy(value: Any) -> None: # noqa: ANN401 -> This is intentional. 

407 """Assert that value is truthy in Python's boolean context. 

408 

409 Args: 

410 value: The value to check. 

411 

412 Raises: 

413 InvariantViolationError: If value is falsy (False, None, 0, "", [], etc.). 

414 

415 Example: 

416 >>> is_truthy(42) # OK 

417 >>> is_truthy("hello") # OK 

418 >>> is_truthy([1]) # OK 

419 >>> is_truthy(0) # Raises InvariantViolationError 

420 >>> is_truthy("") # Raises InvariantViolationError 

421 >>> is_truthy([]) # Raises InvariantViolationError 

422 

423 """ 

424 if not value: 

425 msg = "Value must be truthy." 

426 raise InvariantViolationError(msg) 

427 

428 

429_forbidden_chars = re.compile(r'[\x00-\x1f\\/:*?"<>|]') 

430_windows_reserved = re.compile( 

431 r"^(?:CON|PRN|AUX|NUL|COM[1-9]|LPT[1-9])$", re.IGNORECASE 

432) 

433 

434 

435def is_valid_filename(value: str) -> None: 

436 r"""Assert that the string is a valid filename on all major OSes. 

437 

438 Applies the union of restrictions from Windows, Linux, and macOS: 

439 - No forbidden characters: \\ / : * ? " < > | and control chars (U+0000; U+001F) 

440 - No leading or trailing dot or space 

441 - Not a Windows reserved name (CON, NUL, COM1…COM9, LPT1…LPT9, etc.) 

442 - Length between 1 and 255 characters 

443 

444 Args: 

445 value: The filename string to validate. 

446 

447 Raises: 

448 InvariantViolationError: If the string is not a valid cross-platform filename. 

449 

450 Example: 

451 >>> is_valid_filename("my_test_runner") # OK 

452 >>> is_valid_filename("test/run") # Raises InvariantViolationError 

453 >>> is_valid_filename("CON") # Raises InvariantViolationError 

454 >>> is_valid_filename(".hidden") # Raises InvariantViolationError 

455 

456 """ 

457 if not value: 

458 msg = "Filename must not be empty." 

459 raise InvariantViolationError(msg) 

460 

461 if len(value) > 255: # noqa: PLR2004 

462 msg = f"Filename '{value}' exceeds 255 characters." 

463 raise InvariantViolationError(msg) 

464 

465 if _forbidden_chars.search(value): 

466 msg = ( 

467 f"Filename '{value}' contains forbidden characters" 

468 " " 

469 '(control chars or one of: \\ / : * ? " < > |).' 

470 ) 

471 raise InvariantViolationError(msg) 

472 

473 if value[0] in (".", " ") or value[-1] in (".", " "): 

474 msg = f"Filename '{value}' must not start or end with a dot or a space." 

475 raise InvariantViolationError(msg) 

476 

477 stem = value.split(".", 1)[0] 

478 if _windows_reserved.match(stem): 

479 msg = f"Filename '{value}' uses a reserved Windows device name." 

480 raise InvariantViolationError(msg) 

481 

482 

483def each(predicate: Predicate[Any]) -> Predicate[Iterable[Any]]: 

484 """Create a predicate that applies a predicate to each element of a collection. 

485 

486 Args: 

487 predicate: The predicate to apply to each element. 

488 

489 Returns: 

490 A predicate function that checks each element. 

491 

492 Raises: 

493 InvariantViolationError: If any element fails the predicate. 

494 

495 Example: 

496 >>> check_all_valid = each(is_valid_filename) 

497 >>> check_all_valid(["my_test", "other_test"]) # OK 

498 >>> check_all_valid(["my_test", "CON"]) # Raises InvariantViolationError 

499 

500 """ 

501 

502 def unwrapped(value: Iterable[Any]) -> None: 

503 for item in value: 

504 predicate(item) 

505 

506 return unwrapped