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

1from sqlalchemy import insert, literal, select 

2from sqlalchemy.orm import Session 

3 

4from couchers.models import HostingMeetupStatusHistory, HostingMeetupStatusSource, User 

5 

6 

7def record_hosting_meetup_status(session: Session, user: User, source: HostingMeetupStatusSource) -> None: 

8 """ 

9 Append the user's current hosting and meetup statuses to their history. 

10 

11 Call this after any code path that may change either status; it's a no-op if the statuses are unchanged since the 

12 last recorded row, so it's safe to call unconditionally. 

13 

14 The snapshot is read from the user's row rather than the in-memory object, so a pending status change has to be 

15 flushed first -- the execute below autoflushes, so this only matters under `no_autoflush`. 

16 """ 

17 latest = ( 

18 select(HostingMeetupStatusHistory.hosting_status, HostingMeetupStatusHistory.meetup_status) 

19 .where(HostingMeetupStatusHistory.user_id == user.id) 

20 .order_by(HostingMeetupStatusHistory.time.desc(), HostingMeetupStatusHistory.id.desc()) 

21 .limit(1) 

22 .subquery() 

23 ) 

24 session.execute( 

25 insert(HostingMeetupStatusHistory).from_select( 

26 ["user_id", "source", "hosting_status", "meetup_status"], 

27 select( 

28 User.id, 

29 literal(source, HostingMeetupStatusHistory.source.type), 

30 User.hosting_status, 

31 User.meetup_status, 

32 ) 

33 .where(User.id == user.id) 

34 .where( 

35 ~select(latest) 

36 .where(latest.c.hosting_status == User.hosting_status) 

37 .where(latest.c.meetup_status == User.meetup_status) 

38 .exists() 

39 ), 

40 ) 

41 )