Coverage for tests / test_translate_header.py: 19%

1# This file is part of astro_metadata_translator. 

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 LICENSE file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# Use of this source code is governed by a 3-clause BSD-style 

10# license that can be found in the LICENSE file. 

11 

12import io 

13import logging 

14import os.path 

15import unittest 

16 

17from astro_metadata_translator.bin.translate import translate_or_dump_headers 

18 

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

20TESTDATA = os.path.join(TESTDIR, "data") 

21 

22 

23class TestTranslateHeader(unittest.TestCase): 

24 """Test that the astrometadata translate and dump logic works.""" 

25 

26 def _readlines(self, stream: io.StringIO) -> list[str]: 

27 """Return the lines written to the stream. 

28 

29 Parameters 

30 ---------- 

31 stream : `io.StringIO` 

32 The stream to read. 

33 

34 Returns 

35 ------- 

36 lines : `list` of `str` 

37 The lines contained in the stream. 

38 """ 

39 stream.seek(0) 

40 return [ll.rstrip() for ll in stream.readlines()] 

41 

42 def assert_ok_fail( 

43 self, okay: list[str], failed: list[str], stdout: list[str], expected: tuple[int, int] 

44 ) -> None: 

45 """Check that we have the expected numbers of successes and 

46 failures. 

47 

48 Parameters 

49 ---------- 

50 okay : `list` [`str`] 

51 List of successful translations. 

52 failed : `list` [`str`] 

53 List of failed translations. 

54 stdout : `list` [`str`] 

55 Additional information output. 

56 expected : `tuple` [`int`, `int`] 

57 Expected length of ``okay`` and ``failed``. 

58 """ 

59 # Form message to issue if the test fails. 

60 newline = "\n" # f-string can not accept \ in string. 

61 msg = f"""Converted successfully: 

62{newline.join(okay)} 

63Failed conversions: 

64{newline.join(failed)} 

65Standard output: 

66{newline.join(stdout)} 

67""" 

68 self.assertEqual((len(okay), len(failed)), expected, msg=msg) 

69 

70 def test_translate_header(self) -> None: 

71 """Translate some header files. Use URI for test dir.""" 

72 with io.StringIO() as out: 

73 with self.assertLogs(level=logging.INFO) as cm: 

74 okay, failed = translate_or_dump_headers( 

75 [f"file://localhost{TESTDATA}"], 

76 r"^fitsheader.*yaml$", 

77 0, 

78 False, 

79 outstream=out, 

80 output_mode="none", 

81 ) 

82 output = self._readlines(out) 

83 self.assertEqual(output, []) 

84 lines = [r.getMessage() for r in cm.records] 

85 self.assertEqual(len(lines), 11) 

86 self.assertTrue(lines[0].startswith("Analyzing"), f"Line: '{lines[0]}'") 

87 

88 self.assert_ok_fail(okay, failed, output, (11, 0)) 

89 

90 def test_translate_header_table(self) -> None: 

91 """Translate some header files with table output.""" 

92 with io.StringIO() as out: 

93 with self.assertLogs(level=logging.WARNING) as cm: 

94 logging.getLogger().warning("False warning") 

95 okay, failed = translate_or_dump_headers( 

96 [TESTDATA], 

97 r"^fitsheader.*yaml$", 

98 0, 

99 False, 

100 outstream=out, 

101 ) 

102 output = self._readlines(out) 

103 self.assertIn("ObsId", output[0]) 

104 self.assertTrue(output[2].startswith("-------")) 

105 self.assertEqual(len(output), 14) # 3 header lines for QTable. 

106 self.assertEqual(len(cm.output), 1) # Should only have the warning this test made. 

107 

108 self.assert_ok_fail(okay, failed, output, (11, 0)) 

109 

110 def test_translate_bad_header_table(self) -> None: 

111 """Translate a header that has a bad translation in the table.""" 

112 with io.StringIO() as out: 

113 with self.assertLogs(level=logging.WARNING): 

114 okay, failed = translate_or_dump_headers( 

115 [TESTDATA], "^bad-megaprime.yaml$", 0, False, outstream=out, output_mode="table" 

116 ) 

117 output = self._readlines(out) 

118 self.assertIn(" -- ", output[3]) # String masked value. 

119 self.assertIn(" ———", output[3]) # Quantity masked value. 

