Coverage for app/backend/src/couchers/moderation/utils.py: 100%

1""" 

2Utility functions for the Unified Moderation System (UMS) 

3""" 

4 

5from collections.abc import Callable 

6from typing import Any 

7 

8from sqlalchemy.orm import Session 

9 

10from couchers.metrics import observe_moderation_action, observe_moderation_queue_item_created 

11from couchers.models import ( 

12 ModerationAction, 

13 ModerationLog, 

14 ModerationObjectType, 

15 ModerationQueueItem, 

16 ModerationState, 

17 ModerationTrigger, 

18 ModerationVisibility, 

19 get_moderated_models, 

20) 

21 

22 

23def create_moderation( 

24 session: Session, 

25 object_type: ModerationObjectType, 

26 object_id: int | Callable[[int], int], 

27 # None for objects that are their own creator, whose id isn't known until the callback has run 

28 creator_user_id: int | None = None, 

29 data: Any | None = None, 

30) -> ModerationState: 

31 has_own_visibility_mechanism = get_moderated_models()[object_type].has_own_visibility_mechanism 

32 visibility = None if has_own_visibility_mechanism else ModerationVisibility.shadowed 

33 

34 # Handle callback pattern for circular dependencies 

35 if callable(object_id): 

36 moderation_state = ModerationState( 

37 object_type=object_type, 

38 object_id=0, # Placeholder 

39 visibility=visibility, 

40 ) 

41 session.add(moderation_state) 

42 session.flush() 

43 

44 # Call the callback to create the object and get its ID 

45 actual_object_id = object_id(moderation_state.id) 

46 moderation_state.object_id = actual_object_id 

47 else: 

48 actual_object_id = object_id 

49 moderation_state = ModerationState( 

50 object_type=object_type, 

51 object_id=object_id, 

52 visibility=visibility, 

53 ) 

54 session.add(moderation_state) 

55 session.flush() 

56 

57 session.add( 

58 ModerationLog( 

59 moderation_state_id=moderation_state.id, 

60 action=ModerationAction.create, 

61 moderator_user_id=creator_user_id if creator_user_id is not None else actual_object_id, 

62 new_visibility=visibility, 

63 reason="Object created.", 

64 ) 

65 ) 

66 

67 if not has_own_visibility_mechanism: 

68 session.add( 

69 ModerationQueueItem( 

70 moderation_state_id=moderation_state.id, 

71 trigger=ModerationTrigger.initial_review, 

72 reason="Object created.", 

73 data=data, 

74 ) 

75 ) 

76 observe_moderation_queue_item_created(ModerationTrigger.initial_review, object_type) 

77 

78 observe_moderation_action(ModerationAction.create, object_type) 

79 

80 session.flush() 

81 

82 return moderation_state