1from __future__ import print_function 

2from builtins import str 

3import os 

4import unittest 

5import shutil 

6import yaml 

7from marshallEngine.utKit import utKit 

8from fundamentals import tools 

9from os.path import expanduser 

10home = expanduser("~") 

11 

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

13# settingsFile = packageDirectory + "/test_settings.yaml" 

14settingsFile = home + "/git_repos/_misc_/settings/marshall/test_settings.yaml" 

15 

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 PATHS TO COMMON DIRECTORIES FOR TEST DATA 

27moduleDirectory = os.path.dirname(__file__) 

28pathToInputDir = moduleDirectory + "/input/" 

29pathToOutputDir = moduleDirectory + "/output/" 

30 

31try: 

32 shutil.rmtree(pathToOutputDir) 

33except: 

34 pass 

35# COPY INPUT TO OUTPUT DIR 

36shutil.copytree(pathToInputDir, pathToOutputDir) 

37 

38# Recursively create missing directories 

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

40 os.makedirs(pathToOutputDir) 

41 

42 

43class test_data(unittest.TestCase): 

44 

45 def test_data_function(self): 

46 

47 # ADD A ROW TO BE IMPORTED 

48 # utKit.refresh_database() 

49 from fundamentals.mysql import writequery 

50 sqlQuery = """INSERT IGNORE INTO `fs_user_added` (`id`,`candidateID`,`ra_deg`,`dec_deg`,`mag`,`magErr`,`filter`,`observationMJD`,`discDate`,`discMag`,`suggestedType`,`catalogType`,`hostZ`,`targetImageURL`,`objectURL`,`summaryRow`,`ingested`,`htm16ID`,`survey`,`author`,`dateCreated`,`dateLastModified`,`suggestedClassification`,`htm13ID`,`htm10ID`,`transientBucketId`) VALUES (856,'TestSource',155.125958333,-15.1787369444,20.3,NULL,NULL,57627.5,'2016-08-27',20.3,'SN',NULL,0.34,'http://thespacedoctor.co.uk/images/thespacedoctor_icon_white_circle.png','http://thespacedoctor.co.uk',1,0,NULL,'testSurvey','None','2019-07-30 14:25:39','2019-07-30 14:25:39',NULL,NULL,NULL,NULL);""" % locals() 

51 writequery( 

52 log=log, 

53 sqlQuery=sqlQuery, 

54 dbConn=dbConn 

55 ) 

56 

57 from marshallEngine.feeders.useradded.data import data 

58 ingester = data( 

59 log=log, 

60 settings=settings, 

61 dbConn=dbConn 

62 ).ingest(withinLastDays=3) 

63 

64 def test_data_function2(self): 

65 

66 from marshallEngine.feeders.useradded.data import data 

67 ingester = data( 

68 log=log, 

69 settings=settings, 

70 dbConn=dbConn 

71 ).ingest(withinLastDays=3) 

72 

73 def test_update_summaries_function(self): 

74 

75 from fundamentals.mysql import writequery 

76 sqlQuery = """INSERT IGNORE INTO `fs_user_added` (`id`,`candidateID`,`ra_deg`,`dec_deg`,`mag`,`magErr`,`filter`,`observationMJD`,`discDate`,`discMag`,`suggestedType`,`catalogType`,`hostZ`,`targetImageURL`,`objectURL`,`summaryRow`,`ingested`,`htm16ID`,`survey`,`author`,`dateCreated`,`dateLastModified`,`suggestedClassification`,`htm13ID`,`htm10ID`,`transientBucketId`) VALUES (856,'TestSource',155.125958333,-15.1787369444,20.3,NULL,NULL,57627.5,'2016-08-27',20.3,'SN',NULL,0.34,'http://thespacedoctor.co.uk/images/thespacedoctor_icon_white_circle.png','http://thespacedoctor.co.uk',1,0,NULL,'testSurvey','None','2019-07-30 14:25:39','2019-07-30 14:25:39',NULL,NULL,NULL,NULL);""" % locals() 

77 writequery( 

78 log=log, 

79 sqlQuery=sqlQuery, 

80 dbConn=dbConn 

81 ) 

82 

83 from marshallEngine.feeders.useradded.data import data 

84 ingester = data( 

85 log=log, 

86 settings=settings, 

87 dbConn=dbConn 

88 ).insert_into_transientBucket(updateTransientSummaries=False) 

89 

90 from fundamentals.mysql import readquery 

91 sqlQuery = u""" 

92 SELECT transientBucketId FROM fs_user_added order by dateLastModified desc limit 1; 

93 """ % locals() 

94 rows = readquery( 

95 log=log, 

96 sqlQuery=sqlQuery, 

97 dbConn=dbConn, 

98 ) 

99 

100 transientBucketId = rows[0]["transientBucketId"] 

101 

102 from marshallEngine.feeders.useradded.data import data 

103 ingester = data( 

104 log=log, 

105 settings=settings, 

106 dbConn=dbConn 

107 ).insert_into_transientBucket(updateTransientSummaries=transientBucketId) 

108 

109 def test_data_function_exception(self): 

110 

111 from marshallEngine.feeders.useradded.data import data 

112 try: 

113 this = data( 

114 log=log, 

115 settings=settings, 

116 fakeKey="break the code" 

117 ) 

118 this.get() 

119 assert False 

120 except Exception as e: 

121 assert True 

122 print(str(e)) 

123 

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

125 

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