Coverage for tests/test_schema_plan.py: 100%

1"""Shared schema plan tests.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any, ClassVar 

6 

7from snektest import assert_eq, assert_raises, test 

8 

9from snekql import Index, Model, Pending, SchemaError, Text 

10from snekql._schema_plan import build_schema_plan 

11 

12 

13@test(mark="fast") 

14def schema_plan_preserves_model_order_and_normalizes_indexes() -> None: 

15 """Schema startup derives table names and managed indexes once per model.""" 

16 

17 class User[S = Pending](Model[S, "User[object]"]): 

18 """First table model in a schema plan.""" 

19 

20 email: User.Col[str] = Text(nullable=False, unique=True) 

21 status: User.Col[str] = Text(nullable=False) 

22 

23 __indexes__: ClassVar[list[Index[Any]]] = [Index(status)] 

24 

25 class AuditLog[S = Pending](Model[S, "AuditLog[object]"]): 

26 """Second table model in a schema plan.""" 

27 

28 message: AuditLog.Col[str] = Text(nullable=False) 

29 

30 plan = build_schema_plan([User, AuditLog]) 

31 

32 assert_eq([model.table_name for model in plan.models], ["user", "audit_log"]) 

33 assert_eq( 

34 [index.name for index in plan.models[0].indexes], 

35 ["ux_user_email", "ix_user_status"], 

36 ) 

37 assert_eq([index.name for index in plan.models[1].indexes], []) 

38 

39 

40@test(mark="fast") 

41def schema_plan_rejects_duplicate_resolved_names() -> None: 

42 """Schema startup validates duplicate table and index names in one plan.""" 

43 

44 class First[S = Pending](Model[S, "First[object]"]): 

45 """First table model using a duplicate index name.""" 

46 

47 email: First.Col[str] = Text(nullable=False) 

48 __indexes__: ClassVar[list[Index[Any]]] = [Index(email, name="ix_duplicate")] 

49 

50 class Second[S = Pending](Model[S, "Second[object]"]): 

51 """Second table model using a duplicate index name.""" 

52 

53 email: Second.Col[str] = Text(nullable=False) 

54 __indexes__: ClassVar[list[Index[Any]]] = [Index(email, name="ix_duplicate")] 

55 

56 class DuplicateFirst[S = Pending](Model[S, "DuplicateFirst[object]"]): 

57 """First table model using a duplicate table name.""" 

58 

59 __tablename__ = "duplicate" 

60 email: DuplicateFirst.Col[str] = Text(nullable=False) 

61 

62 class DuplicateSecond[S = Pending](Model[S, "DuplicateSecond[object]"]): 

63 """Second table model using a duplicate table name.""" 

64 

65 __tablename__ = "duplicate" 

66 email: DuplicateSecond.Col[str] = Text(nullable=False) 

67 

68 with assert_raises(SchemaError): 

69 _ = build_schema_plan([First, Second]) 

70 

71 with assert_raises(SchemaError): 

72 _ = build_schema_plan([DuplicateFirst, DuplicateSecond])