Coverage for slidge / util / archive_msg.py: 95%

1import uuid 

2from copy import copy 

3from datetime import UTC, datetime 

4from xml.etree import ElementTree as ET 

5 

6from slixmpp import Message 

7from slixmpp.plugins.xep_0297.stanza import Forwarded 

8from slixmpp.xmlstream import StanzaBase 

9 

10 

11def fix_namespaces( 

12 xml: ET.Element, 

13 old: str = "{jabber:component:accept}", 

14 new: str = "{jabber:client}", 

15) -> None: 

16 """ 

17 Hack to fix namespaces between jabber:component and jabber:client 

18 

19 Acts in-place. 

20 

21 :param xml: 

22 :param old: 

23 :param new: 

24 """ 

25 xml.tag = xml.tag.replace(old, new) 

26 for child in xml: 

27 fix_namespaces(child, old, new) 

28 

29 

30def set_client_namespace(stanza: StanzaBase) -> None: 

31 fix_namespaces(stanza.xml) 

32 

33 

34class HistoryMessage: 

35 def __init__(self, stanza: Message | str, when: datetime | None = None) -> None: 

36 if isinstance(stanza, str): 

37 from_db = True 

38 stanza = Message(xml=ET.fromstring(stanza)) 

39 else: 

40 from_db = False 

41 

42 self.id = stanza["stanza_id"]["id"] or uuid.uuid4().hex 

43 self.when: datetime = when or stanza["delay"]["stamp"] or datetime.now(tz=UTC) 

44 

45 if not from_db: 

46 del stanza["delay"] 

47 del stanza["markable"] 

48 del stanza["hint"] 

49 del stanza["chat_state"] 

50 if not stanza["body"]: 

51 del stanza["body"] 

52 fix_namespaces(stanza.xml) 

53 

54 self.stanza: Message = stanza 

55 

56 @property 

57 def stanza_component_ns(self) -> Message: 

58 stanza = copy(self.stanza) 

59 fix_namespaces( 

60 stanza.xml, old="{jabber:client}", new="{jabber:component:accept}" 

61 ) 

62 return stanza 

63 

64 def forwarded(self) -> Forwarded: 

65 forwarded = Forwarded() 

66 forwarded["delay"]["stamp"] = self.when 

67 forwarded.append(self.stanza) 

68 return forwarded 

69 

70 @property 

71 def occupant_id(self) -> str: 

72 return self.stanza["occupant-id"]["id"]