Coverage for tests/query/test_boundary_validation.py: 100%

1"""Public boundary validation for constrained numeric arguments.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Awaitable, Callable 

6from typing import cast 

7 

8from pydantic import ValidationError 

9from snektest import assert_in, assert_isinstance, assert_raises, test 

10 

11from snekql import ( 

12 Database, 

13 DatabaseRuntimeError, 

14 Fetched, 

15 Model, 

16 Pending, 

17 QueryConstructionError, 

18 Text, 

19 select, 

20) 

21from tests.helpers import NULL_LOGGER 

22 

23 

24class BoundaryUser[S = Pending](Model[S, "BoundaryUser[Fetched]"]): 

25 """Table model used by select boundary validation tests.""" 

26 

27 email: BoundaryUser.Col[str] = Text(nullable=False) 

28 

29 

30@test(mark="fast") 

31def select_limit_and_offset_reject_invalid_values_at_boundary() -> None: 

32 """Constrained select integers are validated by public chain methods.""" 

33 

34 limit_fn = cast("Callable[[object], object]", select(BoundaryUser).all().limit) 

35 offset_fn = cast("Callable[[object], object]", select(BoundaryUser).all().offset) 

36 

37 with assert_raises(QueryConstructionError): 

38 _ = limit_fn(-1) 

39 

40 with assert_raises(QueryConstructionError): 

41 _ = limit_fn(True) 

42 

43 with assert_raises(QueryConstructionError): 

44 _ = offset_fn("1") 

45 

46 

47@test(mark="fast") 

48def boundary_validation_uses_pydantic_message_with_domain_error() -> None: 

49 """Boundary validation preserves Pydantic detail on domain exceptions.""" 

50 

51 limit_fn = cast("Callable[[object], object]", select(BoundaryUser).all().limit) 

52 

53 with assert_raises(QueryConstructionError) as error: 

54 _ = limit_fn(True) 

55 

56 assert_isinstance(error.exception.__cause__, ValidationError) 

57 assert_in("Input should be a valid integer", str(error.exception)) 

58 

59 

60@test(mark="medium") 

61async def database_numeric_configuration_rejects_invalid_values_at_boundary() -> None: 

62 """Runtime numeric configuration rejects invalid values as domain errors.""" 

63 

64 initialize_fn = cast("Callable[..., Awaitable[object]]", Database.initialize) 

65 

66 with assert_raises(DatabaseRuntimeError): 

67 _ = await initialize_fn(logger=NULL_LOGGER, database=":memory:", pool_size=True) 

68 

69 database = await Database.initialize(logger=NULL_LOGGER, database=":memory:") 

70 try: 

71 transaction_fn = cast("Callable[..., object]", database.transaction) 

72 with assert_raises(DatabaseRuntimeError): 

73 _ = transaction_fn(timeout="1") 

74 finally: 

75 await database.close()