Coverage for tests / test_utilsEfd.py: 41%

1# This file is part of cp_pipe. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# This program is free software: you can redistribute it and/or modify 

10# it under the terms of the GNU General Public License as published by 

11# the Free Software Foundation, either version 3 of the License, or 

12# (at your option) any later version. 

13# 

14# This program is distributed in the hope that it will be useful, 

15# but WITHOUT ANY WARRANTY; without even the implied warranty of 

16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

17# GNU General Public License for more details. 

18# 

19# You should have received a copy of the GNU General Public License 

20# along with this program. If not, see <https://www.gnu.org/licenses/>. 

21 

22import os 

23import unittest 

24import vcr 

25 

26import lsst.utils.tests 

27 

28from lsst.cp.pipe.utilsEfd import CpEfdClient 

29 

30 

31def getVcr(): 

32 """Get a VCR object for use in tests. 

33 

34 Adapted from https://github.com/lsst-sitcom/summit_utils/ 

35 blob/main/tests/utils.py 

36 

37 Use record_mode="none" to run tests for normal operation. To update files 

38 or generate new ones, make sure you have a working connection to the EFD at 

39 all the relevant sites, and temporarily run with mode="all" via *both* 

40 python/pytest *and* with scons, as these generate slightly different HTTP 

41 requests for some reason. 

42 """ 

43 dirname = os.path.dirname(__file__) 

44 cassette_library_dir = os.path.join(dirname, "data", "cassettes") 

45 safe_vcr = vcr.VCR( 

46 record_mode="none", 

47 cassette_library_dir=cassette_library_dir, 

48 path_transformer=vcr.VCR.ensure_suffix(".yaml"), 

49 match_on=["method", "scheme", "host", "port", "path", "query", "body"], 

50 ) 

51 return safe_vcr 

52 

53 

54safe_vcr = getVcr() 

55 

56 

57class UtilsEfdTestCase(lsst.utils.tests.TestCase): 

58 """Unit test for EFD access code.""" 

59 

60 @classmethod 

61 @safe_vcr.use_cassette() 

62 def setUpClass(cls): 

63 # Initialize a client. 

64 try: 

65 cls.client = CpEfdClient() 

66 except Exception as e: 

67 cls.client = None 

68 raise unittest.SkipTest(f"Could not initialize EFD client: {e}") 

69 

70 @safe_vcr.use_cassette() 

71 def test_monochromator(self): 

72 data = self.client.getEfdMonochromatorData( 

73 dateMin="2023-12-19T00:00:00", 

74 dateMax="2023-12-19T23:59:59" 

75 ) 

76 

77 indexDate, wavelength = self.client.parseMonochromatorStatus( 

78 data, 

79 "2023-12-19T14:37:19.498" 

80 ) 

81 self.assertEqual(wavelength, 550.0) 

82 self.assertEqual(indexDate, "2023-12-19T14:37:17.799") 

83 

84 @safe_vcr.use_cassette() 

85 def test_electrometer(self): 

86 data = self.client.getEfdElectrometerData( 

87 dateMin="2024-05-30T00:00:00", 

88 dateMax="2024-05-30T05:00:00", 

89 ) 

90 

91 # Test single lookups: 

92 for (iDate, rDate), (iVal, rVal) in zip([("2024-05-30T04:21:48.6", "2024-05-30T04:22:08"), 

93 ("2024-05-30T04:21:50", "2024-05-30T04:22:10"), 

94 ("2024-05-30T04:21:53", "2024-05-30T04:22:13"), 

95 ("2024-05-30T04:21:58", "2024-05-30T04:22:18"), 

96 ("2024-05-30T04:22:17", "2024-05-30T04:22:37")], 

97 [(-1.5269e-07, -1.5244e-07), 

98 (-1.5168e-07, -1.5137e-07), 

99 (-1.5165e-07, -1.5205e-07), 

100 (-1.5223e-07, -1.5147e-07), 

101 (-1.5226e-07, -1.5558e-07)]): 

102 indexDate, intensity, _ = self.client.parseElectrometerStatus( 

103 data, 

104 iDate 

105 ) 

106 self.assertFloatsAlmostEqual(intensity, iVal, atol=1e-10) 

107 

108 indexDateReference, intensityReference, _ = self.client.parseElectrometerStatus( 

109 data, 

110 rDate 

111 ) 

112 self.assertFloatsAlmostEqual(intensityReference, rVal, atol=1e-10) 

113 

114 # Test integrated lookups: 

115 iDate = "2024-05-30T04:21:48.6" 

116 iDateEnd = "2024-05-30T04:22:18" 

117 rDate = "2024-05-30T04:22:08" 

118 rDateEnd = "2024-05-30T04:22:38" 

119 indexDate, intensity, endDate = self.client.parseElectrometerStatus( 

120 data, 

121 iDate, 

122 dateEnd=iDateEnd, 

123 doIntegrateSamples=True, 

124 index=201 

125 ) 

126 self.assertFloatsAlmostEqual(intensity, -1.52297e-7, atol=1e-10) 

127 indexDateReference, intensityReference, endDateReference = self.client.parseElectrometerStatus( 

128 data, 

129 rDate, 

130 dateEnd=rDateEnd, 

131 doIntegrateSamples=True, 

132 index=201 

133 ) 

134 self.assertFloatsAlmostEqual(intensityReference, -1.532977e-07, atol=1e-10) 

135 

136 @safe_vcr.use_cassette() 

137 def test_electrometer_alternate(self): 

138 # This should raise if no dates are passed: 

139 with self.assertRaises(RuntimeError): 

140 data = self.client.getEfdElectrometerData( 

141 dataSeries='lsst.sal.Electrometer.logevent_logMessage') 

142 

143 data = self.client.getEfdElectrometerData( 

144 dataSeries='lsst.sal.Electrometer.logevent_logMessage', 

145 dateMin='2024-07-26T16:30:00', 

146 dateMax='2024-07-26T16:45:00') 

147 

148 # Test single lookups. These should not be integrated. 

149 for iDate, iVal in zip(["2024-07-26T16:38:32.228", 

150 "2024-07-26T16:38:54.581", 

151 "2024-07-26T16:40:56.579", 

152 "2024-07-26T16:42:58.553"], 

153 [-2.24234e-07, 

154 -2.24388e-07, 

155 -2.24105e-07, 

156 -2.23784e-07]): 

157 indexDate, intensity, _ = self.client.parseElectrometerStatus( 

158 data, 

159 iDate 

160 ) 

161 self.assertFloatsAlmostEqual(intensity, iVal, atol=1e-10) 

162 

163 

164class MemoryTester(lsst.utils.tests.MemoryTestCase): 

165 pass 

166 

167 

168def setup_module(module): 

169 lsst.utils.tests.init() 

170 

171 

172if __name__ == "__main__": 172 ↛ 173line 172 didn't jump to line 173 because the condition on line 172 was never true

173 lsst.utils.tests.init() 

174 unittest.main()