Coverage for mindsdb / api / mysql / mysql_proxy / data_types / mysql_packets / resultset_row_package.py: 28%

1""" 

2******************************************************* 

3 * Copyright (C) 2017 MindsDB Inc. <copyright@mindsdb.com> 

4 * 

5 * This file is part of MindsDB Server. 

6 * 

7 * MindsDB Server can not be copied and/or distributed without the express 

8 * permission of MindsDB Inc 

9 ******************************************************* 

10""" 

11 

12from mindsdb.api.mysql.mysql_proxy.data_types.mysql_datum import Datum 

13from mindsdb.api.mysql.mysql_proxy.data_types.mysql_packet import Packet 

14from mindsdb.api.mysql.mysql_proxy.libs.constants.mysql import NULL_VALUE 

15 

16 

17class ResultsetRowPacket(Packet): 

18 """ 

19 Implementation based on: 

20 https://dev.mysql.com/doc/internals/en/com-query-response.html#packet-ProtocolText::ResultsetRow 

21 https://mariadb.com/kb/en/resultset-row/ 

22 """ 

23 

24 def setup(self): 

25 data = self._kwargs.get("data", {}) 

26 self.value = [] 

27 for val in data: 

28 if val is None: 

29 self.value.append(NULL_VALUE) 

30 elif isinstance(val, bytes): 

31 self.value.append(Datum("byte<lenenc>", val)) 

32 else: 

33 self.value.append(Datum("string<lenenc>", str(val))) 

34 

35 @property 

36 def body(self): 

37 string = b"" 

38 for x in self.value: 

39 if x is NULL_VALUE: 

40 string += x 

41 else: 

42 string += x.toStringPacket() 

43 

44 self.setBody(string) 

45 return self._body 

46 

47 @staticmethod 

48 def test(): 

49 import pprint 

50 

51 pprint.pprint(str(ResultsetRowPacket().get_packet_string())) 

52 

53 

54if __name__ == "__main__": 54 ↛ 55line 54 didn't jump to line 55 because the condition on line 54 was never true

55 ResultsetRowPacket.test()