Coverage for tests/test_blockUtils.py: 84%

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 

28from typing import Any 

29 

30import pandas as pd 

31from utils import getVcr 

32 

33import lsst.utils.tests 

34from lsst.summit.utils.blockUtils import BlockParser 

35from lsst.summit.utils.efdUtils import makeEfdClient 

36 

37__all__ = ("writeNewBlockInfoTestTruthValues",) 

38 

39HAS_EFD_CLIENT = True 

40try: 

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

42except ImportError: 

43 HAS_EFD_CLIENT = False 

44 

45vcr = getVcr() 

46 

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

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

49 

50 

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

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

53 

54 Parameters 

55 ---------- 

56 dayObs : `int`, optional 

57 The dayObs to get the truth values for. 

58 

59 Returns 

60 ------- 

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

62 The block info truth data. 

63 """ 

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

65 

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

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

68 

69 data = {} 

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

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

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

73 data[blockNum, blockSeqNum] = line 

74 return data 

75 

76 

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

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

79 

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

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

82 the new values into git. 

83 """ 

84 blockParser = BlockParser(dayObs) 

85 

86 data = {} 

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

88 seqNums = blockParser.getSeqNums(block) 

89 for seqNum in seqNums: 

90 blockInfo = blockParser.getBlockInfo(block, seqNum) 

91 assert blockInfo is not None 

92 line = ( 

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

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

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

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

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

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

99 ) 

100 # must store as string not tuple for json serialization 

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

102 

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

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

105 json.dump(data, f) 

106 

107 

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

109@vcr.use_cassette() 

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

111 # class attributes populated in setUpClass 

112 client: Any 

113 dayObsNoTestCases: int 

114 dayObsWithCases: int 

115 dayObsNoBlocks: int 

116 blockParser: BlockParser 

117 blockNums: list[str] 

118 blockDict: dict[str, list[int]] 

119 

120 @classmethod 

121 @vcr.use_cassette() 

122 def setUpClass(cls) -> None: 

123 try: 

124 cls.client = makeEfdClient(testing=True) 

125 except RuntimeError: 

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

127 

128 cls.dayObsNoTestCases = 20230615 

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

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

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

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

133 cls.blockDict = {} 

134 for block in cls.blockNums: 

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

136 

137 @vcr.use_cassette() 

138 def tearDown(self) -> None: 

139 loop = asyncio.get_event_loop() 

140 if self.client.influx_client is not None: 140 ↛ exitline 140 didn't return from function 'tearDown' because the condition on line 140 was always true

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

142 

143 @vcr.use_cassette() 

144 def test_parsing(self) -> None: 

145 blockNums = self.blockParser.getBlockNums() 

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

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

148 

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

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

151 self.assertIsInstance(seqNums, list) 

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

153 

154 found = self.blockParser.getSeqNums(block) 

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

156 self.assertEqual(found, seqNums) 

157 self.blockParser.printBlockEvolution(block) 

158 

159 for seqNum in seqNums: 

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

161 self.assertIsInstance(data, pd.DataFrame) 

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

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

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

165 

166 @vcr.use_cassette() 

167 def test_notFoundBehavior(self) -> None: 

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

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

170 self.assertIsInstance(blockParser, BlockParser) 

171 

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

173 blocks = blockParser.getBlockNums() 

174 self.assertIsInstance(blocks, list) 

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

176 

177 seqNums = blockParser.getSeqNums(block=123) 

178 self.assertIsInstance(seqNums, list) 

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

180 

181 # just check this doesn't raise 

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

183 

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

185 # when there is data present 

186 blockParser = self.blockParser 

187 seqNums = blockParser.getSeqNums(block=9999999) 

188 self.assertIsInstance(seqNums, list) 

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

190 

191 # just check this doesn't raise 

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

193 

194 @vcr.use_cassette() 

195 def test_actualValues(self) -> None: 

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

197 data = getBlockInfoTestTruthValues(dayObs) 

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

199 

200 for block in blockParser.getBlockNums(): 

201 seqNums = blockParser.getSeqNums(block) 

202 for seqNum in seqNums: 

203 blockInfo = blockParser.getBlockInfo(block, seqNum) 

204 assert blockInfo is not None 

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

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

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

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

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

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

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

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

213 

214 

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

216 pass 

217 

218 

219def setup_module(module: object) -> None: 

220 lsst.utils.tests.init() 

221 

222 

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

224 lsst.utils.tests.init() 

225 unittest.main()