Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

1from __future__ import print_function 

2from builtins import str 

3import os 

4import unittest 

5import shutil 

6import yaml 

7from sherlock.utKit import utKit 

8from fundamentals import tools 

9from os.path import expanduser 

10home = expanduser("~") 

11 

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

13settingsFile = packageDirectory + "/test_settings.yaml" 

14 

15su = tools( 

16 arguments={"settingsFile": settingsFile}, 

17 docString=__doc__, 

18 logLevel="DEBUG", 

19 options_first=False, 

20 projectName=None, 

21 defaultSettingsFile=False 

22) 

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

24 

25# SETUP PATHS TO COMMON DIRECTORIES FOR TEST DATA 

26moduleDirectory = os.path.dirname(__file__) 

27pathToInputDir = moduleDirectory + "/input/" 

28pathToOutputDir = moduleDirectory + "/output/" 

29 

30try: 

31 shutil.rmtree(pathToOutputDir) 

32except: 

33 pass 

34# COPY INPUT TO OUTPUT DIR 

35shutil.copytree(pathToInputDir, pathToOutputDir) 

36 

37# Recursively create missing directories 

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

39 os.makedirs(pathToOutputDir) 

40 

41class test_database(unittest.TestCase): 

42 

43 def test_tunnel(self): 

44 

45 from sherlock import database 

46 db = database( 

47 log=log, 

48 settings=settings 

49 ) 

50 sshPort = db._setup_tunnel( 

51 tunnelParameters=settings["database settings"][ 

52 "static catalogues2"]["tunnel"] 

53 ) 

54 

55 return 

56 

57 def test_database_function(self): 

58 

59 # SETUP ALL DATABASE CONNECTIONS 

60 # SETUP ALL DATABASE CONNECTIONS 

61 from sherlock import database 

62 db = database( 

63 log=log, 

64 settings=settings 

65 ) 

66 dbConns, dbVersions = db.connect() 

67 self.transientsDbConn = dbConns["transients"] 

68 self.cataloguesDbConn = dbConns["catalogues"] 

69 

70 from fundamentals.mysql import readquery 

71 sqlQuery = u""" 

72 SELECT VERSION(); 

73 """ % locals() 

74 rows = readquery( 

75 log=log, 

76 sqlQuery=sqlQuery, 

77 dbConn=self.transientsDbConn, 

78 quiet=False 

79 ) 

80 print(rows) 

81 rows = readquery( 

82 log=log, 

83 sqlQuery=sqlQuery, 

84 dbConn=self.cataloguesDbConn, 

85 quiet=False 

86 ) 

87 print(rows) 

88 

89 def test_database_function_exception(self): 

90 

91 from sherlock import database 

92 try: 

93 this = database( 

94 log=log, 

95 settings=settings, 

96 fakeKey="break the code" 

97 ) 

98 this.get() 

99 assert False 

100 except Exception as e: 

101 assert True 

102 print(str(e)) 

103 

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