Coverage for app/backend/src/tests/test_sql_linkage.py: 97%

1import re 

2from annotationlib import Format, get_annotations 

3from typing import Any 

4 

5import pytest 

6from sqlalchemy import CompoundSelect, Select, select 

7from sqlalchemy.ext.hybrid import hybrid_method 

8from sqlalchemy_utils.view import CreateView 

9 

10import couchers.materialized_views # noqa: F401 -- registers the views 

11from couchers.models import Base, StrongVerificationAttempt, User 

12from tests.sql_linkage import find_unkeyed_joins 

13 

14_MAPPED_CLASSES = {mapper.class_.__name__: mapper.class_ for mapper in Base.registry.mappers} 

15 

16 

17def _materialized_views() -> list[tuple[str, Select[Any] | CompoundSelect[Any]]]: 

18 return sorted( 

19 (listener.name, listener.selectable) 

20 for listener in Base.metadata.dispatch.after_create 

21 if isinstance(listener, CreateView) 

22 ) 

23 

24 

25def _subject_predicates() -> list[tuple[str, Any, Any]]: 

26 found = [] 

27 for model in _MAPPED_CLASSES.values(): 

28 for name, attribute in vars(model).items(): 

29 if name.startswith("_") or not isinstance(attribute, hybrid_method): 

30 continue 

31 # as strings: models import each other under TYPE_CHECKING 

32 annotations = get_annotations(attribute.func, format=Format.STRING) 

33 for parameter, annotation in annotations.items(): 

34 if parameter == "return": 

35 continue 

36 for subject_name in re.findall(r"\w+", annotation): 36 ↛ 33line 36 didn't jump to line 33 because the loop on line 36 didn't complete

37 if subject := _MAPPED_CLASSES.get(subject_name): 37 ↛ 36line 37 didn't jump to line 36 because the condition on line 37 was always true

38 found.append((f"{model.__name__}.{name}", getattr(model, name), subject)) 

39 break 

40 return found 

41 

42 

43_MATERIALIZED_VIEWS = _materialized_views() 

44_SUBJECT_PREDICATES = _subject_predicates() 

45 

46 

47def test_the_checks_below_cover_something(): 

48 assert _MATERIALIZED_VIEWS 

49 assert _SUBJECT_PREDICATES 

50 

51 

52@pytest.mark.parametrize(("name", "selectable"), _MATERIALIZED_VIEWS, ids=[name for name, _ in _MATERIALIZED_VIEWS]) 

53def test_materialized_view_tables_are_linked_by_keys(name, selectable): 

54 assert not (problems := find_unkeyed_joins(selectable, name)), "\n".join(problems) 

55 

56 

57@pytest.mark.parametrize( 

58 ("label", "predicate", "subject"), _SUBJECT_PREDICATES, ids=[label for label, _, _ in _SUBJECT_PREDICATES] 

59) 

60def test_subject_predicates_bind_their_subject(label, predicate, subject): 

61 # no explicit join, so the predicate is the only thing that can link the tables 

62 statement = select(subject.id).where(predicate(subject)) 

63 

64 assert not (problems := find_unkeyed_joins(statement, label)), "\n".join(problems) 

65 

66 

67def test_the_lite_users_bug_is_caught(): 

68 buggy = ( 

69 select(User.id) 

70 .select_from(StrongVerificationAttempt) 

71 .where(StrongVerificationAttempt.passport_date_of_birth == User.birthdate) 

72 ) 

73 

74 problems = find_unkeyed_joins(buggy, "sv_subquery") 

75 

76 assert len(problems) == 1 

77 assert "strong_verification_attempts" in problems[0] 

78 assert "users.birthdate" in problems[0] 

79 

80 

81def test_a_keyed_join_is_accepted(): 

82 fine = ( 

83 select(User.id) 

84 .select_from(StrongVerificationAttempt) 

85 .join(User, User.id == StrongVerificationAttempt.user_id) 

86 .where(StrongVerificationAttempt.passport_date_of_birth == User.birthdate) 

87 ) 

88 

89 assert find_unkeyed_joins(fine, "sv_subquery") == [] 

90 

91 

92def test_a_link_under_an_or_does_not_count(): 

93 either = select(User.id).where( 

94 (StrongVerificationAttempt.user_id == User.id) 

95 | (StrongVerificationAttempt.passport_date_of_birth == User.birthdate) 

96 ) 

97 

98 assert len(find_unkeyed_joins(either, "either")) == 1