1from __future__ import print_function 

2from builtins import str 

3import os 

4import unittest 

5import shutil 

6import unittest 

7import yaml 

8from transientNamer.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 + \ 

16# "/git_repos/_misc_/settings/transientNamer/test_settings.yaml" 

17 

18su = tools( 

19 arguments={"settingsFile": settingsFile}, 

20 docString=__doc__, 

21 logLevel="DEBUG", 

22 options_first=False, 

23 projectName=None, 

24 defaultSettingsFile=False 

25) 

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

27 

28# SETUP PATHS TO COMMON DIRECTORIES FOR TEST DATA 

29moduleDirectory = os.path.dirname(__file__) 

30pathToInputDir = moduleDirectory + "/input/" 

31pathToOutputDir = moduleDirectory + "/output/" 

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 

44 

45try: 

46 shutil.rmtree(settings["astronote-cache"]) 

47except: 

48 pass 

49 

50# xt-setup-unit-testing-files-and-folders 

51# xt-utkit-refresh-database 

52 

53 

54class test_astronotes(unittest.TestCase): 

55 

56 def test_astronotes_create_tables_function(self): 

57 

58 from fundamentals.mysql import writequery 

59 sqlQueries = [ 

60 f"""DROP table astronotes_content""", 

61 f"""DROP table astronotes_transients""", 

62 f"""DROP table astronotes_keywords""" 

63 ] 

64 

65 for sqlQuery in sqlQueries: 

66 try: 

67 writequery( 

68 log=log, 

69 sqlQuery=sqlQuery, 

70 dbConn=dbConn 

71 ) 

72 except: 

73 pass 

74 

75 from transientNamer import astronotes 

76 an = astronotes( 

77 log=log, 

78 dbConn=dbConn, 

79 settings=settings 

80 ) 

81 an._create_db_tables() 

82 # print(noteIds) 

83 

84 def test_astronotes_function(self): 

85 

86 from transientNamer import astronotes 

87 an = astronotes( 

88 log=log, 

89 settings=settings 

90 ) 

91 noteIds = an.get_all_noteids(inLastDays=30) 

92 # print(noteIds) 

93 

94 from transientNamer import astronotes 

95 an = astronotes( 

96 log=log, 

97 dbConn=dbConn, 

98 settings=settings 

99 ) 

100 downloadCount = an.download( 

101 cache_dir=settings["astronote-cache"], inLastDays=30) 

102 print(f"{downloadCount} new astronotes downloaded and cached") 

103 

104 an.notes_to_database() 

105 

106 def test_astronotes_function_exception(self): 

107 

108 from transientNamer import astronotes 

109 try: 

110 this = astronotes( 

111 log=log, 

112 settings=settings, 

113 fakeKey="break the code" 

114 ) 

115 this.get() 

116 assert False 

117 except Exception as e: 

118 assert True 

119 print(str(e)) 

120 

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

122 

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