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

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 

10from ..util.util import fix_namespaces 

11 

12 

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

14 fix_namespaces(stanza.xml, "jabber:component:accept", "jabber:client") 

15 

16 

17class HistoryMessage: 

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

19 if isinstance(stanza, str): 

20 from_db = True 

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

22 else: 

23 from_db = False 

24 

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

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

27 

28 if not from_db: 

29 del stanza["delay"] 

30 del stanza["markable"] 

31 del stanza["hint"] 

32 del stanza["chat_state"] 

33 if not stanza["body"]: 

34 del stanza["body"] 

35 set_client_namespace(stanza) 

36 

37 self.stanza: Message = stanza 

38 

39 @property 

40 def stanza_component_ns(self) -> Message: 

41 stanza = copy(self.stanza) 

42 fix_namespaces(stanza.xml, old="jabber:client", new="jabber:component:accept") 

43 return stanza 

44 

45 def forwarded(self) -> Forwarded: 

46 forwarded = Forwarded() 

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

48 forwarded.append(self.stanza) 

49 return forwarded 

50 

51 @property 

52 def occupant_id(self) -> str: 

53 return self.stanza["occupant-id"]["id"] # type:ignore[no-any-return]