1from __future__ import print_function 

2from builtins import str 

3import os 

4import unittest 

5import shutil 

6import unittest 

7import yaml 

8import time 

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 

44dictList = [] 

45 

46for i in range(100000): 

47 

48 dictList.append({ 

49 "col1": i, 

50 "col2": i + 10., 

51 "col3": i / 2., 

52 "col4": i * 34. 

53 }) 

54 

55dbSettings = dbSettings = { 

56 'host': '127.0.0.1', 

57 'user': 'utuser', 

58 'tunnel': False, 

59 'password': 'utpass', 

60 'db': 'unit_tests' 

61} 

62 

63class test_insert_list_of_dictionaries_into_database_tables(unittest.TestCase): 

64 

65 def test_insert_list_of_dictionaries_into_database_tables_function(self): 

66 

67 t1 = time.time() 

68 

69 from fundamentals.mysql import insert_list_of_dictionaries_into_database_tables 

70 insert_list_of_dictionaries_into_database_tables( 

71 dbConn=dbConn, 

72 log=log, 

73 dictList=dictList, 

74 dbTableName="test_insert_many", 

75 uniqueKeyList=["col1", "col3"], 

76 dateModified=True, 

77 batchSize=10000, 

78 replace=True, 

79 dbSettings=dbSettings 

80 ) 

81 

82 # print(time.time() - t1) 

83 

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

85 

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