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.stanza import Forwarded 

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 

30class HistoryMessage: 

31 def __init__( 

32 self, stanza: Union[Message, str], when: Optional[datetime] = None 

33 ) -> None: 

34 if isinstance(stanza, str): 

35 from_db = True 

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

37 else: 

38 from_db = False 

39 

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

41 self.when: datetime = ( 

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

43 ) 

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