Coverage for tests / test_formatter.py: 14%
156 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-24 08:17 +0000
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/>.
28"""Tests related to the formatter infrastructure."""
30import inspect
31import os.path
32import unittest
34from lsst.daf.butler import (
35 Config,
36 DataCoordinate,
37 DatasetRef,
38 DatasetType,
39 DimensionUniverse,
40 FileDescriptor,
41 Formatter,
42 FormatterFactory,
43 FormatterV2,
44 Location,
45 StorageClass,
46)
47from lsst.daf.butler.tests import DatasetTestHelper
48from lsst.daf.butler.tests.testFormatters import (
49 DoNothingFormatter,
50 MultipleExtensionsFormatter,
51 SingleExtensionFormatter,
52)
53from lsst.resources import ResourcePath
55TESTDIR = os.path.abspath(os.path.dirname(__file__))
58class FormatterFactoryTestCase(unittest.TestCase, DatasetTestHelper):
59 """Tests of the formatter factory infrastructure."""
61 def setUp(self):
62 self.id = 0
63 self.factory = FormatterFactory()
64 self.universe = DimensionUniverse()
65 self.dataId = DataCoordinate.make_empty(self.universe)
67 # Dummy FileDescriptor for testing getFormatter
68 self.fileDescriptor = FileDescriptor(
69 Location("/a/b/c", "d"), StorageClass("DummyStorageClass", dict, None)
70 )
72 def assertIsFormatter(self, formatter):
73 """Check that the supplied parameter is either a Formatter instance
74 or Formatter class.
75 """
76 if inspect.isclass(formatter):
77 self.assertTrue(issubclass(formatter, Formatter | FormatterV2), f"Is {formatter} a Formatter")
78 else:
79 self.assertIsInstance(formatter, Formatter | FormatterV2)
81 def testFormatter(self):
82 """Check basic parameter exceptions"""
83 f = DoNothingFormatter(self.fileDescriptor, dataId=self.dataId)
84 self.assertEqual(f.writeRecipes, {})
85 self.assertEqual(f.writeParameters, {})
86 self.assertIn("DoNothingFormatter", repr(f))
87 self.assertIn("DoNothingFormatter", str(f))
89 with self.assertRaises(TypeError):
90 DoNothingFormatter()
92 with self.assertRaises(ValueError):
93 DoNothingFormatter(self.fileDescriptor, dataId=self.dataId, write_parameters={"param1": 0})
95 with self.assertRaises(RuntimeError):
96 DoNothingFormatter(self.fileDescriptor, dataID=self.dataId, write_recipes={"label": "value"})
98 with self.assertRaises(NotImplementedError):
99 f.write("str")
101 def testExtensionValidation(self):
102 """Test extension validation"""
103 for file, single_ok, multi_ok in (
104 ("e.fits", True, True),
105 ("e.fit", False, True),
106 ("e.fits.fz", False, True),
107 ("e.txt", False, False),
108 ("e.1.4.fits", True, True),
109 ("e.3.fit", False, True),
110 ("e.1.4.fits.gz", False, True),
111 ):
112 loc = Location("/a/b/c", file)
114 for formatter, passes in (
115 (SingleExtensionFormatter, single_ok),
116 (MultipleExtensionsFormatter, multi_ok),
117 ):
118 if passes:
119 formatter.validateExtension(loc)
120 else:
121 with self.assertRaises(ValueError):
122 formatter.validateExtension(loc)
124 def testRegistry(self):
125 """Check that formatters can be stored in the registry."""
126 formatterTypeName = "lsst.daf.butler.tests.deferredFormatter.DeferredFormatter"
127 storageClassName = "Image"
128 self.factory.registerFormatter(storageClassName, formatterTypeName)
129 f = self.factory.getFormatter(storageClassName, self.fileDescriptor, dataId=self.dataId)
130 self.assertIsFormatter(f)
131 self.assertEqual(f.name(), formatterTypeName)
132 self.assertIn(formatterTypeName, str(f))
133 self.assertIn(self.fileDescriptor.location.path, str(f))
135 fcls = self.factory.getFormatterClass(storageClassName)
136 self.assertIsFormatter(fcls)
137 # Defer the import so that we ensure that the infrastructure loaded
138 # it on demand previously
139 from lsst.daf.butler.tests.deferredFormatter import DeferredFormatter
141 self.assertEqual(type(f), DeferredFormatter)
143 with self.assertRaises(TypeError):
144 # Requires a constructor parameter
145 self.factory.getFormatter(storageClassName)
147 with self.assertRaises(KeyError):
148 self.factory.getFormatter("Missing", self.fileDescriptor)
150 # Check that a bad formatter path fails
151 storageClassName = "BadImage"
152 self.factory.registerFormatter(storageClassName, "lsst.daf.butler.tests.deferredFormatter.Unknown")
153 with self.assertRaises(ImportError):
154 self.factory.getFormatter(storageClassName, self.fileDescriptor, dataId=self.dataId)
156 def testRegistryWithStorageClass(self):
157 """Test that the registry can be given a StorageClass object."""
158 formatterTypeName = "lsst.daf.butler.formatters.yaml.YamlFormatter"
159 storageClassName = "TestClass"
160 sc = StorageClass(storageClassName, dict, None)
162 datasetType = DatasetType("calexp", self.universe.empty, sc)
163 ref = DatasetRef(datasetType, self.dataId, "test")
165 # Store using an instance
166 self.factory.registerFormatter(sc, formatterTypeName)
168 # Retrieve using the class
169 f = self.factory.getFormatter(sc, self.fileDescriptor, dataId=self.dataId, ref=ref)
170 self.assertIsFormatter(f)
171 self.assertEqual(f.file_descriptor, self.fileDescriptor)
173 # Retrieve using the DatasetType
174 f2 = self.factory.getFormatter(datasetType, self.fileDescriptor, dataId=self.dataId, ref=ref)
175 self.assertIsFormatter(f2)
176 self.assertEqual(f.name(), f2.name())
178 # Class directly
179 f2cls = self.factory.getFormatterClass(datasetType)
180 self.assertIsFormatter(f2cls)
182 # This might defer the import, pytest may have already loaded it
183 from lsst.daf.butler.formatters.yaml import YamlFormatter
185 self.assertEqual(type(f), YamlFormatter)
187 with self.assertRaises(KeyError):
188 # Attempt to overwrite using a different value
189 self.factory.registerFormatter(storageClassName, "lsst.daf.butler.formatters.json.JsonFormatter")
191 def testRegistryConfig(self):
192 configFile = os.path.join(TESTDIR, "config", "basic", "posixDatastore.yaml")
193 config = Config(configFile)
194 self.factory.registerFormatters(config["datastore", "formatters"], universe=self.universe)
196 # Create a DatasetRef with and without instrument matching the
197 # one in the config file.
198 dimensions = self.universe.conform(("visit", "physical_filter", "instrument"))
199 constant_dataId = {"physical_filter": "v", "visit": 1}
200 sc = StorageClass("DummySC", dict, None)
201 refPviHsc = self.makeDatasetRef(
202 "pvi",
203 dimensions,
204 sc,
205 {"instrument": "DummyHSC", **constant_dataId},
206 )
207 refPviHscFmt = self.factory.getFormatterClass(refPviHsc)
208 self.assertIsFormatter(refPviHscFmt)
209 self.assertIn("JsonFormatter", refPviHscFmt.name())
211 refPviNotHsc = self.makeDatasetRef(
212 "pvi",
213 dimensions,
214 sc,
215 {"instrument": "DummyNotHSC", **constant_dataId},
216 )
217 refPviNotHscFmt = self.factory.getFormatterClass(refPviNotHsc)
218 self.assertIsFormatter(refPviNotHscFmt)
219 self.assertIn("PickleFormatter", refPviNotHscFmt.name())
221 # Create a DatasetRef that should fall back to using Dimensions
222 refPvixHsc = self.makeDatasetRef(
223 "pvix",
224 dimensions,
225 sc,
226 {"instrument": "DummyHSC", **constant_dataId},
227 )
228 refPvixNotHscFmt = self.factory.getFormatterClass(refPvixHsc)
229 self.assertIsFormatter(refPvixNotHscFmt)
230 self.assertIn("PickleFormatter", refPvixNotHscFmt.name())
232 # Create a DatasetRef that should fall back to using StorageClass
233 dimensionsNoV = self.universe.conform(("physical_filter", "instrument"))
234 refPvixNotHscDims = self.makeDatasetRef(
235 "pvix",
236 dimensionsNoV,
237 sc,
238 {"instrument": "DummyHSC", "physical_filter": "v"},
239 )
240 refPvixNotHscDims_fmt = self.factory.getFormatterClass(refPvixNotHscDims)
241 self.assertIsFormatter(refPvixNotHscDims_fmt)
242 self.assertIn("YamlFormatter", refPvixNotHscDims_fmt.name())
244 # Check that parameters are stored
245 refParam = self.makeDatasetRef(
246 "paramtest",
247 dimensions,
248 sc,
249 {"instrument": "DummyNotHSC", **constant_dataId},
250 )
251 lookup, refParam_fmt, kwargs = self.factory.getFormatterClassWithMatch(refParam)
252 self.assertIn("write_parameters", kwargs)
253 expected = {"max": 5, "min": 2, "comment": "Additional commentary", "recipe": "recipe1"}
254 self.assertEqual(kwargs["write_parameters"], expected)
255 self.assertIn("FormatterTest", refParam_fmt.name())
257 f = self.factory.getFormatter(refParam, self.fileDescriptor, dataId=self.dataId, ref=refParam)
258 self.assertEqual(f.writeParameters, expected)
260 f = self.factory.getFormatter(
261 refParam, self.fileDescriptor, dataId=self.dataId, write_parameters={"min": 22, "extra": 50}
262 )
263 self.assertEqual(
264 f.write_parameters,
265 {"max": 5, "min": 22, "comment": "Additional commentary", "extra": 50, "recipe": "recipe1"},
266 )
268 self.assertIn("recipe1", f.write_recipes)
269 self.assertEqual(f.write_parameters["recipe"], "recipe1")
271 with self.assertRaises(ValueError):
272 # "new" is not allowed as a write parameter
273 self.factory.getFormatter(
274 refParam, self.fileDescriptor, dataId=self.dataId, write_parameters={"new": 1}, ref=refParam
275 )
277 with self.assertRaises(RuntimeError):
278 # "mode" is a required recipe parameter
279 self.factory.getFormatter(
280 refParam,
281 self.fileDescriptor,
282 dataId=self.dataId,
283 write_recipes={"recipe3": {"notmode": 1}},
284 ref=refParam,
285 )
288class ZipFormatterTestCase(unittest.TestCase):
289 """Test that files can be read from Zip files via formatter V2."""
291 @classmethod
292 def setUpClass(cls):
293 cls.zip_file = ResourcePath(os.path.join(TESTDIR, "data", "formatter_tests.zip"))
295 # Need a dataset ref but it can be empty.
296 universe = DimensionUniverse()
297 sc = StorageClass("Test", dict, None)
298 datasetType = DatasetType("test", universe.empty, sc)
299 cls.ref = DatasetRef(datasetType, DataCoordinate.make_empty(universe), "test_run")
301 def _make_formatter(self, storage_type, formatter_type, path_in_zip) -> Formatter:
302 storageClass = StorageClass("Something", storage_type)
303 uri = self.zip_file.replace(fragment=f"zip-path={path_in_zip}")
304 descriptor = FileDescriptor(Location(None, uri), storageClass)
305 formatter = formatter_type(descriptor, ref=self.ref)
306 return formatter
308 def test_packages(self):
309 from lsst.daf.butler.formatters.packages import PackagesFormatter
310 from lsst.utils.packages import Packages
312 formatter = self._make_formatter(Packages, PackagesFormatter, "formatter_tests/packages.yaml")
313 packages = formatter.read()
314 self.assertIsInstance(packages, Packages)
315 self.assertEqual(packages["python"], "3.11.8")
317 def test_metrics(self):
318 from lsst.daf.butler.tests import MetricsExample
319 from lsst.daf.butler.tests.testFormatters import MetricsExampleFormatter
321 formatter = self._make_formatter(
322 MetricsExample, MetricsExampleFormatter, "formatter_tests/metrics.yaml"
323 )
324 metrics = formatter.read()
325 self.assertIsInstance(metrics, MetricsExample)
326 self.assertEqual(metrics.summary["key"], 1)
328 def test_logs(self):
329 from lsst.daf.butler.formatters.logs import ButlerLogRecordsFormatter
330 from lsst.daf.butler.logging import ButlerLogRecords
332 formatter = self._make_formatter(
333 ButlerLogRecords, ButlerLogRecordsFormatter, "formatter_tests/logs.json"
334 )
335 logs = formatter.read()
336 self.assertIsInstance(logs, ButlerLogRecords)
337 self.assertEqual(len(logs), 39)
340if __name__ == "__main__":
341 unittest.main()