Coverage for filip/clients/mqtt/encoder/json.py: 95%

1""" 

2Json encoder class for all IoTA-JSON MQTT message encoders 

3""" 

4import json 

5from typing import Any, Dict, Tuple 

6from filip.clients.mqtt.encoder import BaseEncoder 

7from filip.models.mqtt import IoTAMQTTMessageType 

8 

9 

10class Json(BaseEncoder): 

11 """ 

12 Json encoder class for all IoTA-JSON MQTT message encoders 

13 """ 

14 prefix = '/json' 

15 

16 def __init__(self): 

17 super().__init__() 

18 

19 def decode_message(self, msg, decoder='utf-8') -> Tuple[str, str, Dict]: 

20 apikey, device_id, payload = super().decode_message(msg=msg, 

21 decoder=decoder) 

22 payload = json.loads(payload) 

23 return apikey, device_id, payload 

24 

25 def encode_msg(self, 

26 device_id, 

27 payload: Any, 

28 msg_type: IoTAMQTTMessageType) -> str: 

29 if msg_type == IoTAMQTTMessageType.SINGLE: 

30 return payload 

31 elif msg_type == IoTAMQTTMessageType.MULTI: 

32 payload = super()._parse_timestamp(payload=payload) 

33 return json.dumps(payload, default=str) 

34 elif msg_type == IoTAMQTTMessageType.CMDEXE: 

35 return json.dumps(payload) 

36 super()._raise_encoding_error(payload=payload, msg_type=msg_type)