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

1from copy import copy 

2from datetime import datetime, timezone 

3from typing import Optional, Union 

4from uuid import uuid4 

5from xml.etree import ElementTree as ET 

6 

7from slixmpp import Message 

8from slixmpp.plugins.xep_0297 import Forwarded 

9 

10 

11def fix_namespaces(xml, old="{jabber:component:accept}", new="{jabber:client}"): 

12 """ 

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

14 

15 Acts in-place. 

16 

17 :param xml: 

18 :param old: 

19 :param new: 

20 """ 

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

22 for child in xml: 

23 fix_namespaces(child, old, new) 

24 

25 

26class HistoryMessage: 

27 def __init__(self, stanza: Union[Message, str], when: Optional[datetime] = None): 

28 if isinstance(stanza, str): 

29 from_db = True 

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

31 else: 

32 from_db = False 

33 

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

35 self.when: datetime = ( 

36 when or stanza["delay"]["stamp"] or datetime.now(tz=timezone.utc) 

37 ) 

38 

39 if not from_db: 

40 del stanza["delay"] 

41 del stanza["markable"] 

42 del stanza["hint"] 

43 del stanza["chat_state"] 

44 if not stanza["body"]: 

45 del stanza["body"] 

46 fix_namespaces(stanza.xml) 

47 

48 self.stanza: Message = stanza 

49 

50 @property 

51 def stanza_component_ns(self): 

52 stanza = copy(self.stanza) 

53 fix_namespaces( 

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

55 ) 

56 return stanza 

57 

58 def forwarded(self): 

59 forwarded = Forwarded() 

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

61 forwarded.append(self.stanza) 

62 return forwarded