Coverage for slidge/space/space.py: 91%

1from collections.abc import Iterable 

2from pathlib import Path 

3from typing import Any 

4 

5from typing_extensions import TypedDict 

6 

7from ..contact import LegacyContact 

8from ..core.mixins.avatar import ImageMixin 

9from ..db.avatar import CachedAvatar 

10from ..db.models import Space 

11from ..util.types import AnySession, Avatar, ExtraAttrMapping 

12 

13 

14class _SpaceProps(TypedDict, total=False): 

15 name: str | None 

16 description: str | None 

17 member_count: int | None 

18 creator_pk: int | None 

19 

20 

21class LegacySpace[T: ExtraAttrMapping = Any](ImageMixin[T, _SpaceProps]): 

22 """ 

23 A :xep:`0503` space. 

24 """ 

25 

26 _model: Space 

27 _stored: Space 

28 

29 def __init__(self, session: AnySession, *, stored: Space) -> None: 

30 self._session = session 

31 self.log = session.log 

32 self.xmpp = session.xmpp 

33 self._stored = stored 

34 super().__init__() 

35 

36 @property 

37 def name(self) -> str | None: 

38 return self._stored.name 

39 

40 @name.setter 

41 def name(self, name: str | None) -> None: 

42 if self._update_stored_attribute(name=name): 

43 self.__broadcast_meta() 

44 

45 @property 

46 def description(self) -> str | None: 

47 return self._stored.description 

48 

49 @description.setter 

50 def description(self, description: str | None) -> None: 

51 if self._update_stored_attribute(description=description): 

52 self.__broadcast_meta() 

53 

54 @property 

55 def member_count(self) -> int | None: 

56 return self._stored.member_count 

57 

58 @member_count.setter 

59 def member_count(self, member_count: int | None) -> None: 

60 if self._update_stored_attribute(member_count=member_count): 

61 self.__broadcast_meta() 

62 

63 @property 

64 def owners(self) -> list[LegacyContact]: 

65 with self.xmpp.store.session() as orm: 

66 orm.add(self._stored) 

67 return [self._session.contacts._from_store(o) for o in self._stored.owners] 

68 

69 @owners.setter 

70 def owners(self, owners: Iterable[LegacyContact]) -> None: 

71 with self.xmpp.store.session(expire_on_commit=False) as orm: 

72 orm.add(self._stored) 

73 self._stored.owners = [orm.merge(o._stored) for o in owners] 

74 orm.commit() 

75 

76 @property 

77 def creator(self) -> LegacyContact | None: 

78 self._stored.ensure_relation("creator", self.xmpp.store.session) 

79 return self._session.contacts._from_store(self._stored.creator) # type:ignore[no-any-return] 

80 

81 @creator.setter 

82 def creator(self, creator: LegacyContact | None) -> None: 

83 with self.xmpp.store.session(expire_on_commit=False) as orm: 

84 orm.add(self._stored) 

85 if self._stored.creator == creator: 

86 return 

87 self._stored.creator = ( 

88 None if creator is None else orm.merge(creator._stored) 

89 ) 

90 orm.commit() 

91 

92 @property 

93 def banner(self) -> Avatar | None: 

94 """ 

95 This property can be used to set or unset the banner. 

96 """ 

97 stored = self._get_stored_image("banner") 

98 if stored is None: 

99 return None 

100 return stored.to_meta() 

101 

102 @banner.setter 

103 def banner(self, banner: Avatar | Path | str | None) -> None: 

104 self._image_setter(banner, "banner") 

105 

106 def _post_image_update( 

107 self, cached_avatar: CachedAvatar | None, *, attr: str = "avatar" 

108 ) -> None: 

109 if cached_avatar is not None and cached_avatar._stored.http_id is None: 

110 # happens when we couldn't upload the image anywhere 

111 return 

112 self.__broadcast([f"urn:xmpp:spaces:{attr}:metadata:0"]) 

113 

114 @property 

115 def node(self) -> str: 

116 return self._session.spaces.legacy_id_to_node(self.legacy_id) 

117 

118 @property 

119 def legacy_id(self) -> str: 

120 return self._stored.legacy_id 

121 

122 def _on_delete(self) -> None: 

123 msg = self.xmpp.make_message( 

124 mfrom=self.xmpp.boundjid.bare, mto=self._session.user.jid 

125 ) 

126 msg["pubsub_event"]["delete"]["node"] = self.node 

127 msg.send() 

128 

129 def __broadcast_meta(self) -> None: 

130 if not self._session.rooms.is_filling: 

131 self.xmpp.pubsub.broadcast_space_metadata( 

132 self._session, self._stored, self.node 

133 ) 

134 

135 def __broadcast(self, item_ids: list[str]) -> None: 

136 if not self._session.rooms.is_filling: 

137 self._stored.ensure_relation("avatar", self.xmpp.store.session) 

138 self._stored.ensure_relation("banner", self.xmpp.store.session) 

139 self.xmpp.pubsub.broadcast_space( 

140 self._session, self._stored, self.node, item_ids 

141 )