Coverage for mindsdb / api / mysql / mysql_proxy / data_types / mysql_packets / handshake_packet.py: 31%

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.libs.constants.mysql import ( 

14 DEFAULT_AUTH_METHOD, 

15 DEFAULT_COALLITION_ID, 

16 FILLER_FOR_WIRESHARK_DUMP, 

17 SERVER_STATUS_AUTOCOMMIT 

18) 

19from mindsdb.api.mysql.mysql_proxy.classes.server_capabilities import server_capabilities 

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

21 

22 

23class HandshakePacket(Packet): 

24 ''' 

25 Implementation based on: 

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

27 ''' 

28 

29 def setup(self): 

30 capabilities = server_capabilities.value 

31 self.protocol_version = Datum('int<1>', 10) 

32 self.server_version = Datum('string<NUL>', '5.7.1-MindsDB-1.0') 

33 self.connection_id = Datum('int<4>', self.proxy.connection_id) 

34 self.scramble_1st_part = Datum('string<8>', self.proxy.salt[:8]) 

35 self.reserved_byte = Datum('string<1>', '') 

36 self.server_capabilities_1st_part = Datum('int<2>', capabilities) 

37 self.server_default_collation = Datum('int<1>', DEFAULT_COALLITION_ID) 

38 self.status_flags = Datum('int<2>', SERVER_STATUS_AUTOCOMMIT) 

39 self.server_capabilities_2nd_part = Datum('int<2>', capabilities >> 16) 

40 self.wireshark_filler = Datum('int<1>', FILLER_FOR_WIRESHARK_DUMP) 

41 # self.wireshark_filler = Datum('int<1>', len(self.proxy.salt)) 

42 self.reserved_filler1 = Datum('string<6>', '') 

43 self.reserved_filler2 = Datum('string<4>', '') 

44 self.scramble_2nd_part = Datum('string<NUL>', self.proxy.salt[8:]) 

45 self.null_close = Datum('string<NUL>', DEFAULT_AUTH_METHOD) 

46 

47 @property 

48 def body(self): 

49 order = [ 

50 'protocol_version', 

51 'server_version', 

52 'connection_id', 

53 'scramble_1st_part', 

54 'reserved_byte', 

55 'server_capabilities_1st_part', 

56 'server_default_collation', 

57 'status_flags', 

58 'server_capabilities_2nd_part', 

59 'wireshark_filler', 

60 'reserved_filler1', 

61 'reserved_filler2', 

62 'scramble_2nd_part', 

63 'null_close' 

64 ] 

65 string = b'' 

66 for key in order: 

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

68 

69 self.setBody(string) 

70 return self._body 

71 

72 @staticmethod 

73 def test(): 

74 import pprint 

75 pprint.pprint(str(HandshakePacket().get_packet_string())) 

76 

77 

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

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

80 HandshakePacket.test()