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 

43dbSettings = settings["database settings"]["atlasMovers"] 

44dbSettings2 = settings["database settings"]["unit_tests"] 

45 

46 

47class test_database(unittest.TestCase): 

48 

49 # def test_tunnel(self): 

50 

51 # from fundamentals.mysql import database 

52 # db = database( 

53 # log=log, 

54 # dbSettings=dbSettings 

55 # ) 

56 # sshPort = db._setup_tunnel( 

57 # tunnelParameters=dbSettings["tunnel"] 

58 # ) 

59 

60 # return 

61 

62 def test_no_tunnel(self): 

63 

64 from fundamentals.mysql import database 

65 db = database( 

66 log=log, 

67 dbSettings=dbSettings2 

68 ).connect() 

69 

70 return 

71 

72 def test_database_function(self): 

73 

74 # SETUP ALL DATABASE CONNECTIONS 

75 from fundamentals.mysql import database 

76 dbConn = database( 

77 log=log, 

78 dbSettings=dbSettings2 

79 ).connect() 

80 

81 from fundamentals.mysql import readquery 

82 sqlQuery = u""" 

83 SELECT VERSION(); 

84 """ % locals() 

85 rows = readquery( 

86 log=log, 

87 sqlQuery=sqlQuery, 

88 dbConn=dbConn, 

89 quiet=False 

90 ) 

91 # print(rows) 

92 

93 def test_database_function_exception(self): 

94 

95 from fundamentals.mysql import database 

96 try: 

97 this = database( 

98 log=log, 

99 dbSettings=dbSettings2, 

100 fakeKey="break the code" 

101 ) 

102 this.get() 

103 assert False 

104 except Exception as e: 

105 assert True 

106 # print(str(e)) 

107 

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

109 

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