Coverage for src/couchers/helpers/clusters.py: 96%

1from typing import List 

2 

3from geoalchemy2.shape import from_shape 

4 

5from couchers.models import Cluster, ClusterRole, ClusterSubscription, Node, Page, PageType, PageVersion, Thread 

6 

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

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

9 

10 

11def create_node(session, geom, parent_node_id): 

12 node = Node(geom=from_shape(geom), parent_node_id=parent_node_id) 

13 session.add(node) 

14 session.flush() 

15 return node 

16 

17 

18def create_cluster( 

19 session, 

20 parent_node_id: int, 

21 name: str, 

22 description: str, 

23 creator_user_id: int, 

24 admin_ids: List, 

25 is_community: bool, 

26): 

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

28 cluster = Cluster( 

29 name=name, 

30 description=description, 

31 parent_node_id=parent_node_id, 

32 is_official_cluster=is_community, 

33 ) 

34 session.add(cluster) 

35 session.flush() 

36 main_page = Page( 

37 parent_node=cluster.parent_node, 

38 creator_user_id=creator_user_id, 

39 owner_cluster=cluster, 

40 type=PageType.main_page, 

41 thread=Thread(), 

42 ) 

43 session.add(main_page) 

44 session.flush() 

45 page_version = PageVersion( 

46 page=main_page, 

47 editor_user_id=creator_user_id, 

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

49 content=DEFAULT_PAGE_CONTENT, 

50 ) 

51 session.add(page_version) 

52 for admin_id in admin_ids: 

53 cluster.cluster_subscriptions.append( 

54 ClusterSubscription( 

55 user_id=admin_id, 

56 role=ClusterRole.admin, 

57 ) 

58 ) 

59 return cluster