Coverage for app / backend / src / couchers / helpers / clusters.py: 94%

1from collections.abc import Sequence 

2 

3from geoalchemy2.shape import from_shape 

4from shapely.geometry.base import BaseGeometry 

5from sqlalchemy.orm import Session 

6 

7from couchers.models import ( 

8 Cluster, 

9 ClusterRole, 

10 ClusterSubscription, 

11 Node, 

12 NodeType, 

13 Page, 

14 PageType, 

15 PageVersion, 

16 Thread, 

17) 

18 

19CHILD_NODE_TYPE = { 

20 None: NodeType.world, 

21 NodeType.world: NodeType.macroregion, 

22 NodeType.macroregion: NodeType.region, 

23 NodeType.region: NodeType.subregion, 

24 NodeType.subregion: NodeType.locality, 

25 NodeType.locality: NodeType.sublocality, 

26} 

27 

28DEFAULT_PAGE_CONTENT = "There is nothing here yet..." 

29DEFAULT_PAGE_TITLE_TEMPLATE = "Main page for the {name} {type}" 

30 

31 

32def create_node(session: Session, geom: BaseGeometry, parent_node_id: int | None, node_type: NodeType) -> Node: 

33 node = Node(geom=from_shape(geom), parent_node_id=parent_node_id, node_type=node_type) 

34 session.add(node) 

35 session.flush() 

36 return node 

37 

38 

39def create_cluster( 

40 session: Session, 

41 parent_node_id: int, 

42 name: str, 

43 description: str, 

44 creator_user_id: int, 

45 admin_ids: Sequence[int], 

46 is_community: bool, 

47) -> Cluster: 

48 cluster_type = "community" if is_community else "group" 

49 cluster = Cluster( 

50 name=name, 

51 description=description, 

52 parent_node_id=parent_node_id, 

53 is_official_cluster=is_community, 

54 ) 

55 session.add(cluster) 

56 session.flush() 

57 thread = Thread() 

58 session.add(thread) 

59 session.flush() 

60 main_page = Page( 

61 parent_node_id=cluster.parent_node_id, 

62 creator_user_id=creator_user_id, 

63 owner_cluster_id=cluster.id, 

64 type=PageType.main_page, 

65 thread_id=thread.id, 

66 ) 

67 session.add(main_page) 

68 session.flush() 

69 page_version = PageVersion( 

70 page_id=main_page.id, 

71 editor_user_id=creator_user_id, 

72 title=DEFAULT_PAGE_TITLE_TEMPLATE.format(name=name, type=cluster_type), 

73 content=DEFAULT_PAGE_CONTENT, 

74 ) 

75 session.add(page_version) 

76 for admin_id in admin_ids: 76 ↛ 77line 76 didn't jump to line 77 because the loop on line 76 never started

77 cluster.cluster_subscriptions.append( 

78 ClusterSubscription( 

79 user_id=admin_id, 

80 cluster_id=cluster.id, 

81 role=ClusterRole.admin, 

82 ) 

83 ) 

84 return cluster