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

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 Cluster, ClusterRole, ClusterSubscription, Node, Page, PageType, PageVersion, Thread 

8 

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

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

11 

12 

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

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

15 session.add(node) 

16 session.flush() 

17 return node 

18 

19 

20def create_cluster( 

21 session: Session, 

22 parent_node_id: int, 

23 name: str, 

24 description: str, 

25 creator_user_id: int, 

26 admin_ids: Sequence[int], 

27 is_community: bool, 

28) -> Cluster: 

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

30 cluster = Cluster( 

31 name=name, 

32 description=description, 

33 parent_node_id=parent_node_id, 

34 is_official_cluster=is_community, 

35 ) 

36 session.add(cluster) 

37 session.flush() 

38 main_page = Page( 

39 parent_node=cluster.parent_node, 

40 creator_user_id=creator_user_id, 

41 owner_cluster=cluster, 

42 type=PageType.main_page, 

43 thread=Thread(), 

44 ) 

45 session.add(main_page) 

46 session.flush() 

47 page_version = PageVersion( 

48 page=main_page, 

49 editor_user_id=creator_user_id, 

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

51 content=DEFAULT_PAGE_CONTENT, 

52 ) 

53 session.add(page_version) 

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

55 cluster.cluster_subscriptions.append( 

56 ClusterSubscription( 

57 user_id=admin_id, 

58 role=ClusterRole.admin, 

59 ) 

60 ) 

61 return cluster