1from __future__ import print_function 

2from builtins import str 

3import os 

4import unittest 

5import shutil 

6import unittest 

7import yaml 

8from fundamentals.utKit import utKit 

9from fundamentals import tools 

10from os.path import expanduser 

11home = expanduser("~") 

12 

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

14settingsFile = packageDirectory + "/test_settings.yaml" 

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

16su = tools( 

17 arguments={"settingsFile": settingsFile}, 

18 docString=__doc__, 

19 logLevel="DEBUG", 

20 options_first=False, 

21 projectName=None, 

22 defaultSettingsFile=False 

23) 

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

25 

26# SETUP AND TEARDOWN FIXTURE FUNCTIONS FOR THE ENTIRE MODULE 

27moduleDirectory = os.path.dirname(__file__) 

28utKit = utKit(moduleDirectory) 

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

30utKit.tearDownModule() 

31 

32try: 

33 shutil.rmtree(pathToOutputDir) 

34except: 

35 pass 

36# COPY INPUT TO OUTPUT DIR 

37shutil.copytree(pathToInputDir, pathToOutputDir) 

38 

39# Recursively create missing directories 

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

41 os.makedirs(pathToOutputDir) 

42 

43class test_writequery(unittest.TestCase): 

44 

45 def test_writequery_function(self): 

46 

47 from fundamentals.mysql import writequery 

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

49 writequery( 

50 log=log, 

51 sqlQuery=sqlQuery, 

52 dbConn=dbConn, 

53 Force=False, 

54 manyValueList=False 

55 ) 

56 

57 def test_manyvalue_insert(self): 

58 from fundamentals.mysql import writequery 

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

60 writequery( 

61 log=log, 

62 sqlQuery=sqlQuery, 

63 dbConn=dbConn, 

64 Force=False, 

65 manyValueList=False 

66 ) 

67 

68 from fundamentals.mysql import writequery 

69 sqlQuery = """INSERT INTO testing_table (id) values (%s)""" 

70 writequery( 

71 log=log, 

72 sqlQuery=sqlQuery, 

73 dbConn=dbConn, 

74 Force=False, 

75 manyValueList=[(1,), (2,), (3,), (4,), (5,), (6,), (7,), 

76 (8,), (9,), (10,), (11,), (12,), ] 

77 ) 

78 

79 def test_writequery_function_delete(self): 

80 

81 from fundamentals.mysql import writequery 

82 sqlQuery = "DROP TABLE `testing_table`;" 

83 writequery( 

84 log=log, 

85 sqlQuery=sqlQuery, 

86 dbConn=dbConn, 

87 Force=False, 

88 manyValueList=False 

89 ) 

90 

91 def test_writequery_error_force(self): 

92 

93 from fundamentals.mysql import writequery 

94 sqlQuery = "rubbish query;" 

95 writequery( 

96 log=log, 

97 sqlQuery=sqlQuery, 

98 dbConn=dbConn, 

99 Force=True, 

100 manyValueList=False 

101 ) 

102 

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

104 

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