1from __future__ import print_function 

2from builtins import str 

3import os 

4import unittest 

5import shutil 

6import unittest 

7import yaml 

8import re 

9from fundamentals.utKit import utKit 

10from fundamentals import tools 

11from os.path import expanduser 

12home = expanduser("~") 

13 

14packageDirectory = utKit("").get_project_root() 

15settingsFile = packageDirectory + "/test_settings.yaml" 

16# settingsFile = home + "/.config/soxspipe.recipes/soxspipe.recipes.yaml" 

17su = tools( 

18 arguments={"settingsFile": settingsFile}, 

19 docString=__doc__, 

20 logLevel="DEBUG", 

21 options_first=False, 

22 projectName=None, 

23 defaultSettingsFile=False 

24) 

25arguments, settings, log, dbConn = su.setup() 

26 

27# SETUP AND TEARDOWN FIXTURE FUNCTIONS FOR THE ENTIRE MODULE 

28moduleDirectory = os.path.dirname(__file__) 

29utKit = utKit(moduleDirectory) 

30log, dbConn, pathToInputDir, pathToOutputDir = utKit.setupModule() 

31utKit.tearDownModule() 

32 

33try: 

34 shutil.rmtree(pathToOutputDir) 

35except: 

36 pass 

37# COPY INPUT TO OUTPUT DIR 

38shutil.copytree(pathToInputDir, pathToOutputDir) 

39 

40# Recursively create missing directories 

41if not os.path.exists(pathToOutputDir): 

42 os.makedirs(pathToOutputDir) 

43 

44reDatetime = re.compile('^[0-9]{4}-[0-9]{2}-[0-9]{2}T') 

45 

46class test_convert_dictionary_to_mysql_table(unittest.TestCase): 

47 

48 def test_convert_dictionary_to_mysql_table_function(self): 

49 from fundamentals.mysql import writequery 

50 sqlQuery = "DROP TABLE IF EXISTS `testing_table`; CREATE TABLE IF NOT EXISTS `testing_table` (`id` INT NOT NULL AUTO_INCREMENT,`uniquekey1` varchar(45) NOT NULL default 'ha',`uniqueKey2` varchar(45) NOT NULL default 'ha', `dateCreated` TIMESTAMP NOT NULL default CURRENT_TIMESTAMP, `dateModified` TIMESTAMP NOT NULL default CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (`id`) )" 

51 writequery( 

52 log=log, 

53 sqlQuery=sqlQuery, 

54 dbConn=dbConn, 

55 Force=False, 

56 manyValueList=False 

57 ) 

58 

59 dictionary = {"a newKey": "cool", "and another": "super cool", 

60 "uniquekey1": "cheese", "uniqueKey2": "burgers"} 

61 from fundamentals.mysql import convert_dictionary_to_mysql_table 

62 message = convert_dictionary_to_mysql_table( 

63 dbConn=dbConn, 

64 log=log, 

65 dictionary=dictionary, 

66 dbTableName="testing_table", 

67 uniqueKeyList=["uniquekey1", "uniqueKey2"], 

68 createHelperTables=False, 

69 dateModified=True, 

70 returnInsertOnly=False 

71 ) 

72 

73 def test_return_inserts(self): 

74 from fundamentals.mysql import writequery 

75 sqlQuery = "CREATE TABLE IF NOT EXISTS `testing_table` (`id` INT NOT NULL, PRIMARY KEY (`id`))" 

76 writequery( 

77 log=log, 

78 sqlQuery=sqlQuery, 

79 dbConn=dbConn, 

80 Force=False, 

81 manyValueList=False 

82 ) 

83 

84 dictionary = {"a newKey": "cool", "and another": "super cool", 

85 "uniquekey1": "cheese", "uniqueKey2": "burgers"} 

86 from fundamentals.mysql import convert_dictionary_to_mysql_table 

87 message = convert_dictionary_to_mysql_table( 

88 dbConn=dbConn, 

89 log=log, 

90 dictionary=dictionary, 

91 dbTableName="testing_table", 

92 uniqueKeyList=["uniquekey1", "uniqueKey2"], 

93 createHelperTables=False, 

94 dateModified=False, 

95 returnInsertOnly=True 

96 ) 

97 # print(message) 

98 

99 def test_return_inserts_with_datetime_pre_compiled(self): 

100 from fundamentals.mysql import writequery 

101 sqlQuery = "CREATE TABLE IF NOT EXISTS `testing_table` (`id` INT NOT NULL, PRIMARY KEY (`id`))" 

102 writequery( 

103 log=log, 

104 sqlQuery=sqlQuery, 

105 dbConn=dbConn, 

106 Force=False, 

107 manyValueList=False 

108 ) 

109 

110 dictionary = {"a newKey": "cool", "and another": "super cool", 

111 "uniquekey1": "cheese", "uniqueKey2": "burgers"} 

112 from fundamentals.mysql import convert_dictionary_to_mysql_table 

113 message = convert_dictionary_to_mysql_table( 

114 dbConn=dbConn, 

115 log=log, 

116 dictionary=dictionary, 

117 dbTableName="testing_table", 

118 uniqueKeyList=["uniquekey1", "uniqueKey2"], 

119 createHelperTables=False, 

120 dateModified=False, 

121 returnInsertOnly=True, 

122 reDatetime=reDatetime 

123 ) 

124 # print(message) 

125 

126 def test_return_inserts_non_batch(self): 

127 dictionary = {"a newKey": "cool", "and another": "super cool", 

128 "uniquekey1": "cheese", "uniqueKey2": "burgers"} 

129 from fundamentals.mysql import convert_dictionary_to_mysql_table 

130 inserts = convert_dictionary_to_mysql_table( 

131 dbConn=dbConn, 

132 log=log, 

133 dictionary=dictionary, 

134 dbTableName="testing_table", 

135 uniqueKeyList=["uniquekey1", "uniqueKey2"], 

136 dateModified=False, 

137 returnInsertOnly=True, 

138 replace=True, 

139 batchInserts=False 

140 ) 

141 

142 # x-print-testpage-for-pessto-marshall-web-object 

143 

144 # x-class-to-test-named-worker-function