Coverage for app/backend/src/couchers/helpers/references.py: 100%

1from typing import TYPE_CHECKING, Any 

2 

3from sqlalchemy import select 

4from sqlalchemy.orm import InstrumentedAttribute, aliased 

5from sqlalchemy.sql import Select, exists, func, or_ 

6 

7from couchers.models import HostRequest, Reference, ReferenceType, User 

8from couchers.sql import _shadow_clause 

9 

10if TYPE_CHECKING: 

11 from couchers.context import CouchersContext 

12 

13 

14def where_reference_user_visible[T: tuple[Any, ...]]( 

15 statement: Select[T], context: CouchersContext, user_id_column: InstrumentedAttribute[int] 

16) -> Select[T]: 

17 """ 

18 Filters references based on the visibility of the user in the given column (the writer 

19 or the subject of the reference). 

20 

21 Deliberately weaker than users_visible: references involving deleted or blocked users 

22 stay visible so reference history is preserved; only banned or shadowed (to others) 

23 users hide their references. Both the reference list (ListReferences) and the reference 

24 count (get_num_references) must use this, otherwise the count diverges from the list. 

25 """ 

26 return statement.where( 

27 exists( 

28 select(1) 

29 .select_from(User) 

30 .where(User.id == user_id_column) 

31 .where(User.banned_at.is_(None)) 

32 .where(_shadow_clause(context, User)) 

33 .correlate_except(User) 

34 ) 

35 ) 

36 

37 

38def where_references_not_hidden_by_reciprocity[T: tuple[Any, ...]](statement: Select[T]) -> Select[T]: 

39 """ 

40 Filters out references that are still hidden by the reciprocal-reference rule. 

41 

42 A host/surf reference stays hidden until either the recipient has written their 

43 reciprocal reference or the 2-week window to write one has closed; friend 

44 references are always visible. 

45 

46 Apply this to any query that selects from Reference. Both the reference list 

47 (ListReferences) and the reference count (get_num_references) must use it, 

48 otherwise the count includes references the list hides, leaking the existence 

49 of a still-hidden reference. 

50 """ 

51 other_reference = aliased(Reference) 

52 reciprocal_written = exists( 

53 select(other_reference.id) 

54 .where(other_reference.host_request_id == Reference.host_request_id) 

55 .where(other_reference.from_user_id == Reference.to_user_id) 

56 .where(other_reference.reference_type != ReferenceType.friend) 

57 ) 

58 window_closed = exists( 

59 select(HostRequest.conversation_id) 

60 .where(HostRequest.conversation_id == Reference.host_request_id) 

61 .where(HostRequest.end_time_to_write_reference < func.now()) 

62 ) 

63 return statement.where( 

64 or_( 

65 Reference.reference_type == ReferenceType.friend, 

66 reciprocal_written, 

67 window_closed, 

68 ) 

69 )