Coverage for mindsdb / api / mysql / mysql_proxy / data_types / mysql_packets / column_definition_packet.py: 30%

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 CHARSET_NUMBERS, TYPES 

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

15 

16 

17class ColumnDefenitionPacket(Packet): 

18 ''' 

19 Implementation based on: 

20 https://mariadb.com/kb/en/library/resultset/ 

21 ''' 

22 

23 # https://dev.mysql.com/doc/internals/en/com-query-response.html 

24 def setup(self): 

25 self.catalog = Datum('string<lenenc>', 'def') 

26 self.schema = Datum( 

27 'string<lenenc>', 

28 self._kwargs.get('schema', '')) 

29 self.table_alias = Datum( 

30 'string<lenenc>', 

31 self._kwargs.get('table_alias', '') 

32 ) 

33 self.table_name = Datum( 

34 'string<lenenc>', 

35 self._kwargs.get('table_name', '') 

36 ) 

37 self.column_alias = Datum( 

38 'string<lenenc>', 

39 self._kwargs.get('column_alias', '') 

40 ) 

41 self.column_name = Datum( 

42 'string<lenenc>', 

43 self._kwargs.get('column_name', '') 

44 ) 

45 self.fixed_length = Datum('int<lenenc>', 0xC) 

46 charset = self._kwargs.get('charset', CHARSET_NUMBERS["utf8_unicode_ci"]) 

47 self.character_set = Datum('int<2>', charset) 

48 self.column_length = Datum('int<4>', self._kwargs.get('max_length', 0xf)) # may be this? https://books.google.ru/books?id=G2YqBS9CQ0AC&lpg=PP1&hl=ru&pg=PA428#v=onepage&q&f=false 

49 self.column_type = Datum( 

50 'int<1>', 

51 self._kwargs.get( 

52 'column_type', 

53 self._kwargs.get('column_type', TYPES.MYSQL_TYPE_VARCHAR) 

54 ) 

55 ) 

56 

57 self.flags = Datum('int<2>', self._kwargs.get('flags', 0)) 

58 self.decimals = Datum('int<1>', 0) 

59 

60 self.unused = Datum('int<2>', 0) 

61 

62 @property 

63 def body(self): 

64 order = [ 

65 'catalog', 

66 'schema', 

67 'table_alias', 

68 'table_name', 

69 'column_alias', 

70 'column_name', 

71 'fixed_length', 

72 'character_set', 

73 'column_length', 

74 'column_type', 

75 'flags', 

76 'decimals', 

77 'unused' 

78 ] 

79 string = b'' 

80 for key in order: 

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

82 

83 self.setBody(string) 

84 return self._body 

85 

86 @staticmethod 

87 def test(): 

88 import pprint 

89 pprint.pprint(str(ColumnDefenitionPacket().get_packet_string())) 

90 

91 

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

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

94 ColumnDefenitionPacket.test()