Coverage for tests/test_datastore.py: 99%
1314 statements
« prev ^ index » next coverage.py v7.17.2, created at 2026-11-06 09:46 +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/>.
28from __future__ import annotations
30import contextlib
31import os
32import pickle
33import shutil
34import tempfile
35import time
36import unittest
37import unittest.mock
38import uuid
39from collections.abc import Callable, Iterator
40from typing import Any, cast
42import yaml
44import lsst.utils.tests
45from lsst.daf.butler import (
46 Config,
47 DataCoordinate,
48 DatasetIdGenEnum,
49 DatasetRef,
50 DatasetType,
51 DatasetTypeNotSupportedError,
52 Datastore,
53 DimensionUniverse,
54 FileDataset,
55 StorageClass,
56 StorageClassFactory,
57)
58from lsst.daf.butler.datastore import (
59 DatasetRefURIs,
60 DatastoreConfig,
61 DatastoreValidationError,
62 DuplicateIdError,
63 NullDatastore,
64)
65from lsst.daf.butler.datastore.cache_manager import (
66 DatastoreCacheManager,
67 DatastoreCacheManagerConfig,
68 DatastoreDisabledCacheManager,
69)
70from lsst.daf.butler.datastore.record_data import (
71 DatastoreRecordData,
72 DatastoreRecordTable,
73 SerializedDatastoreRecordData,
74)
75from lsst.daf.butler.datastore.stored_file_info import (
76 StoredFileInfo,
77 StoredFileInfoTable,
78 make_datastore_path_relative,
79)
80from lsst.daf.butler.datastores.file_datastore.retrieve_artifacts import retrieve_and_zip
81from lsst.daf.butler.formatters.yaml import YamlFormatter
82from lsst.daf.butler.tests import (
83 BadNoWriteFormatter,
84 BadWriteFormatter,
85 DatasetTestHelper,
86 DatastoreTestHelper,
87 DummyRegistry,
88 MetricsExample,
89 MetricsExampleDataclass,
90 MetricsExampleModel,
91)
92from lsst.daf.butler.tests.dict_convertible_model import DictConvertibleModel
93from lsst.daf.butler.tests.utils import TestCaseMixin
94from lsst.resources import ResourcePath
95from lsst.utils import doImport
96from lsst.utils.introspection import get_full_type_name
98TESTDIR = os.path.dirname(__file__)
101def makeExampleMetrics(use_none: bool = False) -> MetricsExample:
102 """Make example dataset that can be stored in butler."""
103 if use_none:
104 array = None
105 else:
106 array = [563, 234, 456.7, 105, 2054, -1045]
107 return MetricsExample(
108 {"AM1": 5.2, "AM2": 30.6},
109 {"a": [1, 2, 3], "b": {"blue": 5, "red": "green"}},
110 array,
111 )
114class TransactionTestError(Exception):
115 """Specific error for transactions, to prevent misdiagnosing
116 that might otherwise occur when a standard exception is used.
117 """
119 pass
122class DatastoreTestsBase(DatasetTestHelper, DatastoreTestHelper, TestCaseMixin):
123 """Support routines for datastore testing"""
125 root: str | None = None
126 universe: DimensionUniverse
127 storageClassFactory: StorageClassFactory
129 @classmethod
130 def setUpClass(cls) -> None:
131 # Storage Classes are fixed for all datastores in these tests
132 scConfigFile = os.path.join(TESTDIR, "config/basic/storageClasses.yaml")
133 cls.storageClassFactory = StorageClassFactory()
134 cls.storageClassFactory.addFromConfig(scConfigFile)
136 # Read the Datastore config so we can get the class
137 # information (since we should not assume the constructor
138 # name here, but rely on the configuration file itself)
139 datastoreConfig = DatastoreConfig(cls.configFile)
140 cls.datastoreType = cast(type[Datastore], doImport(datastoreConfig["cls"]))
141 cls.universe = DimensionUniverse()
143 def setUp(self) -> None:
144 self.setUpDatastoreTests(DummyRegistry, DatastoreConfig)
146 def tearDown(self) -> None:
147 if self.root is not None and os.path.exists(self.root):
148 shutil.rmtree(self.root, ignore_errors=True)
151class DatastoreTests(DatastoreTestsBase):
152 """Some basic tests of a simple datastore."""
154 hasUnsupportedPut = True
155 rootKeys: tuple[str, ...] | None = None
156 isEphemeral: bool = False
157 validationCanFail: bool = False
159 def testConfigRoot(self) -> None:
160 full = DatastoreConfig(self.configFile)
161 config = DatastoreConfig(self.configFile, mergeDefaults=False)
162 newroot = "/random/location"
163 self.datastoreType.setConfigRoot(newroot, config, full)
164 if self.rootKeys:
165 for k in self.rootKeys:
166 self.assertIn(newroot, config[k])
168 def testConstructor(self) -> None:
169 datastore = self.makeDatastore()
170 self.assertIsNotNone(datastore)
171 self.assertIs(datastore.isEphemeral, self.isEphemeral)
173 def testConfigurationValidation(self) -> None:
174 datastore = self.makeDatastore()
175 sc = self.storageClassFactory.getStorageClass("ThingOne")
176 datastore.validateConfiguration([sc])
178 sc2 = self.storageClassFactory.getStorageClass("ThingTwo")
179 if self.validationCanFail:
180 with self.assertRaises(DatastoreValidationError):
181 datastore.validateConfiguration([sc2], logFailures=True)
183 dimensions = self.universe.conform(("visit", "physical_filter"))
184 dataId = {
185 "instrument": "dummy",
186 "visit": 52,
187 "physical_filter": "V",
188 "band": "v",
189 "day_obs": 20250101,
190 }
191 ref = self.makeDatasetRef("metric", dimensions, sc, dataId)
192 datastore.validateConfiguration([ref])
194 def testParameterValidation(self) -> None:
195 """Check that parameters are validated"""
196 sc = self.storageClassFactory.getStorageClass("ThingOne")
197 dimensions = self.universe.conform(("visit", "physical_filter"))
198 dataId = {
199 "instrument": "dummy",
200 "visit": 52,
201 "physical_filter": "V",
202 "band": "v",
203 "day_obs": 20250101,
204 }
205 ref = self.makeDatasetRef("metric", dimensions, sc, dataId)
206 datastore = self.makeDatastore()
207 data = {1: 2, 3: 4}
208 datastore.put(data, ref)
209 newdata = datastore.get(ref)
210 self.assertEqual(data, newdata)
211 with self.assertRaises(KeyError):
212 newdata = datastore.get(ref, parameters={"missing": 5})
214 def testBasicPutGet(self) -> None:
215 metrics = makeExampleMetrics()
216 datastore = self.makeDatastore()
218 # Create multiple storage classes for testing different formulations
219 storageClasses = [
220 self.storageClassFactory.getStorageClass(sc)
221 for sc in ("StructuredData", "StructuredDataJson", "StructuredDataPickle")
222 ]
224 dimensions = self.universe.conform(("visit", "physical_filter"))
225 dataId = {
226 "instrument": "dummy",
227 "visit": 52,
228 "physical_filter": "V",
229 "band": "v",
230 "day_obs": 20250101,
231 }
232 dataId2 = {
233 "instrument": "dummy",
234 "visit": 53,
235 "physical_filter": "V",
236 "band": "v",
237 "day_obs": 20250101,
238 }
240 for sc in storageClasses:
241 ref = self.makeDatasetRef("metric", dimensions, sc, dataId)
242 ref2 = self.makeDatasetRef("metric", dimensions, sc, dataId2)
244 # Make sure that using getManyURIs without predicting before the
245 # dataset has been put raises.
246 with self.assertRaises(FileNotFoundError):
247 datastore.getManyURIs([ref], predict=False)
249 # Make sure that using getManyURIs with predicting before the
250 # dataset has been put predicts the URI.
251 uris = datastore.getManyURIs([ref, ref2], predict=True)
252 self.assertIn("52", uris[ref].primaryURI.geturl())
253 self.assertIn("#predicted", uris[ref].primaryURI.geturl())
254 self.assertIn("53", uris[ref2].primaryURI.geturl())
255 self.assertIn("#predicted", uris[ref2].primaryURI.geturl())
257 datastore.put(metrics, ref)
259 # Does it exist?
260 self.assertTrue(datastore.exists(ref))
261 self.assertTrue(datastore.knows(ref))
262 multi = datastore.knows_these([ref])
263 self.assertTrue(multi[ref])
264 multi = datastore.mexists([ref, ref2])
265 self.assertTrue(multi[ref])
266 self.assertFalse(multi[ref2])
268 # Get
269 metricsOut = datastore.get(ref, parameters=None)
270 self.assertEqual(metrics, metricsOut)
272 uri = datastore.getURI(ref)
273 self.assertEqual(uri.scheme, self.uriScheme)
275 uris = datastore.getManyURIs([ref])
276 self.assertEqual(len(uris), 1)
277 ref, uri = uris.popitem()
278 self.assertTrue(uri.primaryURI.exists())
279 self.assertFalse(uri.componentURIs)
281 # Get a component -- we need to construct new refs for them
282 # with derived storage classes but with parent ID
283 for comp in ("data", "output"):
284 compRef = ref.makeComponentRef(comp)
285 output = datastore.get(compRef)
286 self.assertEqual(output, getattr(metricsOut, comp))
288 uri = datastore.getURI(compRef)
289 self.assertEqual(uri.scheme, self.uriScheme)
291 uris = datastore.getManyURIs([compRef])
292 self.assertEqual(len(uris), 1)
294 storageClass = sc
296 # Check that we can put a metric with None in a component and
297 # get it back as None
298 metricsNone = makeExampleMetrics(use_none=True)
299 dataIdNone = {
300 "instrument": "dummy",
301 "visit": 54,
302 "physical_filter": "V",
303 "band": "v",
304 "day_obs": 20250101,
305 }
306 refNone = self.makeDatasetRef("metric", dimensions, sc, dataIdNone)
307 datastore.put(metricsNone, refNone)
309 comp = "data"
310 for comp in ("data", "output"):
311 compRef = refNone.makeComponentRef(comp)
312 output = datastore.get(compRef)
313 self.assertEqual(output, getattr(metricsNone, comp))
315 # Check that a put fails if the dataset type is not supported
316 if self.hasUnsupportedPut:
317 sc = StorageClass("UnsupportedSC", pytype=type(metrics))
318 ref = self.makeDatasetRef("unsupportedType", dimensions, sc, dataId)
319 with self.assertRaises(DatasetTypeNotSupportedError):
320 datastore.put(metrics, ref)
322 # These should raise
323 ref = self.makeDatasetRef("metrics", dimensions, storageClass, dataId)
324 with self.assertRaises(FileNotFoundError):
325 # non-existing file
326 datastore.get(ref)
328 # Get a URI from it
329 uri = datastore.getURI(ref, predict=True)
330 self.assertEqual(uri.scheme, self.uriScheme)
332 with self.assertRaises(FileNotFoundError):
333 datastore.getURI(ref)
335 # Calling put twice.
336 dataId = {
337 "instrument": "dummy",
338 "visit": 55,
339 "physical_filter": "V",
340 "band": "v",
341 "day_obs": 20250101,
342 }
343 ref = self.makeDatasetRef("metric", dimensions, storageClass, dataId)
344 datastore.put(metrics, ref)
345 datastore.get(ref)
346 # Second put must fail but dataset will still be there.
347 with self.assertRaises(DuplicateIdError):
348 datastore.put(metricsNone, ref)
349 self.assertEqual(datastore.get(ref), metrics)
351 def testTrustGetRequest(self) -> None:
352 """Check that we can get datasets that registry knows nothing about."""
353 datastore = self.makeDatastore()
355 # Skip test if the attribute is not defined
356 if not hasattr(datastore, "trustGetRequest"):
357 return
359 metrics = makeExampleMetrics()
361 i = 0
362 for sc_name in ("StructuredDataNoComponents", "StructuredData", "StructuredComposite"):
363 i += 1
364 datasetTypeName = f"test_metric{i}" # Different dataset type name each time.
366 if sc_name == "StructuredComposite":
367 disassembled = True
368 else:
369 disassembled = False
371 # Start datastore in default configuration of using registry
372 datastore.trustGetRequest = False
374 # Create multiple storage classes for testing with or without
375 # disassembly
376 sc = self.storageClassFactory.getStorageClass(sc_name)
377 dimensions = self.universe.conform(("visit", "physical_filter"))
379 dataId = {
380 "instrument": "dummy",
381 "visit": 52 + i,
382 "physical_filter": "V",
383 "band": "v",
384 "day_obs": 20250101,
385 }
387 ref = self.makeDatasetRef(datasetTypeName, dimensions, sc, dataId)
388 datastore.put(metrics, ref)
390 # Does it exist?
391 self.assertTrue(datastore.exists(ref))
392 self.assertTrue(datastore.knows(ref))
393 multi = datastore.knows_these([ref])
394 self.assertTrue(multi[ref])
395 multi = datastore.mexists([ref])
396 self.assertTrue(multi[ref])
398 # Get
399 metricsOut = datastore.get(ref)
400 self.assertEqual(metrics, metricsOut)
402 # Get the URI(s)
403 allURIs = datastore.getURIs(ref)
404 primaryURI, componentURIs = allURIs
405 if disassembled:
406 self.assertIsNone(primaryURI)
407 self.assertEqual(len(componentURIs), 3)
408 self.assertEqual(list(allURIs.iter_all()), list(componentURIs.values()))
409 else:
410 self.assertIn(datasetTypeName, primaryURI.path)
411 self.assertFalse(componentURIs)
412 self.assertEqual(list(allURIs.iter_all()), [primaryURI])
414 # Delete registry entry so now we are trusting
415 datastore.removeStoredItemInfo(ref)
417 # Now stop trusting and check that things break
418 datastore.trustGetRequest = False
420 # Does it exist?
421 self.assertFalse(datastore.exists(ref))
422 self.assertFalse(datastore.knows(ref))
423 multi = datastore.knows_these([ref])
424 self.assertFalse(multi[ref])
425 multi = datastore.mexists([ref])
426 self.assertFalse(multi[ref])
428 with self.assertRaises(FileNotFoundError):
429 datastore.get(ref)
431 if sc_name != "StructuredDataNoComponents":
432 with self.assertRaises(FileNotFoundError):
433 datastore.get(ref.makeComponentRef("data"))
435 # URI should fail unless we ask for prediction
436 with self.assertRaises(FileNotFoundError):
437 datastore.getURIs(ref)
439 predicted_primary, predicted_disassembled = datastore.getURIs(ref, predict=True)
440 if disassembled:
441 self.assertIsNone(predicted_primary)
442 self.assertEqual(len(predicted_disassembled), 3)
443 for uri in predicted_disassembled.values():
444 self.assertEqual(uri.fragment, "predicted")
445 self.assertIn(datasetTypeName, uri.path)
446 else:
447 self.assertIn(datasetTypeName, predicted_primary.path)
448 self.assertFalse(predicted_disassembled)
449 self.assertEqual(predicted_primary.fragment, "predicted")
451 # Now enable registry-free trusting mode
452 datastore.trustGetRequest = True
454 # Try again to get it
455 metricsOut = datastore.get(ref)
456 self.assertEqual(metricsOut, metrics)
458 # Does it exist?
459 self.assertTrue(datastore.exists(ref))
461 # Get a component
462 if sc_name != "StructuredDataNoComponents":
463 comp = "data"
464 compRef = ref.makeComponentRef(comp)
465 output = datastore.get(compRef)
466 self.assertEqual(output, getattr(metrics, comp))
468 # Get the URI -- if we trust this should work even without
469 # enabling prediction.
470 primaryURI2, componentURIs2 = datastore.getURIs(ref)
471 self.assertEqual(primaryURI2, primaryURI)
472 self.assertEqual(componentURIs2, componentURIs)
474 # Check for compatible storage class.
475 if sc_name in ("StructuredDataNoComponents", "StructuredData"):
476 # Make new dataset ref with compatible storage class.
477 ref_comp = ref.overrideStorageClass("StructuredDataDictJson")
479 # Without `set_retrieve_dataset_type_method` it will fail to
480 # find correct file.
481 self.assertFalse(datastore.exists(ref_comp))
482 with self.assertRaises(FileNotFoundError):
483 datastore.get(ref_comp)
484 with self.assertRaises(FileNotFoundError):
485 datastore.get(ref, storageClass="StructuredDataDictJson")
487 # Need a special method to generate stored dataset type.
488 def _stored_dataset_type(name: str, ref: DatasetRef = ref) -> DatasetType:
489 if name == ref.datasetType.name: 489 ↛ 491line 489 didn't jump to line 491 because the condition on line 489 was always true
490 return ref.datasetType
491 raise ValueError(f"Unexpected dataset type name {ref.datasetType.name}")
493 datastore.set_retrieve_dataset_type_method(_stored_dataset_type)
495 # Storage class override with original dataset ref.
496 metrics_as_dict = datastore.get(ref, storageClass="StructuredDataDictJson")
497 self.assertIsInstance(metrics_as_dict, dict)
499 # get() should return a dict now.
500 metrics_as_dict = datastore.get(ref_comp)
501 self.assertIsInstance(metrics_as_dict, dict)
503 # exists() should work as well.
504 self.assertTrue(datastore.exists(ref_comp))
506 datastore.set_retrieve_dataset_type_method(None)
508 def testDisassembly(self) -> None:
509 """Test disassembly within datastore."""
510 metrics = makeExampleMetrics()
511 if self.isEphemeral:
512 # in-memory datastore does not disassemble
513 return
515 # Create multiple storage classes for testing different formulations
516 # of composites. One of these will not disassemble to provide
517 # a reference.
518 storageClasses = [
519 self.storageClassFactory.getStorageClass(sc)
520 for sc in (
521 "StructuredComposite",
522 "StructuredCompositeTestA",
523 "StructuredCompositeTestB",
524 "StructuredCompositeReadComp",
525 "StructuredData", # No disassembly
526 "StructuredCompositeReadCompNoDisassembly",
527 )
528 ]
530 # Create the test datastore
531 datastore = self.makeDatastore()
533 # Dummy dataId
534 dimensions = self.universe.conform(("visit", "physical_filter"))
535 dataId = {"instrument": "dummy", "visit": 428, "physical_filter": "R"}
537 for i, sc in enumerate(storageClasses):
538 with self.subTest(storageClass=sc.name):
539 # Create a different dataset type each time round
540 # so that a test failure in this subtest does not trigger
541 # a cascade of tests because of file clashes
542 ref = self.makeDatasetRef(f"metric_comp_{i}", dimensions, sc, dataId)
544 disassembled = sc.name not in {"StructuredData", "StructuredCompositeReadCompNoDisassembly"}
546 datastore.put(metrics, ref)
548 baseURI, compURIs = datastore.getURIs(ref)
549 if disassembled:
550 self.assertIsNone(baseURI)
551 self.assertEqual(set(compURIs), {"data", "output", "summary"})
552 else:
553 self.assertIsNotNone(baseURI)
554 self.assertEqual(compURIs, {})
556 metrics_get = datastore.get(ref)
557 self.assertEqual(metrics_get, metrics)
559 # Retrieve the composite with read parameter
560 stop = 4
561 metrics_get = datastore.get(ref, parameters={"slice": slice(stop)})
562 self.assertEqual(metrics_get.summary, metrics.summary)
563 self.assertEqual(metrics_get.output, metrics.output)
564 self.assertEqual(metrics_get.data, metrics.data[:stop])
566 # Retrieve a component
567 data = datastore.get(ref.makeComponentRef("data"))
568 self.assertEqual(data, metrics.data)
570 # On supported storage classes attempt to access a read
571 # only component
572 if "ReadComp" in sc.name:
573 cRef = ref.makeComponentRef("counter")
574 counter = datastore.get(cRef)
575 self.assertEqual(counter, len(metrics.data))
577 counter = datastore.get(cRef, parameters={"slice": slice(stop)})
578 self.assertEqual(counter, stop)
580 datastore.remove(ref)
582 def prepDeleteTest(self, n_refs: int = 1) -> tuple[Datastore, tuple[DatasetRef, ...]]:
583 metrics = makeExampleMetrics()
584 datastore = self.makeDatastore()
585 # Put
586 dimensions = self.universe.conform(("visit", "physical_filter"))
587 sc = self.storageClassFactory.getStorageClass("StructuredData")
588 refs = []
589 for i in range(n_refs):
590 dataId = {
591 "instrument": "dummy",
592 "visit": 638 + i,
593 "physical_filter": "U",
594 "band": "u",
595 "day_obs": 20250101,
596 }
597 ref = self.makeDatasetRef("metric", dimensions, sc, dataId)
598 datastore.put(metrics, ref)
600 # Does it exist?
601 self.assertTrue(datastore.exists(ref))
603 # Get
604 metricsOut = datastore.get(ref)
605 self.assertEqual(metrics, metricsOut)
606 refs.append(ref)
608 return datastore, *refs
610 def testRemove(self) -> None:
611 datastore, ref = self.prepDeleteTest()
613 # Remove
614 datastore.remove(ref)
616 # Does it exist?
617 self.assertFalse(datastore.exists(ref))
619 # Do we now get a predicted URI?
620 uri = datastore.getURI(ref, predict=True)
621 self.assertEqual(uri.fragment, "predicted")
623 # Get should now fail
624 with self.assertRaises(FileNotFoundError):
625 datastore.get(ref)
626 # Can only delete once
627 with self.assertRaises(FileNotFoundError):
628 datastore.remove(ref)
630 def testForget(self) -> None:
631 datastore, ref = self.prepDeleteTest()
633 # Remove
634 datastore.forget([ref])
636 # Does it exist (as far as we know)?
637 self.assertFalse(datastore.exists(ref))
639 # Do we now get a predicted URI?
640 uri = datastore.getURI(ref, predict=True)
641 self.assertEqual(uri.fragment, "predicted")
643 # Get should now fail
644 with self.assertRaises(FileNotFoundError):
645 datastore.get(ref)
647 # Forgetting again is a silent no-op
648 datastore.forget([ref])
650 # Predicted URI should still point to the file.
651 self.assertTrue(uri.exists())
653 def testTransfer(self) -> None:
654 metrics = makeExampleMetrics()
656 dimensions = self.universe.conform(("visit", "physical_filter"))
657 dataId = {
658 "instrument": "dummy",
659 "visit": 2048,
660 "physical_filter": "Uprime",
661 "band": "u",
662 "day_obs": 20250101,
663 }
665 sc = self.storageClassFactory.getStorageClass("StructuredData")
666 ref = self.makeDatasetRef("metric", dimensions, sc, dataId)
668 inputDatastore = self.makeDatastore("test_input_datastore")
669 outputDatastore = self.makeDatastore("test_output_datastore")
671 inputDatastore.put(metrics, ref)
672 outputDatastore.transfer(inputDatastore, ref)
674 metricsOut = outputDatastore.get(ref)
675 self.assertEqual(metrics, metricsOut)
677 def testBasicTransaction(self) -> None:
678 datastore = self.makeDatastore()
679 storageClass = self.storageClassFactory.getStorageClass("StructuredData")
680 dimensions = self.universe.conform(("visit", "physical_filter"))
681 nDatasets = 6
682 dataIds = [
683 {"instrument": "dummy", "visit": i, "physical_filter": "V", "band": "v", "day_obs": 20250101}
684 for i in range(nDatasets)
685 ]
686 data = [
687 (
688 self.makeDatasetRef("metric", dimensions, storageClass, dataId),
689 makeExampleMetrics(),
690 )
691 for dataId in dataIds
692 ]
693 succeed = data[: nDatasets // 2]
694 fail = data[nDatasets // 2 :]
695 # All datasets added in this transaction should continue to exist
696 with datastore.transaction():
697 for ref, metrics in succeed:
698 datastore.put(metrics, ref)
699 # Whereas datasets added in this transaction should not
700 with self.assertRaises(TransactionTestError):
701 with datastore.transaction():
702 for ref, metrics in fail:
703 datastore.put(metrics, ref)
704 raise TransactionTestError("This should propagate out of the context manager")
705 # Check for datasets that should exist
706 for ref, metrics in succeed:
707 # Does it exist?
708 self.assertTrue(datastore.exists(ref))
709 # Get
710 metricsOut = datastore.get(ref, parameters=None)
711 self.assertEqual(metrics, metricsOut)
712 # URI
713 uri = datastore.getURI(ref)
714 self.assertEqual(uri.scheme, self.uriScheme)
715 # Check for datasets that should not exist
716 for ref, _ in fail:
717 # These should raise
718 with self.assertRaises(FileNotFoundError):
719 # non-existing file
720 datastore.get(ref)
721 with self.assertRaises(FileNotFoundError):
722 datastore.getURI(ref)
724 def testNestedTransaction(self) -> None:
725 datastore = self.makeDatastore()
726 storageClass = self.storageClassFactory.getStorageClass("StructuredData")
727 dimensions = self.universe.conform(("visit", "physical_filter"))
728 metrics = makeExampleMetrics()
730 dataId = {"instrument": "dummy", "visit": 0, "physical_filter": "V", "band": "v", "day_obs": 20250101}
731 refBefore = self.makeDatasetRef("metric", dimensions, storageClass, dataId)
732 datastore.put(metrics, refBefore)
733 with self.assertRaises(TransactionTestError):
734 with datastore.transaction():
735 dataId = {
736 "instrument": "dummy",
737 "visit": 1,
738 "physical_filter": "V",
739 "band": "v",
740 "day_obs": 20250101,
741 }
742 refOuter = self.makeDatasetRef("metric", dimensions, storageClass, dataId)
743 datastore.put(metrics, refOuter)
744 with datastore.transaction():
745 dataId = {
746 "instrument": "dummy",
747 "visit": 2,
748 "physical_filter": "V",
749 "band": "v",
750 "day_obs": 20250101,
751 }
752 refInner = self.makeDatasetRef("metric", dimensions, storageClass, dataId)
753 datastore.put(metrics, refInner)
754 # All datasets should exist
755 for ref in (refBefore, refOuter, refInner):
756 metricsOut = datastore.get(ref, parameters=None)
757 self.assertEqual(metrics, metricsOut)
758 raise TransactionTestError("This should roll back the transaction")
759 # Dataset(s) inserted before the transaction should still exist
760 metricsOut = datastore.get(refBefore, parameters=None)
761 self.assertEqual(metrics, metricsOut)
762 # But all datasets inserted during the (rolled back) transaction
763 # should be gone
764 with self.assertRaises(FileNotFoundError):
765 datastore.get(refOuter)
766 with self.assertRaises(FileNotFoundError):
767 datastore.get(refInner)
769 def _prepareIngestTest(self) -> tuple[MetricsExample, DatasetRef]:
770 storageClass = self.storageClassFactory.getStorageClass("StructuredData")
771 dimensions = self.universe.conform(("visit", "physical_filter"))
772 metrics = makeExampleMetrics()
773 dataId = {"instrument": "dummy", "visit": 0, "physical_filter": "V", "band": "v", "day_obs": 20250101}
774 ref = self.makeDatasetRef("metric", dimensions, storageClass, dataId)
775 return metrics, ref
777 def runIngestTest(self, func: Callable[[MetricsExample, str, DatasetRef], None]) -> None:
778 metrics, ref = self._prepareIngestTest()
779 # The file will be deleted after the test.
780 # For symlink tests this leads to a situation where the datastore
781 # points to a file that does not exist. This will make os.path.exist
782 # return False but then the new symlink will fail with
783 # FileExistsError later in the code so the test still passes.
784 with _temp_yaml_file(metrics._asdict()) as path:
785 func(metrics, path, ref)
787 def testIngestNoTransfer(self) -> None:
788 """Test ingesting existing files with no transfer."""
789 for mode in (None, "auto"):
790 # Some datastores have auto but can't do in place transfer
791 if mode == "auto" and "auto" in self.ingestTransferModes and not self.canIngestNoTransferAuto:
792 continue
794 with self.subTest(mode=mode):
795 datastore = self.makeDatastore()
797 def succeed(
798 obj: MetricsExample,
799 path: str,
800 ref: DatasetRef,
801 mode: str | None = mode,
802 datastore: Datastore = datastore,
803 ) -> None:
804 """Ingest a file already in the datastore root."""
805 # first move it into the root, and adjust the path
806 # accordingly.
807 # In the case of a ChainedDatastore, we have multiple
808 # roots, all of which will accept the file, so we
809 # have to copy it into all the roots.
810 relative_path = None
811 for root in datastore.roots.values():
812 if root is not None:
813 copied_path = shutil.copy(path, root.ospath)
814 relative_path = os.path.relpath(copied_path, start=root.ospath)
815 assert relative_path is not None, (
816 "Running a FileDatastore test on a Datastore instance without any roots"
817 )
818 datastore.ingest(FileDataset(path=relative_path, refs=ref), transfer=mode)
819 self.assertEqual(obj, datastore.get(ref))
821 def failInputDoesNotExist(
822 obj: MetricsExample,
823 path: str,
824 ref: DatasetRef,
825 mode: str | None = mode,
826 datastore: Datastore = datastore,
827 ) -> None:
828 """Can't ingest files if we're given a bad path."""
829 with self.assertRaises(FileNotFoundError):
830 datastore.ingest(
831 FileDataset(path="this-file-does-not-exist.yaml", refs=ref), transfer=mode
832 )
833 self.assertFalse(datastore.exists(ref))
835 def failOutsideRoot(
836 obj: MetricsExample,
837 path: str,
838 ref: DatasetRef,
839 mode: str | None = mode,
840 datastore: Datastore = datastore,
841 ) -> None:
842 """Can't ingest files outside of datastore root unless
843 auto.
844 """
845 if mode == "auto":
846 datastore.ingest(FileDataset(path=os.path.abspath(path), refs=ref), transfer=mode)
847 self.assertTrue(datastore.exists(ref))
848 else:
849 with self.assertRaises(RuntimeError):
850 datastore.ingest(FileDataset(path=os.path.abspath(path), refs=ref), transfer=mode)
851 self.assertFalse(datastore.exists(ref))
853 def failNotImplemented(
854 obj: MetricsExample,
855 path: str,
856 ref: DatasetRef,
857 mode: str | None = mode,
858 datastore: Datastore = datastore,
859 ) -> None:
860 with self.assertRaises(NotImplementedError):
861 datastore.ingest(FileDataset(path=path, refs=ref), transfer=mode)
863 if mode in self.ingestTransferModes:
864 self.runIngestTest(failOutsideRoot)
865 self.runIngestTest(failInputDoesNotExist)
866 self.runIngestTest(succeed)
867 else:
868 self.runIngestTest(failNotImplemented)
870 def testIngestTransfer(self) -> None:
871 """Test ingesting existing files after transferring them."""
872 for mode in ("copy", "move", "link", "hardlink", "symlink", "relsymlink", "auto"):
873 with self.subTest(mode=mode):
874 datastore = self.makeDatastore(mode)
876 def succeed(
877 obj: MetricsExample,
878 path: str,
879 ref: DatasetRef,
880 mode: str | None = mode,
881 datastore: Datastore = datastore,
882 ) -> None:
883 """Ingest a file by transferring it to the template
884 location.
885 """
886 datastore.ingest(FileDataset(path=os.path.abspath(path), refs=ref), transfer=mode)
887 self.assertEqual(obj, datastore.get(ref))
888 file_exists = os.path.exists(path)
889 if mode == "move":
890 self.assertFalse(file_exists)
891 else:
892 self.assertTrue(file_exists)
894 def failInputDoesNotExist(
895 obj: MetricsExample,
896 path: str,
897 ref: DatasetRef,
898 mode: str | None = mode,
899 datastore: Datastore = datastore,
900 ) -> None:
901 """Can't ingest files if we're given a bad path."""
902 with self.assertRaises(FileNotFoundError):
903 # Ensure the file does not look like it is in
904 # datastore for auto mode
905 datastore.ingest(
906 FileDataset(path="../this-file-does-not-exist.yaml", refs=ref), transfer=mode
907 )
908 self.assertFalse(datastore.exists(ref), f"Checking not in datastore using mode {mode}")
910 def failNotImplemented(
911 obj: MetricsExample,
912 path: str,
913 ref: DatasetRef,
914 mode: str | None = mode,
915 datastore: Datastore = datastore,
916 ) -> None:
917 with self.assertRaises(NotImplementedError):
918 datastore.ingest(FileDataset(path=os.path.abspath(path), refs=ref), transfer=mode)
920 if mode in self.ingestTransferModes:
921 self.runIngestTest(failInputDoesNotExist)
922 self.runIngestTest(succeed)
923 else:
924 self.runIngestTest(failNotImplemented)
926 def testIngestSymlinkOfSymlink(self) -> None:
927 """Special test for symlink to a symlink ingest"""
928 metrics, ref = self._prepareIngestTest()
929 # The aim of this test is to create a dataset on disk, then
930 # create a symlink to it and finally ingest the symlink such that
931 # the symlink in the datastore points to the original dataset.
932 for mode in ("symlink", "relsymlink"):
933 if mode not in self.ingestTransferModes:
934 continue
936 print(f"Trying mode {mode}")
937 with _temp_yaml_file(metrics._asdict()) as realpath:
938 with tempfile.TemporaryDirectory() as tmpdir:
939 sympath = os.path.join(tmpdir, "symlink.yaml")
940 os.symlink(os.path.realpath(realpath), sympath)
942 datastore = self.makeDatastore()
943 datastore.ingest(FileDataset(path=os.path.abspath(sympath), refs=ref), transfer=mode)
945 uri = datastore.getURI(ref)
946 self.assertTrue(uri.isLocal, f"Check {uri.scheme}")
947 self.assertTrue(os.path.islink(uri.ospath), f"Check {uri} is a symlink")
949 linkTarget = os.readlink(uri.ospath)
950 if mode == "relsymlink":
951 self.assertFalse(os.path.isabs(linkTarget))
952 else:
953 self.assertTrue(os.path.samefile(linkTarget, realpath))
955 # Check that we can get the dataset back regardless of mode
956 metric2 = datastore.get(ref)
957 self.assertEqual(metric2, metrics)
959 # Cleanup the file for next time round loop
960 # since it will get the same file name in store
961 datastore.remove(ref)
963 def _populate_export_datastore(self, name: str) -> tuple[Datastore, list[DatasetRef]]:
964 datastore = self.makeDatastore(name)
966 # For now only the FileDatastore can be used for this test.
967 # ChainedDatastore that only includes InMemoryDatastores have to be
968 # skipped as well.
969 for name in datastore.names:
970 if not name.startswith("InMemoryDatastore"):
971 break
972 else:
973 raise unittest.SkipTest("in-memory datastore does not support record export/import")
975 metrics = makeExampleMetrics()
976 dimensions = self.universe.conform(("visit", "physical_filter"))
977 sc = self.storageClassFactory.getStorageClass("StructuredData")
979 refs = []
980 for visit in (2048, 2049, 2050):
981 dataId = {
982 "instrument": "dummy",
983 "visit": visit,
984 "physical_filter": "Uprime",
985 "band": "u",
986 "day_obs": 20250101,
987 }
988 ref = self.makeDatasetRef("metric", dimensions, sc, dataId)
989 datastore.put(metrics, ref)
990 refs.append(ref)
991 return datastore, refs
993 def testExportImportRecords(self) -> None:
994 """Test for export_records and import_records methods."""
995 datastore, refs = self._populate_export_datastore("test_datastore")
996 for exported_refs in (refs, refs[1:]):
997 n_refs = len(exported_refs)
998 records = datastore.export_records(exported_refs)
999 self.assertGreater(len(records), 0)
1000 self.assertTrue(set(records.keys()) <= set(datastore.names))
1001 # In a ChainedDatastore each FileDatastore will have a complete set
1002 for datastore_name in records:
1003 record_data = records[datastore_name]
1004 self.assertEqual(len(record_data.records), n_refs)
1006 # Check that subsetting works, include non-existing dataset ID.
1007 dataset_ids = {exported_refs[0].id, uuid.uuid4()}
1008 subset = record_data.subset(dataset_ids)
1009 assert subset is not None
1010 self.assertEqual(len(subset.records), 1)
1011 subset = record_data.subset({uuid.uuid4()})
1012 self.assertIsNone(subset)
1014 # Use the same datastore name to import relative path.
1015 datastore2 = self.makeDatastore("test_datastore")
1017 records = datastore.export_records(refs[1:])
1018 datastore2.import_records(records)
1020 with self.assertRaises(FileNotFoundError):
1021 data = datastore2.get(refs[0])
1022 data = datastore2.get(refs[1])
1023 self.assertIsNotNone(data)
1024 data = datastore2.get(refs[2])
1025 self.assertIsNotNone(data)
1027 def testExportImportTable(self) -> None:
1028 datastore, refs = self._populate_export_datastore("test_datastore")
1029 table = datastore.export_table([ref.id for ref in refs])
1030 datastore2 = self.makeDatastore("test_datastore")
1031 datastore2.import_table(table)
1033 for ref in refs:
1034 self.assertIsNotNone(datastore2.get(ref))
1035 self.assertEqual(datastore.getURI(ref), datastore2.getURI(ref))
1036 original_info = datastore.get_file_info_for_transfer([ref.id])[ref.id][0]
1037 imported_info_list = datastore.get_file_info_for_transfer([ref.id]).get(ref.id)
1038 self.assertIsNotNone(imported_info_list)
1039 self.assertEqual(len(imported_info_list), 1)
1040 imported_info = imported_info_list[0]
1041 self.assertEqual(imported_info.file_info.formatter, original_info.file_info.formatter)
1042 self.assertEqual(
1043 imported_info.file_info.storage_class_name, original_info.file_info.storage_class_name
1044 )
1045 self.assertEqual(imported_info.file_info.file_size, original_info.file_info.file_size)
1046 self.assertEqual(imported_info.file_info.checksum, original_info.file_info.checksum)
1047 self.assertEqual(imported_info.file_info.component, original_info.file_info.component)
1049 def testExportPredictedRecords(self):
1050 if self.isEphemeral:
1051 raise unittest.SkipTest("in-memory datastore does not support record export/import")
1052 sc = self.storageClassFactory.getStorageClass("ThingOne")
1053 dimensions = self.universe.conform(("visit", "physical_filter"))
1054 dataId = {
1055 "instrument": "dummy",
1056 "visit": 52,
1057 "physical_filter": "V",
1058 "band": "v",
1059 "day_obs": 20250101,
1060 }
1061 ref = self.makeDatasetRef("metric", dimensions, sc, dataId)
1063 datastore = self.makeDatastore("test_datastore")
1064 names = {n for n in datastore.names if not n.startswith("InMemory")}
1065 records = datastore.export_predicted_records([ref])
1067 # Expect predicted records from all datastores.
1068 self.assertEqual(set(records.keys()), names)
1070 for record_data in records.values():
1071 self.assertEqual(len(record_data.records), 1)
1073 def testExport(self) -> None:
1074 datastore, refs = self._populate_export_datastore("test_datastore")
1076 datasets = list(datastore.export(refs))
1077 self.assertEqual(len(datasets), 3)
1079 for transfer in (None, "auto"):
1080 # Both will default to None
1081 datasets = list(datastore.export(refs, transfer=transfer))
1082 self.assertEqual(len(datasets), 3)
1084 with self.assertRaises(TypeError):
1085 list(datastore.export(refs, transfer="copy"))
1087 with self.assertRaises(TypeError):
1088 list(datastore.export(refs, directory="exportDir", transfer="move"))
1090 # Create a new ref that is not known to the datastore and try to
1091 # export it.
1092 sc = self.storageClassFactory.getStorageClass("ThingOne")
1093 dimensions = self.universe.conform(("visit", "physical_filter"))
1094 dataId = {
1095 "instrument": "dummy",
1096 "visit": 52,
1097 "physical_filter": "V",
1098 "band": "v",
1099 "day_obs": 20250101,
1100 }
1101 ref = self.makeDatasetRef("metric", dimensions, sc, dataId)
1102 with self.assertRaises(FileNotFoundError):
1103 list(datastore.export(refs + [ref], transfer=None))
1105 def test_pydantic_dict_storage_class_conversions(self) -> None:
1106 """Test converting a dataset stored as a pydantic model into a dict on
1107 read.
1108 """
1109 datastore = self.makeDatastore()
1110 store_as_model = self.makeDatasetRef(
1111 "store_as_model",
1112 dimensions=self.universe.empty,
1113 storageClass="DictConvertibleModel",
1114 dataId=DataCoordinate.make_empty(self.universe),
1115 )
1116 content = {"a": "one", "b": "two"}
1117 model = DictConvertibleModel.from_dict(content, extra="original content")
1118 datastore.put(model, store_as_model)
1119 retrieved_model = datastore.get(store_as_model)
1120 self.assertEqual(retrieved_model, model)
1121 loaded = datastore.get(store_as_model.overrideStorageClass("NativeDictForConvertibleModel"))
1122 self.assertEqual(type(loaded), dict)
1123 self.assertEqual(loaded, content)
1125 def test_simple_class_put_get(self) -> None:
1126 """Test that we can put and get a simple class with dict()
1127 constructor.
1128 """
1129 datastore = self.makeDatastore()
1130 data = MetricsExample(summary={"a": 1}, data=[1, 2, 3], output={"b": 2})
1131 self._assert_different_puts(datastore, "MetricsExample", data)
1133 def test_dataclass_put_get(self) -> None:
1134 """Test that we can put and get a simple dataclass."""
1135 datastore = self.makeDatastore()
1136 data = MetricsExampleDataclass(summary={"a": 1}, data=[1, 2, 3], output={"b": 2})
1137 self._assert_different_puts(datastore, "MetricsExampleDataclass", data)
1139 def test_pydantic_put_get(self) -> None:
1140 """Test that we can put and get a simple Pydantic model."""
1141 datastore = self.makeDatastore()
1142 data = MetricsExampleModel(summary={"a": 1}, data=[1, 2, 3], output={"b": 2})
1143 self._assert_different_puts(datastore, "MetricsExampleModel", data)
1145 def test_tuple_put_get(self) -> None:
1146 """Test that we can put and get a tuple."""
1147 datastore = self.makeDatastore()
1148 data = ("a", "b", 1)
1149 self._assert_different_puts(datastore, "TupleExample", data)
1151 def _assert_different_puts(self, datastore: Datastore, storageClass_root: str, data: Any) -> None:
1152 refs = {
1153 x: self.makeDatasetRef(
1154 f"stora_as_{x}",
1155 dimensions=self.universe.empty,
1156 storageClass=f"{storageClass_root}{x}",
1157 dataId=DataCoordinate.make_empty(self.universe),
1158 )
1159 for x in ["A", "B"]
1160 }
1162 for ref in refs.values():
1163 datastore.put(data, ref)
1165 self.assertEqual(datastore.get(refs["A"]), datastore.get(refs["B"]))
1168class PosixDatastoreTestCase(DatastoreTests, unittest.TestCase):
1169 """PosixDatastore specialization"""
1171 configFile = os.path.join(TESTDIR, "config/basic/butler.yaml")
1172 uriScheme = "file"
1173 canIngestNoTransferAuto = True
1174 ingestTransferModes = (None, "copy", "move", "link", "hardlink", "symlink", "relsymlink", "auto")
1175 isEphemeral = False
1176 rootKeys = ("root",)
1177 validationCanFail = True
1179 def setUp(self) -> None:
1180 # The call to os.path.realpath is necessary because Mac temporary files
1181 # can end up in either /private/var/folders or /var/folders, which
1182 # refer to the same location but don't appear to.
1183 # This matters for "relsymlink" transfer mode, because it needs to be
1184 # able to read the file through a relative symlink, but some of the
1185 # intermediate directories are not traversable if you try to get from a
1186 # tempfile in /var/folders to one in /private/var/folders via a
1187 # relative path.
1188 self.root = os.path.realpath(self.enterContext(tempfile.TemporaryDirectory()))
1189 super().setUp()
1191 def testAtomicWrite(self) -> None:
1192 """Test that we write to a temporary and then rename"""
1193 datastore = self.makeDatastore()
1194 storageClass = self.storageClassFactory.getStorageClass("StructuredData")
1195 dimensions = self.universe.conform(("visit", "physical_filter"))
1196 metrics = makeExampleMetrics()
1198 dataId = {"instrument": "dummy", "visit": 0, "physical_filter": "V", "band": "v", "day_obs": 20250101}
1199 ref = self.makeDatasetRef("metric", dimensions, storageClass, dataId)
1201 with self.assertLogs("lsst.resources", "DEBUG") as cm:
1202 datastore.put(metrics, ref)
1203 move_logs = [ll for ll in cm.output if "transfer=" in ll]
1204 self.assertIn("transfer=move", move_logs[0])
1206 # And the transfer should be file to file.
1207 self.assertEqual(move_logs[0].count("file://"), 2)
1209 def testCanNotDeterminePutFormatterLocation(self) -> None:
1210 """Verify that the expected exception is raised if the FileDatastore
1211 can not determine the put formatter location.
1212 """
1213 _ = makeExampleMetrics()
1214 datastore = self.makeDatastore()
1216 # Create multiple storage classes for testing different formulations
1217 storageClass = self.storageClassFactory.getStorageClass("StructuredData")
1219 sccomp = StorageClass("Dummy")
1220 compositeStorageClass = StorageClass(
1221 "StructuredComposite", components={"dummy": sccomp, "dummy2": sccomp}
1222 )
1224 dimensions = self.universe.conform(("visit", "physical_filter"))
1225 dataId = {
1226 "instrument": "dummy",
1227 "visit": 52,
1228 "physical_filter": "V",
1229 "band": "v",
1230 "day_obs": 20250101,
1231 }
1233 ref = self.makeDatasetRef("metric", dimensions, storageClass, dataId)
1234 compRef = self.makeDatasetRef("metric", dimensions, compositeStorageClass, dataId)
1236 def raiser(ref: DatasetRef) -> None:
1237 raise DatasetTypeNotSupportedError()
1239 with unittest.mock.patch.object(
1240 lsst.daf.butler.datastores.fileDatastore.FileDatastore,
1241 "_determine_put_formatter_location",
1242 side_effect=raiser,
1243 ):
1244 # verify the non-composite ref execution path:
1245 with self.assertRaises(DatasetTypeNotSupportedError):
1246 datastore.getURIs(ref, predict=True)
1248 # verify the composite-ref execution path:
1249 with self.assertRaises(DatasetTypeNotSupportedError):
1250 datastore.getURIs(compRef, predict=True)
1252 def test_roots(self):
1253 datastore = self.makeDatastore()
1255 self.assertEqual(set(datastore.names), set(datastore.roots.keys()))
1256 for root in datastore.roots.values():
1257 if root is not None:
1258 self.assertTrue(root.exists())
1260 def test_prepare_get_for_external_client(self):
1261 datastore = self.makeDatastore()
1262 storageClass = self.storageClassFactory.getStorageClass("StructuredData")
1263 dimensions = self.universe.conform(("visit", "physical_filter"))
1264 dataId = {"instrument": "dummy", "visit": 52, "physical_filter": "V", "band": "v"}
1265 ref = self.makeDatasetRef("metric", dimensions, storageClass, dataId)
1266 # Most of the coverage for this function is in test_server.py,
1267 # because it requires a file backend that supports URL signing.
1268 self.assertIsNone(datastore.prepare_get_for_external_client(ref))
1271class PosixDatastoreNoChecksumsTestCase(PosixDatastoreTestCase):
1272 """Posix datastore tests but with checksums disabled."""
1274 configFile = os.path.join(TESTDIR, "config/basic/posixDatastoreNoChecksums.yaml")
1276 def testChecksum(self) -> None:
1277 """Ensure that checksums have not been calculated."""
1278 datastore = self.makeDatastore()
1279 storageClass = self.storageClassFactory.getStorageClass("StructuredData")
1280 dimensions = self.universe.conform(("visit", "physical_filter"))
1281 metrics = makeExampleMetrics()
1283 dataId = {"instrument": "dummy", "visit": 0, "physical_filter": "V", "band": "v", "day_obs": 20250101}
1284 ref = self.makeDatasetRef("metric", dimensions, storageClass, dataId)
1286 # Configuration should have disabled checksum calculation
1287 datastore.put(metrics, ref)
1288 infos = datastore.getStoredItemsInfo(ref)
1289 self.assertIsNone(infos[0].checksum)
1291 # Remove put back but with checksums enabled explicitly
1292 datastore.remove(ref)
1293 datastore.useChecksum = True
1294 datastore.put(metrics, ref)
1296 infos = datastore.getStoredItemsInfo(ref)
1297 self.assertIsNotNone(infos[0].checksum)
1299 def test_repeat_ingest(self):
1300 """Test that repeatedly ingesting the same file in direct mode
1301 is allowed.
1303 Test can only run with FileDatastore since that is the only one
1304 supporting "direct" ingest.
1305 """
1306 metrics, v4ref = self._prepareIngestTest()
1307 datastore = self.makeDatastore()
1308 v5ref = DatasetRef(
1309 v4ref.datasetType, v4ref.dataId, v4ref.run, id_generation_mode=DatasetIdGenEnum.DATAID_TYPE_RUN
1310 )
1312 with _temp_yaml_file(metrics._asdict()) as path:
1313 datastore.ingest(FileDataset(path=path, refs=v4ref), transfer="direct")
1315 # This will fail because the ref is using UUIDv4.
1316 with self.assertRaises(DuplicateIdError):
1317 datastore.ingest(FileDataset(path=path, refs=v4ref), transfer="direct")
1319 # UUIDv5 can be repeatedly ingested in direct mode.
1320 datastore.ingest(FileDataset(path=path, refs=v5ref), transfer="direct")
1321 datastore.ingest(FileDataset(path=path, refs=v5ref), transfer="direct")
1323 with self.assertRaises(DuplicateIdError):
1324 datastore.ingest(FileDataset(path=path, refs=v5ref), transfer="copy")
1326 def test_repeat_ingest_zip(self):
1327 """Check repeated calls to ingest_zip."""
1328 metrics, ref = self._prepareIngestTest()
1329 datastore = self.makeDatastore()
1331 with _temp_yaml_file(metrics._asdict()) as path:
1332 datastore.ingest(FileDataset(path=path, refs=ref), transfer="direct")
1333 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
1334 # Dump to zip file.
1335 zip_path = retrieve_and_zip([ref], tmpdir, datastore.retrieveArtifacts)
1337 with self.assertRaises(DuplicateIdError):
1338 datastore.ingest_zip(zip_path, "copy")
1341class TrashDatastoreTestCase(PosixDatastoreTestCase):
1342 """Restrict trash test to FileDatastore."""
1344 configFile = os.path.join(TESTDIR, "config/basic/butler.yaml")
1346 def testTrash(self) -> None:
1347 datastore, *refs = self.prepDeleteTest(n_refs=10)
1349 # Trash one of them.
1350 ref = refs.pop()
1351 uri = datastore.getURI(ref)
1352 datastore.trash(ref)
1353 self.assertTrue(uri.exists(), uri) # Not deleted yet
1354 datastore.emptyTrash()
1355 self.assertFalse(uri.exists(), uri)
1357 # Trash it again should be fine.
1358 datastore.trash(ref)
1360 # Trash multiple items at once.
1361 subset = [refs.pop(), refs.pop()]
1362 datastore.trash(subset)
1363 datastore.emptyTrash()
1365 # Remove a record and trash should do nothing.
1366 # This is execution butler scenario.
1367 ref = refs.pop()
1368 uri = datastore.getURI(ref)
1369 datastore._table.delete(["dataset_id"], {"dataset_id": ref.id})
1370 self.assertTrue(uri.exists())
1371 datastore.trash(ref)
1372 datastore.emptyTrash()
1373 self.assertTrue(uri.exists())
1375 # Switch on trust and it should delete the file.
1376 datastore.trustGetRequest = True
1377 datastore.trash([ref])
1378 self.assertFalse(uri.exists())
1380 # Remove multiples at once in trust mode.
1381 subset = [refs.pop() for i in range(3)]
1382 datastore.trash(subset)
1383 datastore.trash(refs.pop()) # Check that a single ref can trash
1385 def test_empty_trash(self) -> None:
1386 """Test parameters and return value for empty trash."""
1387 datastore, *refs = self.prepDeleteTest(n_refs=10)
1389 # Trash one of them.
1390 ref = refs.pop()
1391 uri = datastore.getURI(ref)
1392 datastore.trash(ref)
1393 self.assertTrue(uri.exists(), uri) # Not deleted yet
1395 # Empty trash but with a list of refs that does not include the
1396 # one in the trash table.
1397 removed = datastore.emptyTrash(refs=refs)
1398 self.assertEqual(len(removed), 0)
1399 self.assertTrue(uri.exists(), uri)
1401 # Empty the entire trash but in dry_run mode.
1402 removed = datastore.emptyTrash(dry_run=True)
1403 self.assertEqual(len(removed), 1)
1404 self.assertEqual(removed.pop(), uri)
1405 self.assertTrue(uri.exists(), uri)
1407 # Empty the trash specifying the actual ref.
1408 removed = datastore.emptyTrash(refs=[ref])
1409 self.assertEqual(len(removed), 1)
1410 self.assertEqual(removed.pop(), uri)
1411 self.assertFalse(uri.exists(), uri)
1413 # Trash everything and empty.
1414 datastore.trash(refs)
1415 removed = datastore.emptyTrash(dry_run=True)
1416 for u in removed:
1417 self.assertTrue(u.exists())
1418 removed = datastore.emptyTrash()
1419 for u in removed:
1420 self.assertFalse(u.exists())
1423class CleanupPosixDatastoreTestCase(DatastoreTestsBase, unittest.TestCase):
1424 """Test datastore cleans up on failure."""
1426 configFile = os.path.join(TESTDIR, "config/basic/butler.yaml")
1428 def setUp(self) -> None:
1429 # Override the working directory before calling the base class
1430 self.root = tempfile.mkdtemp()
1431 super().setUp()
1433 def testCleanup(self) -> None:
1434 """Test that a failed formatter write does cleanup a partial file."""
1435 metrics = makeExampleMetrics()
1436 datastore = self.makeDatastore()
1438 storageClass = self.storageClassFactory.getStorageClass("StructuredData")
1440 dimensions = self.universe.conform(("visit", "physical_filter"))
1441 dataId = {
1442 "instrument": "dummy",
1443 "visit": 52,
1444 "physical_filter": "V",
1445 "band": "v",
1446 "day_obs": 20250101,
1447 }
1449 ref = self.makeDatasetRef("metric", dimensions, storageClass, dataId)
1451 # Determine where the file will end up (we assume Formatters use
1452 # the same file extension)
1453 expectedUri = datastore.getURI(ref, predict=True)
1454 self.assertEqual(expectedUri.fragment, "predicted")
1456 self.assertEqual(expectedUri.getExtension(), ".yaml", f"Is there a file extension in {expectedUri}")
1458 # Try formatter that fails and formatter that fails and leaves
1459 # a file behind
1460 for formatter in (BadWriteFormatter, BadNoWriteFormatter):
1461 with self.subTest(formatter=str(formatter)):
1462 # Monkey patch the formatter
1463 datastore.formatterFactory.registerFormatter(ref.datasetType, formatter, overwrite=True)
1465 # Try to put the dataset, it should fail
1466 with self.assertRaises(RuntimeError):
1467 datastore.put(metrics, ref)
1469 # Check that there is no file on disk
1470 self.assertFalse(expectedUri.exists(), f"Check for existence of {expectedUri}")
1472 # Check that there is a directory
1473 dir = expectedUri.dirname()
1474 self.assertTrue(dir.exists(), f"Check for existence of directory {dir}")
1476 # Force YamlFormatter and check that this time a file is written
1477 datastore.formatterFactory.registerFormatter(ref.datasetType, YamlFormatter, overwrite=True)
1478 datastore.put(metrics, ref)
1479 self.assertTrue(expectedUri.exists(), f"Check for existence of {expectedUri}")
1480 datastore.remove(ref)
1481 self.assertFalse(expectedUri.exists(), f"Check for existence of now removed {expectedUri}")
1484class InMemoryDatastoreTestCase(DatastoreTests, unittest.TestCase):
1485 """PosixDatastore specialization"""
1487 configFile = os.path.join(TESTDIR, "config/basic/inMemoryDatastore.yaml")
1488 uriScheme = "mem"
1489 hasUnsupportedPut = False
1490 ingestTransferModes = ()
1491 isEphemeral = True
1492 rootKeys = None
1493 validationCanFail = False
1496class ChainedDatastoreTestCase(PosixDatastoreTestCase):
1497 """ChainedDatastore specialization using a POSIXDatastore"""
1499 configFile = os.path.join(TESTDIR, "config/basic/chainedDatastore.yaml")
1500 hasUnsupportedPut = False
1501 canIngestNoTransferAuto = False
1502 ingestTransferModes = (None, "copy", "move", "hardlink", "symlink", "relsymlink", "link", "auto")
1503 isEphemeral = False
1504 rootKeys = (".datastores.1.root", ".datastores.2.root")
1505 validationCanFail = True
1508class ChainedDatastoreMemoryTestCase(InMemoryDatastoreTestCase):
1509 """ChainedDatastore specialization using all InMemoryDatastore"""
1511 configFile = os.path.join(TESTDIR, "config/basic/chainedDatastore2.yaml")
1512 validationCanFail = False
1513 isEphemeral = True
1516class DatastoreConstraintsTests(DatastoreTestsBase):
1517 """Basic tests of constraints model of Datastores."""
1519 def testConstraints(self) -> None:
1520 """Test constraints model. Assumes that each test class has the
1521 same constraints.
1522 """
1523 metrics = makeExampleMetrics()
1524 datastore = self.makeDatastore()
1526 sc1 = self.storageClassFactory.getStorageClass("StructuredData")
1527 sc2 = self.storageClassFactory.getStorageClass("StructuredDataJson")
1528 dimensions = self.universe.conform(("visit", "physical_filter", "instrument"))
1529 dataId = {
1530 "visit": 52,
1531 "physical_filter": "V",
1532 "band": "v",
1533 "instrument": "DummyCamComp",
1534 "day_obs": 20250101,
1535 }
1537 # Write empty file suitable for ingest check (JSON and YAML variants)
1538 testfile_y = tempfile.NamedTemporaryFile(suffix=".yaml")
1539 testfile_j = tempfile.NamedTemporaryFile(suffix=".json")
1540 for datasetTypeName, sc, accepted in (
1541 ("metric", sc1, True),
1542 ("metric5", sc1, False),
1543 ("metric33", sc1, True),
1544 ("metric5", sc2, True),
1545 ):
1546 # Choose different temp file depending on StorageClass
1547 testfile = testfile_j if sc.name.endswith("Json") else testfile_y
1549 with self.subTest(datasetTypeName=datasetTypeName, storageClass=sc.name, file=testfile.name):
1550 ref = self.makeDatasetRef(datasetTypeName, dimensions, sc, dataId)
1551 if accepted:
1552 datastore.put(metrics, ref)
1553 self.assertTrue(datastore.exists(ref))
1554 datastore.remove(ref)
1556 # Try ingest
1557 if self.canIngest:
1558 datastore.ingest(FileDataset(testfile.name, [ref]), transfer="link")
1559 self.assertTrue(datastore.exists(ref))
1560 datastore.remove(ref)
1561 else:
1562 with self.assertRaises(DatasetTypeNotSupportedError):
1563 datastore.put(metrics, ref)
1564 self.assertFalse(datastore.exists(ref))
1566 # Again with ingest
1567 if self.canIngest:
1568 with self.assertRaises(DatasetTypeNotSupportedError):
1569 datastore.ingest(FileDataset(testfile.name, [ref]), transfer="link")
1570 self.assertFalse(datastore.exists(ref))
1573class PosixDatastoreConstraintsTestCase(DatastoreConstraintsTests, unittest.TestCase):
1574 """PosixDatastore specialization"""
1576 configFile = os.path.join(TESTDIR, "config/basic/posixDatastoreP.yaml")
1577 canIngest = True
1579 def setUp(self) -> None:
1580 # Override the working directory before calling the base class
1581 self.root = tempfile.mkdtemp()
1582 super().setUp()
1585class InMemoryDatastoreConstraintsTestCase(DatastoreConstraintsTests, unittest.TestCase):
1586 """InMemoryDatastore specialization."""
1588 configFile = os.path.join(TESTDIR, "config/basic/inMemoryDatastoreP.yaml")
1589 canIngest = False
1592class ChainedDatastoreConstraintsNativeTestCase(PosixDatastoreConstraintsTestCase):
1593 """ChainedDatastore specialization using a POSIXDatastore and constraints
1594 at the ChainedDatstore.
1595 """
1597 configFile = os.path.join(TESTDIR, "config/basic/chainedDatastorePa.yaml")
1600class ChainedDatastoreConstraintsTestCase(PosixDatastoreConstraintsTestCase):
1601 """ChainedDatastore specialization using a POSIXDatastore."""
1603 configFile = os.path.join(TESTDIR, "config/basic/chainedDatastoreP.yaml")
1606class ChainedDatastoreMemoryConstraintsTestCase(InMemoryDatastoreConstraintsTestCase):
1607 """ChainedDatastore specialization using all InMemoryDatastore."""
1609 configFile = os.path.join(TESTDIR, "config/basic/chainedDatastore2P.yaml")
1610 canIngest = False
1613class ChainedDatastorePerStoreConstraintsTests(DatastoreTestsBase, unittest.TestCase):
1614 """Test that a chained datastore can control constraints per-datastore
1615 even if child datastore would accept.
1616 """
1618 configFile = os.path.join(TESTDIR, "config/basic/chainedDatastorePb.yaml")
1620 def setUp(self) -> None:
1621 # Override the working directory before calling the base class
1622 self.root = tempfile.mkdtemp()
1623 super().setUp()
1625 def testConstraints(self) -> None:
1626 """Test chained datastore constraints model."""
1627 metrics = makeExampleMetrics()
1628 datastore = self.makeDatastore()
1630 sc1 = self.storageClassFactory.getStorageClass("StructuredData")
1631 sc2 = self.storageClassFactory.getStorageClass("StructuredDataJson")
1632 dimensions = self.universe.conform(("visit", "physical_filter", "instrument"))
1633 dataId1 = {
1634 "visit": 52,
1635 "physical_filter": "V",
1636 "band": "v",
1637 "instrument": "DummyCamComp",
1638 "day_obs": 20250101,
1639 }
1640 dataId2 = {"visit": 52, "physical_filter": "V", "band": "v", "instrument": "HSC", "day_obs": 20250101}
1642 # Write empty file suitable for ingest check (JSON and YAML variants)
1643 testfile_y = tempfile.NamedTemporaryFile(suffix=".yaml")
1644 testfile_j = tempfile.NamedTemporaryFile(suffix=".json")
1646 for typeName, dataId, sc, accept, ingest in (
1647 ("metric", dataId1, sc1, (False, True, False), True),
1648 ("metric5", dataId1, sc1, (False, False, False), False),
1649 ("metric5", dataId2, sc1, (True, False, False), False),
1650 ("metric33", dataId2, sc2, (True, True, False), True),
1651 ("metric5", dataId1, sc2, (False, True, False), True),
1652 ):
1653 # Choose different temp file depending on StorageClass
1654 testfile = testfile_j if sc.name.endswith("Json") else testfile_y
1656 with self.subTest(datasetTypeName=typeName, dataId=dataId, sc=sc.name):
1657 ref = self.makeDatasetRef(typeName, dimensions, sc, dataId)
1658 if any(accept):
1659 datastore.put(metrics, ref)
1660 self.assertTrue(datastore.exists(ref))
1662 # Check each datastore inside the chained datastore
1663 for childDatastore, expected in zip(datastore.datastores, accept, strict=True):
1664 self.assertEqual(
1665 childDatastore.exists(ref),
1666 expected,
1667 f"Testing presence of {ref} in datastore {childDatastore.name}",
1668 )
1670 datastore.remove(ref)
1672 # Check that ingest works
1673 if ingest:
1674 datastore.ingest(FileDataset(testfile.name, [ref]), transfer="link")
1675 self.assertTrue(datastore.exists(ref))
1677 # Check each datastore inside the chained datastore
1678 for childDatastore, expected in zip(datastore.datastores, accept, strict=True):
1679 # Ephemeral datastores means InMemory at the moment
1680 # and that does not accept ingest of files.
1681 if childDatastore.isEphemeral:
1682 expected = False
1683 self.assertEqual(
1684 childDatastore.exists(ref),
1685 expected,
1686 f"Testing presence of ingested {ref} in datastore {childDatastore.name}",
1687 )
1689 datastore.remove(ref)
1690 else:
1691 with self.assertRaises(DatasetTypeNotSupportedError):
1692 datastore.ingest(FileDataset(testfile.name, [ref]), transfer="link")
1694 else:
1695 with self.assertRaises(DatasetTypeNotSupportedError):
1696 datastore.put(metrics, ref)
1697 self.assertFalse(datastore.exists(ref))
1699 # Again with ingest
1700 with self.assertRaises(DatasetTypeNotSupportedError):
1701 datastore.ingest(FileDataset(testfile.name, [ref]), transfer="link")
1702 self.assertFalse(datastore.exists(ref))
1705@unittest.mock.patch.dict(os.environ, {}, clear=True)
1706class DatastoreCacheTestCase(DatasetTestHelper, unittest.TestCase):
1707 """Tests for datastore caching infrastructure."""
1709 @classmethod
1710 def setUpClass(cls) -> None:
1711 cls.storageClassFactory = StorageClassFactory()
1712 cls.universe = DimensionUniverse()
1714 # Ensure that we load the test storage class definitions.
1715 scConfigFile = os.path.join(TESTDIR, "config/basic/storageClasses.yaml")
1716 cls.storageClassFactory.addFromConfig(scConfigFile)
1718 def setUp(self) -> None:
1719 self.id = 0
1721 # Create a root that we can use for caching tests.
1722 self.root = tempfile.mkdtemp()
1724 # Create some test dataset refs and associated test files
1725 sc = self.storageClassFactory.getStorageClass("StructuredDataDict")
1726 dimensions = self.universe.conform(("visit", "physical_filter"))
1727 dataId = {
1728 "instrument": "dummy",
1729 "visit": 52,
1730 "physical_filter": "V",
1731 "band": "v",
1732 "day_obs": 20250101,
1733 }
1735 # Create list of refs and list of temporary files
1736 n_datasets = 10
1737 self.refs = [self.makeDatasetRef(f"metric{n}", dimensions, sc, dataId) for n in range(n_datasets)]
1739 root_uri = ResourcePath(self.root, forceDirectory=True)
1740 self.files = [root_uri.join(f"file{n}.txt") for n in range(n_datasets)]
1742 # Create test files.
1743 for uri in self.files:
1744 uri.write(b"0123456789")
1746 # Create some composite refs with component files.
1747 sc = self.storageClassFactory.getStorageClass("StructuredData")
1748 self.composite_refs = [self.makeDatasetRef(f"composite{n}", dimensions, sc, dataId) for n in range(3)]
1749 self.comp_files = []
1750 self.comp_refs = []
1751 for n, ref in enumerate(self.composite_refs):
1752 component_refs = []
1753 component_files = []
1754 for component in sc.components:
1755 component_ref = ref.makeComponentRef(component)
1756 file = root_uri.join(f"composite_file-{n}-{component}.txt")
1757 component_refs.append(component_ref)
1758 component_files.append(file)
1759 file.write(b"9876543210")
1761 self.comp_files.append(component_files)
1762 self.comp_refs.append(component_refs)
1764 def tearDown(self) -> None:
1765 if self.root is not None and os.path.exists(self.root): 1765 ↛ exitline 1765 didn't return from function 'tearDown' because the condition on line 1765 was always true
1766 shutil.rmtree(self.root, ignore_errors=True)
1768 def _make_cache_manager(self, config_str: str) -> DatastoreCacheManager:
1769 config = Config.fromYaml(config_str)
1770 return DatastoreCacheManager(DatastoreCacheManagerConfig(config), universe=self.universe)
1772 def testNoCacheDir(self) -> None:
1773 config_str = """
1774cached:
1775 root: null
1776 cacheable:
1777 metric0: true
1778 """
1779 cache_manager = self._make_cache_manager(config_str)
1781 # Look inside to check we don't have a cache directory
1782 self.assertIsNone(cache_manager._cache_directory)
1784 self.assertCache(cache_manager)
1786 # Test that the cache directory is marked temporary
1787 self.assertTrue(cache_manager.cache_directory.isTemporary)
1789 def testNoCacheDirReversed(self) -> None:
1790 """Use default caching status and metric1 to false"""
1791 config_str = """
1792cached:
1793 root: null
1794 default: true
1795 cacheable:
1796 metric1: false
1797 """
1798 cache_manager = self._make_cache_manager(config_str)
1800 self.assertCache(cache_manager)
1802 def testEnvvarCacheDir(self) -> None:
1803 config_str = f"""
1804cached:
1805 root: '{self.root}'
1806 cacheable:
1807 metric0: true
1808 """
1810 root = ResourcePath(self.root, forceDirectory=True)
1811 env_dir = root.join("somewhere", forceDirectory=True)
1812 elsewhere = root.join("elsewhere", forceDirectory=True)
1814 # Environment variable should override the config value.
1815 with unittest.mock.patch.dict(os.environ, {"DAF_BUTLER_CACHE_DIRECTORY": env_dir.ospath}):
1816 cache_manager = self._make_cache_manager(config_str)
1817 self.assertEqual(cache_manager.cache_directory, env_dir)
1819 # This environment variable should not override the config value.
1820 with unittest.mock.patch.dict(os.environ, {"DAF_BUTLER_CACHE_DIRECTORY_IF_UNSET": env_dir.ospath}):
1821 cache_manager = self._make_cache_manager(config_str)
1822 self.assertEqual(cache_manager.cache_directory, root)
1824 # No default setting.
1825 config_str = """
1826cached:
1827 root: null
1828 default: true
1829 cacheable:
1830 metric1: false
1831 """
1832 cache_manager = self._make_cache_manager(config_str)
1834 # This environment variable should override the config value.
1835 with unittest.mock.patch.dict(os.environ, {"DAF_BUTLER_CACHE_DIRECTORY_IF_UNSET": env_dir.ospath}):
1836 cache_manager = self._make_cache_manager(config_str)
1837 self.assertEqual(cache_manager.cache_directory, env_dir)
1839 # If both environment variables are set the main (not IF_UNSET)
1840 # variable should win.
1841 with unittest.mock.patch.dict(
1842 os.environ,
1843 {
1844 "DAF_BUTLER_CACHE_DIRECTORY": env_dir.ospath,
1845 "DAF_BUTLER_CACHE_DIRECTORY_IF_UNSET": elsewhere.ospath,
1846 },
1847 ):
1848 cache_manager = self._make_cache_manager(config_str)
1849 self.assertEqual(cache_manager.cache_directory, env_dir)
1851 # Use the API to set the environment variable, making sure that the
1852 # variable is reset on exit.
1853 with unittest.mock.patch.dict(
1854 os.environ,
1855 {"DAF_BUTLER_CACHE_DIRECTORY_IF_UNSET": ""},
1856 ):
1857 defined, cache_dir = DatastoreCacheManager.set_fallback_cache_directory_if_unset()
1858 self.assertTrue(defined)
1859 cache_manager = self._make_cache_manager(config_str)
1860 self.assertEqual(cache_manager.cache_directory, ResourcePath(cache_dir, forceDirectory=True))
1862 # Now create the cache manager ahead of time and set the fallback
1863 # later.
1864 cache_manager = self._make_cache_manager(config_str)
1865 self.assertIsNone(cache_manager._cache_directory)
1866 with unittest.mock.patch.dict(
1867 os.environ,
1868 {"DAF_BUTLER_CACHE_DIRECTORY_IF_UNSET": ""},
1869 ):
1870 defined, cache_dir = DatastoreCacheManager.set_fallback_cache_directory_if_unset()
1871 self.assertTrue(defined)
1872 self.assertEqual(cache_manager.cache_directory, ResourcePath(cache_dir, forceDirectory=True))
1874 def testExplicitCacheDir(self) -> None:
1875 config_str = f"""
1876cached:
1877 root: '{self.root}'
1878 cacheable:
1879 metric0: true
1880 """
1881 cache_manager = self._make_cache_manager(config_str)
1883 # Look inside to check we do have a cache directory.
1884 self.assertEqual(cache_manager.cache_directory, ResourcePath(self.root, forceDirectory=True))
1886 self.assertCache(cache_manager)
1888 # Test that the cache directory is not marked temporary
1889 self.assertFalse(cache_manager.cache_directory.isTemporary)
1891 def testUnexpectedFilesInCacheDir(self) -> None:
1892 """Test for regression of a bug where extraneous files in a cache
1893 directory would cause all cache lookups to raise an exception.
1894 """
1895 config_str = f"""
1896cached:
1897 root: '{self.root}'
1898 cacheable:
1899 metric0: true
1900 """
1902 for filename in ["unexpected.txt", "unexpected", "un_expected", "un_expected.txt"]:
1903 unexpected_file = os.path.join(self.root, filename)
1904 with open(unexpected_file, "w") as fh:
1905 fh.write("test")
1907 cache_manager = self._make_cache_manager(config_str)
1908 cache_manager.scan_cache()
1909 self.assertCache(cache_manager)
1911 def assertCache(self, cache_manager: DatastoreCacheManager) -> None:
1912 self.assertTrue(cache_manager.should_be_cached(self.refs[0]))
1913 self.assertFalse(cache_manager.should_be_cached(self.refs[1]))
1915 uri = cache_manager.move_to_cache(self.files[0], self.refs[0])
1916 self.assertIsInstance(uri, ResourcePath)
1917 self.assertIsNone(cache_manager.move_to_cache(self.files[1], self.refs[1]))
1919 # Check presence in cache using ref and then using file extension.
1920 self.assertFalse(cache_manager.known_to_cache(self.refs[1]))
1921 self.assertTrue(cache_manager.known_to_cache(self.refs[0]))
1922 self.assertFalse(cache_manager.known_to_cache(self.refs[1], self.files[1].getExtension()))
1923 self.assertTrue(cache_manager.known_to_cache(self.refs[0], self.files[0].getExtension()))
1925 # Cached file should no longer exist but uncached file should be
1926 # unaffected.
1927 self.assertFalse(self.files[0].exists())
1928 self.assertTrue(self.files[1].exists())
1930 # Should find this file and it should be within the cache directory.
1931 with cache_manager.find_in_cache(self.refs[0], ".txt") as found:
1932 self.assertTrue(found.exists())
1933 self.assertIsNotNone(found.relative_to(cache_manager.cache_directory))
1935 # Should not be able to find these in cache
1936 with cache_manager.find_in_cache(self.refs[0], ".fits") as found:
1937 self.assertIsNone(found)
1938 with cache_manager.find_in_cache(self.refs[1], ".fits") as found:
1939 self.assertIsNone(found)
1941 def testNoCache(self) -> None:
1942 cache_manager = DatastoreDisabledCacheManager("", universe=self.universe)
1943 for uri, ref in zip(self.files, self.refs, strict=True):
1944 self.assertFalse(cache_manager.should_be_cached(ref))
1945 self.assertIsNone(cache_manager.move_to_cache(uri, ref))
1946 self.assertFalse(cache_manager.known_to_cache(ref))
1947 with cache_manager.find_in_cache(ref, ".txt") as found:
1948 self.assertIsNone(found, msg=f"{cache_manager}")
1950 def _expiration_config(self, mode: str, threshold: int) -> str:
1951 return f"""
1952cached:
1953 default: true
1954 expiry:
1955 mode: {mode}
1956 threshold: {threshold}
1957 cacheable:
1958 unused: true
1959 """
1961 def testCacheExpiryFiles(self) -> None:
1962 threshold = 2 # Keep at least 2 files.
1963 mode = "files"
1964 config_str = self._expiration_config(mode, threshold)
1966 cache_manager = self._make_cache_manager(config_str)
1968 # Check that an empty cache returns unknown for arbitrary ref
1969 self.assertFalse(cache_manager.known_to_cache(self.refs[0]))
1971 # Should end with datasets: 2, 3, 4
1972 self.assertExpiration(cache_manager, 5, threshold + 1)
1973 self.assertIn(f"{mode}={threshold}", str(cache_manager))
1975 # Check that we will not expire a file that is actively in use.
1976 with cache_manager.find_in_cache(self.refs[2], ".txt") as found:
1977 self.assertIsNotNone(found)
1979 # Trigger cache expiration that should remove the file
1980 # we just retrieved. Should now have: 3, 4, 5
1981 cached = cache_manager.move_to_cache(self.files[5], self.refs[5])
1982 self.assertIsNotNone(cached)
1984 # Cache should still report the standard file count.
1985 self.assertEqual(cache_manager.file_count, threshold + 1)
1987 # Add additional entry to cache.
1988 # Should now have 4, 5, 6
1989 cached = cache_manager.move_to_cache(self.files[6], self.refs[6])
1990 self.assertIsNotNone(cached)
1992 # Is the file still there?
1993 self.assertTrue(found.exists())
1995 # Can we read it?
1996 data = found.read()
1997 self.assertGreater(len(data), 0)
1999 # Outside context the file should no longer exist.
2000 self.assertFalse(found.exists())
2002 # File count should not have changed.
2003 self.assertEqual(cache_manager.file_count, threshold + 1)
2005 # Dataset 2 was in the exempt directory but because hardlinks
2006 # are used it was deleted from the main cache during cache expiry
2007 # above and so should no longer be found.
2008 with cache_manager.find_in_cache(self.refs[2], ".txt") as found:
2009 self.assertIsNone(found)
2011 # And the one stored after it is also gone.
2012 with cache_manager.find_in_cache(self.refs[3], ".txt") as found:
2013 self.assertIsNone(found)
2015 # But dataset 4 is present.
2016 with cache_manager.find_in_cache(self.refs[4], ".txt") as found:
2017 self.assertIsNotNone(found)
2019 # Adding a new dataset to the cache should now delete it.
2020 cache_manager.move_to_cache(self.files[7], self.refs[7])
2022 with cache_manager.find_in_cache(self.refs[2], ".txt") as found:
2023 self.assertIsNone(found)
2025 def testCacheExpiryDatasets(self) -> None:
2026 threshold = 2 # Keep 2 datasets.
2027 mode = "datasets"
2028 config_str = self._expiration_config(mode, threshold)
2030 cache_manager = self._make_cache_manager(config_str)
2031 self.assertExpiration(cache_manager, 5, threshold + 1)
2032 self.assertIn(f"{mode}={threshold}", str(cache_manager))
2034 def testCacheExpiryDatasetsFromDisabled(self) -> None:
2035 threshold = 2
2036 mode = "datasets"
2037 with unittest.mock.patch.dict(
2038 os.environ,
2039 {"DAF_BUTLER_CACHE_EXPIRATION_MODE": f"{mode}={threshold}"},
2040 ):
2041 cache_manager = DatastoreCacheManager.create_disabled(universe=DimensionUniverse())
2042 self.assertExpiration(cache_manager, 5, threshold + 1)
2043 self.assertIn(f"{mode}={threshold}", str(cache_manager))
2045 def testExpirationModeOverride(self) -> None:
2046 threshold = 2 # Keep 2 datasets.
2047 mode = "datasets"
2048 config_str = self._expiration_config(mode, threshold)
2050 mode = "size"
2051 threshold = 55
2052 with unittest.mock.patch.dict(
2053 os.environ,
2054 {"DAF_BUTLER_CACHE_EXPIRATION_MODE": f"{mode}={threshold}"},
2055 ):
2056 cache_manager = self._make_cache_manager(config_str)
2057 self.assertExpiration(cache_manager, 10, 6)
2058 self.assertIn(f"{mode}={threshold}", str(cache_manager))
2060 # Check we get a warning with unrecognized form.
2061 with unittest.mock.patch.dict(
2062 os.environ,
2063 {"DAF_BUTLER_CACHE_EXPIRATION_MODE": "something"},
2064 ):
2065 with self.assertLogs(level="WARNING") as cm:
2066 self._make_cache_manager(config_str)
2067 self.assertIn("Unrecognized form (something)", cm.output[0])
2069 with unittest.mock.patch.dict(
2070 os.environ,
2071 {"DAF_BUTLER_CACHE_EXPIRATION_MODE": "something=5"},
2072 ):
2073 with self.assertRaises(ValueError) as cm:
2074 self._make_cache_manager(config_str)
2075 self.assertIn("Unrecognized value", str(cm.exception))
2077 def testMissingThreshold(self) -> None:
2078 threshold = ""
2079 mode = "datasets"
2080 config_str = self._expiration_config(mode, threshold)
2082 with self.assertRaises(ValueError) as cm:
2083 self._make_cache_manager(config_str)
2084 self.assertIn("Cache expiration threshold", str(cm.exception))
2086 def testCacheExpiryDatasetsComposite(self) -> None:
2087 threshold = 2 # Keep 2 datasets.
2088 mode = "datasets"
2089 config_str = self._expiration_config(mode, threshold)
2091 cache_manager = self._make_cache_manager(config_str)
2093 n_datasets = 3
2094 for i in range(n_datasets):
2095 for component_file, component_ref in zip(self.comp_files[i], self.comp_refs[i], strict=True):
2096 cached = cache_manager.move_to_cache(component_file, component_ref)
2097 self.assertIsNotNone(cached)
2098 self.assertTrue(cache_manager.known_to_cache(component_ref))
2099 self.assertTrue(cache_manager.known_to_cache(component_ref.makeCompositeRef()))
2100 self.assertTrue(cache_manager.known_to_cache(component_ref, component_file.getExtension()))
2102 self.assertEqual(cache_manager.file_count, 6) # 2 datasets each of 3 files
2104 # Write two new non-composite and the number of files should drop.
2105 self.assertExpiration(cache_manager, 2, 5)
2107 def testCacheExpirySize(self) -> None:
2108 threshold = 55 # Each file is 10 bytes
2109 mode = "size"
2110 config_str = self._expiration_config(mode, threshold)
2112 cache_manager = self._make_cache_manager(config_str)
2113 self.assertExpiration(cache_manager, 10, 6)
2114 self.assertIn(f"{mode}={threshold}", str(cache_manager))
2116 def testDisabledCache(self) -> None:
2117 # Configure an active cache but disable via environment.
2118 threshold = 2
2119 mode = "datasets"
2120 config_str = self._expiration_config(mode, threshold)
2122 with unittest.mock.patch.dict(
2123 os.environ,
2124 {"DAF_BUTLER_CACHE_EXPIRATION_MODE": "disabled"},
2125 ):
2126 env_cache_manager = self._make_cache_manager(config_str)
2128 # Configure to be disabled
2129 threshold = 0
2130 mode = "disabled"
2131 config_str = self._expiration_config(mode, threshold)
2132 cfg_cache_manager = self._make_cache_manager(config_str)
2134 for cache_manager in (
2135 cfg_cache_manager,
2136 env_cache_manager,
2137 DatastoreCacheManager.create_disabled(universe=DimensionUniverse()),
2138 ):
2139 for uri, ref in zip(self.files, self.refs, strict=True):
2140 self.assertFalse(cache_manager.should_be_cached(ref))
2141 self.assertIsNone(cache_manager.move_to_cache(uri, ref))
2142 self.assertFalse(cache_manager.known_to_cache(ref))
2143 with cache_manager.find_in_cache(ref, ".txt") as found:
2144 self.assertIsNone(found, msg=f"{cache_manager}")
2145 self.assertIn("disabled", str(cache_manager))
2147 def assertExpiration(
2148 self, cache_manager: DatastoreCacheManager, n_datasets: int, n_retained: int
2149 ) -> None:
2150 """Insert the datasets and then check the number retained."""
2151 for i in range(n_datasets):
2152 cached = cache_manager.move_to_cache(self.files[i], self.refs[i])
2153 self.assertIsNotNone(cached)
2155 self.assertEqual(cache_manager.file_count, n_retained)
2157 # The oldest file should not be in the cache any more.
2158 for i in range(n_datasets):
2159 with cache_manager.find_in_cache(self.refs[i], ".txt") as found:
2160 if i >= n_datasets - n_retained:
2161 self.assertIsInstance(found, ResourcePath)
2162 else:
2163 self.assertIsNone(found)
2165 def testCacheExpiryAge(self) -> None:
2166 threshold = 1 # Expire older than 2 seconds
2167 mode = "age"
2168 config_str = self._expiration_config(mode, threshold)
2170 cache_manager = self._make_cache_manager(config_str)
2171 self.assertIn(f"{mode}={threshold}", str(cache_manager))
2173 # Insert 3 files, then sleep, then insert more.
2174 for i in range(2):
2175 cached = cache_manager.move_to_cache(self.files[i], self.refs[i])
2176 self.assertIsNotNone(cached)
2177 time.sleep(2.0)
2178 for j in range(4):
2179 i = 2 + j # Continue the counting
2180 cached = cache_manager.move_to_cache(self.files[i], self.refs[i])
2181 self.assertIsNotNone(cached)
2183 # Only the files written after the sleep should exist.
2184 self.assertEqual(cache_manager.file_count, 4)
2185 with cache_manager.find_in_cache(self.refs[1], ".txt") as found:
2186 self.assertIsNone(found)
2187 with cache_manager.find_in_cache(self.refs[2], ".txt") as found:
2188 self.assertIsInstance(found, ResourcePath)
2191class NullDatastoreTestCase(DatasetTestHelper, unittest.TestCase):
2192 """Test the null datastore."""
2194 storageClassFactory = StorageClassFactory()
2196 def test_basics(self) -> None:
2197 storageClass = self.storageClassFactory.getStorageClass("StructuredDataDict")
2198 ref = self.makeDatasetRef("metric", DimensionUniverse().empty, storageClass, {})
2200 null = NullDatastore(None, None)
2202 self.assertFalse(null.exists(ref))
2203 self.assertFalse(null.knows(ref))
2204 knows = null.knows_these([ref])
2205 self.assertFalse(knows[ref])
2206 null.validateConfiguration(ref)
2208 with self.assertRaises(FileNotFoundError):
2209 null.get(ref)
2210 with self.assertRaises(NotImplementedError):
2211 null.put("", ref)
2212 with self.assertRaises(FileNotFoundError):
2213 null.getURI(ref)
2214 with self.assertRaises(FileNotFoundError):
2215 null.getURIs(ref)
2216 with self.assertRaises(FileNotFoundError):
2217 null.getManyURIs([ref])
2218 with self.assertRaises(NotImplementedError):
2219 null.getLookupKeys()
2220 with self.assertRaises(NotImplementedError):
2221 null.import_records({})
2222 with self.assertRaises(NotImplementedError):
2223 null.export_records([])
2224 with self.assertRaises(NotImplementedError):
2225 null.export_predicted_records([])
2226 with self.assertRaises(NotImplementedError):
2227 null.export([ref])
2228 with self.assertRaises(NotImplementedError):
2229 null.transfer(null, ref)
2230 with self.assertRaises(NotImplementedError):
2231 null.emptyTrash()
2232 with self.assertRaises(NotImplementedError):
2233 null.trash(ref)
2234 with self.assertRaises(NotImplementedError):
2235 null.forget([ref])
2236 with self.assertRaises(NotImplementedError):
2237 null.remove(ref)
2238 with self.assertRaises(NotImplementedError):
2239 null.retrieveArtifacts([ref], ResourcePath("."))
2240 with self.assertRaises(NotImplementedError):
2241 null.transfer_from(null, [ref])
2242 with self.assertRaises(NotImplementedError):
2243 null.ingest()
2246class DatasetRefURIsTestCase(unittest.TestCase):
2247 """Tests for DatasetRefURIs."""
2249 def testSequenceAccess(self) -> None:
2250 """Verify that DatasetRefURIs can be treated like a two-item tuple."""
2251 uris = DatasetRefURIs()
2253 self.assertEqual(len(uris), 2)
2254 self.assertEqual(uris[0], None)
2255 self.assertEqual(uris[1], {})
2257 primaryURI = ResourcePath("1/2/3")
2258 componentURI = ResourcePath("a/b/c")
2260 # affirm that DatasetRefURIs does not support MutableSequence functions
2261 with self.assertRaises(TypeError):
2262 uris[0] = primaryURI
2263 with self.assertRaises(TypeError):
2264 uris[1] = {"foo": componentURI}
2266 # but DatasetRefURIs can be set by property name:
2267 uris.primaryURI = primaryURI
2268 uris.componentURIs = {"foo": componentURI}
2269 self.assertEqual(uris.primaryURI, primaryURI)
2270 self.assertEqual(uris[0], primaryURI)
2272 primary, components = uris
2273 self.assertEqual(primary, primaryURI)
2274 self.assertEqual(components, {"foo": componentURI})
2276 def testRepr(self) -> None:
2277 """Verify __repr__ output."""
2278 uris = DatasetRefURIs(ResourcePath("/1/2/3"), {"comp": ResourcePath("/a/b/c")})
2279 self.assertEqual(
2280 repr(uris),
2281 'DatasetRefURIs(ResourcePath("file:///1/2/3"), {\'comp\': ResourcePath("file:///a/b/c")})',
2282 )
2285class StoredFileInfoTestCase(DatasetTestHelper, unittest.TestCase):
2286 """Test the StoredFileInfo class."""
2288 storageClassFactory = StorageClassFactory()
2290 def test_StoredFileInfo(self) -> None:
2291 storageClass = self.storageClassFactory.getStorageClass("StructuredDataDict")
2292 ref = self.makeDatasetRef("metric", DimensionUniverse().empty, storageClass, {})
2294 record = dict(
2295 storage_class="StructuredDataDict",
2296 formatter="lsst.daf.butler.Formatter",
2297 path="a/b/c.txt",
2298 component="component",
2299 checksum=None,
2300 file_size=5,
2301 )
2302 info = StoredFileInfo.from_record(record)
2304 self.assertEqual(info.to_record(), record)
2306 ref2 = self.makeDatasetRef("metric", DimensionUniverse().empty, storageClass, {})
2307 rebased = info.rebase(ref2)
2308 self.assertEqual(rebased.rebase(ref), info)
2310 with self.assertRaises(TypeError):
2311 rebased.update(formatter=42)
2313 with self.assertRaises(ValueError):
2314 rebased.update(something=42, new="42")
2316 # Check that pickle works on StoredFileInfo.
2317 pickled_info = pickle.dumps(info)
2318 unpickled_info = pickle.loads(pickled_info)
2319 self.assertEqual(unpickled_info, info)
2321 def test_make_datastore_path_relative(self):
2322 self.assertEqual(make_datastore_path_relative("a/relative/path"), "a/relative/path")
2323 self.assertEqual(make_datastore_path_relative("path/with#fragment"), "path/with#fragment")
2324 self.assertEqual(make_datastore_path_relative("http://server.com/some/path"), "some/path")
2325 self.assertEqual(make_datastore_path_relative("http://server.com/some/path#frag"), "some/path#frag")
2327 def test_datastore_record_data_json_types(self):
2328 """Test that we don't round-trip checksums to UUIDs when deserializing
2329 datastore record data.
2330 """
2331 test_json = """
2332 {
2333 "dataset_ids": [
2334 "74478304-abf1-4a9c-9eb2-926090a84446"
2335 ],
2336 "records": {
2337 "lsst.daf.butler.datastore.stored_file_info.StoredFileInfo": {
2338 "74478304abf14a9c9eb2926090a84446": {
2339 "file_datastore_records": [
2340 {
2341 "formatter": "lsst.daf.butler.formatters.yaml.YamlFormatter",
2342 "path": "gain_factors/base-2025-158/gain_factors_spx_base-2025-158.yaml",
2343 "storage_class": "GainFactors",
2344 "component": "__NULL_STRING__",
2345 "checksum": "cab515f6-ab67-0484-393f-aaa525dd526f",
2346 "file_size": 5412
2347 }
2348 ]
2349 }
2350 }
2351 }
2352 }
2353 """
2354 id_str = "74478304abf14a9c9eb2926090a84446"
2355 s = SerializedDatastoreRecordData.model_validate_json(test_json)
2356 self.assertIsInstance(
2357 s.records[get_full_type_name(StoredFileInfo)][id_str]["file_datastore_records"][0]["checksum"],
2358 str,
2359 )
2360 id = uuid.UUID(id_str)
2361 d = DatastoreRecordData.from_simple(s)
2362 self.assertIsInstance(d.records[id]["file_datastore_records"][0], StoredFileInfo)
2363 self.assertIsInstance(d.records[id]["file_datastore_records"][0].checksum, str)
2366class TestDatastoreRecordTable(unittest.TestCase):
2367 """Test DatastoreRecordTable and StoredFileInfoTable."""
2369 def test_empty_datastore_records_table(self) -> None:
2370 file_info_table = StoredFileInfoTable.from_records([])
2371 self.assertEqual(0, len(file_info_table))
2373 self.assertEqual(
2374 0, len(DatastoreRecordTable.from_stored_file_info_table("datastore_name", file_info_table))
2375 )
2377 self.assertEqual(0, len(DatastoreRecordTable.create_empty()))
2378 self.assertEqual(0, len(DatastoreRecordTable.combine([])))
2379 # Doesn't throw because there is no mismatch in datastore names.
2380 DatastoreRecordTable.create_empty().validate_datastore_names("arbitrary_name")
2382 def test_stored_file_info_table_records(self) -> None:
2383 uuid1 = uuid.UUID("019e1892-7b9b-736d-8248-0e031723646c")
2384 uuid2 = uuid.UUID("019e1895-9ec3-7431-bed9-8ae60096103f")
2385 uuid3 = uuid.UUID("13d13272-454c-4bc4-94d5-3e322982eee8")
2386 checksum = (
2387 "021ced8799518305c451cde3e921515ef315ee7ba8937"
2388 "a92697a20c571c776afa4102744bc28d2d99d35f44e073cde80cf96e387f65f3967cca45b0d015f5a6b"
2389 )
2390 input_records = [
2391 {
2392 "dataset_id": uuid1,
2393 "path": "a/relative/path.fits",
2394 "formatter": "lsst.obs.base.formatters.fitsExposure.FitsExposureFormatter",
2395 "storage_class": "ExposureF",
2396 "component": "__NULL_STRING__",
2397 "checksum": None,
2398 "file_size": 123,
2399 },
2400 {
2401 "dataset_id": uuid2,
2402 "path": "file:///an/absolute/path.fits",
2403 "formatter": "lsst.obs.base.formatters.fitsExposure.FitsExposureFormatter",
2404 "storage_class": "ExposureF",
2405 "component": "comp",
2406 "checksum": checksum,
2407 "file_size": -1,
2408 },
2409 ]
2410 table = StoredFileInfoTable.from_records(input_records)
2411 self.assertEqual(len(table), 2)
2413 def _check_records(records: list[dict]) -> None:
2414 rec0 = records[0]
2415 self.assertEqual(rec0["dataset_id"], uuid1)
2416 self.assertEqual(rec0["path"], "a/relative/path.fits")
2417 self.assertEqual(rec0["formatter"], "lsst.obs.base.formatters.fitsExposure.FitsExposureFormatter")
2418 self.assertEqual(rec0["storage_class"], "ExposureF")
2419 self.assertIsNone(rec0["component"])
2420 self.assertIsNone(rec0["checksum"])
2421 self.assertEqual(rec0["file_size"], 123)
2422 rec1 = records[1]
2423 self.assertEqual(rec1["dataset_id"], uuid2)
2424 self.assertEqual(rec1["path"], "file:///an/absolute/path.fits")
2425 self.assertEqual(rec1["formatter"], "lsst.obs.base.formatters.fitsExposure.FitsExposureFormatter")
2426 self.assertEqual(rec1["storage_class"], "ExposureF")
2427 self.assertEqual(rec1["component"], "comp")
2428 self.assertEqual(rec1["checksum"], checksum)
2429 self.assertIsNone(rec1["file_size"])
2431 arrow_records = table.to_arrow().to_pylist()
2432 self.assertEqual(len(arrow_records), 2)
2433 _check_records(arrow_records)
2434 self.assertEqual(table.to_records(), input_records)
2436 datastore_table = DatastoreRecordTable.from_stored_file_info_table("name_of_datastore", table)
2437 datastore_arrow_records = datastore_table.to_arrow().to_pylist()
2438 self.assertEqual(len(datastore_arrow_records), 2)
2439 _check_records(datastore_arrow_records)
2440 self.assertEqual(datastore_arrow_records[0]["datastore_name"], "name_of_datastore")
2441 self.assertEqual(datastore_arrow_records[1]["datastore_name"], "name_of_datastore")
2442 _check_records(datastore_table.to_stored_file_info_table().to_arrow().to_pylist())
2443 # Check round-tripping to_arrow() through from_arrow()
2444 _check_records(
2445 datastore_table.from_arrow(datastore_table.to_arrow())
2446 .to_stored_file_info_table()
2447 .to_arrow()
2448 .to_pylist()
2449 )
2451 with self.assertRaisesRegex(ValueError, "do not match known datastores"):
2452 datastore_table.validate_datastore_names(["not_the_same_datastore"])
2453 datastore_table.validate_datastore_names(["not_the_same_datastore", "name_of_datastore"])
2455 second_table = DatastoreRecordTable.from_stored_file_info_table(
2456 "other_datastore_name",
2457 StoredFileInfoTable.from_records(
2458 [
2459 {
2460 "dataset_id": uuid3,
2461 "path": "a/relative/path2.fits",
2462 "formatter": "lsst.obs.lsst.rawFormatter.LsstCamRawFormatter",
2463 "storage_class": "Exposure",
2464 "component": "__NULL_STRING__",
2465 "checksum": None,
2466 "file_size": 1000,
2467 },
2468 ]
2469 ),
2470 )
2471 combined_table = DatastoreRecordTable.combine([datastore_table, second_table])
2472 combined_records = combined_table.to_arrow().to_pylist()
2473 _check_records(combined_records)
2474 rec2 = combined_records[2]
2475 self.assertEqual(rec2["dataset_id"], uuid3)
2476 self.assertEqual(rec2["path"], "a/relative/path2.fits")
2477 self.assertEqual(rec2["formatter"], "lsst.obs.lsst.rawFormatter.LsstCamRawFormatter")
2478 self.assertEqual(rec2["storage_class"], "Exposure")
2479 self.assertIsNone(rec2["component"])
2480 self.assertIsNone(rec2["checksum"])
2481 self.assertEqual(rec2["file_size"], 1000)
2483 self.assertEqual(0, len(datastore_table.filter_by_datastore_name("unknown_datastore")))
2484 filtered_table = combined_table.filter_by_datastore_name("name_of_datastore")
2485 self.assertEqual(len(filtered_table), 2)
2486 _check_records(filtered_table.to_arrow().to_pylist())
2487 self.assertEqual(filtered_table.to_stored_file_info_table().to_records(), input_records)
2490@contextlib.contextmanager
2491def _temp_yaml_file(data: Any) -> Iterator[str]:
2492 fh = tempfile.NamedTemporaryFile(mode="w", suffix=".yaml")
2493 try:
2494 yaml.dump(data, stream=fh)
2495 fh.flush()
2496 yield fh.name
2497 finally:
2498 # Some tests delete the file
2499 with contextlib.suppress(FileNotFoundError):
2500 fh.close()
2503if __name__ == "__main__":
2504 unittest.main()