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

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 

9from slixmpp.xmlstream import StanzaBase 

10 

11 

12def fix_namespaces( 

13 xml: ET.Element, 

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

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

16) -> None: 

17 """ 

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

19 

20 Acts in-place. 

21 

22 :param xml: 

23 :param old: 

24 :param new: 

25 """ 

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

27 for child in xml: 

28 fix_namespaces(child, old, new) 

29 

30 

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

32 fix_namespaces(stanza.xml) 

33 

34 

35class HistoryMessage: 

36 def __init__( 

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

38 ) -> None: 

39 if isinstance(stanza, str): 

40 from_db = True 

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

42 else: 

43 from_db = False 

44 

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

46 self.when: datetime = ( 

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

48 ) 

49 

50 if not from_db: 

51 del stanza["delay"] 

52 del stanza["markable"] 

53 del stanza["hint"] 

54 del stanza["chat_state"] 

55 if not stanza["body"]: 

56 del stanza["body"] 

57 fix_namespaces(stanza.xml) 

58 

59 self.stanza: Message = stanza 

60 

61 @property 

62 def stanza_component_ns(self) -> Message: 

63 stanza = copy(self.stanza) 

64 fix_namespaces( 

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

66 ) 

67 return stanza 

68 

69 def forwarded(self) -> Forwarded: 

70 forwarded = Forwarded() 

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

72 forwarded.append(self.stanza) 

73 return forwarded