Coverage for tests / test_blockUtils.py: 27%

1# This file is part of summit_utils. 

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 

22"""Test cases for utils.""" 

23 

24import asyncio 

25import json 

26import os 

27import unittest 

28 

29import pandas as pd 

30from utils import getVcr 

31 

32import lsst.utils.tests 

33from lsst.summit.utils.blockUtils import BlockParser 

34from lsst.summit.utils.efdUtils import makeEfdClient 

35 

36__all__ = ("writeNewBlockInfoTestTruthValues",) 

37 

38HAS_EFD_CLIENT = True 

39try: 

40 import lsst_efd_client # noqa: F401 just need to check this is available 

41except ImportError: 

42 HAS_EFD_CLIENT = False 

43 

44vcr = getVcr() 

45 

46DELIMITER = "||" # don't use a comma, as str(list) will naturally contain commas 

47TESTDIR = os.path.abspath(os.path.dirname(__file__)) 

48 

49 

50def getBlockInfoTestTruthValues(dayObs: int) -> dict[tuple[str, int], str]: 

51 """Get the current truth values for the block information. 

52 

53 Parameters 

54 ---------- 

55 dayObs : `int`, optional 

56 The dayObs to get the truth values for. 

57 

58 Returns 

59 ------- 

60 data : `dict` [`tuple` [`int`, `int`], `str`] 

61 The block info truth data. 

62 """ 

63 dataFilename = os.path.join(TESTDIR, "data", f"blockInfoData_{dayObs}.json") 

64 

65 with open(dataFilename, "r") as f: 

66 loaded = json.loads(f.read()) 

67 

68 data = {} 

69 for dayObsSeqNumStr, line in loaded.items(): 

70 blockNum = str(dayObsSeqNumStr.split(f"{DELIMITER}")[0]) 

71 blockSeqNum = int(dayObsSeqNumStr.split(f"{DELIMITER}")[1]) 

72 data[blockNum, blockSeqNum] = line 

73 return data 

74 

75 

76def writeNewBlockInfoTestTruthValues(dayObs: int) -> None: 

77 """This function is used to write out the truth values for the test cases. 

78 

79 If bugs are found in the parsing, it's possible these values could change, 

80 and would need to be updated. If that happens, run this function, and check 

81 the new values into git. 

82 """ 

83 blockParser = BlockParser(dayObs) 

84 

85 data = {} 

86 for block in (blockParser).getBlockNums(): 

87 seqNums = blockParser.getSeqNums(block) 

88 for seqNum in seqNums: 

89 blockInfo = blockParser.getBlockInfo(block, seqNum) 

90 assert blockInfo is not None 

91 line = ( 

92 f"{blockInfo.blockId}{DELIMITER}" 

93 f"{blockInfo.begin}{DELIMITER}" 

94 f"{blockInfo.end}{DELIMITER}" 

95 f"{blockInfo.salIndices}{DELIMITER}" 

96 f"{blockInfo.tickets}{DELIMITER}" 

97 f"{len(blockInfo.states)}" 

98 ) 

99 # must store as string not tuple for json serialization 

100 data[f"{block}{DELIMITER}{seqNum}"] = line 

101 

102 dataFilename = os.path.join(TESTDIR, "data", f"blockInfoData_{dayObs}.json") 

103 with open(dataFilename, "w") as f: 

104 json.dump(data, f) 

105 

106 

107@unittest.skipIf(not HAS_EFD_CLIENT, "No EFD client available") 

108@vcr.use_cassette() 

109class BlockParserTestCase(lsst.utils.tests.TestCase): 

110 @classmethod 

111 @vcr.use_cassette() 

112 def setUpClass(cls): 

113 try: 

114 cls.client = makeEfdClient(testing=True) 

115 except RuntimeError: 

116 raise unittest.SkipTest("Could not instantiate an EFD client") 

117 

118 cls.dayObsNoTestCases = 20230615 

119 cls.dayObsWithCases = 20250420 # blocks = ['365', 'T282', 'T3', 'T379', 'T380', 'T4', 'T454'] 

120 cls.dayObsNoBlocks = 20230531 # contains data but no blocks 

121 cls.blockParser = BlockParser(dayObs=cls.dayObsNoTestCases, client=cls.client) 

