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

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: 91 ↛ exitline 91 didn't return from function 'is_not_none' because the condition on line 91 was always true

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: 313 ↛ 314line 313 didn't jump to line 314 because the condition on line 313 was never true

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: 322 ↛ 323line 322 didn't jump to line 323 because the condition on line 322 was never true

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

324 raise InvariantViolationError(msg) 

325 

326 if "." not in domain_part: 326 ↛ 327line 326 didn't jump to line 327 because the condition on line 326 was never true

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: 373 ↛ 375line 373 didn't jump to line 375 because the condition on line 373 was always true

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 

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

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

431 

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

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

434 - No leading or trailing dot or space 

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

436 - Length between 1 and 255 characters 

437 

438 Args: 

439 value: The filename string to validate. 

440 

441 Raises: 

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

443 

444 Example: 

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

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

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

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

449 

450 """ 

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

452 _windows_reserved = re.compile( 

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

454 ) 

455 

456 if not value: 456 ↛ 457line 456 didn't jump to line 457 because the condition on line 456 was never true

457 msg = "Filename must not be empty." 

458 raise InvariantViolationError(msg) 

459 

460 if len(value) > 255: # noqa: PLR2004 460 ↛ 461line 460 didn't jump to line 461 because the condition on line 460 was never true

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

462 raise InvariantViolationError(msg) 

463 

464 if _forbidden_chars.search(value): 464 ↛ 465line 464 didn't jump to line 465 because the condition on line 464 was never true

465 msg = ( 

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

467 " " 

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

469 ) 

470 raise InvariantViolationError(msg) 

471 

472 if value[0] in (".", " ") or value[-1] in (".", " "): 472 ↛ 473line 472 didn't jump to line 473 because the condition on line 472 was never true

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

474 raise InvariantViolationError(msg) 

475 

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

477 if _windows_reserved.match(stem): 477 ↛ 478line 477 didn't jump to line 478 because the condition on line 477 was never true

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

479 raise InvariantViolationError(msg) 

480 

481 

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

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

484 

485 Args: 

486 predicate: The predicate to apply to each element. 

487 

488 Returns: 

489 A predicate function that checks each element. 

490 

491 Raises: 

492 InvariantViolationError: If any element fails the predicate. 

493 

494 Example: 

495 >>> check_all_valid = each(is_valid_filename) 

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

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

498 

499 """ 

500 

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

502 for item in value: 

503 predicate(item) 

504 

505 return unwrapped