120 

121 def test_translate_header_fails(self) -> None: 

122 """Translate some header files that fail.""" 

123 with io.StringIO() as out: 

124 with self.assertLogs(level=logging.INFO) as cm: 

125 okay, failed = translate_or_dump_headers( 

126 [TESTDATA], r"^.*yaml$", 0, False, outstream=out, output_mode="none" 

127 ) 

128 

129 out_lines = self._readlines(out) 

130 self.assertEqual(len(out_lines), len(failed)) 

131 self.assertTrue(out_lines[0].startswith("Failure processing"), f"Line: '{out_lines[0]}'") 

132 self.assertIn("not a mapping", out_lines[0], f"Line: '{out_lines[0]}'") 

133 

134 err_lines = [r.getMessage() for r in cm.records] 

135 # Filter out warnings. 

136 analyzed = [e for e in err_lines if e.startswith("Analyzing")] 

137 self.assertEqual(len(analyzed), 16) # The number of files analyzed 

138 self.assertTrue(err_lines[0].startswith("Analyzing"), f"Line: '{err_lines[0]}'") 

139 

140 self.assert_ok_fail(okay, failed, out_lines, (12, 4)) 

141 

142 def test_translate_header_traceback(self) -> None: 

143 """Translate some header files that fail and trigger traceback.""" 

144 with io.StringIO() as out: 

145 with self.assertLogs(level=logging.INFO) as cm: 

146 okay, failed = translate_or_dump_headers( 

147 [TESTDATA], r"^.*yaml$", 0, True, outstream=out, output_mode="none" 

148 ) 

149 

150 out_lines = self._readlines(out) 

151 self.assertGreaterEqual(len(out_lines), 22, "\n".join(out_lines)) 

152 self.assertTrue(out_lines[0].startswith("Traceback"), f"Line '{out_lines[0]}'") 

153 

154 lines = [r.getMessage() for r in cm.records] 

155 self.assertGreaterEqual(len(lines), 13, "\n".join(lines)) 

156 self.assertTrue(lines[0].startswith("Analyzing"), f"Line: '{lines[0]}'") 

157 

158 self.assert_ok_fail(okay, failed, out_lines, (12, 4)) 

159 

160 def test_translate_header_dump(self) -> None: 

161 """Check that a header is dumped.""" 

162 with io.StringIO() as out: 

163 with self.assertLogs(level=logging.INFO) as cm: 

164 okay, failed = translate_or_dump_headers( 

165 [os.path.join(TESTDATA, "fitsheader-decam.yaml")], 

166 r"^fitsheader.*yaml$", 

167 0, 

168 False, 

169 outstream=out, 

170 output_mode="yaml", 

171 ) 

172 

173 out_lines = self._readlines(out) 

174 # Look for a DECam header in the output 

175 header = "\n".join(out_lines) 

176 self.assertIn("DTINSTRU", header) 

177 

178 lines = [r.getMessage() for r in cm.records] 

179 self.assertEqual(len(lines), 1) 

180 self.assertTrue(lines[0], "Analyzing tests/data/fitsheader-decam.yaml...") 

181 

182 self.assert_ok_fail(okay, failed, out_lines, (1, 0)) 

183 

184 def test_translate_header_loud(self) -> None: 

185 """Check that ObservationInfo content is displayed.""" 

186 with io.StringIO() as out: 

187 with self.assertLogs(level=logging.INFO) as cm: 

188 okay, failed = translate_or_dump_headers( 

189 [os.path.join(TESTDATA, "fitsheader-decam.yaml")], 

190 r"^fitsheader.*yaml$", 

191 0, 

192 False, 

193 outstream=out, 

194 output_mode="verbose", 

195 ) 

196 

197 out_lines = self._readlines(out) 

198 # Look for the translated DECam header in the output 

199 self.assertEqual(out_lines[2], "datetime_begin: 2013-09-01T06:02:55.754") 

200 

201 lines = [r.getMessage() for r in cm.records] 

202 self.assertEqual(len(lines), 1) 

203 self.assertTrue(lines[0], "Analyzing tests/data/fitsheader-decam.yaml...") 

204 

205 self.assert_ok_fail(okay, failed, out_lines, (1, 0)) 

206 

207 

208if __name__ == "__main__": 

209 unittest.main()