122 cls.blockNums = cls.blockParser.getBlockNums() 

123 cls.blockDict = {} 

124 for block in cls.blockNums: 

125 cls.blockDict[block] = cls.blockParser.getSeqNums(block) 

126 

127 @vcr.use_cassette() 

128 def tearDown(self): 

129 loop = asyncio.get_event_loop() 

130 if self.client.influx_client is not None: 

131 loop.run_until_complete(self.client.influx_client.close()) 

132 

133 @vcr.use_cassette() 

134 def test_parsing(self): 

135 blockNums = self.blockParser.getBlockNums() 

136 self.assertTrue(all(isinstance(n, str)) for n in blockNums) 

137 self.assertEqual(blockNums, list(self.blockDict.keys())) 

138 

139 for block, seqNums in self.blockDict.items(): 

140 self.assertTrue(isinstance(block, str)) 

141 self.assertIsInstance(seqNums, list) 

142 self.assertTrue(all(isinstance(s, int)) for s in seqNums) 

143 

144 found = self.blockParser.getSeqNums(block) 

145 self.assertTrue(all(isinstance(s, int) for s in found)) 

146 self.assertEqual(found, seqNums) 

147 self.blockParser.printBlockEvolution(block) 

148 

149 for seqNum in seqNums: 

150 data = self.blockParser.getRows(block, seqNum) 

151 self.assertIsInstance(data, pd.DataFrame) 

152 self.assertGreater(len(data), 0) 

153 self.blockParser.getBlockInfo(block=block, seqNum=seqNum) 

154 self.blockParser.printBlockEvolution(block, seqNum=seqNum) 

155 

156 @vcr.use_cassette() 

157 def test_notFoundBehavior(self): 

158 # no block data on this day so check init doesn't raise 

159 blockParser = BlockParser(dayObs=self.dayObsNoBlocks, client=self.client) 

160 self.assertIsInstance(blockParser, BlockParser) 

161 

162 # check the queries which return nothing give nothing back gracefully 

163 blocks = blockParser.getBlockNums() 

164 self.assertIsInstance(blocks, list) 

165 self.assertEqual(len(blocks), 0) 

166 

167 seqNums = blockParser.getSeqNums(block=123) 

168 self.assertIsInstance(seqNums, list) 

169 self.assertEqual(len(seqNums), 0) 

170 

171 # just check this doesn't raise 

172 blockParser.getBlockInfo(block=1, seqNum=1) 

173 

174 # now switch back to one with data, and make sure the same is true 

175 # when there is data present 

176 blockParser = self.blockParser 

177 seqNums = blockParser.getSeqNums(block=9999999) 

178 self.assertIsInstance(seqNums, list) 

179 self.assertEqual(len(seqNums), 0) 

180 

181 # just check this doesn't raise 

182 blockParser.getBlockInfo(block=9999999, seqNum=9999999) 

183 

184 @vcr.use_cassette() 

185 def test_actualValues(self): 

186 for dayObs in [self.dayObsNoTestCases, self.dayObsWithCases]: 

187 data = getBlockInfoTestTruthValues(dayObs) 

188 blockParser = BlockParser(dayObs, client=self.client) 

189 

190 for block in blockParser.getBlockNums(): 

191 seqNums = blockParser.getSeqNums(block) 

192 for seqNum in seqNums: 

193 blockInfo = blockParser.getBlockInfo(block, seqNum) 

194 line = data[blockInfo.blockNumber, blockInfo.seqNum] 

195 items = line.split(f"{DELIMITER}") 

196 self.assertEqual(items[0], blockInfo.blockId) 

197 self.assertEqual(items[1], str(blockInfo.begin.value)) 

198 self.assertEqual(items[2], str(blockInfo.end.value)) 

199 self.assertEqual(items[3], str(blockInfo.salIndices)) 

200 self.assertEqual(items[4], str(blockInfo.tickets)) 

201 self.assertEqual(items[5], str(len(blockInfo.states))) 

202 

203 

204class TestMemory(lsst.utils.tests.MemoryTestCase): 

205 pass 

206 

207 

208def setup_module(module): 

209 lsst.utils.tests.init() 

210 

211 

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

213 lsst.utils.tests.init() 

214 unittest.main()