Coverage for tests/test_logFormatter.py: 21%

1# This file is part of daf_butler. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (http://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 software is dual licensed under the GNU General Public License and also 

10# under a 3-clause BSD license. Recipients may choose which of these licenses 

11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt, 

12# respectively. If you choose the GPL option then the following text applies 

13# (but note that there is still no warranty even if you opt for BSD instead): 

14# 

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

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

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

18# (at your option) any later version. 

19# 

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

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

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

23# GNU General Public License for more details. 

24# 

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

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

27 

28"""Tests for ButlerLogRecordsFormatter.""" 

29 

30import json 

31import logging 

32import os 

33import sys 

34import tempfile 

35import unittest 

36from logging import FileHandler 

37 

38from lsst.daf.butler import Butler, DatasetRef, DatasetType, FileDataset 

39from lsst.daf.butler.logging import ( 

40 ButlerLogRecordHandler, 

41 ButlerLogRecords, 

42 JsonLogFormatter, 

43 _ButlerLogRecordsModelV1, 

44) 

45from lsst.daf.butler.tests.utils import makeTestTempDir, removeTestTempDir 

46 

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

48 

49 

50class ButlerLogRecordsFormatterTestCase(unittest.TestCase): 

51 """Test for ButlerLogRecords put/get.""" 

52 

53 def setUp(self): 

54 self.root = makeTestTempDir(TESTDIR) 

55 Butler.makeRepo(self.root) 

56 

57 self.run = "testrun" 

58 self.butler = Butler.from_config(self.root, run=self.run) 

59 self.datasetType = DatasetType("test_logs", [], "ButlerLogRecords", universe=self.butler.dimensions) 

60 

61 self.butler.registry.registerDatasetType(self.datasetType) 

62 

63 def tearDown(self): 

64 removeTestTempDir(self.root) 

65 

66 def testButlerLogRecordsFormatter(self): 

67 """Test that we can get and put records handled by 

68 ButlerLogRecordHandler. 

69 """ 

70 handler = ButlerLogRecordHandler() 

71 

72 log = logging.getLogger(self.id()) 

73 log.setLevel(logging.INFO) 

74 log.addHandler(handler) 

75 

76 log.info("An INFO message") 

77 log.debug("A DEBUG message") 

78 log.warning("A WARNING message") 

79 

80 ref = self.butler.put(handler.records, self.datasetType) 

81 records = self.butler.get(ref) 

82 

83 self.assertEqual(records, handler.records) 

84 self.assertEqual(len(records), 2) 

85 

86 def testButlerLogRecordsFormatterExtra(self): 

87 """Test that we can get and put records handled by 

88 ButlerLogRecordHandler with extra JSON data attached. 

89 """ 

90 handler = ButlerLogRecordHandler() 

91 

92 log = logging.getLogger(self.id()) 

93 log.setLevel(logging.INFO) 

94 log.addHandler(handler) 

95 

96 log.info("An INFO message") 

97 log.debug("A DEBUG message") 

98 log.warning("A WARNING message") 

99 

100 extra_data = {"extra1": 1, "extra2": 2} 

101 handler.records.extra = extra_data 

102 ref = self.butler.put(handler.records, self.datasetType) 

103 records = self.butler.get(ref) 

104 

105 self.assertEqual(records, handler.records) 

106 self.assertEqual(len(records), 2) 

107 

108 @unittest.skipIf( 

109 sys.version_info < (3, 12, 0), 

110 "This test requires NamedTemporaryFile behavior not available in older Python versions.", 

111 ) 

112 def testButlerLogRecordsFormatterV1(self): 

113 """Test that we can read log records stored via the old V1 format 

114 (just a JSON list). 

115 """ 

116 handler = ButlerLogRecordHandler() 

117 

118 log = logging.getLogger(self.id()) 

119 log.setLevel(logging.INFO) 

120 log.addHandler(handler) 

121 

122 log.info("An INFO message") 

123 log.debug("A DEBUG message") 

124 log.warning("A WARNING message") 

125 

126 with tempfile.NamedTemporaryFile( 

127 mode="w", suffix=".json", prefix="butler-log-", delete_on_close=False 

128 ) as tmp: 

129 # We can't use butler.put since that always writes the new format, 

130 # so we write manually and ingest. 

131 tmp.write( 

132 _ButlerLogRecordsModelV1(root=handler.records._records).model_dump_json( 

133 exclude_unset=True, exclude_defaults=True 

134 ) 

135 ) 

136 tmp.close() 

137 ref = DatasetRef(self.datasetType, dataId={}, run=self.run) 

138 dataset = FileDataset(path=os.path.abspath(tmp.name), refs=ref) 

139 self.butler.ingest(dataset, transfer="move") 

140 

141 records = self.butler.get(ref) 

142 self.assertEqual(records, handler.records) 

143 self.assertEqual(len(records), 2) 

144 

145 @unittest.skipIf( 

146 sys.version_info < (3, 12, 0), 

147 "This test requires NamedTemporaryFile behavior not available in older Python versions.", 

148 ) 

149 def testJsonLogRecordsFormatter(self): 

150 """Test that externally created JSON format stream files work.""" 

151 log = logging.getLogger(self.id()) 

152 log.setLevel(logging.INFO) 

153 

154 with tempfile.NamedTemporaryFile(mode="w", suffix=".json", prefix="butler-log-") as tmp: 

155 handler = FileHandler(tmp.name) 

156 handler.setFormatter(JsonLogFormatter()) 

157 log.addHandler(handler) 

158 

159 log.info("An INFO message") 

160 log.debug("A DEBUG message") 

161 log.warning("A WARNING message") 

162 

163 handler.close() 

164 

165 # Now ingest the file. 

166 ref = DatasetRef(self.datasetType, dataId={}, run=self.run) 

167 dataset = FileDataset(path=tmp.name, refs=ref) 

168 self.butler.ingest(dataset, transfer="move") 

169 

170 records = self.butler.get(ref) 

171 self.assertEqual(len(records), 2) 

172 

173 @unittest.skipIf( 

174 sys.version_info < (3, 12, 0), 

175 "This test requires NamedTemporaryFile behavior not available in older Python versions.", 

176 ) 

177 def testJsonLogRecordsFormatterExtra(self): 

178 """Test that externally created JSON format stream files work with 

179 extra JSON data appended. 

180 """ 

181 log = logging.getLogger(self.id()) 

182 log.setLevel(logging.INFO) 

183 

184 with tempfile.NamedTemporaryFile(mode="w", suffix=".json", prefix="butler-log-") as tmp: 

185 handler = FileHandler(tmp.name) 

186 handler.setFormatter(JsonLogFormatter()) 

187 log.addHandler(handler) 

188 

189 log.info("An INFO message") 

190 log.debug("A DEBUG message") 

191 log.warning("A WARNING message") 

192 

193 handler.close() 

194 extra_data = {"extra1": 1, "extra2": 2} 

195 with open(tmp.name, "a") as stream: 

196 ButlerLogRecords.write_streaming_extra(stream, json.dumps(extra_data)) 

197 

198 # Now ingest the file. 

199 ref = DatasetRef(self.datasetType, dataId={}, run=self.run) 

200 dataset = FileDataset(path=tmp.name, refs=ref) 

201 self.butler.ingest(dataset, transfer="move") 

202 

203 records: ButlerLogRecords = self.butler.get(ref) 

204 self.assertEqual(len(records), 2) 

205 self.assertEqual(records.extra, extra_data) 

206 

207 

208if __name__ == "__main__": 

209 unittest.main()