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.enterContext(self.butler) 

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

61 

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

63 

64 def tearDown(self): 

65 removeTestTempDir(self.root) 

66 

67 def testButlerLogRecordsFormatter(self): 

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

69 ButlerLogRecordHandler. 

70 """ 

71 handler = ButlerLogRecordHandler() 

72 

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

74 log.setLevel(logging.INFO) 

75 log.addHandler(handler) 

76 

77 log.info("An INFO message") 

78 log.debug("A DEBUG message") 

79 log.warning("A WARNING message") 

80 

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

82 records = self.butler.get(ref) 

83 

84 self.assertEqual(records, handler.records) 

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

86 

87 def testButlerLogRecordsFormatterExtra(self): 

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

89 ButlerLogRecordHandler with extra JSON data attached. 

90 """ 

91 handler = ButlerLogRecordHandler() 

92 

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

94 log.setLevel(logging.INFO) 

95 log.addHandler(handler) 

96 

97 log.info("An INFO message") 

98 log.debug("A DEBUG message") 

99 log.warning("A WARNING message") 

100 

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

102 handler.records.extra = extra_data 

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

104 records = self.butler.get(ref) 

105 

106 self.assertEqual(records, handler.records) 

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

108 

109 @unittest.skipIf( 

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

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

112 ) 

113 def testButlerLogRecordsFormatterV1(self): 

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

115 (just a JSON list). 

116 """ 

117 handler = ButlerLogRecordHandler() 

118 

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

120 log.setLevel(logging.INFO) 

121 log.addHandler(handler) 

122 

123 log.info("An INFO message") 

124 log.debug("A DEBUG message") 

125 log.warning("A WARNING message") 

126 

127 with tempfile.NamedTemporaryFile( 

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

129 ) as tmp: 

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

131 # so we write manually and ingest. 

132 tmp.write( 

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

134 exclude_unset=True, exclude_defaults=True 

135 ) 

136 ) 

137 tmp.close() 

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

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

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

141 

142 records = self.butler.get(ref) 

143 self.assertEqual(records, handler.records) 

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

145 

146 @unittest.skipIf( 

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

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

149 ) 

150 def testJsonLogRecordsFormatter(self): 

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

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

153 log.setLevel(logging.INFO) 

154 

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

156 handler = FileHandler(tmp.name) 

157 handler.setFormatter(JsonLogFormatter()) 

158 log.addHandler(handler) 

159 

160 log.info("An INFO message") 

161 log.debug("A DEBUG message") 

162 log.warning("A WARNING message") 

163 

164 handler.close() 

165 

166 # Now ingest the file. 

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

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

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

170 

171 records = self.butler.get(ref) 

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

173 

174 @unittest.skipIf( 

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

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

177 ) 

178 def testJsonLogRecordsFormatterExtra(self): 

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

180 extra JSON data appended. 

181 """ 

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

183 log.setLevel(logging.INFO) 

184 

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

186 handler = FileHandler(tmp.name) 

187 handler.setFormatter(JsonLogFormatter()) 

188 log.addHandler(handler) 

189 

190 log.info("An INFO message") 

191 log.debug("A DEBUG message") 

192 log.warning("A WARNING message") 

193 

194 handler.close() 

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

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

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

198 

199 # Now ingest the file. 

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

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

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

203 

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

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

206 self.assertEqual(records.extra, extra_data) 

207 

208 

209if __name__ == "__main__": 

210 unittest.main()