Coverage for mindsdb / api / mysql / mysql_proxy / data_types / mysql_packets / err_packet.py: 29%

27 statements  

« prev     ^ index     » next       coverage.py v7.13.1, created at 2026-01-21 00:36 +0000

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_packet import Packet 

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

14 

15 

16class ErrPacket(Packet): 

17 ''' 

18 Implementation based on: 

19 https://mariadb.com/kb/en/library/1-connecting-connecting/#initial-handshake-packet 

20 ''' 

21 

22 def setup(self): 

23 err_code = 0 

24 if 'err_code' in self._kwargs: 

25 err_code = self._kwargs['err_code'] 

26 

27 msg = 'ERROR' 

28 if 'msg' in self._kwargs: 

29 msg = self._kwargs['msg'] 

30 

31 self.err_header = Datum('int<1>', 255) 

32 self.err_code = Datum('int<2>', err_code) 

33 self.msg = Datum('string<EOF>', msg) 

34 

35 @property 

36 def body(self): 

37 

38 order = [ 

39 'err_header', 

40 'err_code', 

41 'msg' 

42 ] 

43 string = b'' 

44 for key in order: 

45 string += getattr(self, key).toStringPacket() 

46 

47 self.setBody(string) 

48 return self._body 

49 

50 @staticmethod 

51 def test(): 

52 import pprint 

53 pprint.pprint(str(ErrPacket().get_packet_string())) 

54 

55 

56# only run the test if this file is called from debugger 

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

58 ErrPacket.test()