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
« 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"""
12from mindsdb.api.mysql.mysql_proxy.data_types.mysql_packet import Packet
13from mindsdb.api.mysql.mysql_proxy.data_types.mysql_datum import Datum
16class ErrPacket(Packet):
17 '''
18 Implementation based on:
19 https://mariadb.com/kb/en/library/1-connecting-connecting/#initial-handshake-packet
20 '''
22 def setup(self):
23 err_code = 0
24 if 'err_code' in self._kwargs:
25 err_code = self._kwargs['err_code']
27 msg = 'ERROR'
28 if 'msg' in self._kwargs:
29 msg = self._kwargs['msg']
31 self.err_header = Datum('int<1>', 255)
32 self.err_code = Datum('int<2>', err_code)
33 self.msg = Datum('string<EOF>', msg)
35 @property
36 def body(self):
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()
47 self.setBody(string)
48 return self._body
50 @staticmethod
51 def test():
52 import pprint
53 pprint.pprint(str(ErrPacket().get_packet_string()))
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()